diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 3a2af872db..19ce7fd727 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -2444,6 +2444,80 @@ describe("sandbox adapter execution targets", () => { } }); + it("forwards the host indeterminate-outcome header so the sandbox server maps the 504 to a non-retryable 409", async () => { + // The host marks a possibly-committed mutation with a 504 and the + // `x-paperclip-bridge-outcome: indeterminate` header. The forward must keep + // that header, so the in-sandbox server maps the 504 to a non-retryable 409. + // If the forward drops the header, the client sees a retryable 504 and a + // retry repeats a mutation that already committed. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-outcome-")); + cleanupDirs.push(rootDir); + const remoteCwd = path.join(rootDir, "workspace"); + const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex"); + await mkdir(runtimeRootDir, { recursive: true }); + + const responseBody = JSON.stringify({ error: "Mutation outcome is indeterminate.", outcome: "indeterminate", retryable: false }); + const apiServer = createServer((_req, res) => { + res.writeHead(504, { + "content-type": "application/json", + "x-paperclip-bridge-outcome": "indeterminate", + }); + res.end(responseBody); + }); + await new Promise((resolve, reject) => { + apiServer.once("error", reject); + apiServer.listen(0, "127.0.0.1", () => resolve()); + }); + const address = apiServer.address(); + if (!address || typeof address === "string") { + throw new Error("Expected the bridge outcome test API server to listen on a TCP port."); + } + + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "e2b", + environmentId: "env-1", + leaseId: "lease-1", + remoteCwd, + runner: createLocalSandboxRunner(), + timeoutMs: 30_000, + }; + + const bridge = await startAdapterExecutionTargetPaperclipBridge({ + runId: "run-bridge-outcome", + target, + runtimeRootDir, + adapterKey: "codex", + hostApiToken: "real-run-jwt", + hostApiUrl: `http://127.0.0.1:${address.port}`, + }); + try { + const response = await fetch(`${bridge!.env.PAPERCLIP_API_URL}/api/issues/issue-1/comments`, { + method: "POST", + headers: { + authorization: `Bearer ${bridge!.env.PAPERCLIP_API_KEY}`, + "content-type": "application/json", + }, + body: JSON.stringify({ body: "Status update." }), + }); + + // The sandbox server maps the indeterminate 504 to a non-retryable 409. + expect(response.status).toBe(409); + // The outcome header and body still reach the client, so a caller that + // reads them still sees the indeterminate result. + expect(response.headers.get("x-paperclip-bridge-outcome")).toBe("indeterminate"); + await expect(response.json()).resolves.toEqual({ + error: "Mutation outcome is indeterminate.", + outcome: "indeterminate", + retryable: false, + }); + } finally { + await bridge?.stop(); + await new Promise((resolve) => apiServer.close(() => resolve())); + } + }); + it("forwards bridge traffic to the local listen origin even when public API URLs are configured", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-local-")); cleanupDirs.push(rootDir); diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 4c21a33acc..606016222c 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1246,7 +1246,12 @@ export function runtimeAssetDir( function buildBridgeResponseHeaders(response: Response): Record { const out: Record = {}; - for (const key of ["content-type", "etag", "last-modified"]) { + // Keep `x-paperclip-bridge-outcome` in this list. The host marks a + // possibly-committed mutation with the `indeterminate` outcome. The in-sandbox + // server reads that header to map the 504 to a terminal 409. If the forward + // drops the header, the server keeps the retryable 504 and a caller that + // retries 5xx can repeat a mutation that already committed. + for (const key of ["content-type", "etag", "last-modified", "x-paperclip-bridge-outcome"]) { const value = response.headers.get(key); if (value && value.trim().length > 0) out[key] = value.trim(); } @@ -2162,7 +2167,7 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { maxBodyBytes, getRuntimeParentContext: input.getRuntimeParentContext, runtimeSpan: input.runtimeSpan, - handleRequest: async (request) => { + handleRequest: async (request, options) => { const method = request.method.trim().toUpperCase() || "GET"; if (bridgeDebugEnabled) { await onLog( @@ -2177,11 +2182,19 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { } headers.set("authorization", `Bearer ${hostApiToken}`); headers.set("x-paperclip-run-id", input.runId); + // Abort the forward when the worker aborts the request (its per-iteration + // timeout or watchdog fired), or after the 30s ceiling, whichever comes + // first. The worker abort lets the bridge fail a hung forward fast + // instead of stranding the request until the 30s ceiling. + const timeoutSignal = AbortSignal.timeout(30_000); + const forwardSignal = options?.signal + ? AbortSignal.any([options.signal, timeoutSignal]) + : timeoutSignal; const response = await fetch(buildBridgeForwardUrl(hostApiUrl, request), { method, headers, ...(method === "GET" || method === "HEAD" ? {} : { body: request.body }), - signal: AbortSignal.timeout(30_000), + signal: forwardSignal, }); if (bridgeDebugEnabled) { await onLog( diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index e69fef32c8..028ec83407 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -13,12 +13,15 @@ import { createFileSystemSandboxCallbackBridgeQueueClient, createSandboxCallbackBridgeAsset, createSandboxCallbackBridgeToken, + getSandboxCallbackBridgeServerSource, sandboxCallbackBridgeDirectories, syncRemoteTextFileWithHashSkip, syncSandboxCallbackBridgeEntrypoint, startSandboxCallbackBridgeServer, startSandboxCallbackBridgeWorker, } from "./sandbox-callback-bridge.js"; +import type { SandboxCallbackBridgeQueueClient } from "./sandbox-callback-bridge.js"; +import type { RuntimeSpanRunner } from "./acpx-engine/startup-timing.js"; import type { RunProcessResult } from "./server-utils.js"; const execFile = promisify(execFileCallback); @@ -1460,4 +1463,1168 @@ describe("sandbox callback bridge", () => { expect(script).toContain("/workspace/b"); expect(script).toContain("/workspace/c"); }); + + // Capture the run-level error the worker surfaces through `runtimeSpan`. The + // worker runs a throwing function under the `sandbox.callbackBridge.workerFailed` + // span; this double records that error and re-throws, like the real runner. + function createWorkerErrorCapture(): { + runtimeSpan: RuntimeSpanRunner; + workerErrors: string[]; + } { + const workerErrors: string[] = []; + const runtimeSpan: RuntimeSpanRunner = async (name, work) => { + try { + return await work(); + } catch (error) { + if (name === "sandbox.callbackBridge.workerFailed") { + workerErrors.push(error instanceof Error ? error.message : String(error)); + } + throw error; + } + }; + return { runtimeSpan, workerErrors }; + } + + function bridgeRequestJson(id: string): string { + return `${JSON.stringify({ + id, + method: "GET", + path: "/api/agents/me", + query: "", + headers: {}, + body: "", + createdAt: new Date().toISOString(), + })}\n`; + } + + it("times out a stalled poll, writes a 503, and surfaces a run-level error", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-hang-")); + cleanupDirs.push(rootDir); + + const queueDir = path.posix.join(rootDir, "queue"); + const directories = sandboxCallbackBridgeDirectories(queueDir); + await mkdir(directories.requestsDir, { recursive: true }); + await writeFile(path.posix.join(directories.requestsDir, "req-a.json"), bridgeRequestJson("req-a"), "utf8"); + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + + const base = createFileSystemSandboxCallbackBridgeQueueClient(); + let listCalls = 0; + const client: SandboxCallbackBridgeQueueClient = { + ...base, + // The first poll never resolves — a silently unresponsive sandbox channel. + // The per-iteration timeout must convert the hang into a caught error. The + // request never reaches the handler, so the recovery path can safely 503 + // it. Later calls (the recovery's failPendingRequests) resolve. + listJsonFiles: async (dir) => { + listCalls += 1; + if (listCalls === 1) { + return await new Promise(() => {}); + } + return await base.listJsonFiles(dir); + }, + }; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async () => ({ status: 200, body: "ok" }), + }); + + const responseFile = await waitForJsonFile(directories.responsesDir, 3_000); + const responseBody = await readFile(path.posix.join(directories.responsesDir, responseFile), "utf8"); + expect(JSON.parse(responseBody).status).toBe(503); + expect(workerErrors.length).toBeGreaterThan(0); + expect(workerErrors[0]).toContain("timed out"); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("does not abandon an in-flight handler, so a started mutation is not applied twice", async () => { + // Prove the completion fence for a request whose handler already started. The + // per-iteration timeout rejects the wrapper but does not stop the handler, so + // the host operation (a mutation) is still in flight. The recovery path must + // not write a 503 there; a 503 makes the caller retry while the original + // mutation still completes, applying it twice. The test proves no 503 lands + // and the handler's real response is delivered exactly once. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-late.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + // An in-memory queue client with no file-existence guards. A response write + // always lands here, so only the completion fence controls the outcome. The + // real filesystem and command clients add their own existence guards; this + // client removes them, so the test isolates the fence. + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-late")); + const responseWrites: Array<{ path: string; status: number }> = []; + const requestRemovals: string[] = []; + + const handlerControl: { release: (() => void) | null } = { release: null }; + let handlerCompleted = false; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + // The handler stays pending until the test releases it, well after the + // per-iteration timeout fires. It records that it finished, so the test + // proves the in-flight handler truly ran. + handleRequest: () => + new Promise<{ status: number; body?: string }>((resolve) => { + handlerControl.release = () => { + handlerCompleted = true; + resolve({ status: 200, body: "req-late" }); + }; + }), + }); + + // Wait until the per-iteration timeout surfaced a run error and the handler + // is pending. The recovery path ran, so it already skipped the in-flight + // request. + await waitFor( + () => workerErrors.some((message) => message.includes("timed out")) && handlerControl.release !== null, + 3_000, + ); + // The recovery path wrote no 503 for the in-flight request. + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + + // Release the handler after the timeout. It delivers the real response. + handlerControl.release?.(); + await waitFor(() => handlerCompleted, 3_000); + // Give the finalize write and remove time to land. + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(handlerCompleted).toBe(true); + // The handler committed exactly one response for the request: the real 200. + expect(responseWrites.filter((write) => write.path === responsePath)).toEqual([ + { path: responsePath, status: 200 }, + ]); + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + // The handler removed the request file exactly once, after it finalized. + expect(requestRemovals).toEqual([requestPath]); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("aborts an in-flight handler on timeout, so a cooperating handler finalizes instead of stranding the request", async () => { + // A handler that threads the worker signal into its work must stop when the + // per-iteration timeout fires. It then finalizes with its own error + // response, so the request does not strand with no response. The recovery + // path still writes no 503 for the handler-owned request, so a started + // mutation is not applied twice. The abort reaches the handler after the host + // operation started, so the handler finalizes a non-retryable 504, not a + // retryable 502; the caller then does not retry a mutation that may have + // committed. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-abort.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-abort")); + const responseWrites: Array<{ path: string; status: number }> = []; + const requestRemovals: string[] = []; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + let handlerAborted = false; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + // The handler stays pending until the worker aborts the signal. It then + // rejects, so the request finalizes with the handler's own error response. + handleRequest: (_request, options) => + new Promise<{ status: number; body?: string }>((_resolve, reject) => { + options?.signal.addEventListener("abort", () => { + handlerAborted = true; + reject(new Error("aborted by worker")); + }); + }), + }); + + // The handler finalizes only after the worker aborts it. The request gets a + // terminal response (a 502 from the handler failure), never stranded. + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + expect(handlerAborted).toBe(true); + expect(workerErrors.some((message) => message.includes("timed out"))).toBe(true); + // Exactly one response landed: the handler's non-retryable 504. The recovery + // path wrote no competing 503, and the aborted handler wrote no retryable 502. + expect(responseWrites.filter((write) => write.path === responsePath)).toEqual([ + { path: responsePath, status: 504 }, + ]); + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + // The handler removed the request file after it finalized. The recovery path + // may issue a redundant idempotent remove for the same path when the handler + // finalized first, so assert the removal happened rather than a fixed count. + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("finalizes an aborted handler that ignores the signal and never settles with a non-retryable 504 backstop", async () => { + // A handler that does not thread the worker signal into its work and never + // settles must not strand the request. The recovery path aborts the handler, + // waits the grace, then writes a non-retryable 504 backstop, so the request + // gets a terminal response even when the handler ignores the abort. The 504 + // is non-retryable, so the caller does not retry a mutation that may have + // committed. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-stuck.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-stuck")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + let handlerStarted = false; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + // A short grace, so the backstop fires quickly after the abort. + abortedHandlerGraceMs: 30, + runtimeSpan, + authorizeRequest: async () => null, + // The handler ignores the worker signal and never settles. Without the + // backstop, the request would stay without a response forever. + handleRequest: () => { + handlerStarted = true; + return new Promise<{ status: number; body?: string }>(() => {}); + }, + }); + + // The backstop writes the terminal response after the grace. + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + expect(handlerStarted).toBe(true); + expect(workerErrors.some((message) => message.includes("timed out"))).toBe(true); + // Exactly one response landed: the non-retryable 504 backstop. The recovery + // path wrote no retryable 503 or 502 for the handler-owned request. + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + expect(requestResponses[0]?.status).toBe(504); + const parsed = JSON.parse(requestResponses[0]!.body.trim()); + const responseBody = JSON.parse(parsed.body); + expect(responseBody.outcome).toBe("indeterminate"); + expect(responseBody.retryable).toBe(false); + expect(parsed.headers?.["x-paperclip-bridge-outcome"]).toBe("indeterminate"); + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + // The backstop removed the request file, so it does not strand in the queue. + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("finalizes a late worker-aborted handler with a non-retryable 504, so the caller does not retry a committed mutation", async () => { + // Prove the completion status for a mutating request that the worker aborts. + // The per-iteration timeout aborts a handler that already started its host + // operation, so the mutation may have committed. The bridge cannot cancel a + // host operation that is in flight. The handler completes late, after the + // abort. Its response must be a non-retryable 504, never a retryable 502 or + // 503; a retry of either would apply the mutation twice. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-late-abort.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-late-abort")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + let handlerCompletedLate = false; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + // The handler mirrors a mutating forward that already committed on the + // host. It rejects only after the worker aborts it, so its completion is + // late. The mutation stays committed; the abort cannot undo it. + handleRequest: (_request, options) => + new Promise<{ status: number; body?: string }>((_resolve, reject) => { + options?.signal.addEventListener("abort", () => { + handlerCompletedLate = true; + reject(new Error("aborted by worker after the mutation committed")); + }); + }), + }); + + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + expect(handlerCompletedLate).toBe(true); + expect(workerErrors.some((message) => message.includes("timed out"))).toBe(true); + + // Exactly one response landed for the request: the non-retryable 504. + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + expect(requestResponses[0]?.status).toBe(504); + + // The bridge wrote no retryable status for the possibly-committed mutation. + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + + // The response marks the outcome indeterminate and non-retryable, so the + // caller does not retry. + const parsed = JSON.parse(requestResponses[0]?.body ?? "{}"); + expect(parsed.status).toBe(504); + const responseBody = JSON.parse(parsed.body); + expect(responseBody.outcome).toBe("indeterminate"); + expect(responseBody.retryable).toBe(false); + expect(parsed.headers?.["x-paperclip-bridge-outcome"]).toBe("indeterminate"); + + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("retries a failed 504 backstop write and keeps the request until it lands, so the caller is not stranded", async () => { + // A backstop write can fail transiently, or it can exceed the iteration + // timeout. The recovery path must not drop the request file on that failure. A + // dropped file leaves no terminal 504 for the caller and fences out a late + // handler, so the caller waits to its own deadline and gets a generic 502. The + // recovery path retries the write and removes the request file only after the + // write lands. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-retry.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-retry")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: Array<{ path: string; afterWrites: number }> = []; + let writeAttempts = 0; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + writeAttempts += 1; + // The first backstop write fails; the retry then succeeds. + if (writeAttempts === 1) { + throw new Error("simulated transient write failure"); + } + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push({ path: remotePath, afterWrites: writeAttempts }); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + // A short grace, so the backstop fires quickly after the abort. + abortedHandlerGraceMs: 30, + runtimeSpan, + authorizeRequest: async () => null, + // The handler ignores the worker signal and never settles, so only the + // backstop can finalize the request. + handleRequest: () => new Promise<{ status: number; body?: string }>(() => {}), + }); + + // The backstop retries the write, so the terminal 504 lands after the failure. + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + expect(requestResponses[0]?.status).toBe(504); + // The write failed once, so the backstop wrote on a later attempt. + expect(writeAttempts).toBeGreaterThanOrEqual(2); + // The recovery path removed the request file only after the write landed. The + // failed first attempt kept the file, so the request never dropped with no + // response. + const requestPathRemovals = requestRemovals.filter((entry) => entry.path === requestPath); + expect(requestPathRemovals.length).toBeGreaterThanOrEqual(1); + for (const entry of requestPathRemovals) { + expect(entry.afterWrites).toBeGreaterThanOrEqual(2); + } + // The bridge wrote no retryable status for the possibly-committed mutation. + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("retries a handler response write that fails transiently and delivers the real response, not a retryable 503", async () => { + // A handler settles with its own response, but the terminal write fails + // once. `finalize` must retry the write and deliver the real response. It + // must not fence out recovery and leave the request for a retryable 503, + // which would let the caller repeat a possibly-committed mutation. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-write-retry.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-write-retry")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + let writeAttempts = 0; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + writeAttempts += 1; + // The first handler-response write fails; the retry then succeeds. + if (writeAttempts === 1) { + throw new Error("simulated transient response write failure"); + } + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 500, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + // The handler settles with its own 200 response. + handleRequest: async () => ({ status: 200, body: JSON.stringify({ ok: true }) }), + }); + + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + // Exactly one response landed: the handler's real 200. The retry delivered it + // after the transient failure. + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + expect(requestResponses[0]?.status).toBe(200); + expect(writeAttempts).toBeGreaterThanOrEqual(2); + // The bridge wrote no retryable status, so the caller does not repeat the + // request. + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + // The handler removed the request file after the write landed. + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("re-arms the 504 backstop when every handler response write fails, so the caller gets a terminal 504 instead of stranding", async () => { + // A handler settles with its own response, but every terminal write fails. + // `finalize` must roll back the fence and re-arm the 504 backstop, so the + // recovery still delivers a non-retryable terminal response. Without the + // re-arm, a rejected write leaves the request for a retryable 503 and a hung + // write strands the caller until its own deadline. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-write-backstop.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-write-backstop")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + let writeAttempts = 0; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + writeAttempts += 1; + // Fail every `finalize` write attempt. The re-armed 504 backstop then + // writes on a later attempt. + if (writeAttempts <= 3) { + throw new Error("simulated persistent response write failure"); + } + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 500, + watchdogTimeoutMs: 10_000, + // A short grace, so the re-armed backstop fires quickly. + abortedHandlerGraceMs: 30, + runtimeSpan, + authorizeRequest: async () => null, + // The handler settles with its own 200 response, a committed mutation. + handleRequest: async () => ({ status: 200, body: JSON.stringify({ ok: true }) }), + }); + + // The re-armed backstop writes the terminal 504 after the finalize writes fail. + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + // The backstop delivered a non-retryable 504 with an indeterminate outcome. + expect(requestResponses[0]?.status).toBe(504); + const parsed = JSON.parse(requestResponses[0]?.body ?? "{}"); + const responseBody = JSON.parse(parsed.body); + expect(responseBody.outcome).toBe("indeterminate"); + expect(responseBody.retryable).toBe(false); + expect(parsed.headers?.["x-paperclip-bridge-outcome"]).toBe("indeterminate"); + // The finalize writes failed before the backstop wrote, so it took a later + // attempt. + expect(writeAttempts).toBeGreaterThanOrEqual(4); + // The bridge wrote no retryable status for the possibly-committed mutation. + expect(responseWrites.some((write) => write.status === 503)).toBe(false); + expect(responseWrites.some((write) => write.status === 502)).toBe(false); + // The backstop removed the request file, so it does not strand in the queue. + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("retries a recovery 503 write that fails transiently, delivers the 503, and removes the request", async () => { + // The poll times out, so the recovery path aborts the queued request with a + // 503. The first 503 write fails, so the recovery must retry it inside the + // same pass. It then delivers the 503 and removes the request. The request is + // unclaimed, so its host mutation never ran; the 503 stays retry-safe. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-503-retry.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-503-retry")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + let listCalls = 0; + let writeAttempts = 0; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + // The first poll never resolves — a silently unresponsive sandbox channel. + // The per-iteration timeout converts the hang into a caught error, so the + // loop `catch` runs the recovery path. Later listings resolve, so the + // recovery enumerates and aborts the request. + listJsonFiles: async (dir) => { + if (dir !== directories.requestsDir) { + return []; + } + listCalls += 1; + if (listCalls === 1) { + return await new Promise(() => {}); + } + return [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort(); + }, + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + writeAttempts += 1; + // The first recovery 503 write fails; the retry then succeeds. + if (writeAttempts === 1) { + throw new Error("simulated transient recovery 503 write failure"); + } + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status, body }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 200, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async () => ({ status: 200, body: "ok" }), + }); + + await waitFor(() => responseWrites.some((write) => write.path === responsePath), 3_000); + + // Exactly one response landed: the retry-safe 503. The retry delivered it + // after the transient failure. + const requestResponses = responseWrites.filter((write) => write.path === responsePath); + expect(requestResponses).toHaveLength(1); + expect(requestResponses[0]?.status).toBe(503); + expect(writeAttempts).toBeGreaterThanOrEqual(2); + // The recovery removed the request file only after the 503 write landed. + expect(requestRemovals).toContain(requestPath); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("keeps the queued request when every recovery 503 write fails, so a later pass can still deliver a terminal 503", async () => { + // The poll times out, so the recovery path aborts the queued request with a + // 503. Every 503 write fails. The recovery must keep the request file instead + // of dropping it. Without the fix, the recovery removed the request before its + // single write attempt, so a failed write left no queue state; a later + // recovery pass found nothing and the caller waited until its own deadline. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-503-keep.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-503-keep")); + const responseWrites: Array<{ path: string; status: number; body: string }> = []; + const requestRemovals: string[] = []; + let listCalls = 0; + let writeAttempts = 0; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => { + if (dir !== directories.requestsDir) { + return []; + } + listCalls += 1; + if (listCalls === 1) { + return await new Promise(() => {}); + } + return [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort(); + }, + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async () => { + writeAttempts += 1; + // Fail every recovery 503 write attempt. + throw new Error("simulated persistent recovery 503 write failure"); + }, + rename: async () => {}, + remove: async (remotePath) => { + requestRemovals.push(remotePath); + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 200, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async () => ({ status: 200, body: "ok" }), + }); + + // Wait until the recovery pass has exhausted its bounded 503 write attempts + // (three, the same bound as the finalize and 504 backstop writes). + await waitFor(() => writeAttempts >= 3, 3_000); + + // The recovery could not write the 503, so it kept the request file. The + // request still sits in the queue for a later recovery pass to deliver a + // terminal 503. A drop here would strand the caller until its own deadline. + expect(requestRemovals).not.toContain(requestPath); + expect(requestBodies.has(requestPath)).toBe(true); + // No response landed, because every write failed. The recovery wrote no + // partial or retryable status for the possibly-committed mutation. + expect(responseWrites).toHaveLength(0); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("maps an indeterminate host outcome to a non-retryable 409 in the in-sandbox bridge server", () => { + // The in-sandbox server returns the host response to the sandbox caller. A 5xx + // status is retryable by convention, so the server must not forward the + // indeterminate 504 as a retry-safe status. It maps the indeterminate outcome + // to a non-retryable 409, so a caller that retries 5xx does not repeat a + // possibly-committed mutation. The outcome header and body still forward, so a + // caller that reads them still sees the indeterminate result. + const source = getSandboxCallbackBridgeServerSource(); + expect(source).toContain("x-paperclip-bridge-outcome"); + expect(source).toContain('=== "indeterminate"'); + expect(source).toContain("res.statusCode = 409"); + }); + + it("abandons a request before its handler starts, so the mutation never runs", async () => { + // Prove the reverse race. When the recovery path claims a request before the + // handler starts its host operation, the handler must bail without running + // the mutation. A retry after the 503 then applies the mutation once. + const waitFor = async (predicate: () => boolean, timeoutMs: number) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error("waitFor timed out"); + }; + + const queueDir = "/virtual-bridge/queue"; + const directories = sandboxCallbackBridgeDirectories(queueDir); + const requestFile = "req-early.json"; + const requestPath = path.posix.join(directories.requestsDir, requestFile); + const responsePath = path.posix.join(directories.responsesDir, requestFile); + + const requestBodies = new Map(); + requestBodies.set(requestPath, bridgeRequestJson("req-early")); + const responseWrites: Array<{ path: string; status: number }> = []; + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async (dir) => + dir === directories.requestsDir + ? [...requestBodies.keys()].map((entry) => path.posix.basename(entry)).sort() + : [], + readTextFile: async (remotePath) => { + const body = requestBodies.get(remotePath); + if (body === undefined) { + throw new Error(`missing request ${remotePath}`); + } + return body; + }, + writeTextFile: async () => {}, + writeResponseFile: async (remotePath, body) => { + responseWrites.push({ path: remotePath, status: JSON.parse(body.trim()).status }); + return { wrote: true }; + }, + rename: async () => {}, + remove: async (remotePath) => { + requestBodies.delete(remotePath); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + + // The authorize step stays pending until the test releases it, so the handler + // does not start before the per-iteration timeout fires and the recovery path + // claims the request. + const authorizeControl: { release: (() => void) | null } = { release: null }; + let handlerCalls = 0; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + iterationTimeoutMs: 50, + watchdogTimeoutMs: 10_000, + runtimeSpan, + authorizeRequest: () => + new Promise((resolve) => { + authorizeControl.release = () => resolve(null); + }), + handleRequest: async () => { + handlerCalls += 1; + return { status: 200, body: "req-early" }; + }, + }); + + // Wait until the recovery path wrote the 503 and the authorize step is + // pending. + await waitFor( + () => responseWrites.some((write) => write.status === 503) && authorizeControl.release !== null, + 3_000, + ); + expect(responseWrites.some((write) => write.path === responsePath && write.status === 503)).toBe(true); + + // Release authorize after the 503. The handler must bail at its claim. + authorizeControl.release?.(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + // The handler never ran, so the mutation never applied. + expect(handlerCalls).toBe(0); + // No competing 200 landed over the 503. + expect(responseWrites.some((write) => write.path === responsePath && write.status === 200)).toBe(false); + expect(workerErrors.some((message) => message.includes("timed out"))).toBe(true); + + await worker.stop({ drainTimeoutMs: 10 }); + }); + + it("trips the watchdog on a stalled poll, writes a 503, and surfaces a run-level error", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-watchdog-")); + cleanupDirs.push(rootDir); + + const queueDir = path.posix.join(rootDir, "queue"); + const directories = sandboxCallbackBridgeDirectories(queueDir); + await mkdir(directories.requestsDir, { recursive: true }); + await writeFile(path.posix.join(directories.requestsDir, "req-w.json"), bridgeRequestJson("req-w"), "utf8"); + + const base = createFileSystemSandboxCallbackBridgeQueueClient(); + let listCalls = 0; + const client: SandboxCallbackBridgeQueueClient = { + ...base, + listJsonFiles: async (dir) => { + listCalls += 1; + // The first poll (the loop) never resolves — a stalled channel that the + // per-iteration timeout does not catch soon enough. Later calls (the + // watchdog's failPendingRequests) resolve, so it enumerates and 503s. + if (listCalls === 1) { + return await new Promise(() => {}); + } + return await base.listJsonFiles(dir); + }, + }; + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + // The per-iteration timeout is far larger than the watchdog threshold, so + // the watchdog — not the per-iteration timeout — is the mechanism proven. + iterationTimeoutMs: 400, + watchdogTimeoutMs: 50, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async () => ({ status: 200, body: "ok" }), + }); + + const responseFile = await waitForJsonFile(directories.responsesDir, 3_000); + const responseBody = await readFile(path.posix.join(directories.responsesDir, responseFile), "utf8"); + expect(JSON.parse(responseBody).status).toBe(503); + expect(workerErrors.some((message) => message.includes("no successful poll iteration"))).toBe(true); + + await worker.stop({ drainTimeoutMs: 400 }); + }); + + it("processes a fast request with no false-positive timeout and no run-level error", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-fast-")); + cleanupDirs.push(rootDir); + + const queueDir = path.posix.join(rootDir, "queue"); + const directories = sandboxCallbackBridgeDirectories(queueDir); + await mkdir(directories.requestsDir, { recursive: true }); + await writeFile(path.posix.join(directories.requestsDir, "req-ok.json"), bridgeRequestJson("req-ok"), "utf8"); + + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + const processed: string[] = []; + + const worker = await startSandboxCallbackBridgeWorker({ + client: createFileSystemSandboxCallbackBridgeQueueClient(), + queueDir, + iterationTimeoutMs: 200, + watchdogTimeoutMs: 1_000, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async (request) => { + processed.push(request.id); + return { status: 200, body: request.id }; + }, + }); + + const responseFile = await waitForJsonFile(directories.responsesDir, 3_000); + const responseBody = await readFile(path.posix.join(directories.responsesDir, responseFile), "utf8"); + expect(JSON.parse(responseBody).status).toBe(200); + expect(JSON.parse(responseBody).body).toBe("req-ok"); + + // Let several idle poll iterations and watchdog checks pass. A healthy idle + // loop must never trip the watchdog and never surface a run-level error. + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(processed).toEqual(["req-ok"]); + expect(workerErrors).toEqual([]); + + await worker.stop({ drainTimeoutMs: 50 }); + }); }); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index 9a373b869d..4d9cd0b014 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -19,6 +19,37 @@ const DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS = 30_000; const DEFAULT_BRIDGE_STOP_TIMEOUT_MS = 2_000; const DEFAULT_BRIDGE_MAX_QUEUE_DEPTH = 64; const DEFAULT_BRIDGE_MAX_BODY_BYTES = 256 * 1024; +// Per-iteration timeout for one poll-loop client call. A healthy control-plane +// round trip finishes in well under one second, so 10s is far above a normal +// iteration and never false-fires on a slow-but-live call. It is also well +// under the in-sandbox 30s response deadline +// (PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS), so the host loop fails fast and writes +// 503 responses before the in-sandbox client gives up. A silently unresponsive +// sandbox channel makes a client call hang with no reject; this timeout turns +// that hang into a caught error, so the loop `catch` runs `failPendingRequests`. +const DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS = 10_000; +// Watchdog backstop for a hang that the per-iteration timeout does not catch +// (for example many slow-but-under-timeout calls, or a stall outside the awaited +// calls). It is larger than one iteration timeout, so a single slow iteration +// never trips it, and it stays under the in-sandbox 30s response deadline. +const DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS = 20_000; +// Grace period the recovery path gives an aborted in-flight handler to finalize +// its own response. The recovery path aborts the handler, then waits this long. +// A cooperating handler threads the abort signal into its work, rejects, and +// writes its own response inside the grace, so its accurate result wins. A +// handler that ignores the signal and never settles does not write inside the +// grace; the recovery path then writes a non-retryable 504 backstop, so the +// request never strands with no response. The grace is well under the in-sandbox +// 30s response deadline, so the backstop lands before the in-sandbox client +// gives up. +const DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS = 5_000; +// The recovery path retries the 504 backstop write this many times before it +// gives up. A single transient write failure, or one that exceeds the iteration +// timeout, then does not strand the caller with no terminal response. +const MAX_BACKSTOP_WRITE_ATTEMPTS = 3; +// The delay between two 504 backstop write attempts. It is short, so all retries +// finish well under the in-sandbox 30s response deadline. +const BACKSTOP_WRITE_RETRY_MS = 50; const REMOTE_WRITE_BASE64_CHUNK_SIZE = 32 * 1024; const SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT = "paperclip-bridge-server.mjs"; const SANDBOX_EXEC_CHANNEL_ENV = "PAPERCLIP_SANDBOX_EXEC_CHANNEL"; @@ -28,6 +59,12 @@ const SANDBOX_EXEC_CHANNEL_BRIDGE = "bridge"; * write the response, and remove the request file. */ const CALLBACK_BRIDGE_RELAY_REQUEST_SPAN = "sandbox.callbackBridge.relayRequest"; +/** Span name for a failed or hung bridge worker. The worker runs a throwing + * function under this span through `input.runtimeSpan`, so the failure lands in + * the run trace. The run and the orchestrator then see the hang, not only + * stdout. */ +const CALLBACK_BRIDGE_WORKER_FAILED_SPAN = "sandbox.callbackBridge.workerFailed"; + export const DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES = DEFAULT_BRIDGE_MAX_BODY_BYTES; export interface SandboxCallbackBridgeRouteRule { @@ -203,6 +240,27 @@ function normalizeTimeoutMs(value: number | null | undefined, fallback: number): return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : fallback; } +/** + * Race a promise against a timeout. On timeout the returned promise rejects with + * a clear error. The helper clears the timer on every settle path, so it leaks + * no `setTimeout`. The wrapped promise is not cancelable; when it never settles, + * it keeps running in the background, but the caller already moved on through + * the reject. + */ +function withTimeout(promise: Promise, timeoutMs: number, label: string): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => { + reject(new Error(`${label} timed out after ${timeoutMs}ms.`)); + }, timeoutMs); + }); + return Promise.race([promise, timeout]).finally(() => { + if (timer !== undefined) { + clearTimeout(timer); + } + }); +} + function toBuffer(bytes: Buffer | Uint8Array | ArrayBuffer): Buffer { if (Buffer.isBuffer(bytes)) return bytes; if (bytes instanceof ArrayBuffer) return Buffer.from(bytes); @@ -642,8 +700,30 @@ export async function startSandboxCallbackBridgeWorker(input: { client: SandboxCallbackBridgeQueueClient; queueDir: string; pollIntervalMs?: number | null; + // Per-iteration timeout for one poll-loop client call (the `listJsonFiles` + // poll and one `processRequestFile`). On timeout the loop `catch` runs + // `failPendingRequests`. Defaults to DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS. + iterationTimeoutMs?: number | null; + // Watchdog threshold. When the loop makes no successful iteration within this + // time, the watchdog runs `failPendingRequests` and surfaces a run-level error + // through `runtimeSpan`. Defaults to DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS. + watchdogTimeoutMs?: number | null; + // Grace the recovery path gives an aborted in-flight handler to finalize its + // own response before the recovery path writes a non-retryable 504 backstop. + // Defaults to DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS. + abortedHandlerGraceMs?: number | null; authorizeRequest?: (request: SandboxCallbackBridgeRequest) => string | null | Promise; - handleRequest: (request: SandboxCallbackBridgeRequest) => Promise<{ + // Handle one bridge request. The worker passes an `AbortSignal` through + // `options.signal`. The per-iteration timeout, the watchdog, and worker + // failure recovery abort it, so a handler that threads the signal into its + // work (for example a `fetch`) stops and rejects instead of running forever. + // The handler then finalizes with its own error response, so the request does + // not strand with no response. A handler that ignores the signal keeps its + // earlier behavior. + handleRequest: ( + request: SandboxCallbackBridgeRequest, + options?: { signal: AbortSignal }, + ) => Promise<{ status: number; headers?: Record; body?: string; @@ -663,6 +743,12 @@ export async function startSandboxCallbackBridgeWorker(input: { runtimeSpan?: RuntimeSpanRunner; }): Promise { const pollIntervalMs = normalizeTimeoutMs(input.pollIntervalMs, DEFAULT_BRIDGE_POLL_INTERVAL_MS); + const iterationTimeoutMs = normalizeTimeoutMs(input.iterationTimeoutMs, DEFAULT_BRIDGE_ITERATION_TIMEOUT_MS); + const watchdogTimeoutMs = normalizeTimeoutMs(input.watchdogTimeoutMs, DEFAULT_BRIDGE_WATCHDOG_TIMEOUT_MS); + const abortedHandlerGraceMs = normalizeTimeoutMs( + input.abortedHandlerGraceMs, + DEFAULT_BRIDGE_ABORTED_HANDLER_GRACE_MS, + ); const maxBodyBytes = normalizeTimeoutMs(input.maxBodyBytes, DEFAULT_BRIDGE_MAX_BODY_BYTES); const directories = sandboxCallbackBridgeDirectories(input.queueDir); const queueDirectories = [ @@ -694,96 +780,519 @@ export async function startSandboxCallbackBridgeWorker(input: { const buildWorkerFailureMessage = (error: unknown) => `Sandbox callback bridge worker failed: ${error instanceof Error ? error.message : String(error)}`; + // Per-attempt finalization guard. Each `processRequestFile` call registers one, + // keyed by the request file name. The guard is the completion fence between the + // request handler and the per-iteration timeout or watchdog recovery. Node runs + // one event loop, so a synchronous check-and-set of `claim` is atomic. The + // first path to move `claim` off `unclaimed` wins. + // + // The `claim` value has three states: + // - `unclaimed`: no path owns the request yet. + // - `handler`: the request handler owns finalization. It set this before it + // started the host operation, or when it wrote a 400/403/response. It will + // write the real response. + // - `abandon`: the recovery path owns the request. It writes a 503. + // + // The recovery path must never write a 503 for a request whose handler already + // started. The per-iteration timeout and the watchdog cannot cancel a host + // operation that is in flight. A 503 there makes the caller retry while the + // original mutation still completes, so the mutation applies twice. So the + // recovery path abandons only a request that the handler did not yet claim; the + // handler, when it later reaches the host-operation claim, sees the abandon and + // does not run the mutation. This keeps a retry after the 503 exactly-once. + // + // Each guard also holds an `AbortController`. The recovery path aborts it, so a + // handler that already started (claim `handler`) stops its work and finalizes + // with its own error response. The abort turns a stranded request into a prompt + // error response for a handler that threads the signal into its work. + // + // A worker abort reaches the handler only after the host operation started, so + // the mutation may have committed. The bridge cannot cancel a host operation + // that is in flight. So the handler finalizes a worker-aborted request with a + // non-retryable 504, not a retryable 502. The caller must not retry a 504, so + // it never re-applies a mutation that already committed. A retry-safe 503 comes + // only from the recovery path, and only before the host operation starts. + // + // A handler that ignores the abort signal and never settles would still keep + // the request without a response, because the recovery path must not write a + // competing 503 for an in-flight mutation. So the recovery path also arms a + // backstop timer for each handler-owned request. It aborts the handler, then + // waits `abortedHandlerGraceMs`. A cooperating handler finalizes inside the + // grace, so its own response wins and `finalize` clears the timer. A handler + // that never settles does not finalize inside the grace; the timer then writes + // a non-retryable 504 backstop, so the request never strands. The backstop is + // non-retryable for the same reason the handler's own 504 is: the recovery + // cannot cancel a committed mutation, so the caller must not retry. + // + // The `finalized` flag is the single-writer fence between the handler's own + // `finalize` and the backstop timer. Node runs one event loop, so the + // synchronous check-and-set is atomic. The first path to set it writes the + // terminal response; the other bails. + type RequestFinalizeGuard = { + claim: "unclaimed" | "handler" | "abandon"; + controller: AbortController; + finalized: boolean; + backstopTimer?: ReturnType; + }; + const inFlightRequestGuards = new Map(); + const processRequestFile = async (fileName: string) => { + // Skip a request that already has an active attempt. The guard map holds only + // in-flight attempts; the attempt's finally removes its guard when it ends. A + // file that still has a guard is in flight, or it waits for its aborted-handler + // 504 backstop. A second attempt would register a new guard and re-run the host + // mutation, so the mutation could apply twice. + if (inFlightRequestGuards.has(fileName)) { + return; + } const requestPath = path.posix.join(directories.requestsDir, fileName); const responsePath = path.posix.join(directories.responsesDir, fileName); - const raw = await input.client.readTextFile(requestPath); - let request: SandboxCallbackBridgeRequest; - try { - request = JSON.parse(raw) as SandboxCallbackBridgeRequest; - } catch { - const requestId = fileName.replace(/\.json$/i, "") || randomUUID(); - await writeBridgeResponse(input.client, requestPath, responsePath, { - id: requestId, - status: 400, - headers: { "content-type": "application/json" }, - body: JSON.stringify({ error: "Invalid bridge request payload." }), - completedAt: new Date().toISOString(), - }); - await input.client.remove(requestPath); - return; - } - - const denialReason = await authorizeRequest(request); - if (denialReason) { - await writeBridgeResponse(input.client, requestPath, responsePath, { - id: request.id, - status: 403, - headers: { "content-type": "application/json" }, - body: JSON.stringify({ error: denialReason }), - completedAt: new Date().toISOString(), - }); - await input.client.remove(requestPath); - return; - } - - try { - const result = await input.handleRequest(request); - const responseBody = result.body ?? ""; - if (Buffer.byteLength(responseBody, "utf8") > maxBodyBytes) { - throw new Error(`Bridge response body exceeded the configured size limit of ${maxBodyBytes} bytes.`); + const guard: RequestFinalizeGuard = { + claim: "unclaimed", + controller: new AbortController(), + finalized: false, + }; + inFlightRequestGuards.set(fileName, guard); + // Claim the request for the handler. Return `false` when the recovery path + // already claimed it; the caller must then not run the mutation and must not + // write a response, because the recovery path writes a 503 and the caller + // may retry. + const claimForHandler = (): boolean => { + if (guard.claim === "abandon") { + return false; } - await writeBridgeResponse(input.client, requestPath, responsePath, { - id: request.id, - status: result.status, - headers: result.headers ?? {}, - body: responseBody, - completedAt: new Date().toISOString(), - }); - } catch (error) { - console.warn( - `[paperclip] sandbox callback bridge handler failed for ${request.id}: ${error instanceof Error ? error.message : String(error)}`, - ); - await writeBridgeResponse(input.client, requestPath, responsePath, { - id: request.id, - status: 502, - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - error: error instanceof Error ? error.message : String(error), - }), - completedAt: new Date().toISOString(), - }); + guard.claim = "handler"; + return true; + }; + // Finalize the request exactly once. Claim it for the handler first. When the + // recovery path already won the claim, skip both the write and the remove. + // The `finalized` fence stops a double write when the backstop timer already + // wrote a 504 for a handler the recovery path aborted. The handler wins when + // it settles inside the grace; the backstop wins when the handler never + // settles. + const finalize = async (response: SandboxCallbackBridgeResponse) => { + if (!claimForHandler()) { + return; + } + if (guard.finalized) { + return; + } + guard.finalized = true; + // This finalize now owns delivery for the request, so drop a pending + // backstop timer. The handler settled inside the grace, so the backstop no + // longer needs to wait; `finalize` delivers the terminal response itself, + // inline, and never leaves the request file for a detached timer that the + // busy poll loop could starve. + if (guard.backstopTimer !== undefined) { + clearTimeout(guard.backstopTimer); + guard.backstopTimer = undefined; + } + // Write the handler response, bounded by the per-iteration timeout so a + // hung sandbox channel never strands the caller until its own generic + // deadline. Retry a transient write failure, exactly like the 504 backstop + // write. + let lastWriteError = "Sandbox callback bridge could not write the handler response."; + for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) { + try { + await withTimeout( + writeBridgeResponse(input.client, requestPath, responsePath, response), + iterationTimeoutMs, + `Sandbox callback bridge write response for ${response.id}`, + ); + await input.client.remove(requestPath).catch(() => undefined); + return; + } catch (error) { + lastWriteError = error instanceof Error ? error.message : String(error); + console.warn( + `[paperclip] sandbox callback bridge failed to write response for ${response.id} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${lastWriteError}`, + ); + if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS)); + } + } + } + // Every handler-response write failed. Deliver a non-retryable 504 backstop + // inline, so the caller gets a terminal response instead of a retryable 503 + // that repeats a possibly-committed mutation, or a strand until its own + // deadline. Roll the `finalized` fence back so `writeAbortedHandlerBackstop` + // can proceed; it re-fences, retries the 504 write, and removes the request + // file. Await it, so the request file does not linger for the poll loop + // while the loop still runs. + guard.finalized = false; + await writeAbortedHandlerBackstop(fileName, guard, lastWriteError); + }; + try { + const raw = await input.client.readTextFile(requestPath); + let request: SandboxCallbackBridgeRequest; + try { + request = JSON.parse(raw) as SandboxCallbackBridgeRequest; + } catch { + const requestId = fileName.replace(/\.json$/i, "") || randomUUID(); + await finalize({ + id: requestId, + status: 400, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: "Invalid bridge request payload." }), + completedAt: new Date().toISOString(), + }); + return; + } + + const denialReason = await authorizeRequest(request); + if (denialReason) { + await finalize({ + id: request.id, + status: 403, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: denialReason }), + completedAt: new Date().toISOString(), + }); + return; + } + + // Claim the request for the handler before the host operation starts. When + // the recovery path already claimed it, it writes a 503 and the caller may + // retry, so do not run the mutation; the retry then applies it once. When + // the handler claims first, the recovery path leaves the request alone and + // the handler writes the real response. + if (!claimForHandler()) { + return; + } + + // Build the response, then finalize once. The handler already holds the + // claim, so `finalize` writes the real response. + let response: SandboxCallbackBridgeResponse; + try { + const result = await input.handleRequest(request, { signal: guard.controller.signal }); + const responseBody = result.body ?? ""; + if (Buffer.byteLength(responseBody, "utf8") > maxBodyBytes) { + throw new Error(`Bridge response body exceeded the configured size limit of ${maxBodyBytes} bytes.`); + } + response = { + id: request.id, + status: result.status, + headers: result.headers ?? {}, + body: responseBody, + completedAt: new Date().toISOString(), + }; + } catch (error) { + console.warn( + `[paperclip] sandbox callback bridge handler failed for ${request.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + // Tell a worker abort apart from a normal handler failure. The recovery + // path aborts `guard.controller` when the per-iteration timeout or the + // watchdog fires. The abort reaches this catch only after the handler + // claimed the request and started the host operation. The bridge cannot + // cancel a host operation that is in flight, so the mutation may have + // committed. A 502 (or 503) is a retryable status: the caller retries it + // and applies the mutation twice. So return a non-retryable 504 and mark + // the outcome indeterminate. The caller must not retry a 504 from the + // bridge, unlike the retry-safe 503 that the recovery path writes only + // before the host operation starts. + if (guard.controller.signal.aborted) { + response = { + id: request.id, + status: 504, + headers: { + "content-type": "application/json", + "x-paperclip-bridge-outcome": "indeterminate", + }, + body: JSON.stringify({ + error: error instanceof Error ? error.message : String(error), + outcome: "indeterminate", + retryable: false, + }), + completedAt: new Date().toISOString(), + }; + } else { + response = { + id: request.id, + status: 502, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + error: error instanceof Error ? error.message : String(error), + }), + completedAt: new Date().toISOString(), + }; + } + } + await finalize(response); } finally { - await input.client.remove(requestPath); + // Drop the guard only when it still points to this attempt. A retry can + // register a new attempt under the same file name; that new guard must + // stay in the map. Keep the guard when a backstop is still pending: a + // failed terminal write re-arms the backstop and keeps the request file, + // so the guard must stay in the map. The poll loop then skips the file and + // does not re-run a possibly-committed mutation before the backstop writes + // its 504. + if (guard.backstopTimer === undefined && inFlightRequestGuards.get(fileName) === guard) { + inFlightRequestGuards.delete(fileName); + } } }; - const failPendingRequests = async (message: string) => { - const fileNames = await input.client.listJsonFiles(directories.requestsDir).catch(() => []); + // Write the non-retryable 504 backstop for an aborted handler that did not + // finalize inside the grace. The `finalized` fence makes this a no-op when the + // handler already wrote its own response. The request file name is the request + // id plus `.json`, so derive the id from it without another client read that + // could hang on the same dead channel. + const writeAbortedHandlerBackstop = async ( + fileName: string, + guard: RequestFinalizeGuard, + message: string, + ) => { + guard.backstopTimer = undefined; + if (guard.finalized) { + return; + } + guard.finalized = true; + const requestPath = path.posix.join(directories.requestsDir, fileName); + const responsePath = path.posix.join(directories.responsesDir, fileName); + const requestId = fileName.replace(/\.json$/i, "") || randomUUID(); + for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) { + try { + await withTimeout( + writeBridgeResponse(input.client, requestPath, responsePath, { + id: requestId, + status: 504, + headers: { + "content-type": "application/json", + "x-paperclip-bridge-outcome": "indeterminate", + }, + body: JSON.stringify({ error: message, outcome: "indeterminate", retryable: false }), + completedAt: new Date().toISOString(), + }, { + requireRequestPath: false, + }), + iterationTimeoutMs, + `Sandbox callback bridge write 504 backstop for ${requestId}`, + ); + // The 504 backstop reached the caller. Remove the request file, so the + // poll loop does not list it again and re-run the mutation. + await input.client.remove(requestPath).catch(() => undefined); + return; + } catch (error) { + console.warn( + `[paperclip] sandbox callback bridge failed to write 504 backstop for ${requestId} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${error instanceof Error ? error.message : String(error)}`, + ); + if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS)); + } + } + } + // Every backstop write failed. Keep the request file and clear the fence, so a + // late handler can still finalize its own 504. The guard stays in the map, so + // the poll loop skips the file and does not re-run the mutation. A removed + // file plus a set fence would strand the caller until its own deadline and + // give it a generic 502 instead of the terminal 504. + // + // Re-arm the backstop directly. A stuck handler never settles, so its + // `processRequestFile` never runs the `finally` that drops the guard. The + // poll loop then skips the file on each iteration, so every iteration + // succeeds and the watchdog never trips again. No later recovery pass runs, + // so a re-arm here is the only path that retries the 504 write. Clear the + // fence before the re-arm, because `scheduleAbortedHandlerBackstop` bails on + // a set fence. The re-arm only re-writes the 504 response; it never re-runs + // the mutation, so a retry cannot apply the mutation twice. + guard.finalized = false; + scheduleAbortedHandlerBackstop(fileName, guard, message); + }; + + // Arm the backstop timer for a handler the recovery path just aborted. It is + // idempotent: a second recovery pass (the watchdog and the loop catch both run + // `failPendingRequests`) does not re-arm a live timer or one that already + // finalized. + const scheduleAbortedHandlerBackstop = ( + fileName: string, + guard: RequestFinalizeGuard, + message: string, + ) => { + if (guard.finalized || guard.backstopTimer !== undefined) { + return; + } + guard.backstopTimer = setTimeout(() => { + void writeAbortedHandlerBackstop(fileName, guard, message); + }, abortedHandlerGraceMs); + if (typeof guard.backstopTimer.unref === "function") { + guard.backstopTimer.unref(); + } + }; + + // Abort every queued request with a 503. The `abandonInFlight` option controls + // the completion fence for a request a `processRequestFile` attempt still owns. + // The timeout and watchdog recovery pass `true`: the loop already gave up on + // the request. When the handler did not yet start the host operation, claim the + // request so a later handler claim bails, then write the 503. When the handler + // already started, skip the 503; the recovery cannot cancel an in-flight host + // operation, and a 503 there would make the caller retry and apply the mutation + // twice. The stop drain passes `false` (the default): a request the loop + // already picked up keeps its normal completion, so a late handler result still + // wins over the drain 503, exactly like the earlier stop behavior. + const failPendingRequests = async ( + message: string, + options: { abandonInFlight?: boolean } = {}, + ) => { + if (options.abandonInFlight) { + // Abort every in-flight handler first, then arm its 504 backstop. The loop + // already gave up on the request. A handler that threads the signal into + // its work stops, rejects, and finalizes with its own error response inside + // the grace, so the request does not strand. A handler that ignores the + // signal and never settles does not finalize inside the grace; the backstop + // timer then writes a non-retryable 504, so the request still never + // strands. This reads the guard map directly, so it runs even when the + // request listing below fails on the same dead channel. It never writes a + // 503 for a handler-owned request: the recovery cannot cancel a committed + // host mutation, so a retryable status there could apply the mutation twice. + for (const [fileName, guard] of inFlightRequestGuards.entries()) { + if (guard.claim === "handler") { + guard.controller.abort(new Error(message)); + scheduleAbortedHandlerBackstop(fileName, guard, message); + } + } + } + // Wrap each client call in the per-iteration timeout. When the sandbox + // channel is unresponsive, a client call hangs with no reject. The timeout + // keeps this recovery path fail-fast, so it never re-hangs on the same dead + // channel that triggered the recovery. + const fileNames = await withTimeout( + input.client.listJsonFiles(directories.requestsDir), + iterationTimeoutMs, + "Sandbox callback bridge list pending requests", + ).catch(() => [] as string[]); for (const fileName of fileNames) { + const guard = inFlightRequestGuards.get(fileName); + if (guard && guard.claim === "handler" && (options.abandonInFlight || guard.controller.signal.aborted)) { + // The handler already started this request's host operation, or it already + // finalized the request. The timeout and watchdog cannot cancel a host + // operation that is in flight. A competing 503 here makes the caller retry + // while the original mutation still completes, so the mutation applies + // twice. Leave the request for the handler to finalize, or for the 504 + // backstop to finalize when the handler never settles. The recovery path + // reaches this skip through `abandonInFlight`. The stop drain reaches it + // only when the recovery path already aborted the handler, so a request + // whose 504 backstop write failed and kept its file never gets a competing + // 503 on stop. A normal in-flight handler at a graceful stop is not + // aborted, so it still gets the stop drain 503. + continue; + } + if (options.abandonInFlight && guard) { + // The handler did not start the host operation yet. Claim the request, so + // the handler bails at its host-operation claim instead of running the + // mutation. A retry after the 503 then applies the mutation once. + guard.claim = "abandon"; + } const requestPath = path.posix.join(directories.requestsDir, fileName); const responsePath = path.posix.join(directories.responsesDir, fileName); const requestId = fileName.replace(/\.json$/i, "") || randomUUID(); + let responseId = requestId; try { - const raw = await input.client.readTextFile(requestPath); - const parsed = JSON.parse(raw) as Partial; - await input.client.remove(requestPath).catch(() => undefined); - await writeBridgeResponse(input.client, requestPath, responsePath, { - id: typeof parsed.id === "string" && parsed.id.length > 0 ? parsed.id : requestId, - status: 503, - headers: { "content-type": "application/json" }, - body: JSON.stringify({ error: message }), - completedAt: new Date().toISOString(), - }, { - requireRequestPath: false, - }); - } catch (error) { - console.warn( - `[paperclip] sandbox callback bridge failed to abort pending request ${requestId}: ${error instanceof Error ? error.message : String(error)}`, + const raw = await withTimeout( + input.client.readTextFile(requestPath), + iterationTimeoutMs, + `Sandbox callback bridge read pending request ${requestId}`, ); - } finally { - await input.client.remove(requestPath).catch(() => undefined); + const parsed = JSON.parse(raw) as Partial; + if (typeof parsed.id === "string" && parsed.id.length > 0) { + responseId = parsed.id; + } + } catch (error) { + // The read or the parse failed, most likely on the same dead channel that + // triggered this recovery. Keep the request file, so a later recovery pass + // can still read it and deliver a terminal 503. A remove here drops the + // request and strands the caller until its own deadline. + console.warn( + `[paperclip] sandbox callback bridge could not read pending request ${requestId}: ${error instanceof Error ? error.message : String(error)}`, + ); + continue; } + // Write the 503 first, then remove the request file only after the write + // lands. Retry a transient failure, bounded by the per-iteration timeout, + // exactly like the finalize and 504 backstop writes. When every attempt + // fails, keep the request file. A later recovery pass, or the caller retry, + // then still finds the queued request and delivers a terminal 503, instead + // of a silent drop that strands the caller until its own deadline. The + // request is unclaimed or abandoned, so its host mutation never ran; a later + // 503 stays exactly-once. + let wrote503 = false; + let lastWriteError = "Sandbox callback bridge could not write the recovery 503."; + for (let attempt = 1; attempt <= MAX_BACKSTOP_WRITE_ATTEMPTS; attempt += 1) { + try { + await withTimeout( + writeBridgeResponse(input.client, requestPath, responsePath, { + id: responseId, + status: 503, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: message }), + completedAt: new Date().toISOString(), + }, { + requireRequestPath: false, + }), + iterationTimeoutMs, + `Sandbox callback bridge write 503 for ${requestId}`, + ); + wrote503 = true; + break; + } catch (error) { + lastWriteError = error instanceof Error ? error.message : String(error); + console.warn( + `[paperclip] sandbox callback bridge failed to write recovery 503 for ${requestId} (attempt ${attempt}/${MAX_BACKSTOP_WRITE_ATTEMPTS}): ${lastWriteError}`, + ); + if (attempt < MAX_BACKSTOP_WRITE_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, BACKSTOP_WRITE_RETRY_MS)); + } + } + } + if (wrote503) { + // The 503 landed. Remove the request file, so the poll loop does not + // re-process it. + await input.client.remove(requestPath).catch(() => undefined); + } else { + // Every 503 write failed. Keep the request file for a later recovery pass. + console.warn( + `[paperclip] sandbox callback bridge kept queued request ${requestId} after every recovery 503 write failed: ${lastWriteError}`, + ); + } + } + }; + + // Surface a bridge-worker failure through the run trace, not only stdout. A + // failed span under `input.runtimeSpan` records the error against the live run + // span, so the run and the orchestrator see the hang. When no `runtimeSpan` + // runner is wired (no injected tracer), the helper still writes a warn line, + // so the failure is never silent. + const surfaceRunError = async (error: Error) => { + if (input.runtimeSpan) { + try { + await input.runtimeSpan(CALLBACK_BRIDGE_WORKER_FAILED_SPAN, async () => { + throw error; + }); + } catch { + // `runtimeSpan` re-throws after it records the failed span. The error is + // now on the trace; swallow it here so the worker recovery continues. + } + } + console.warn(`[paperclip] ${error.message}`); + }; + + // The timestamp of the last successful loop iteration. The watchdog compares + // it to the current time. The idle branch and every processed request update + // it, so steady progress keeps the watchdog re-armed. + let lastSuccessfulIterationAt = Date.now(); + let watchdogTrippedAt: number | null = null; + let watchdogTripInFlight = false; + // Check often enough to fire soon after the threshold, but not so often that + // the check adds load. One fifth of the threshold, with a 10ms floor. + const watchdogCheckIntervalMs = Math.max(10, Math.floor(watchdogTimeoutMs / 5)); + + const handleWatchdogTrip = async (idleMs: number) => { + const message = `Sandbox callback bridge made no successful poll iteration for ${idleMs}ms; the sandbox connection is unresponsive.`; + await surfaceRunError(new Error(message)); + try { + await failPendingRequests(message, { abandonInFlight: true }); + } catch (error) { + console.warn( + `[paperclip] sandbox callback bridge watchdog failed to abort queued requests: ${error instanceof Error ? error.message : String(error)}`, + ); } }; @@ -794,10 +1303,37 @@ export async function startSandboxCallbackBridgeWorker(input: { // its `criticalPath` flag. `runWithoutActiveStep` empties the store for the // loop only; Node keeps the empty store on every later poll continuation. const loop = runWithoutActiveStep(() => (async () => { + // The watchdog runs on its own timer, so it fires even while the loop is + // stuck on an awaited client call. It is the backstop for a hang the + // per-iteration timeout does not catch. `unref` keeps it from holding the + // process open. The loop `finally` clears it on every exit. + const watchdogTimer = setInterval(() => { + if (settled || stopping) return; + const idleMs = Date.now() - lastSuccessfulIterationAt; + if (idleMs < watchdogTimeoutMs) return; + // Fire once per hang period. Re-arm only after the loop advances + // `lastSuccessfulIterationAt` past the last trip (a new successful + // iteration), so a persistent hang never fires the watchdog repeatedly. + if (watchdogTrippedAt !== null && watchdogTrippedAt >= lastSuccessfulIterationAt) return; + if (watchdogTripInFlight) return; + watchdogTrippedAt = Date.now(); + watchdogTripInFlight = true; + void handleWatchdogTrip(idleMs).finally(() => { + watchdogTripInFlight = false; + }); + }, watchdogCheckIntervalMs); + if (typeof watchdogTimer.unref === "function") { + watchdogTimer.unref(); + } try { while (true) { - const fileNames = await input.client.listJsonFiles(directories.requestsDir); + const fileNames = await withTimeout( + input.client.listJsonFiles(directories.requestsDir), + iterationTimeoutMs, + "Sandbox callback bridge list requests", + ); if (fileNames.length === 0) { + lastSuccessfulIterationAt = Date.now(); if (stopping) { break; } @@ -815,32 +1351,41 @@ export async function startSandboxCallbackBridgeWorker(input: { // live parent switches to `agent.turn` during the turn and back to // `task.run` after it. Without a runner, the request runs under the // run parent with no wrapper span, exactly like the earlier behavior. - await (input.runtimeSpan - ? input.runtimeSpan(CALLBACK_BRIDGE_RELAY_REQUEST_SPAN, () => - processRequestFile(fileName), - ) - : runWithRuntimeParent(input.getRuntimeParentContext?.(), () => - processRequestFile(fileName), - )); + // The per-iteration timeout wraps the whole request, so a hung + // request rejects and the loop `catch` runs `failPendingRequests`. + await withTimeout( + input.runtimeSpan + ? input.runtimeSpan(CALLBACK_BRIDGE_RELAY_REQUEST_SPAN, () => + processRequestFile(fileName), + ) + : runWithRuntimeParent(input.getRuntimeParentContext?.(), () => + processRequestFile(fileName), + ), + iterationTimeoutMs, + `Sandbox callback bridge process request ${fileName}`, + ); + lastSuccessfulIterationAt = Date.now(); } finally { inFlight -= 1; } } + lastSuccessfulIterationAt = Date.now(); if (stopping && Date.now() >= stopDeadline) { break; } } } catch (error) { const message = buildWorkerFailureMessage(error); - console.warn(`[paperclip] ${message}`); + await surfaceRunError(new Error(message)); try { - await failPendingRequests(message); + await failPendingRequests(message, { abandonInFlight: true }); } catch (failPendingError) { console.warn( `[paperclip] sandbox callback bridge failed to abort queued requests after worker failure: ${failPendingError instanceof Error ? failPendingError.message : String(failPendingError)}`, ); } } finally { + clearInterval(watchdogTimer); settled = true; if (settleResolve) { settleResolve(); @@ -1153,7 +1698,7 @@ export async function startSandboxCallbackBridgeServer(input: { }; } -function getSandboxCallbackBridgeServerSource(): string { +export function getSandboxCallbackBridgeServerSource(): string { return `import { randomUUID, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; import { promises as fs } from "node:fs"; @@ -1278,8 +1823,21 @@ const server = createServer(async (req, res) => { await fs.rename(tempPath, requestPath); const response = await waitForResponse(requestId); - res.statusCode = typeof response.status === "number" ? response.status : 200; - for (const [key, value] of Object.entries(response.headers || {})) { + const responseHeaders = response.headers || {}; + // The host marks a possibly-committed mutation with an indeterminate outcome. + // The host cannot cancel a host operation that is in flight, so the mutation + // may have committed before the worker aborted the handler. A 5xx status is + // retryable by convention, so a caller that retries 5xx would apply the + // mutation twice. Map the indeterminate outcome to a non-retryable 409, so a + // standard retry policy does not repeat the request. The outcome header and + // body stay, so a caller that reads them still sees the indeterminate result. + const bridgeOutcome = responseHeaders["x-paperclip-bridge-outcome"]; + if (bridgeOutcome === "indeterminate") { + res.statusCode = 409; + } else { + res.statusCode = typeof response.status === "number" ? response.status : 200; + } + for (const [key, value] of Object.entries(responseHeaders)) { if (typeof value !== "string" || key.toLowerCase() === "content-length") continue; res.setHeader(key, value); }