From ed8075b535da6a39ef0abdd602a61b2a6d767718 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 14 Aug 2026 18:45:31 -0700 Subject: [PATCH] fix(adapter-utils): order stdin file writes in the sandbox process-session bridge (#11406) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agent adapters can run through a sandbox process-session bridge > - The bridge writes streamed standard input to files before the remote process reads them > - Concurrent file writes can make a later chunk visible before an earlier chunk > - The remote process can then parse a tail fragment and wait forever for the missing head > - This pull request serializes host writes and makes an unexpected file gap a loud error > - The benefit is ordered input with a bounded failure path for sandbox ACP sessions ## Linked Issues or Issue Description No public GitHub issue exists for this change. The description below follows `.github/ISSUE_TEMPLATE/bug_report.yml`. **What happened?** A sandbox ACP process-session bridge could stall after its handshake when a run sent a large prompt. The host sent one un-awaited file write for each standard input chunk. A small later chunk could finish before a large earlier chunk. The remote poller then sent the tail bytes first. The agent parser raised an error on the tail fragment, and the head bytes stayed buffered without a newline. **Expected behavior** The bridge must expose standard input files in sequence. The remote poller must report a clear error when an earlier file remains missing beyond the retry budget. **Steps to reproduce** 1. Start an ACP session through a sandbox process-session bridge. 2. Send a prompt that produces multiple standard input file chunks. 3. Delay finalization of an earlier chunk while a later chunk completes. 4. Observe that the remote parser can receive the later chunk first and the session can stop without a clear error. **Paperclip version or commit** The change targets the current `master` branch at the submitted commit. **Deployment mode** The bug affects sandbox execution. **Agent adapter(s) involved** The failure affects the ACP process-session bridge. **Database mode** Not database-related. **Additional context** The fix keeps the existing per-file atomic write behavior. It adds ordering at the host write boundary and a bounded ordering check in the shared wrapper poll tail. ## What Changed - Add a per-session promise chain for host standard input file writes. - Keep a failed write from blocking later chain entries. - Track the next expected sequence number in the shared wrapper poll tail. - Hold later files while an earlier file is missing within the existing retry budget. - Emit a loud error and advance after the retry budget expires. - Add regression tests for host ordering, gap holding, and the loud error path. ## Verification - `npx vitest run packages/adapter-utils/src/execution-target-stdin-race.test.ts` — 9 tests passed. - `npx vitest run packages/adapter-utils/src/execution-target-sandbox.test.ts` — 43 tests passed. - `pnpm --filter @paperclipai/adapter-utils typecheck` — clean. - With the source fix reverted, the 3 new tests fail and the 6 original tests pass. - CI must pass on the pull request before merge. ## Risks Low risk. - The host now serializes writes for each session, which can reduce write parallelism. - A failed write still emits one error and destroys the socket, as before. - The wrapper can emit a loud error after the existing retry budget when a file gap persists. - The change does not alter the atomic per-file write behavior. ## Model Used OpenAI GPT-5, model ID `gpt-5`, with tool use and code execution. The context window and internal reasoning details are not disclosed. ## 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 | 273 +++++++++++++++++- .../adapter-utils/src/execution-target.ts | 81 +++++- 2 files changed, 345 insertions(+), 9 deletions(-) diff --git a/packages/adapter-utils/src/execution-target-stdin-race.test.ts b/packages/adapter-utils/src/execution-target-stdin-race.test.ts index 6fcb6085b9..26fb6a0ff9 100644 --- a/packages/adapter-utils/src/execution-target-stdin-race.test.ts +++ b/packages/adapter-utils/src/execution-target-stdin-race.test.ts @@ -1,13 +1,18 @@ import { execFile as execFileCallback, spawn } from "node:child_process"; import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; +import net from "node:net"; 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 { + getProcessSessionRemoteSource, + startAdapterExecutionTargetProcessSessionBridge, + type AdapterSandboxExecutionTarget, +} from "./execution-target.js"; import { createCommandManagedSandboxCallbackBridgeQueueClient } from "./sandbox-callback-bridge.js"; -import type { RunProcessResult } from "./server-utils.js"; +import { runChildProcess, type RunProcessResult } from "./server-utils.js"; const execFile = promisify(execFileCallback); @@ -222,6 +227,270 @@ describe("stdin file race (parent PAP-4037)", () => { expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); }); + it("holds a later stdin file until the missing earlier file appears", async () => { + const poller = await startPollerWrapper(); + + // File 2 is complete, but file 1 has not appeared yet (a host reordering). + // The poller must not deliver file 2 ahead of the missing file 1. It holds + // the send order and waits for the earlier file. + await poller.writeFileAtomic("000000000002.json", stdinMessage("second-payload")); + await delay(300); + // File 2 is still on disk and nothing was delivered: the poller holds it. + const afterHold = await readdir(poller.stdinDir); + expect(afterHold).toContain("000000000002.json"); + expect(collectDelivered(poller.frames)).toBe(""); + + // File 1 arrives. The poller now delivers file 1 then file 2, in send order. + await poller.writeFileAtomic("000000000001.json", stdinMessage("first-payload")); + await waitFor(() => collectDelivered(poller.frames).includes("second-payload")); + expect(collectDelivered(poller.frames)).toBe("first-payloadsecond-payload"); + + await poller.writeFileAtomic("000000000003.json", stdinEndMessage); + await poller.exited; + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + it("fails loud and advances past a missing stdin file after the retry limit", async () => { + const poller = await startPollerWrapper({ maxRetries: 3 }); + + // File 1 never appears. File 2 is complete. After the retry limit the poller + // writes a loud error event and advances past the gap, then delivers file 2. + // So a permanent reordering fails loud, never silently. + await poller.writeFileAtomic("000000000002.json", stdinMessage("after-gap")); + + await waitFor(() => + poller.frames.some( + (frame) => + frame.type === "error" && + typeof frame.message === "string" && + frame.message.includes("Advanced past missing stdin files"), + ), + ); + await waitFor(() => collectDelivered(poller.frames).includes("after-gap")); + + await poller.writeFileAtomic("000000000003.json", stdinEndMessage); + await poller.exited; + expect(collectDelivered(poller.frames)).toBe("after-gap"); + expect(poller.frames.some((frame) => frame.type === "exit")).toBe(true); + }); + + // ---- Host serialization test (drives the real bridge) ----------------- + + // A runner that runs each bridge shell script as a real child process, so the + // test drives the whole legacy-poll bridge: the socket handler, the command- + // managed `writeTextFile` script, the nohup wrapper, and the output poll. + function createLocalSandboxRunner( + onExecute?: (script: string) => Promise, + ) { + let counter = 0; + return { + execute: async (input: { + command: string; + args?: string[]; + cwd?: string; + env?: Record; + stdin?: string; + timeoutMs?: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + }): Promise => { + counter += 1; + const script = input.args?.[1] ?? ""; + if (onExecute) await onExecute(script); + const command = + input.command === "bash" ? "/bin/bash" : input.command === "sh" ? "/bin/sh" : input.command; + return runChildProcess(`stdin-order-run-${counter}`, command, input.args ?? [], { + cwd: input.cwd ?? process.cwd(), + env: input.env ?? {}, + stdin: input.stdin, + timeoutSec: Math.max(1, Math.ceil((input.timeoutMs ?? 30_000) / 1000)), + graceSec: 5, + onLog: input.onLog ?? (async () => {}), + }); + }, + }; + } + + it("serializes host stdin writes so a slow earlier write still lands first", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-host-order-")); + cleanupDirs.push(rootDir); + // The child echoes every stdin byte to stdout, so the wrapper reports the + // exact bytes and order the child received on its stdin. + const childPath = path.join(rootDir, "echo-child.mjs"); + await writeFile(childPath, "process.stdin.on('data', (c) => process.stdout.write(c));\n", "utf8"); + + // Record the send-order-relevant event: the completion of each stdin file's + // finalize (atomic rename). Delay the finalize of the FIRST file, so its + // write resolves slower than the second. Without serialization the second + // rename would land first; the per-session chain must keep the send order. + const finalizeOrder: string[] = []; + const runner = createLocalSandboxRunner(async (script) => { + const finalizeMatch = /base64 -d[\s\S]*mv '[^']*\.decoded' '([^']+\.json)'/.exec(script); + if (finalizeMatch) { + const remotePath = finalizeMatch[1]; + if (remotePath.endsWith("000000000001.json")) await delay(300); + finalizeOrder.push(path.posix.basename(remotePath)); + } + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner, + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stdin-host-order", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + }); + expect(bridge).not.toBeNull(); + + let peer: net.Socket | null = null; + try { + const proxySource = await readFile(bridge!.agentCommand, "utf8"); + const port = Number(/port: (\d+)/.exec(proxySource)?.[1] ?? Number.NaN); + const tokenLiteral = /const token = (".*?");/.exec(proxySource)?.[1]; + expect(Number.isFinite(port)).toBe(true); + const token = JSON.parse(tokenLiteral as string) as string; + + const peerSocket = net.createConnection({ host: "127.0.0.1", port }); + peer = peerSocket; + peerSocket.setEncoding("utf8"); + peerSocket.on("error", () => undefined); + const delivered: string[] = []; + let peerBuffer = ""; + peerSocket.on("data", (chunk: string) => { + peerBuffer += chunk; + const lines = peerBuffer.split("\n"); + peerBuffer = lines.pop() || ""; + for (const line of lines) { + if (!line.trim()) continue; + const frame = JSON.parse(line) as { type?: string; stream?: string; data?: string }; + if (frame.type === "data" && frame.stream === "stdout" && typeof frame.data === "string") { + delivered.push(Buffer.from(frame.data, "base64").toString("utf8")); + } + } + }); + await new Promise((resolve, reject) => { + peerSocket.once("connect", () => resolve()); + peerSocket.once("error", reject); + }); + + // Send two stdin messages back to back. The first authenticates and writes + // file 1; the second writes file 2. Both are scheduled before file 1's + // delayed finalize resolves, so an un-chained handler would race them. + const head = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("HEAD_ONE_", "utf8").toString("base64") })}\n`; + const tail = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("TAIL_TWO", "utf8").toString("base64") })}\n`; + peerSocket.write(head); + peerSocket.write(tail); + + // The two finalize renames complete in send order, not in the order the + // delayed and fast writes would otherwise finish. + await waitFor(() => finalizeOrder.length >= 2, 8_000); + expect(finalizeOrder.slice(0, 2)).toEqual(["000000000001.json", "000000000002.json"]); + + // End to end: the child receives the two payloads intact and in send + // order, so the prompt is byte-identical on the child stdin. + await waitFor(() => delivered.join("").includes("TAIL_TWO"), 8_000); + expect(delivered.join("")).toBe("HEAD_ONE_TAIL_TWO"); + } finally { + peer?.destroy(); + await bridge?.stop(); + } + }); + + it("holds stdinEnd on stop until an earlier pending stdin write lands first", async () => { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-stdin-stop-order-")); + cleanupDirs.push(rootDir); + const childPath = path.join(rootDir, "echo-child.mjs"); + await writeFile(childPath, "process.stdin.on('data', (c) => process.stdout.write(c));\n", "utf8"); + + // Record each stdin file finalize (atomic rename). `finalizeStarted` marks + // the start; `finalizeOrder` marks the completion. Delay the FIRST chunk's + // finalize, so its write is still pending when `stop()` runs. `stop()` must + // chain the `stdinEnd` write after the pending chunk, so file 2 (stdinEnd) + // never finishes its rename before file 1. + const finalizeStarted: string[] = []; + const finalizeOrder: string[] = []; + const runner = createLocalSandboxRunner(async (script) => { + const finalizeMatch = /base64 -d[\s\S]*mv '[^']*\.decoded' '([^']+\.json)'/.exec(script); + if (finalizeMatch) { + const name = path.posix.basename(finalizeMatch[1]); + finalizeStarted.push(name); + if (name === "000000000001.json") await delay(300); + finalizeOrder.push(name); + } + }); + const target: AdapterSandboxExecutionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "local-test", + remoteCwd: rootDir, + timeoutMs: 30_000, + runner, + }; + + const bridge = await startAdapterExecutionTargetProcessSessionBridge({ + runId: "run-stdin-stop-order", + target, + runtimeRootDir: path.posix.join(rootDir, ".paperclip-runtime", "acpx"), + adapterKey: "acpx", + command: process.execPath, + args: [childPath], + cwd: rootDir, + env: {}, + timeoutSec: 5, + onLog: async () => {}, + }); + expect(bridge).not.toBeNull(); + + let peer: net.Socket | null = null; + let stopped = false; + try { + const proxySource = await readFile(bridge!.agentCommand, "utf8"); + const port = Number(/port: (\d+)/.exec(proxySource)?.[1] ?? Number.NaN); + const tokenLiteral = /const token = (".*?");/.exec(proxySource)?.[1]; + expect(Number.isFinite(port)).toBe(true); + const token = JSON.parse(tokenLiteral as string) as string; + + const peerSocket = net.createConnection({ host: "127.0.0.1", port }); + peer = peerSocket; + peerSocket.setEncoding("utf8"); + peerSocket.on("error", () => undefined); + peerSocket.on("data", () => undefined); + await new Promise((resolve, reject) => { + peerSocket.once("connect", () => resolve()); + peerSocket.once("error", reject); + }); + + // Send one stdin message. It authenticates and writes file 1, whose + // finalize the runner delays. Wait until that finalize has started, so the + // write is in flight when `stop()` runs. + const head = `${JSON.stringify({ token, type: "stdin", data: Buffer.from("HEAD_ONE_", "utf8").toString("base64") })}\n`; + peerSocket.write(head); + await waitFor(() => finalizeStarted.includes("000000000001.json"), 8_000); + + // Stop the bridge while file 1's write is still pending. `stop()` awaits + // the chained `stdinEnd` write, so both finalizes are complete when it + // returns, in send order. + await bridge!.stop(); + stopped = true; + expect(finalizeOrder).toEqual(["000000000001.json", "000000000002.json"]); + } finally { + peer?.destroy(); + if (!stopped) await bridge?.stop(); + } + }); + // ---- Host atomic-write tests ------------------------------------------ // A runner that executes each bridge shell script on the local filesystem, diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 606016222c..fab8ce9227 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -1532,6 +1532,14 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { signalStopped = resolve; }); let stdinSeq = 0; + // One promise chain per session that serializes the stdin file writes. Each + // write is multi-exec on the command-managed client: prepare, append per 32 + // KiB, then an atomic rename. The chain makes the rename for file N finish + // before the write for file N+1 starts, so the files land in send order. + // Without it the writes overlap. A small later chunk can then rename ahead of + // a big earlier chunk, so the wrapper reads the stdin bytes out of order and + // corrupts a large prompt on the stdin path. + let stdinWriteChain: Promise = Promise.resolve(); let pollTimer: NodeJS.Timeout | null = null; const pendingRemoteEvents: Array<{ type?: string; @@ -1642,9 +1650,21 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { if (stdinPayload) { stdinSeq += 1; const name = `${String(stdinSeq).padStart(12, "0")}.json`; - void runRuntimeWork(AGENT_SESSION_SEND_INPUT_SPAN, () => - client.writeTextFile(path.posix.join(stdinDir, name), jsonLine(stdinPayload)), - ).catch((error) => { + const filePath = path.posix.join(stdinDir, name); + // Chain this write after the previous one, so the atomic rename for + // file N finishes before the write for file N+1 starts. Keep the + // per-message `sandbox.agentSession.sendInput` span inside the chain. + const write = stdinWriteChain.then(() => + runRuntimeWork(AGENT_SESSION_SEND_INPUT_SPAN, () => + client.writeTextFile(filePath, jsonLine(stdinPayload)), + ), + ); + // The next message chains after this write on success or failure, so a + // failed write never blocks the chain. This mirrors the wrapper + // `writeChain` pattern for its event files. + stdinWriteChain = write.then(() => undefined, () => undefined); + // Keep the failure behavior: send one error line, then destroy the socket. + write.catch((error) => { nextSocket.write(jsonLine({ type: "error", message: error instanceof Error ? error.message : String(error) })); nextSocket.destroy(); }); @@ -1830,10 +1850,21 @@ export async function startAdapterExecutionTargetProcessSessionBridge(input: { if (pollTimer) clearTimeout(pollTimer); for (const liveSocket of liveSockets) liveSocket.destroy(); await new Promise((resolve) => server.close(() => resolve())).catch(() => undefined); - await client.writeTextFile( - path.posix.join(stdinDir, `${String(stdinSeq + 1).padStart(12, "0")}.json`), - jsonLine({ type: "stdinEnd" }), - ).catch(() => undefined); + // Wait for every accepted stdin write before `stdinEnd`. The socket handler + // fires each chunk write un-awaited through `stdinWriteChain`, so an earlier + // chunk can still be pending here. Chain the `stdinEnd` write onto the same + // per-session chain, so its file rename never finishes before an earlier + // chunk. `stdinSeq` is stable now, because the sockets are destroyed and the + // server is closed, so no new message can increment it. + const stdinEndPath = path.posix.join( + stdinDir, + `${String(stdinSeq + 1).padStart(12, "0")}.json`, + ); + const stdinEndWrite = stdinWriteChain.then(() => + client.writeTextFile(stdinEndPath, jsonLine({ type: "stdinEnd" })), + ); + stdinWriteChain = stdinEndWrite.then(() => undefined, () => undefined); + await stdinEndWrite.catch(() => undefined); await client.remove(sessionDir).catch(() => undefined); await fs.rm(proxyDir, { recursive: true, force: true }).catch(() => undefined); }, @@ -1913,12 +1944,40 @@ const stdinMaxParseRetries = (() => { return Number.isFinite(raw) && raw > 0 ? raw : 100; })(); const stdinParseRetries = new Map(); +// Track the next expected sequence number. The host writes the stdin files in +// send order and pads the number to 12 digits, starting at 1. The files sort in +// send order. When the smallest present number is greater than expected, an +// earlier file has not appeared yet: a missing file, not an unreadable one. Hold +// the send order and wait for it, bounded by the same retry budget as the +// unreadable-file path. This turns a reordering into a loud error, never silent +// corruption. +let stdinExpectedSeq = 1; +let stdinGapRetries = 0; 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 entrySeq = Number.parseInt(name, 10); + // Hold the send order when an earlier file has not appeared. Do not consume + // this later file: wait for the missing file on a later cycle, bounded by + // the retry budget. After the budget, fail loud and advance past the gap, + // so the present file can run. + if (Number.isFinite(entrySeq) && entrySeq > stdinExpectedSeq) { + stdinGapRetries += 1; + if (stdinGapRetries < stdinMaxParseRetries) { + break; + } + await writeEvent({ + type: "error", + message: + "Advanced past missing stdin files " + stdinExpectedSeq + " to " + (entrySeq - 1) + + " after " + stdinMaxParseRetries + " retries.", + }); + stdinGapRetries = 0; + stdinExpectedSeq = entrySeq; + } const file = path.posix.join(stdinDir, name); let message; try { @@ -1941,6 +2000,10 @@ async function pollStdin() { "Dropped unreadable stdin file after " + stdinMaxParseRetries + " retries: " + name + ": " + (error instanceof Error ? error.message : String(error)), }); + // The file is resolved (dropped). Advance the expected number and reset + // the gap budget, then let the loop go on to the next entry. + if (Number.isFinite(entrySeq)) stdinExpectedSeq = entrySeq + 1; + stdinGapRetries = 0; continue; } // The file is not readable yet and is not past the retry limit. Keep it @@ -1954,6 +2017,10 @@ async function pollStdin() { // 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); + // The file is handled. Advance the expected number and reset the gap + // budget, so the next expected file starts fresh. + if (Number.isFinite(entrySeq)) stdinExpectedSeq = entrySeq + 1; + stdinGapRetries = 0; if (message.type === "stdin" && typeof message.data === "string") { if (!stdinClosed) child.stdin.write(Buffer.from(message.data, "base64")); } else if (message.type === "stdinEnd") {