diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts index a3140e9b64..7d67f87336 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.test.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.test.ts @@ -1,5 +1,6 @@ -import { execFile as execFileCallback } from "node:child_process"; -import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; +import { execFile as execFileCallback, spawn } from "node:child_process"; +import { createServer } from "node:http"; +import { mkdir, mkdtemp, readFile, readdir, rm, utimes, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { promisify } from "node:util"; @@ -2706,4 +2707,237 @@ describe("sandbox callback bridge", () => { await worker.stop({ drainTimeoutMs: 50 }); }); + + async function prepareGatewayFixture(prefix: string) { + const rootDir = await mkdtemp(path.join(os.tmpdir(), prefix)); + cleanupDirs.push(rootDir); + const localWorkspaceDir = path.join(rootDir, "local-workspace"); + const remoteWorkspaceDir = path.join(rootDir, "remote-workspace"); + await mkdir(localWorkspaceDir, { recursive: true }); + await mkdir(remoteWorkspaceDir, { recursive: true }); + await writeFile(path.join(localWorkspaceDir, "README.md"), "bridge test\n", "utf8"); + + const runner = createExecRunner(); + const bridgeAsset = await createSandboxCallbackBridgeAsset(); + cleanupFns.push(bridgeAsset.cleanup); + const prepared = await prepareCommandManagedRuntime({ + runner, + spec: { remoteCwd: remoteWorkspaceDir, timeoutMs: 30_000 }, + adapterKey: "codex", + workspaceLocalDir: localWorkspaceDir, + assets: [{ key: "bridge", localDir: bridgeAsset.localDir }], + }); + const queueDir = path.posix.join(prepared.runtimeRootDir, "paperclip-bridge"); + return { + runner, + remoteWorkspaceDir, + assetRemoteDir: prepared.assetDirs.bridge, + queueDir, + directories: sandboxCallbackBridgeDirectories(queueDir), + bridgeToken: createSandboxCallbackBridgeToken(), + }; + } + + it("cleans up a timed-out request file and keeps serving after the host recovers", async () => { + const fixture = await prepareGatewayFixture("paperclip-bridge-timeout-clean-"); + + const bridge = await startSandboxCallbackBridgeServer({ + runner: fixture.runner, + remoteCwd: fixture.remoteWorkspaceDir, + assetRemoteDir: fixture.assetRemoteDir, + queueDir: fixture.queueDir, + bridgeToken: fixture.bridgeToken, + timeoutMs: 30_000, + responseTimeoutMs: 600, + pollIntervalMs: 50, + }); + cleanupFns.push(async () => { + await bridge.stop(); + }); + + // No worker runs, so the request times out at the gateway. + const timedOut = await fetch(`${bridge.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${fixture.bridgeToken}` }, + }); + expect(timedOut.status).toBe(502); + await expect(timedOut.json()).resolves.toMatchObject({ + error: expect.stringContaining("Timed out"), + }); + + // The gateway cleaned its own request file, so nothing counts toward the + // queue-depth cap after the caller gave up. + const leftover = (await readdir(fixture.directories.requestsDir).catch(() => [])).filter((name) => + name.endsWith(".json"), + ); + expect(leftover).toEqual([]); + + // A worker that comes up afterwards serves the next request normally — + // the timeout neither wedged nor killed the gateway. + const worker = await startSandboxCallbackBridgeWorker({ + client: createFileSystemSandboxCallbackBridgeQueueClient(), + queueDir: fixture.queueDir, + authorizeRequest: async () => null, + handleRequest: async () => ({ + status: 200, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ok: true }), + }), + }); + cleanupFns.push(async () => { + await worker.stop(); + }); + + const recovered = await fetch(`${bridge.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${fixture.bridgeToken}` }, + }); + expect(recovered.status).toBe(200); + await expect(recovered.json()).resolves.toMatchObject({ ok: true }); + }, 30_000); + + it("sweeps stale request files before rejecting at the queue-depth cap", async () => { + const fixture = await prepareGatewayFixture("paperclip-bridge-stale-sweep-"); + + const bridge = await startSandboxCallbackBridgeServer({ + runner: fixture.runner, + remoteCwd: fixture.remoteWorkspaceDir, + assetRemoteDir: fixture.assetRemoteDir, + queueDir: fixture.queueDir, + bridgeToken: fixture.bridgeToken, + timeoutMs: 30_000, + responseTimeoutMs: 500, + pollIntervalMs: 50, + maxQueueDepth: 1, + }); + cleanupFns.push(async () => { + await bridge.stop(); + }); + + // Plant an orphaned request file (a killed caller, or a previous gateway + // process's leftover) and backdate it beyond the response deadline. + const orphanPath = path.join(fixture.directories.requestsDir, "orphan.json"); + await writeFile(orphanPath, bridgeRequestJson("orphan"), "utf8"); + const staleTime = new Date(Date.now() - 60_000); + await utimes(orphanPath, staleTime, staleTime); + + // The queue sits at the cap, but the only entry is stale: the gateway must + // sweep it and admit the request instead of answering 503. With no worker + // the admitted request then times out (502) — proof it entered the queue. + const response = await fetch(`${bridge.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${fixture.bridgeToken}` }, + }); + expect(response.status).toBe(502); + const body = (await response.json()) as { error?: string }; + expect(body.error ?? "").not.toContain("queue is full"); + + const leftover = (await readdir(fixture.directories.requestsDir).catch(() => [])).filter((name) => + name.endsWith(".json"), + ); + expect(leftover).toEqual([]); + }, 30_000); + + it("skips a request file that vanished before the read instead of escalating", async () => { + // The gateway deletes a request file when its caller stops waiting. The + // worker's read then races the deletion; a vanished file must be a quiet + // skip, not a worker failure with a recovery pass. + const queueDir = "/virtual-bridge/vanished"; + let listCalls = 0; + const handled: string[] = []; + const responseStatuses: number[] = []; + const { runtimeSpan, workerErrors } = createWorkerErrorCapture(); + + const client: SandboxCallbackBridgeQueueClient = { + makeDir: async () => {}, + makeDirs: async () => {}, + listJsonFiles: async () => { + listCalls += 1; + return listCalls === 1 ? ["ghost.json"] : []; + }, + readTextFile: async () => { + throw new Error("cat: ghost.json: No such file or directory"); + }, + writeTextFile: async () => {}, + writeResponseFile: async (_remotePath, body) => { + responseStatuses.push((JSON.parse(body.trim()) as { status: number }).status); + return { wrote: true }; + }, + rename: async () => {}, + remove: async () => {}, + }; + + const worker = await startSandboxCallbackBridgeWorker({ + client, + queueDir, + runtimeSpan, + authorizeRequest: async () => null, + handleRequest: async (request) => { + handled.push(request.id); + return { status: 200, body: "ok" }; + }, + }); + + await new Promise((resolve) => setTimeout(resolve, 250)); + await worker.stop({ drainTimeoutMs: 50 }); + + expect(handled).toEqual([]); + expect(responseStatuses).toEqual([]); + expect(workerErrors).toEqual([]); + }); + + it("embeds crash handlers and queue hygiene in the generated gateway source", () => { + // A crashed gateway is a dead loopback port for the rest of the run, so + // the generated source must keep its crash handlers and its stale-queue + // sweep. The readiness gate matters too: survival applies only after the + // gateway is adoptable, so a startup fault still fails fast. This pins + // their presence; the behavior is proven above and below. + const source = getSandboxCallbackBridgeServerSource(); + expect(source).toContain('process.on("uncaughtException"'); + expect(source).toContain('process.on("unhandledRejection"'); + expect(source).toContain("gatewayReady"); + expect(source).toContain("sweepStaleRequests"); + }); + + it("exits fast when the gateway cannot bind its port instead of lingering un-ready", async () => { + // Before readiness, the crash handlers must not keep the process alive: a + // failed bind means the gateway can never serve, and surviving would only + // leave an un-ready zombie while the host waits out its readiness poll. + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-bind-fail-")); + cleanupDirs.push(rootDir); + const entrypoint = path.join(rootDir, "paperclip-bridge-server.mjs"); + await writeFile(entrypoint, getSandboxCallbackBridgeServerSource(), "utf8"); + const queueDir = path.join(rootDir, "queue"); + await mkdir(queueDir, { recursive: true }); + + const blocker = createServer(); + await new Promise((resolve) => { + blocker.listen(0, "127.0.0.1", resolve); + }); + cleanupFns.push( + () => + new Promise((resolve) => { + blocker.close(() => resolve()); + }), + ); + const blockedPort = (blocker.address() as { port: number }).port; + + const child = spawn(process.execPath, [entrypoint], { + env: { + ...process.env, + PAPERCLIP_BRIDGE_QUEUE_DIR: queueDir, + PAPERCLIP_BRIDGE_TOKEN: "test-token", + PAPERCLIP_BRIDGE_PORT: String(blockedPort), + }, + stdio: ["ignore", "ignore", "pipe"], + }); + let stderr = ""; + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + const exitCode = await new Promise((resolve) => { + child.on("close", resolve); + }); + + expect(exitCode).toBe(1); + expect(stderr).toContain("[paperclip-bridge] server error"); + expect(stderr).toContain("EADDRINUSE"); + }, 15_000); }); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index b198d6d9cb..42e5888e2b 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -983,7 +983,22 @@ export async function startSandboxCallbackBridgeWorker(input: { await writeAbortedHandlerBackstop(fileName, guard, lastWriteError); }; try { - const raw = await input.client.readTextFile(requestPath); + let raw: string; + try { + raw = await input.client.readTextFile(requestPath); + } catch (error) { + // The gateway deletes a request file when its caller stops waiting + // (client-side timeout cleanup). A read that fails because the file is + // gone is that benign race, not a channel fault: confirm the file + // vanished and skip quietly instead of escalating into a recovery + // pass. A file that is still listed rethrows, so a real read fault + // keeps its existing handling. + const remaining = await input.client.listJsonFiles(directories.requestsDir).catch(() => null); + if (remaining !== null && !remaining.includes(fileName)) { + return; + } + throw error; + } let request: SandboxCallbackBridgeRequest; try { request = JSON.parse(raw) as SandboxCallbackBridgeRequest; @@ -2118,6 +2133,32 @@ if (bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" && !queueDir) { throw new Error("PAPERCLIP_BRIDGE_QUEUE_DIR and PAPERCLIP_BRIDGE_TOKEN are required."); } +// A crashed gateway is a dead loopback port for the rest of the run: nothing +// inside the sandbox respawns this process, and every later agent API call +// then fails at the connection level. Once the gateway is ready, log an +// uncaught fault to stderr (the host redirects it into logs/bridge.log) and +// keep serving — the relay holds no state a fault can corrupt beyond the one +// request it interrupted. Before readiness the same fault means the gateway +// can never become usable (a failed bind, a failed readiness write), so exit +// instead: surviving there only leaves an un-ready zombie behind while the +// host waits out its readiness poll. +let gatewayReady = false; +process.on("uncaughtException", (error) => { + process.stderr.write( + "[paperclip-bridge] uncaught exception: " + (error && error.stack ? error.stack : String(error)) + "\\n", + ); + if (!gatewayReady) { + process.exit(1); + } +}); +process.on("unhandledRejection", (reason) => { + const detail = reason && typeof reason === "object" && "stack" in reason ? reason.stack : String(reason); + process.stderr.write("[paperclip-bridge] unhandled rejection: " + detail + "\\n"); + if (!gatewayReady) { + process.exit(1); + } +}); + // The embedded zero-dependency frame codec. The duplex gateway uses it; the file // gateway ignores it. ${DUPLEX_GATEWAY_CODEC_SOURCE} @@ -2177,6 +2218,24 @@ async function runFileGateway() { return entries.filter((entry) => entry.isFile() && entry.name.endsWith(".json")).length; } + // Delete request files older than the response deadline. Every live caller + // cleans its own request file when it times out, so a file this old is an + // orphan: its writer was killed mid-wait, or a previous gateway process died + // and left its queue behind. Orphans otherwise count toward the queue-depth + // cap forever and wedge the gateway at a permanent 503. + async function sweepStaleRequests() { + const staleBefore = Date.now() - responseTimeoutMs - 2000; + const entries = await fs.readdir(requestsDir, { withFileTypes: true }).catch(() => []); + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith(".json")) continue; + const filePath = path.posix.join(requestsDir, entry.name); + const stats = await fs.stat(filePath).catch(() => null); + if (stats && stats.mtimeMs < staleBefore) { + await fs.rm(filePath, { force: true }).catch(() => undefined); + } + } + } + async function waitForResponse(requestId) { const responsePath = path.posix.join(responsesDir, \`\${requestId}.json\`); const deadline = Date.now() + responseTimeoutMs; @@ -2201,8 +2260,13 @@ async function runFileGateway() { } if (await queueDepth() >= maxQueueDepth) { - writeJsonResponse(res, 503, { error: "Bridge request queue is full." }); - return; + // Reclaim orphaned request files before rejecting; only a queue that + // is genuinely full of live requests gets the 503. + await sweepStaleRequests(); + if (await queueDepth() >= maxQueueDepth) { + writeJsonResponse(res, 503, { error: "Bridge request queue is full." }); + return; + } } const url = new URL(req.url || "/", "http://127.0.0.1"); @@ -2227,7 +2291,19 @@ async function runFileGateway() { await fs.writeFile(tempPath, \`\${JSON.stringify(payload)}\\n\`, "utf8"); await fs.rename(tempPath, requestPath); - const response = await waitForResponse(requestId); + let response; + try { + response = await waitForResponse(requestId); + } catch (error) { + // The host never delivered a response inside the deadline. Remove this + // request's file so it cannot pile up toward the queue-depth cap. The + // host's normal response write is guarded on the request file, so the + // removal also tells the host that no caller waits anymore. Without + // this cleanup a stalled host wedges the gateway at the cap and every + // later request gets an immediate 503 until run end. + await fs.rm(requestPath, { force: true }).catch(() => undefined); + throw error; + } 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 @@ -2265,7 +2341,24 @@ async function runFileGateway() { await fs.mkdir(responsesDir, { recursive: true }); await fs.mkdir(logsDir, { recursive: true }); + // Newer Node runtimes do not reliably surface a failed bind through + // uncaughtException here: with nothing else keeping the event loop alive, + // the process can drain and exit 0 before the error event is delivered + // (observed on Node 24/25; Node 22 delivered it). Attach an explicit error + // listener and pin the loop with a keepalive until the bind settles, so a + // startup failure exits 1 with the fault on stderr on every runtime. + const bindKeepalive = setInterval(() => {}, 1000); + server.once("error", (error) => { + clearInterval(bindKeepalive); + process.stderr.write( + "[paperclip-bridge] server error: " + (error && error.stack ? error.stack : String(error)) + "\\n", + ); + if (!gatewayReady) { + process.exit(1); + } + }); server.listen(port, host, async () => { + clearInterval(bindKeepalive); const address = server.address(); if (!address || typeof address === "string") { throw new Error("Bridge server did not expose a TCP address."); @@ -2280,6 +2373,9 @@ async function runFileGateway() { const tempReadyFile = \`\${readyFile}.tmp\`; await fs.writeFile(tempReadyFile, JSON.stringify(ready), "utf8"); await fs.rename(tempReadyFile, readyFile); + // The readiness file is on disk, so the host will adopt this process. + // From here on an uncaught fault must not kill the listener. + gatewayReady = true; }); } @@ -2621,6 +2717,9 @@ function runDuplexGateway() { type: "ready", nonce: bridgeNonce, }); + // READY is on the wire, so the host will adopt this process. From here on + // an uncaught fault must not kill the listener. + gatewayReady = true; }); }