fix: harden cache resilience and support external task inputs

packages/tooling/task-runner/__tests__/integration/e2e-runner.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { promisify } from "node:util";
 
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import { defaultTaskRunner } from "../../src/default-task-runner";
 import { EmptyLifeCycle } from "../../src/life-cycle";
@@ -125,7 +125,7 @@ describe("defaultTaskRunner (E2E via real subprocess)", () => {
     });
 
     afterEach(async () => {
-        await rm(workspaceRoot, { force: true, recursive: true });
+        await rm(workspaceRoot, { force: true, maxRetries: 20, recursive: true, retryDelay: 100 });
     });
 
     it("round-trips a real subprocess through the cache (cold miss → warm hit restores output)", async () => {
@@ -350,3 +350,62 @@ describe("defaultTaskRunner (E2E via real subprocess)", () => {
         expect(checkCount).toBe(2);
     });
 });
+
+describe("remote cache failures degrade to a miss", () => {
+    let workspaceRoot: string;
+
+    beforeEach(async () => {
+        workspaceRoot = await createTemporaryDirectory();
+
+        await mkdir(join(workspaceRoot, "packages/app/src"), { recursive: true });
+        await writeFile(join(workspaceRoot, "packages/app/src/index.ts"), "export const x = 1;\n");
+        await writeFile(join(workspaceRoot, "packages/app/package.json"), JSON.stringify({ name: "app" }));
+    });
+
+    afterEach(async () => {
+        await rm(workspaceRoot, { force: true, maxRetries: 20, recursive: true, retryDelay: 100 });
+    });
+
+    it("runs the tasks normally and warns once when the remote cache is unreachable", async () => {
+        expect.assertions(5);
+
+        const build = makeTask("app:build", "packages/app");
+        const check = makeTask("app:check", "packages/app");
+
+        const executor = createNodeExecutor(workspaceRoot, {
+            "app:build": "process.stdout.write('built');",
+            "app:check": "process.stdout.write('checked');",
+        });
+        const context = createContext(workspaceRoot, [build, check], { "app:build": [], "app:check": [] }, executor);
+        const diagnostics: { message: string; taskId: string }[] = [];
+        const onUploadError = vi.fn<(hash: string, error: unknown) => void>();
+
+        const results = await defaultTaskRunner(
+            [build, check],
+            {
+                onDiagnostic: (taskId, message) => {
+                    diagnostics.push({ message, taskId });
+                },
+                remoteCache: {
+                    mode: "read",
+                    onUploadError,
+                    timeout: 100,
+                    url: "http://127.0.0.1:1",
+                },
+            },
+            context,
+        );
+
+        expect(results.get("app:build")?.status).toBe("success");
+        expect(results.get("app:check")?.status).toBe("success");
+
+        const remoteWarnings = diagnostics.filter((entry) => entry.taskId === "remote-cache");
+
+        expect(remoteWarnings).toHaveLength(1);
+        expect(remoteWarnings[0]?.message).toMatch(/Remote cache unavailable/);
+
+        // The caller's detail hook still sees every failed operation, so a
+        // debug logger can retain the full sequence.
+        expect(onUploadError).toHaveBeenCalledTimes(2);
+    });
+});

packages/tooling/task-runner/__tests__/unit/affected-external-inputs.test.ts

@@ -0,0 +1,150 @@
+import { execFileSync } from "node:child_process";
+import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+
+import { getAffectedProjects } from "../../src/affected";
+import { collectExternalInputs, computeExternalInputHashes } from "../../src/external-inputs";
+import type { ProjectConfiguration, ProjectGraph } from "../../src/types";
+
+const git = (cwd: string, ...arguments_: string[]): void => {
+    execFileSync("git", arguments_, { cwd, stdio: "pipe" });
+};
+
+describe("affected detection with external inputs", () => {
+    let workspaceRoot: string;
+    let externalRoot: string;
+    let externalFile: string;
+
+    beforeEach(async () => {
+        const base = await mkdtemp(join(tmpdir(), "tr-affected-external-"));
+
+        workspaceRoot = join(base, "workspace");
+        externalRoot = join(base, "shared");
+
+        await mkdir(join(workspaceRoot, "packages/lib-a"), { recursive: true });
+        await mkdir(join(workspaceRoot, "packages/app"), { recursive: true });
+        await mkdir(externalRoot, { recursive: true });
+
+        await writeFile(join(workspaceRoot, "packages/lib-a/index.ts"), "export const a = 1;\n");
+        await writeFile(join(workspaceRoot, "packages/app/index.ts"), "export const app = 1;\n");
+
+        externalFile = join(externalRoot, "constants.ts");
+        await writeFile(externalFile, "export const shared = 1;\n");
+
+        git(workspaceRoot, "init", "--initial-branch=main");
+        git(workspaceRoot, "config", "user.email", "test@example.com");
+        git(workspaceRoot, "config", "user.name", "Test");
+        git(workspaceRoot, "add", ".");
+        git(workspaceRoot, "commit", "-m", "initial");
+    });
+
+    afterEach(async () => {
+        await rm(join(workspaceRoot, ".."), { force: true, recursive: true });
+    });
+
+    it("hashes external glob inputs and detects a change between snapshots", async () => {
+        expect.assertions(4);
+
+        const globPattern = `${externalRoot}/**/*.ts`;
+        const externalInputs = [{ pattern: globPattern, projects: ["lib-a"] }];
+        const projects: Record<string, ProjectConfiguration> = { "lib-a": { root: "packages/lib-a" } };
+        const projectGraph: ProjectGraph = {
+            dependencies: { "lib-a": [] },
+            nodes: { "lib-a": { data: { root: "packages/lib-a" }, name: "lib-a", type: "library" } },
+        };
+
+        const before = await computeExternalInputHashes(externalInputs);
+
+        expect(before[globPattern]).toBeTypeOf("string");
+
+        await writeFile(join(externalRoot, "extra.ts"), "export const extra = 1;\n");
+
+        const after = await computeExternalInputHashes(externalInputs);
+
+        expect(after[globPattern]).not.toBe(before[globPattern]);
+
+        const firstRun = await getAffectedProjects({
+            base: "HEAD",
+            externalInputs,
+            head: "HEAD",
+            projectGraph,
+            projects,
+            workspaceRoot,
+        });
+
+        // No previous snapshot: the owning project must not be reported clean.
+        expect(firstRun.changedExternalInputs).toStrictEqual([globPattern]);
+
+        const secondRun = await getAffectedProjects({
+            base: "HEAD",
+            externalInputs,
+            head: "HEAD",
+            previousExternalInputHashes: firstRun.externalInputHashes,
+            projectGraph,
+            projects,
+            workspaceRoot,
+        });
+
+        expect(secondRun.changedExternalInputs).toStrictEqual([]);
+    });
+
+    it("marks the owning project affected when an external input changes and its dependents downstream", async () => {
+        expect.assertions(4);
+
+        const externalInputs = collectExternalInputs(workspaceRoot, {
+            "lib-a": { root: "packages/lib-a", targets: { build: { inputs: [`file://${externalFile}`] } } },
+        });
+        const graph: ProjectGraph = {
+            dependencies: { app: [{ source: "app", target: "lib-a", type: "static" }], "lib-a": [] },
+            nodes: {
+                app: { data: { root: "packages/app" }, name: "app", type: "application" },
+                "lib-a": { data: { root: "packages/lib-a" }, name: "lib-a", type: "library" },
+            },
+        };
+        const projects: Record<string, ProjectConfiguration> = {
+            app: { root: "packages/app" },
+            "lib-a": { root: "packages/lib-a" },
+        };
+
+        const baseline = await getAffectedProjects({
+            base: "HEAD",
+            externalInputs,
+            head: "HEAD",
+            projectGraph: graph,
+            projects,
+            workspaceRoot,
+        });
+
+        expect(baseline.changedProjects).toStrictEqual(["lib-a"]);
+
+        const unchanged = await getAffectedProjects({
+            base: "HEAD",
+            externalInputs,
+            head: "HEAD",
+            previousExternalInputHashes: baseline.externalInputHashes,
+            projectGraph: graph,
+            projects,
+            workspaceRoot,
+        });
+
+        expect(unchanged.changedProjects).toStrictEqual([]);
+
+        await writeFile(externalFile, "export const shared = 3;\n");
+
+        const changed = await getAffectedProjects({
+            base: "HEAD",
+            externalInputs,
+            head: "HEAD",
+            previousExternalInputHashes: baseline.externalInputHashes,
+            projectGraph: graph,
+            projects,
+            workspaceRoot,
+        });
+
+        expect(changed.changedProjects).toStrictEqual(["lib-a"]);
+        expect(changed.affectedProjects).toStrictEqual(["lib-a", "app"]);
+    });
+});

packages/tooling/task-runner/__tests__/unit/backends/http.test.ts

@@ -5,9 +5,11 @@ import type { IncomingMessage, Server, ServerResponse } from "node:http";
 import { createServer } from "node:http";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
+import { Readable } from "node:stream";
 
 import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 
+import { createRemoteCacheBackend } from "../../../src/backends/factory";
 import { actionDigestForTaskHash, containsByTaskHash, retrieveByTaskHash, storeByTaskHash } from "../../../src/backends/hash-bridge";
 import { HttpRemoteCache } from "../../../src/backends/http";
 
@@ -1197,6 +1199,83 @@ describe(HttpRemoteCache, () => {
         });
     });
 
+    describe("remote cache error resilience", () => {
+        it("treats a 502 read as a miss and reports each failure without throwing", async () => {
+            expect.assertions(3);
+
+            const { server, url } = await startMockServer((_request, response) => {
+                response.writeHead(502);
+                response.end();
+            });
+
+            try {
+                const errors: unknown[] = [];
+                const cache = createRemoteCacheBackend({ localCasRoot: cacheDirectory, timeout: 250, url }, (error) => errors.push(error));
+
+                const first = await retrieveByTaskHash(cache, "first", cacheDirectory);
+                const second = await retrieveByTaskHash(cache, "second", cacheDirectory);
+
+                expect(first).toBe(false);
+                expect(second).toBe(false);
+                expect(errors).toHaveLength(2);
+
+                await cache.close();
+            } finally {
+                await closeServer(server);
+            }
+        });
+
+        it("falls back to the caller's onUploadError hook when no reporter is supplied", async () => {
+            expect.assertions(2);
+
+            const { server, url } = await startMockServer((_request, response) => {
+                response.writeHead(500);
+                response.end();
+            });
+
+            try {
+                const onUploadError = vi.fn<(hash: string, error: unknown) => void>();
+                const cache = createRemoteCacheBackend({ localCasRoot: cacheDirectory, onUploadError, timeout: 250, url });
+
+                await retrieveByTaskHash(cache, "first", cacheDirectory);
+                await retrieveByTaskHash(cache, "second", cacheDirectory);
+
+                expect(onUploadError).toHaveBeenCalledTimes(2);
+                expect(onUploadError.mock.calls[0]?.[0]).toBe("remote-cache");
+            } finally {
+                await closeServer(server);
+            }
+        });
+
+        it("treats a 502 write as a failed upload without throwing", async () => {
+            expect.assertions(2);
+
+            const { server, url } = await startMockServer((_request, response) => {
+                response.writeHead(502);
+                response.end();
+            });
+
+            try {
+                const errors: unknown[] = [];
+                const cache = createRemoteCacheBackend({ timeout: 250, url }, (error) => errors.push(error));
+                const bytes = Buffer.from("payload", "utf8");
+                const digest = { hash: createHash("sha256").update(bytes).digest("hex"), sizeBytes: bytes.byteLength };
+                const stored = await cache.storeAction(
+                    actionDigestForTaskHash("write-502"),
+                    { exitCode: 0, outputDirectories: [], outputFiles: [{ digest, isExecutable: false, path: "vis-entry.tar.gz" }] },
+                    [{ digest, open: () => Promise.resolve(Readable.from(bytes)) }],
+                );
+
+                expect(stored).toBe(false);
+                expect(errors).toHaveLength(1);
+
+                await cache.close();
+            } finally {
+                await closeServer(server);
+            }
+        });
+    });
+
     describe(actionDigestForTaskHash, () => {
         it("derives a stable sha256 digest from the task hash", () => {
             expect.assertions(3);

packages/tooling/task-runner/__tests__/unit/backends/reapi.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { Readable } from "node:stream";
 
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 
 import { actionDigestForTaskHash, containsByTaskHash, retrieveByTaskHash, storeByTaskHash } from "../../../src/backends/hash-bridge";
 import { ReapiRemoteCache } from "../../../src/backends/reapi";
@@ -422,6 +422,27 @@ describe(ReapiRemoteCache, () => {
 
             expect(result).toBeNull();
         });
+
+        it("treats a non-NOT_FOUND gRPC error as a cache miss through the hash bridge", async () => {
+            expect.assertions(3);
+
+            fixture.server.getActionResultErrorOverride = { code: 13, message: "internal" };
+
+            const cache = new ReapiRemoteCache({ url: fixture.server.address });
+            const directory = await createTemporaryDirectory();
+            const onError = vi.fn<(hash: string, error: unknown) => void>();
+
+            try {
+                // The backend method still surfaces the transport error; the
+                // orchestrator's bridge is where a failed read becomes a miss
+                // so the task can run normally.
+                await expect(cache.retrieveAction({ hash: "c".repeat(64), sizeBytes: 1 })).rejects.toMatchObject({ code: 13 });
+                await expect(retrieveByTaskHash(cache, "unreachable", directory, onError)).resolves.toBe(false);
+                expect(onError).toHaveBeenCalledTimes(1);
+            } finally {
+                await rm(directory, { force: true, recursive: true });
+            }
+        });
     });
 
     describe("containsAction", () => {