From 8f478242f1b19d997d193400c7adc32b0d91aa56 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Tue, 11 Aug 2026 16:05:41 -0700 Subject: [PATCH] fix(adapter-utils): close sandbox stdin file race with atomic write and fault-tolerant poller (#11235) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agent adapters use execution targets to exchange input and output with sandbox processes > - The sandbox input path can expose partial files, and the poller can delete or stop on invalid input > - These timing windows can lose agent input without a clear error > - This pull request makes host writes atomic and makes the poller retry invalid files before it drops them > - The benefit is reliable sandbox input delivery with visible failure after bounded retries ## Linked Issues or Issue Description Closes #10874 ## What Changed - Decode host input into a temporary file, then rename it onto the final JSON path. - Apply the same atomic write pattern to the filesystem client. - Parse each input file before deletion. - Retry parse failures and drop a file after the bounded retry limit with an error event. - Add regression tests for empty, partial, and permanently malformed input files. ## Verification - `pnpm exec vitest run packages/adapter-utils/src/execution-target-stdin-race.test.ts` passes 5/5 tests. - Related sandbox callback, execution target, and sandbox execution suites pass 71/71 tests. - TypeScript checks pass for the changed files. - The regression suite fails on the old code and passes on this change. ## Risks - Low risk. The change affects sandbox input file handling and adds bounded retry behavior. - A permanently malformed file now creates an error event after the retry limit. > This bug fix does not add a core feature, so a roadmap change is not needed. ## Model Used Codex, OpenAI GPT-5, current agent runtime, large context window, tool use and code review support. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../src/execution-target-stdin-race.test.ts | 354 ++++++++++++++++++ .../adapter-utils/src/execution-target.ts | 54 ++- .../src/sandbox-callback-bridge.ts | 19 +- 3 files changed, 420 insertions(+), 7 deletions(-) create mode 100644 packages/adapter-utils/src/execution-target-stdin-race.test.ts diff --git a/packages/adapter-utils/src/execution-target-stdin-race.test.ts b/packages/adapter-utils/src/execution-target-stdin-race.test.ts new file mode 100644 index 0000000000..6fcb6085b9 --- /dev/null +++ b/packages/adapter-utils/src/execution-target-stdin-race.test.ts @@ -0,0 +1,354 @@ +import { execFile as execFileCallback, spawn } from "node:child_process"; +import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { afterEach, describe, expect, it } from "vitest"; + +import { getProcessSessionRemoteSource } from "./execution-target.js"; +import { createCommandManagedSandboxCallbackBridgeQueueClient } from "./sandbox-callback-bridge.js"; +import type { RunProcessResult } from "./server-utils.js"; + +const execFile = promisify(execFileCallback); + +// Regression coverage for the stdin file race (parent PAP-4037): the host sends +// each ACP message as a file in the sandbox stdin directory, and a poller in +// the sandbox reads the file and writes the data to the child. Two defects lost +// a message. The host write was not atomic, so the poller could read an empty +// or partial `.json` file. The poller deleted the file before it validated the +// content, so an empty read was lost and a partial read stopped the loop. +describe("stdin file race (parent PAP-4037)", () => { + const cleanupDirs: string[] = []; + + afterEach(async () => { + while (cleanupDirs.length > 0) { + const dir = cleanupDirs.pop(); + if (dir) await rm(dir, { recursive: true, force: true }).catch(() => undefined); + } + }); + + // ---- Poller wrapper harness ------------------------------------------- + + type DeliveredFrame = { seq: number; type: string; stream?: string; data?: string; message?: string }; + + // Run the real emitted poller wrapper as a node process. The streamed variant + // writes one JSON frame per line to its stdout, so the test reads the frames + // directly. The child command is `cat`, so every byte the poller writes to + // the child stdin comes back as a `data` frame. + async function startPollerWrapper(options?: { maxRetries?: number }) { + const sessionDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-poll-")); + cleanupDirs.push(sessionDir); + const stdinDir = path.join(sessionDir, "stdin"); + await mkdir(stdinDir, { recursive: true }); + + const wrapperPath = path.join(sessionDir, "wrapper.mjs"); + await writeFile(wrapperPath, getProcessSessionRemoteSource({ outputToStdout: true }), "utf8"); + + const config = { command: "cat", args: [] as string[], cwd: sessionDir, env: {} }; + const commandPayload = Buffer.from(JSON.stringify(config), "utf8").toString("base64"); + + const env: Record = { + ...process.env, + PAPERCLIP_PROCESS_SESSION_DIR: sessionDir, + PAPERCLIP_PROCESS_SESSION_COMMAND_B64: commandPayload, + }; + if (options?.maxRetries != null) { + env.PAPERCLIP_PROCESS_SESSION_STDIN_MAX_RETRIES = String(options.maxRetries); + } + + const child = spawn(process.execPath, [wrapperPath], { + cwd: sessionDir, + env, + stdio: ["ignore", "pipe", "pipe"], + }); + + const frames: DeliveredFrame[] = []; + let stdoutBuffer = ""; + child.stdout.on("data", (chunk: Buffer) => { + stdoutBuffer += chunk.toString("utf8"); + const lines = stdoutBuffer.split("\n"); + stdoutBuffer = lines.pop() || ""; + for (const line of lines) { + if (!line.trim()) continue; + frames.push(JSON.parse(line) as DeliveredFrame); + } + }); + + const exited = new Promise((resolve) => child.on("close", () => resolve())); + + return { + sessionDir, + stdinDir, + frames, + // Write a complete stdin file with an atomic rename, so the test never + // creates its own partial-write race. + writeFileAtomic: async (name: string, content: string) => { + const finalPath = path.join(stdinDir, name); + const tempPath = `${finalPath}.writing`; + await writeFile(tempPath, content, "utf8"); + await rename(tempPath, finalPath); + }, + // Write a `.json` file directly, so a reader can observe it before the + // content lands. This simulates the non-atomic-write window. + writeFileRaw: async (name: string, content: string) => { + await writeFile(path.join(stdinDir, name), content, "utf8"); + }, + exited, + kill: () => child.kill("SIGKILL"), + }; + } + + function stdinMessage(text: string): string { + return `${JSON.stringify({ type: "stdin", data: Buffer.from(text, "utf8").toString("base64") })}\n`; + } + + const stdinEndMessage = `${JSON.stringify({ type: "stdinEnd" })}\n`; + + // Concatenate every stdout `data` frame and decode it back to text. + function collectDelivered(frames: DeliveredFrame[]): string { + return frames + .filter((frame) => frame.type === "data" && frame.stream === "stdout" && typeof frame.data === "string") + .map((frame) => Buffer.from(frame.data as string, "base64").toString("utf8")) + .join(""); + } + + function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + async function waitFor(check: () => boolean, timeoutMs = 4_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (check()) return; + await delay(20); + } + throw new Error("Timed out waiting for condition."); + } + + // ---- Poller tests ----------------------------------------------------- + + it("delivers a stdin file that appears empty first and then gets content", async () => { + const poller = await startPollerWrapper(); + + // The file appears empty first (the non-atomic-write window). The poller + // must keep it and retry, not delete it and lose the message. + await poller.writeFileRaw("000000000001.json", ""); + await delay(200); + // The poller keeps the empty file for a later retry. A poller that deletes + // before it validates would drop the file here and lose the message. + const afterEmpty = await readdir(poller.stdinDir); + expect(afterEmpty).toContain("000000000001.json"); + + // The content lands in the same file. The poller must deliver it on a later + // cycle, because it kept the file across the empty read. + await poller.writeFileAtomic("000000000001.json", stdinMessage("late-payload")); + + await waitFor(() => collectDelivered(poller.frames).includes("late-payload")); + + await poller.writeFileAtomic("000000000002.json", stdinEndMessage); + await poller.exited; + + expect(collectDelivered(poller.frames)).toBe("late-payload"); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + it("keeps polling after a malformed file and still delivers a later valid file", async () => { + const poller = await startPollerWrapper({ maxRetries: 3 }); + + // A malformed file sorts before the valid file. The poller keeps the send + // order: it holds the later file until it drops the malformed file after the + // retry limit, then it delivers the later valid file. So one bad file blocks + // the loop only until the retry limit, not forever. + await poller.writeFileRaw("000000000001.json", "{ this is not valid json"); + await poller.writeFileAtomic("000000000002.json", stdinMessage("valid-after-bad")); + + await waitFor(() => collectDelivered(poller.frames).includes("valid-after-bad")); + + await poller.writeFileAtomic("000000000003.json", stdinEndMessage); + await poller.exited; + + expect(collectDelivered(poller.frames)).toBe("valid-after-bad"); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + it("does not close the stream on a later stdinEnd while an earlier file awaits retry", async () => { + const poller = await startPollerWrapper(); + + // An earlier stdin file is momentarily unreadable (the non-atomic-write + // window). A later stdinEnd file is already complete. The poller must keep + // the send order: it must not read the stdinEnd ahead of the earlier file + // and close the stream. It must hold the stream open until the earlier file + // is readable. + await poller.writeFileRaw("000000000001.json", ""); + await poller.writeFileAtomic("000000000002.json", stdinEndMessage); + + // Give the poller time to scan. The stream stays open, so the child does not + // exit and no exit frame appears yet. + await delay(300); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(false); + + // The earlier file's content lands. The poller delivers it, then reads the + // stdinEnd and closes the stream. + await poller.writeFileAtomic("000000000001.json", stdinMessage("early-payload")); + + await waitFor(() => collectDelivered(poller.frames).includes("early-payload")); + await poller.exited; + + expect(collectDelivered(poller.frames)).toBe("early-payload"); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + it("drops a permanently malformed file after the retry limit and writes an error event", async () => { + const poller = await startPollerWrapper({ maxRetries: 3 }); + + // This file never becomes valid. After the retry limit the poller drops it + // and writes an error event, so the lost message fails loudly. + await poller.writeFileRaw("000000000001.json", "{ permanently broken"); + + await waitFor(() => + poller.frames.some( + (frame) => frame.type === "error" && typeof frame.message === "string" && frame.message.includes("Dropped unreadable stdin file"), + ), + ); + + // The loop still works after the drop: a later valid file is delivered. + await poller.writeFileAtomic("000000000002.json", stdinMessage("still-alive")); + await waitFor(() => collectDelivered(poller.frames).includes("still-alive")); + + await poller.writeFileAtomic("000000000003.json", stdinEndMessage); + await poller.exited; + + expect(collectDelivered(poller.frames)).toBe("still-alive"); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + // ---- Host atomic-write tests ------------------------------------------ + + // A runner that executes each bridge shell script on the local filesystem, + // so the test exercises the real command-managed `writeTextFile` script. + function createLocalShellRunner(scripts: string[]) { + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + }): Promise => { + const args = input.args ?? []; + if ((input.command === "sh" || input.command === "bash") && args[0] === "-c" && typeof args[1] === "string") { + scripts.push(args[1]); + } + const command = input.command === "sh" ? "/bin/sh" : input.command === "bash" ? "/bin/bash" : input.command; + try { + const result = await execFile(command, args, { + cwd: input.cwd, + env: { ...process.env, ...input.env }, + maxBuffer: 32 * 1024 * 1024, + }); + return { + exitCode: 0, + signal: null, + timedOut: false, + stdout: result.stdout, + stderr: result.stderr, + pid: null, + startedAt: null, + }; + } catch (error) { + const err = error as NodeJS.ErrnoException & { stdout?: string; stderr?: string; code?: string | number | null }; + return { + exitCode: typeof err.code === "number" ? err.code : null, + signal: null, + timedOut: false, + stdout: err.stdout ?? "", + stderr: err.stderr ?? "", + pid: null, + startedAt: null, + }; + } + }, + }; + } + + it("finalizes the command-managed host write with an atomic rename onto the .json path", async () => { + const remoteRoot = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-host-cmd-")); + cleanupDirs.push(remoteRoot); + const stdinDir = path.join(remoteRoot, "stdin"); + await mkdir(stdinDir, { recursive: true }); + + const scripts: string[] = []; + const client = createCommandManagedSandboxCallbackBridgeQueueClient({ + runner: createLocalShellRunner(scripts), + remoteCwd: remoteRoot, + timeoutMs: 30_000, + }); + + const jsonPath = path.join(stdinDir, "000000000001.json"); + const body = `${JSON.stringify({ type: "stdin", data: Buffer.from("host-payload", "utf8").toString("base64") })}\n`; + await client.writeTextFile(jsonPath, body); + + // The final file holds the complete body. + expect(await readFile(jsonPath, "utf8")).toBe(body); + // No temporary upload file remains next to the final file. + const entries = await readdir(stdinDir); + expect(entries).toEqual(["000000000001.json"]); + + // The finalize script renames a non-`.json` temporary file onto the final + // path. It never redirects the decode output straight into the `.json` + // file, so a reader never sees an empty or partial `.json` file. + const finalizeScript = scripts.find((script) => script.includes("base64 -d")); + expect(finalizeScript).toBeDefined(); + expect(finalizeScript).toContain(`mv `); + expect(finalizeScript).not.toContain(`> '${jsonPath}'`); + expect(finalizeScript).toContain(`> '${jsonPath}.paperclip-upload.decoded'`); + }); + + it("never exposes a partial .json file under a concurrent reader (command-managed host write)", async () => { + const remoteRoot = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-host-race-")); + cleanupDirs.push(remoteRoot); + const stdinDir = path.join(remoteRoot, "stdin"); + await mkdir(stdinDir, { recursive: true }); + + const client = createCommandManagedSandboxCallbackBridgeQueueClient({ + runner: createLocalShellRunner([]), + remoteCwd: remoteRoot, + timeoutMs: 30_000, + }); + + const jsonPath = path.join(stdinDir, "000000000001.json"); + // A large body needs many decode bytes, so the write window is wide. + const bigText = "x".repeat(64 * 1024); + const body = `${JSON.stringify({ type: "stdin", data: Buffer.from(bigText, "utf8").toString("base64") })}\n`; + + let stop = false; + const readerErrors: string[] = []; + let observedComplete = 0; + const reader = (async () => { + while (!stop) { + const raw = await readFile(jsonPath, "utf8").catch(() => null); + if (raw) { + try { + JSON.parse(raw); + observedComplete += 1; + } catch (error) { + readerErrors.push(error instanceof Error ? error.message : String(error)); + } + } + } + })(); + + for (let round = 0; round < 6; round += 1) { + await client.remove(jsonPath); + await client.writeTextFile(jsonPath, body); + } + stop = true; + await reader; + + // Every read of the final file parsed as complete JSON. The reader never + // saw an empty or partial `.json` file. + expect(readerErrors).toEqual([]); + expect(observedComplete).toBeGreaterThan(0); + }); +}); diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index a525b2f290..614131cd0c 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1843,7 +1843,7 @@ socket.on("close", () => { `; } -function getProcessSessionRemoteSource(input?: { outputToStdout?: boolean }): string { +export function getProcessSessionRemoteSource(input?: { outputToStdout?: boolean }): string { return input?.outputToStdout === true ? getProcessSessionRemoteStreamSource() : getProcessSessionRemoteEventFileSource(); @@ -1855,15 +1855,61 @@ function getProcessSessionRemoteSource(input?: { outputToStdout?: boolean }): st // event, so the wrapper installs a no-op handler at the call site. const PROCESS_SESSION_STDIN_POLL_TAIL = `child.stdin.on("error", () => {}); +// A stdin file can appear before the host finishes the write. An empty read is +// the non-atomic-write window; a partial read makes JSON.parse throw. The +// poller must not delete a file before it validates the content. So read and +// parse each file first, and delete it only after a successful parse. The files +// sort in send order. If an earlier file is not readable yet, stop the cycle and +// keep the order: a later file (for example stdinEnd) must not run ahead of it. +// Retry the earlier file on a later cycle. After the retry limit, drop the file +// and write an error event, so a lost message fails loud, and let later files +// run. +const stdinMaxParseRetries = (() => { + const raw = Number.parseInt(process.env.PAPERCLIP_PROCESS_SESSION_STDIN_MAX_RETRIES || "", 10); + return Number.isFinite(raw) && raw > 0 ? raw : 100; +})(); +const stdinParseRetries = new Map(); + async function pollStdin() { while (!stdinClosed) { const entries = (await fs.readdir(stdinDir).catch(() => [])).filter((name) => name.endsWith(".json")).sort(); for (const name of entries) { + if (stdinClosed) break; const file = path.posix.join(stdinDir, name); - const raw = await fs.readFile(file, "utf8").catch(() => null); + let message; + try { + const raw = await fs.readFile(file, "utf8"); + // An empty read means the content is not on disk yet. Treat it the same + // as a parse failure: keep the file and retry on a later cycle. + if (!raw) throw new Error("stdin file is empty"); + message = JSON.parse(raw); + } catch (error) { + const retries = (stdinParseRetries.get(name) || 0) + 1; + if (retries >= stdinMaxParseRetries) { + // The retry limit is reached. Drop the file and write an error event, + // so the lost message fails loud. The file is resolved now, so let the + // loop go on to the next entry. + stdinParseRetries.delete(name); + await fs.rm(file, { force: true }).catch(() => undefined); + await writeEvent({ + type: "error", + message: + "Dropped unreadable stdin file after " + stdinMaxParseRetries + " retries: " + name + ": " + + (error instanceof Error ? error.message : String(error)), + }); + continue; + } + // The file is not readable yet and is not past the retry limit. Keep it + // and stop this cycle to hold the send order. A later file (for example + // stdinEnd) must not run before this earlier file. A later cycle reads + // from the start again. + stdinParseRetries.set(name, retries); + break; + } + // The parse succeeded, so the content is complete. Delete the file first, + // then act on the message. A later cycle never re-reads a handled file. + stdinParseRetries.delete(name); await fs.rm(file, { force: true }).catch(() => undefined); - if (!raw) continue; - const message = JSON.parse(raw); if (message.type === "stdin" && typeof message.data === "string") { if (!stdinClosed) child.stdin.write(Buffer.from(message.data, "base64")); } else if (message.type === "stdinEnd") { diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index 35ee47db70..9a373b869d 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -396,7 +396,13 @@ export function createFileSystemSandboxCallbackBridgeQueueClient(): SandboxCallb readTextFile: async (remotePath) => await fs.readFile(remotePath, "utf8"), writeTextFile: async (remotePath, body) => { await fs.mkdir(path.posix.dirname(remotePath), { recursive: true }); - await fs.writeFile(remotePath, body, "utf8"); + // Write to a temporary path that does NOT end in `.json`, then rename it + // onto the final `.json` path. A direct `writeFile` truncates the final + // path first, so a `.json`-only reader (the stdin poller) can see an + // empty or partial file. The atomic rename never exposes partial content. + const tempPath = `${remotePath}.paperclip-upload.decoded`; + await fs.writeFile(tempPath, body, "utf8"); + await fs.rename(tempPath, remotePath); }, writeResponseFile: async (responsePath, body, options = {}) => { const responseDir = path.posix.dirname(responsePath); @@ -532,10 +538,17 @@ export function createCommandManagedSandboxCallbackBridgeQueueClient(input: { }, writeTextFile: async (remotePath, body) => { const remoteDir = path.posix.dirname(remotePath); + // Two temporary paths that do NOT end in `.json`, so a `.json`-only + // reader (the stdin poller) never lists them. The base64 upload lands in + // `tempPath`. The decode result lands in `decodedPath`. An atomic rename + // then moves the complete decoded content onto the final `.json` path. + // A direct `> remotePath` redirect truncates the final path before the + // decode writes it, so a reader can see an empty or partial file. const tempPath = `${remotePath}.paperclip-upload.b64`; + const decodedPath = `${remotePath}.paperclip-upload.decoded`; await runChecked( `prepare upload ${remotePath}`, - `mkdir -p ${shellQuote(remoteDir)} && rm -f ${shellQuote(tempPath)} && : > ${shellQuote(tempPath)}`, + `mkdir -p ${shellQuote(remoteDir)} && rm -f ${shellQuote(tempPath)} ${shellQuote(decodedPath)} && : > ${shellQuote(tempPath)}`, ); const base64Body = toBuffer(Buffer.from(body, "utf8")).toString("base64"); for (const chunk of base64Chunks(base64Body)) { @@ -546,7 +559,7 @@ export function createCommandManagedSandboxCallbackBridgeQueueClient(input: { } await runChecked( `finalize upload ${remotePath}`, - `base64 -d < ${shellQuote(tempPath)} > ${shellQuote(remotePath)} && rm -f ${shellQuote(tempPath)}`, + `base64 -d < ${shellQuote(tempPath)} > ${shellQuote(decodedPath)} && mv ${shellQuote(decodedPath)} ${shellQuote(remotePath)} && rm -f ${shellQuote(tempPath)}`, ); }, writeResponseFile: async (responsePath, body, options = {}) => {