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") {