From 96516a86221da65cfb4f920119f3977d0c62bd7d Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 17:44:00 -0800 Subject: [PATCH] Keep ACPX output while Core reconnects (#19365) --- .../runner/DURABLE_TRANSPORT.md | 11 +- .../runner-core/src/acpx_provider_backend.rs | 54 ++++- .../runner-core/src/acpx_provider_session.rs | 41 +++- .../runner-core/src/acpx_sidecar_transport.rs | 5 + .../src/cli/acpx-runtime-sidecar.test.ts | 200 +++++++++++++++++- .../src/cli/acpx-runtime-sidecar.ts | 29 ++- .../acpx-controller-recovery.test.ts | 134 +++++++++--- .../test-support/acpx-blocked-output-host.mjs | 67 ++++++ 8 files changed, 486 insertions(+), 55 deletions(-) create mode 100644 packages/paperclip-runner/src/test-support/acpx-blocked-output-host.mjs diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index a852fe0f07b..047f25aa087 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -42,9 +42,11 @@ atomically from durable state. After disconnect, every remaining event is sent again with the same identity and source sequence. After disconnect, runnerd tries Core when reconnect is due. Between attempts, -live Codex processes can drain local output into the outbox. Work that +live Codex and ACPX processes can drain local output into the outbox. Work that needs a provider request waits for Core, including managed-provider history -reads and unanswered tools or approvals. Runnerd suspends +reads and unanswered tools or approvals. An ACPX sidecar can include a fresh +`goalAtTerminal` snapshot in `runtime.turn_terminal` to settle the turn offline. +A missing or stale snapshot waits for a connected goal read. Runnerd suspends intake during terminal delivery or an authority transition. If Codex hits its tool-receipt limit during reconnect, runnerd stops its local @@ -85,7 +87,10 @@ P0 reserve is an explicit unrecoverable condition. While disconnected, soft-limit backpressure retains the rejected event and its suffix at the provider owner until replay ACKs free outbox space. Oversized events or exhausted hard limits stop the owned provider and preserve the -accepted durable prefix. +accepted durable prefix. The ACPX sidecar separately caps queued stdout at +4 MiB; exceeding that bound requests owned provider shutdown and records a +failure rather than allowing unbounded process memory while the Runner cannot +read. ## Recovering Codex between turns diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index dc10623db86..67013b60dc2 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -1144,7 +1144,7 @@ impl AcpxCommandExecutor { "status": "failed", "turnTerminalState": "failed", "runTerminalState": "failed", - "reportedWorkDisposition": "unknown", + "reportedWorkDisposition": "needs_review", "provider": "acpx", }), })?; @@ -1657,7 +1657,7 @@ impl AcpxCommandExecutor { )); } } - let mut projection = self + let projection = self .session .as_mut() .ok_or_else(|| { @@ -1667,6 +1667,15 @@ impl AcpxCommandExecutor { .map_err(|error| { DurableRunnerError::invalid(format!("ACPX goal control failed: {error}")) })?; + self.accept_goal_projection(command, payload, projection) + } + + fn accept_goal_projection( + &mut self, + command: &str, + payload: &Value, + mut projection: Value, + ) -> Result { if projection.get("schema").and_then(Value::as_str) != Some("paperclip.session_goal.snapshot.v1") { @@ -2276,8 +2285,7 @@ impl AcpxCommandExecutor { Ok(CommandExecution::result(json!({"status": "completed"}))) } - fn poll_provider(&mut self) -> Result<(), DurableRunnerError> { - self.restore()?; + fn poll_provider(&mut self, disconnected: bool) -> Result<(), DurableRunnerError> { if self .state .as_ref() @@ -2291,7 +2299,13 @@ impl AcpxCommandExecutor { .session .as_mut() .expect("ACPX session remains available while polling") - .poll_event(Duration::from_millis(1)); + .poll_output( + Duration::from_millis(1), + disconnected, + self.state + .as_ref() + .and_then(|state| state.goal_source_revision), + ); let events = match events { Ok(events) => events, Err(error) => { @@ -2310,7 +2324,7 @@ impl AcpxCommandExecutor { "code":"acpx_provider_transport_failed","message":message, "error":{"code":"acpx_provider_transport_failed","message":message,"recoverable":false}}) })?; state.push(NormalizedProviderEvent { event_type: "run.terminal".to_owned(), priority: EventPriority::P0, - payload: json!({"schema":"paperclip.prp.terminal.v1","status":"failed","turnTerminalState":"failed","runTerminalState":"failed","reportedWorkDisposition":"unknown","provider":"acpx"}) })?; + payload: json!({"schema":"paperclip.prp.terminal.v1","status":"failed","turnTerminalState":"failed","runTerminalState":"failed","reportedWorkDisposition":"needs_review","provider":"acpx"}) })?; } self.context.provider_turn_id = None; // A failed transport cannot attest escaped provider cleanup. @@ -2320,7 +2334,9 @@ impl AcpxCommandExecutor { return Ok(()); } }; - let Some(events) = events else { break }; + let Some((events, mut terminal_goal)) = events else { + break; + }; let mut provider_turn_settled = false; for event in events { let provider_process_lost = matches!( @@ -2342,7 +2358,12 @@ impl AcpxCommandExecutor { // terminal receipt must remain publishable without another // protocol request, and must revoke future warm admission. if terminal.is_some() && !provider_process_lost { - let snapshot = self.goal_control("session.goal.get", &json!({}))?; + let snapshot = match terminal_goal.take() { + Some(goal) => { + self.accept_goal_projection("session.goal.get", &Value::Null, goal)? + } + None => self.goal_control("session.goal.get", &json!({}))?, + }; let state = self.state.as_mut().expect("ACPX session has durable state"); for (event_type, priority, payload) in snapshot.events { state.push(NormalizedProviderEvent { @@ -2563,7 +2584,21 @@ impl CommandExecutor for AcpxCommandExecutor { { return Ok(Vec::new()); } - self.poll_provider()?; + self.poll_provider(false)?; + self.retained_events() + } + + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + self.ensure_persistence_healthy()?; + if self.session.is_none() + || self + .state + .as_ref() + .is_none_or(|state| state.descriptor.run_id != self.context.run_id) + { + return Ok(Vec::new()); + } + self.poll_provider(true)?; self.retained_events() } @@ -4895,6 +4930,7 @@ mod tests { assert_eq!(events[2].event_type, "run.terminal"); assert_eq!(events[2].payload["schema"], "paperclip.prp.terminal.v1"); assert_eq!(events[2].payload["turnTerminalState"], "failed"); + assert_eq!(events[2].payload["reportedWorkDisposition"], "needs_review"); assert_eq!(recovered.poll_events().unwrap(), events); let cleanup_error = recovered .shutdown() diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index 550469d08a9..86adf7dd148 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -742,6 +742,17 @@ impl AcpxProviderSession { &mut self, timeout: Duration, ) -> Result>, LocalRunnerError> { + Ok(self + .poll_output(timeout, false, None)? + .map(|(events, _)| events)) + } + + pub(crate) fn poll_output( + &mut self, + timeout: Duration, + disconnected: bool, + last_goal_revision: Option, + ) -> Result, Option)>, LocalRunnerError> { self.ensure_open()?; let event = match self.transport.poll_event(timeout) { Ok(event) => event, @@ -750,6 +761,34 @@ impl AcpxProviderSession { let Some(mut event) = event else { return Ok(None); }; + let mut terminal_goal = None; + if disconnected + && event.event_type == GeneratedAcpxSidecarEventType::RuntimeTurnTerminal + && event.payload.pointer("/error/code").and_then(Value::as_str) + != Some("ACPX_PROVIDER_PROCESS_LOST") + { + let snapshot = event.payload.get("goalAtTerminal").filter(|snapshot| { + snapshot["schema"] == "paperclip.session_goal.snapshot.v1" + && snapshot["workingNow"] == false + && snapshot["providerRevision"] + .as_u64() + .is_some_and(|revision| revision > last_goal_revision.unwrap_or(0)) + && snapshot["sessionGoals"]["availability"] + .as_str() + .is_some_and(|availability| { + matches!(availability, "available" | "unsupported") + }) + && (goal_is_dormant(snapshot) + || snapshot.pointer("/goal/status").and_then(Value::as_str) + == Some("active")) + }); + let Some(snapshot) = snapshot else { + // Older sidecars need a connected goal read before settling this turn. + self.transport.defer_polled_event(event); + return Ok(None); + }; + terminal_goal = Some(snapshot.clone()); + } if event.event_type == GeneratedAcpxSidecarEventType::RuntimePermissionRequested { // Authority comes from this admitted connection's profile. A sidecar // claim cannot relabel another provider; old frames may omit origin. @@ -916,7 +955,7 @@ impl AcpxProviderSession { self.state = next_state; self.tool_bridge = next_bridge; self.reserved_tool_bridge = next_reserved_bridge; - Ok(Some(reconciled_events)) + Ok(Some((reconciled_events, terminal_goal))) } pub fn deliver_tool_result(&mut self, result: &ToolResult) -> Result<(), LocalRunnerError> { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs index a712f59f430..f402a48248d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -454,6 +454,11 @@ impl AcpxSidecarTransport { result } + pub(crate) fn defer_polled_event(&mut self, event: AcpxSidecarEvent) { + // poll_event already validated its sequence and freed its FIFO slot. + self.buffered_events.push_front(event); + } + pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> { self.poisoned = true; self.process.terminate_group().map(|_| ()) diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts index 57685b851c7..90d49bb600c 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts @@ -14,11 +14,15 @@ import { stripTypeScriptTypes } from "node:module"; import { cursorPlanToolIdentity, cursorToolIdentity } from "../drivers/acpx/cursor-plan-tool-identity.js"; import { createHash, randomUUID } from "node:crypto"; import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { readFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { once } from "node:events"; import { fileURLToPath } from "node:url"; import { createAcpxSidecarHostEnvironment, createSanitizedAcpxSpawnInput } from "../drivers/acpx/environment.js"; import { createKimiMessageProjection } from "../drivers/acpx/kimi-message-boundary.js"; import { classifyClaudeStatusBanner } from "../drivers/acpx/claude-status-banners.js"; +import { acpxGoalProjection } from "../drivers/acpx/session-goals.js"; import { afterEach, describe, expect, it, vi } from "vitest"; @@ -116,6 +120,74 @@ afterEach(async () => { }); describe("qualified ACPX runtime sidecar", () => { + it.each(["snapshot", "unavailable", "oversized", "payload-under-limit", "payload-limit", "payload-over-limit"])("emits the real completed terminal with optional goal observation: %s", async mode => { + const root = mkdtempSync(join(tmpdir(), "acpx-terminal-goal-")); + const loader = join(root, "provider-loader.mjs"), host = join(root, "terminal-host.mjs"); + const baseHost = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href; + const capability = { version: 1, controlMethod: "_session/goal", actions: ["set", "clear"] as Array<"set" | "clear"> }; + const goal = { objective: mode === "oversized" ? "x".repeat(ACPX_SIDECAR_MAX_FRAME_BYTES) : "Retain the completed goal", + status: "paused" as const, iterations: 1, lastReason: "Unicode é🙂, quote \" and slash \\ and newline\n" }; + if (mode.startsWith("payload-")) { + const bytes = Buffer.byteLength(JSON.stringify({status:"completed", goalAtTerminal:{ + ...acpxGoalProjection(capability, goal, false), providerRevision:1 }})); + const offset = mode === "payload-under-limit" ? -1 : mode === "payload-over-limit" ? 1 : 0; + goal.lastReason += "x".repeat(256 * 1024 + offset - bytes); + } + writeFileSync(host, `import { AcpxRuntimeHost as BaseHost } from ${JSON.stringify(baseHost)}; + export class AcpxRuntimeHost extends BaseHost { + static async open(options) { const host = await BaseHost.open(options); Object.setPrototypeOf(host, this.prototype); return host; } + goalCapability() { return ${JSON.stringify(capability)}; } + goalSnapshot() { + if (${JSON.stringify(mode)} === "unavailable") throw new Error("goal observation unavailable"); + return ${JSON.stringify(goal)}; + } + startTurn({ requestId }) { return { requestId, promptStarted: Promise.resolve(), + result: Promise.resolve({ status: "completed" }), events: (async function* () {})() }; } + }`); + writeFileSync(loader, `import { registerHooks } from "node:module"; + import { pathToFileURL } from "node:url"; + registerHooks({ resolve(specifier, context, next) { + if (context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js")) + return { url: pathToFileURL(${JSON.stringify(host)}).href, shortCircuit: true }; + return next(specifier, context); + } });`); + const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}}); + children.add(sidecar); + const send = (id:number, command:string, params:Record) => sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params}); + try { + sidecar.write(initializeRequest(1,"codex")); + expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true}); + send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root, + normalizedSessionId:"terminal-goal-session",permissionMode:"deny-all",systemInstructions:"Complete the local fixture turn.",tools:[]}); + expect(await sidecar.next(frame=>frame.id === 2)).toMatchObject({ok:true}); + send(3,"run.attach",{runId:"terminal-goal-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 3)).toMatchObject({ok:true}); + send(4,"turn.start",{turnId:"terminal-goal-turn",message:"Complete the fixture turn."}); + expect(await sidecar.next(frame=>frame.id === 4)).toMatchObject({ok:true}); + const terminal = await sidecar.next(frame=>frame.eventType === "runtime.turn_terminal"); + expect(terminal).toMatchObject({runId:"terminal-goal-run",turnId:"terminal-goal-turn",payload:{status:"completed"}}); + const payload = terminal.payload as Record; + expect(Buffer.byteLength(JSON.stringify(payload))).toBeLessThanOrEqual(256 * 1024); + if (mode === "snapshot" || mode === "payload-limit" || mode === "payload-under-limit") { + expect(payload.goalAtTerminal).toMatchObject({schema:"paperclip.session_goal.snapshot.v1",workingNow:false, + sessionGoals:{availability:"available"},goal:{objective:"Retain the completed goal",status:"paused",workingNow:false}}); + if (mode === "payload-limit") expect(Buffer.byteLength(JSON.stringify(payload))).toBe(256 * 1024); + if (mode === "payload-under-limit") expect(Buffer.byteLength(JSON.stringify(payload))).toBe(256 * 1024 - 1); + send(5,"session.goal.get",{}); + const observed = await sidecar.next(frame=>frame.id === 5); + expect(observed.ok).toBe(true); + const {providerRevision:_terminalRevision,...atTerminal} = payload.goalAtTerminal as Record; + const {providerRevision:_readRevision,...ordinary} = observed.result as Record; + expect(atTerminal).toEqual(ordinary); + } else { + expect(payload).toEqual({status:"completed"}); + } + send(6,"session.snapshot",{}); + expect(await sidecar.next(frame=>frame.id === 6)).toMatchObject({ok:true,result:{runId:"terminal-goal-run",turnId:null}}); + expect(sidecar.pending(frame=>frame.eventType === "runtime.turn_terminal")).toBe(0); + } finally {await sidecar.close();children.delete(sidecar);rmSync(root,{recursive:true,force:true});} + }, 15_000); + it("gives native Claude HTTP tools the Runner lease budget without changing sibling providers", () => { const source = readFileSync(new URL("./acpx-runtime-sidecar.ts", import.meta.url), "utf8"); const start = source.indexOf(" environment:"); @@ -509,6 +581,107 @@ describe("qualified ACPX runtime sidecar", () => { } }); + it("bounds aggregate blocked output and requests shutdown without consuming rejected event identities", () => { + const wire = loadActualSidecarWriter("backpressure"); + expect(wire.emit("runtime.diagnostic", { message: "accepted" })).toBe(true); + wire.bufferedBytes(4 * ACPX_SIDECAR_MAX_FRAME_BYTES - 1); + expect(wire.emit("runtime.diagnostic", { message: "cannot fit" })).toBe(false); + expect(wire.sequence()).toBe(1); + expect(wire.shutdownReasons).toEqual(["ACPX sidecar output buffer limit exceeded"]); + expect(wire.errors).toEqual(["[paperclip-acpx-sidecar] output_buffer_limit_exceeded\n"]); + wire.bufferedBytes(0); + expect(wire.emit("runtime.turn_terminal", { status: "completed" })).toBe(false); + expect(wire.writes).toHaveLength(1); + expect(wire.shutdownReasons).toHaveLength(1); + }); + + it("stops its owned provider under real stdout backpressure and exits with failure after draining", async () => { + const root = mkdtempSync(join(tmpdir(), "acpx-sidecar-output-pressure-")); + const loader = join(root, "provider-loader.mjs"); + const fakeHost = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href; + writeFileSync(loader, `import { registerHooks } from "node:module"; + registerHooks({ resolve(specifier, context, next) { + if (context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js")) { + return { url: ${JSON.stringify(fakeHost)}, shortCircuit: true }; + } + return next(specifier, context); + } });`); + const child = spawn(process.execPath, ["--import", "tsx", "--import", loader, + fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url))], + { stdio: ["pipe", "pipe", "pipe"], env: { PATH: process.env.PATH, PAPERCLIP_TEST_OUTPUT_ROOT: root } }); + const exited = once(child, "exit"); + let stderr = "", stdout = ""; + child.stderr.setEncoding("utf8").on("data", text => { stderr += text; }); + child.stdout.setEncoding("utf8").on("data", text => { stdout += text; }); + const send = (id: number, command: string, params: Record) => { + child.stdin.write(`${JSON.stringify({ protocolVersion: ACPX_SIDECAR_PROTOCOL_VERSION, id, command, params })}\n`); + }; + const response = async (id: number) => { + await vi.waitFor(() => { + expect(child.exitCode, stderr).toBeNull(); + expect(stdout.split("\n").some(line => line && JSON.parse(line).id === id), stderr).toBe(true); + }, { timeout: 5000 }); + const frame = stdout.split("\n").filter(Boolean).map(line => JSON.parse(line)).find(frame => frame.id === id); + expect(frame.ok, JSON.stringify(frame)).toBe(true); + }; + let providerPid: number | undefined; + try { + send(1, "initialize", { agent: "codex", model: "gpt-5.6-sol" }); await response(1); + send(2, "session.open", { agent: "codex", model: "gpt-5.6-sol", runtimeDirectory: root, + workingDirectory: root, normalizedSessionId: "blocked-output-session", permissionMode: "deny-all", + systemInstructions: "Generate synthetic local output.", tools: [] }); await response(2); + providerPid = Number(readFileSync(join(root, "provider-pid"), "utf8")); + send(3, "run.attach", { runId: "blocked-output-run", catalogRevision: 1, tools: [] }); await response(3); + child.stdout.pause(); + send(4, "turn.start", { turnId: "blocked-output-turn", message: "Fill the blocked pipe." }); + await vi.waitFor(() => expect(existsSync(join(root, "provider-cleanup.json")), stderr).toBe(true), { timeout: 8000 }); + const cleanup = JSON.parse(readFileSync(join(root, "provider-cleanup.json"), "utf8")); + expect(cleanup).toMatchObject({ reason: "ACPX sidecar output buffer limit exceeded", pid: providerPid, + closeCount: 1, code: 0, signal: null }); + expect(cleanup.bufferedBytes).toBeGreaterThan(3 * ACPX_SIDECAR_MAX_FRAME_BYTES); + expect(cleanup.bufferedBytes).toBeLessThanOrEqual(4 * ACPX_SIDECAR_MAX_FRAME_BYTES); + expect(() => process.kill(providerPid!, 0)).toThrow(); + expect(child.exitCode).toBeNull(); + expect(stderr).toContain("output_buffer_limit_exceeded"); + child.stdout.resume(); + expect(await exited).toEqual([1, null]); + const events = stdout.trim().split("\n").map(line => JSON.parse(line)).filter(frame => frame.sequence); + expect(events.length).toBeGreaterThan(256); + expect(events.map(frame => frame.sequence)).toEqual(events.map((_frame, index) => index + 1)); + expect(events.some(frame => frame.eventType === "runtime.turn_terminal" && frame.payload.status === "completed")).toBe(false); + } finally { + child.stdout.resume(); + child.stdin.end(); + if (providerPid && !existsSync(join(root, "provider-cleanup.json"))) { try { process.kill(providerPid, "SIGKILL"); } catch {} } + if (child.exitCode === null) child.kill("SIGKILL"); + await exited.catch(() => undefined); + rmSync(root, { recursive: true, force: true }); + } + }, 15_000); + + it("reaps the output-pressure fixture provider when its owner dies before cleanup", async () => { + const fixture = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href; + const owner = spawn(process.execPath, ["--input-type=module", "-e", ` + import { AcpxRuntimeHost } from ${JSON.stringify(fixture)}; + const host = new AcpxRuntimeHost({}); + host.provider.once("message", () => process.send(host.provider.pid)); + setInterval(() => {}, 1000); + `], { stdio: ["ignore", "ignore", "ignore", "ipc"], env: { PATH: process.env.PATH } }); + const exited = once(owner, "exit"); + let providerPid: number | undefined; + try { + [providerPid] = await once(owner, "message"); + expect(typeof providerPid).toBe("number"); + owner.kill("SIGKILL"); + await exited; + await vi.waitFor(() => expect(() => process.kill(providerPid!, 0)).toThrow()); + } finally { + if (providerPid) { try { process.kill(providerPid, "SIGKILL"); } catch {} } + if (owner.exitCode === null && owner.signalCode === null) owner.kill("SIGKILL"); + await exited.catch(() => undefined); + } + }); + it.each(["success", "accepted-false", "error", "cancel", "timeout", "receiver-rejected"])( "commits normalized delivery only after the actual tool.resolve handler succeeds: %s", @@ -1488,11 +1661,11 @@ class SidecarProcess { #stderr = ""; #closed = false; - constructor() { + constructor(options: {loader?:string;environment?:Record} = {}) { this.#child = spawn( - fileURLToPath(new URL("../../node_modules/.bin/tsx", import.meta.url)), - [fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url))], - { stdio: ["pipe", "pipe", "pipe"], env: { PATH: process.env.PATH, LANG: "C.UTF-8" } }, + options.loader ? process.execPath : fileURLToPath(new URL("../../node_modules/.bin/tsx", import.meta.url)), + [...(options.loader ? ["--import","tsx","--import",options.loader] : []), fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url))], + { stdio: ["pipe", "pipe", "pipe"], env: { PATH: process.env.PATH, LANG: "C.UTF-8", ...options.environment } }, ); let stdout = ""; this.#child.stdout.setEncoding("utf8"); @@ -1522,6 +1695,10 @@ class SidecarProcess { return this.#stderr; } + pending(predicate: (frame: Record) => boolean): number { + return this.#frames.filter(predicate).length; + } + async next( predicate: (frame: Record) => boolean, ): Promise> { @@ -1612,21 +1789,22 @@ function loadWaitForTool(input: { /** Actual producer and encoder, with only the destination stream replaced. */ function loadActualSidecarWriter(mode: "normal" | "backpressure" | "throws" = "normal"): { emit: (eventType: string, payload: unknown) => boolean; sequence: () => number; writes: string[]; errors: string[]; + bufferedBytes: (bytes: number) => void; shutdownReasons: string[]; } { const source = readFileSync(new URL("./acpx-runtime-sidecar.ts", import.meta.url), "utf8"); const emitStart = source.indexOf("function emit("), emitEnd = source.indexOf("\nfunction diagnostic", emitStart); const writeStart = source.indexOf("function writeFrame("), writeEnd = source.indexOf("\nfunction requireHost", writeStart); if (emitStart < 0 || emitEnd < 0 || writeStart < 0 || writeEnd < 0) throw new Error("Actual sidecar writer source not found"); - const writes: string[] = [], errors: string[] = []; - const process = { stdout: { write: (line: string) => { if (mode === "throws") throw new Error("fixture write failed"); writes.push(line); return mode !== "backpressure"; } }, + const writes: string[] = [], errors: string[] = [], shutdownReasons: string[] = []; + const process = { stdout: { writableLength: 0, write: (line: string) => { if (mode === "throws") throw new Error("fixture write failed"); writes.push(line); return mode !== "backpressure"; } }, stderr: { write: (line: string) => { errors.push(line); return true; } } }; - const writer = new Function("process", "stringifyAcpxSidecarFrame", "ACPX_SIDECAR_PROTOCOL_VERSION", "ACPX_SIDECAR_MAX_FRAME_BYTES", ` - let sequence=0; const runId="test-run", turnId="test-turn"; + const writer = new Function("process", "stringifyAcpxSidecarFrame", "ACPX_SIDECAR_PROTOCOL_VERSION", "ACPX_SIDECAR_MAX_FRAME_BYTES", "requestShutdown", ` + let sequence=0, outputFailed=false; const runId="test-run", turnId="test-turn", MAX_BUFFERED_OUTPUT_BYTES=4*ACPX_SIDECAR_MAX_FRAME_BYTES; ${stripTypeScriptTypes(source.slice(emitStart, emitEnd))} ${stripTypeScriptTypes(source.slice(writeStart, writeEnd))} return { emit, sequence: () => sequence }; - `)(process, stringifyAcpxSidecarFrame, ACPX_SIDECAR_PROTOCOL_VERSION, ACPX_SIDECAR_MAX_FRAME_BYTES); - return { ...writer, writes, errors }; + `)(process, stringifyAcpxSidecarFrame, ACPX_SIDECAR_PROTOCOL_VERSION, ACPX_SIDECAR_MAX_FRAME_BYTES, (reason: string) => shutdownReasons.push(reason)); + return { ...writer, writes, errors, bufferedBytes: bytes => { process.stdout.writableLength = bytes; }, shutdownReasons }; } /** Actual parsed command dispatcher; only the owned pending map/turn are supplied. */ diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index c03c2063334..082a9e93ed3 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -117,6 +117,9 @@ import { AcpxTurnControlLedger, parseAcpxTurnControl } from "../drivers/acpx/tur const turnControls = new AcpxTurnControlLedger(); const MAX_PENDING_TOOLS = 512; const MAX_PENDING_INPUTS = 16; +const MAX_BUFFERED_OUTPUT_BYTES = 4 * ACPX_SIDECAR_MAX_FRAME_BYTES; +const MAX_EVENT_PAYLOAD_BYTES = 256 * 1024; +let outputFailed = false; let goalSourceRevision = 0; function observedGoalProjection(...args: Parameters) { return { ...acpxGoalProjection(...args), providerRevision: ++goalSourceRevision }; @@ -795,6 +798,20 @@ async function pumpTurn( } // A terminal frame is also runnerd's permission to recycle this provider. // Publish it only after no callback can inherit this turn's mutable binding. + // runnerd can settle this turn offline using the same observation as session.goal.get. + // Fall back to the original terminal frame if the observation is unavailable or too large. + let goalAtTerminal: ReturnType | undefined; + try { + goalAtTerminal = observedGoalProjection(activeHost.goalCapability(), activeHost.goalSnapshot(), turnId !== null); + } catch { + // The terminal stays authoritative even if the optional observation fails. + } + if (goalAtTerminal) { + const payload = { ...terminal, goalAtTerminal }; + // Rust admits the payload separately from the larger sidecar frame. + if (Buffer.byteLength(JSON.stringify(payload)) <= MAX_EVENT_PAYLOAD_BYTES + && emit("runtime.turn_terminal", payload, currentTurnId)) return; + } emit("runtime.turn_terminal", terminal, currentTurnId); } @@ -1628,11 +1645,19 @@ function response( } function writeFrame(value: AcpxSidecarEvent | AcpxSidecarResponse): boolean { + if (outputFailed) return false; const line = stringifyAcpxSidecarFrame(value); - if (Buffer.byteLength(line) > ACPX_SIDECAR_MAX_FRAME_BYTES) { + const bytes = Buffer.byteLength(line); + if (bytes > ACPX_SIDECAR_MAX_FRAME_BYTES) { process.stderr.write("[paperclip-acpx-sidecar] output_frame_too_large\n"); return false; } + if (process.stdout.writableLength + bytes + 1 > MAX_BUFFERED_OUTPUT_BYTES) { + outputFailed = true; + process.stderr.write("[paperclip-acpx-sidecar] output_buffer_limit_exceeded\n"); + requestShutdown("ACPX sidecar output buffer limit exceeded"); + return false; + } process.stdout.write(`${line}\n`); // Writable.write(false) accepted the bytes into its queue; it signals // backpressure, not a rejected frame. A synchronous write error still throws. @@ -1815,7 +1840,7 @@ async function shutdown(reason: string): Promise { turnId = null; lines.close(); process.stdin.pause(); - process.exitCode = cleanupIncomplete ? 1 : 0; + process.exitCode = cleanupIncomplete || outputFailed ? 1 : 0; } function retainActiveHostCleanup( diff --git a/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts b/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts index e44b6c7caf5..9c6e8924fea 100644 --- a/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts +++ b/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts @@ -2,7 +2,7 @@ import { execFileSync, spawn, type ChildProcess } from "node:child_process"; import { once } from "node:events"; import { createHash } from "node:crypto"; import { createServer as createNetServer } from "node:net"; -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; @@ -39,7 +39,8 @@ function sidecarSource(root: string, selectedProfile = profile, agent = "claude" "const profileDigest = " + JSON.stringify(selectedProfile.commandDigest) + ";", "const protocolVersion = " + GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION + ";", "const send = value => process.stdout.write(JSON.stringify(value) + '\\n');", - "let identity, runId, turnId = null, sequence = 0, lastAction = '';", + "let identity, runId, turnId = null, sequence = 0, lastAction = '', lastOfflineAction = '', pendingPermission = false;", + "let goalRevision=0; const goalSnapshot=()=>({schema:'paperclip.session_goal.snapshot.v1',providerRevision:++goalRevision,goal:null,workingNow:false,sessionGoals:{availability:'unsupported',actions:[],autonomousUpdates:false,persistentAcrossResume:false,maxObjectiveChars:4000,tokenBudgetControl:false,usageReporting:false}});", "const event = (eventType, payload) => send({protocolVersion, sequence:++sequence, eventType, runId, turnId, payload});", "(async () => {", " const lease = async () => {for(let attempt=0;attempt<64;attempt++){const s=net.createServer();try{await new Promise((resolve,reject)=>{s.once('error',reject);s.listen(randomInt(49152,65536),'127.0.0.1',resolve);});return s;}catch(error){if(error.code!=='EADDRINUSE')throw error;}}throw new Error('No lifetime fence port available');};", @@ -49,21 +50,24 @@ function sidecarSource(root: string, selectedProfile = profile, agent = "claude" " const reply = (r,result) => send({protocolVersion,id:r.id,ok:true,result});", " const lines=rl.createInterface({input:process.stdin}); lines.on('close',()=>process.exit(0)); lines.on('line', line => {", " const r=JSON.parse(line), p=r.params || {};", + " fs.appendFileSync(root+'/commands.jsonl',r.command+'\\n');", " if(r.command==='session.open'&&fs.existsSync(root+'/startup-failure.json')){send({protocolVersion,id:r.id,ok:false,error:{code:'ACPX_SESSION_ENSURE_FAILED',message:'Internal error',retryable:false,providerFailure:JSON.parse(fs.readFileSync(root+'/startup-failure.json','utf8'))}});return;}", " switch(r.command) {", " case 'initialize': reply(r,{protocolVersion,sidecarPid:process.pid,profile:{agent:" + JSON.stringify(agent) + "},capabilities:{persistentSessions:true,exactModelVerification:true,permissions:'runner_policy',semanticTools:'runner_bridge',structuredInput:'paperclip.question_set.v1'}});break;", " case 'session.open': identity={kind:'acpx',normalizedSessionId:p.normalizedSessionId,acpxRecordId:'record-1',backendSessionId:'backend-1',agentSessionId:'agent-1',profileDigest,workspaceDigest:'sha256:'+'2'.repeat(64),requestedModel:p.model,effectiveModel:p.model,permissionMode:p.permissionMode,mode:p.mode,piThinkingLevel:p.piThinkingLevel,claudeEffort:p.claudeEffort,grokEffort:p.grokEffort,grokDisableWebSearch:p.grokDisableWebSearch,providerLifetimeFenceCandidates:ports};reply(r,{sidecarPid:process.pid,identity,status:{},turnControls:{steering:false,queuedFollowUp:false}});break;", " case 'run.attach': runId=p.runId; reply(r,{runId,catalogRevision:p.catalogRevision});break;", - " case 'session.goal.get': reply(r,{schema:'paperclip.session_goal.snapshot.v1',goal:null,workingNow:false,sessionGoals:{availability:'unsupported',actions:[],autonomousUpdates:false,persistentAcrossResume:false,maxObjectiveChars:4000,tokenBudgetControl:false,usageReporting:false}});break;", + " case 'session.goal.get': if(fs.existsSync(root+'/hold-goal')){const timer=setInterval(()=>{if(fs.existsSync(root+'/release-goal')){clearInterval(timer);reply(r,goalSnapshot());}},10);}else reply(r,goalSnapshot());break;", " case 'turn.start': turnId=p.turnId;reply(r,{turnId,turnControls:{steering:false,queuedFollowUp:false}});event('runtime.event',{type:'text_delta',text:'started'});break;", - " case 'session.snapshot': reply(r,{identity,runId,turnId,pendingRuntimeRequests:[]});break;", + " case 'session.snapshot': reply(r,{identity,runId,turnId,pendingRuntimeRequests:pendingPermission?[{requestId:'permission-1',type:'permission',turnId}]:[]});break;", + " case 'permission.resolve': if(!pendingPermission)throw new Error('duplicate approval effect');pendingPermission=false;fs.appendFileSync(root+'/approval-effects','approved\\n');reply(r,{resolved:true});break;", " case 'turn.cancel': reply(r,{cancelled:true});event('runtime.turn_terminal',{status:'interrupted'});turnId=null;break;", " case 'session.suspend': reply(r,{identity});break;", " case 'session.close': reply(r,{closed:true});break;", " default: reply(r,{});", " }", " });", - " setInterval(()=>{if(!turnId)return;let action;try{action=fs.readFileSync(root+'/action','utf8');}catch{return;}if(action===lastAction)return;lastAction=action;if(action==='progress'){event('runtime.event',{type:'text_delta',text:'work-during-outage'});fs.writeFileSync(root+'/progress-receipt',String(process.pid));}if(action==='complete'){event('runtime.turn_terminal',{status:'completed'});turnId=null;fs.writeFileSync(root+'/complete-receipt',String(process.pid));}if(action==='exit'){fs.writeFileSync(root+'/exit-receipt',String(process.pid));process.exit(9);}},10);", + " setInterval(()=>{if(!turnId)return;let action;try{action=fs.readFileSync(root+'/action','utf8');}catch{return;}if(action===lastAction)return;lastAction=action;if(action==='progress'){event('runtime.event',{type:'text_delta',text:'work-during-outage'});fs.writeFileSync(root+'/progress-receipt',String(process.pid));}if(action==='complete'){event('runtime.turn_terminal',{status:'completed',...(fs.existsSync(root+'/stale-goal')?{goalAtTerminal:{...goalSnapshot(),providerRevision:0}}:!fs.existsSync(root+'/hold-goal')?{goalAtTerminal:goalSnapshot()}:{})});turnId=null;fs.writeFileSync(root+'/complete-receipt',String(process.pid));}if(action==='exit'){fs.writeFileSync(root+'/exit-receipt',String(process.pid));process.exit(9);}},10);", + " setInterval(()=>{if(!turnId)return;let action;try{action=fs.readFileSync(root+'/offline-action','utf8');}catch{return;}if(action===lastOfflineAction)return;lastOfflineAction=action;if(action==='burst'){for(let i=0;i<288;i++)event('runtime.event',{type:'text_delta',text:'offline-'+i+':'+'x'.repeat(512)});}if(action==='approval'){pendingPermission=true;event('runtime.permission_requested',{requestId:'permission-1',kind:'edit',title:'Write the file?',choices:[{key:'accept',label:'Allow once'},{key:'cancel',label:'Deny'}]});}},10);", "})().catch(e=>{process.stderr.write(String(e));process.exit(1);});", ].join("\n"); } @@ -83,9 +87,9 @@ it("preserves a startup provider failure through the real sidecar transport and result: { result: { providerFailure } } }); expect(run.launches()).toHaveLength(1); } finally { await run.cleanup(); } -}, 20_000); +}, 60_000); -async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidateStartupAdmission, beforeLaunchDelayMs = 0) { +async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidateStartupAdmission, beforeLaunchDelayMs = 0, reconnectGraceMs = 15_000) { const root = mkdtempSync(resolve(tmpdir(), "acpx-controller-recovery-")); const { executable: nodeCommand } = await copyOwnedNodeRuntime(resolve(root, "node")); const stateDirectory = resolve(root, "control-plane"); @@ -122,7 +126,7 @@ async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidat "--runner-id",identity.runnerInstanceId,"--environment-lease-id",identity.environmentLeaseId, "--run-id",identity.runId,"--session-id",identity.normalizedSessionId,"--turn-id",identity.turnId,"--item-id",identity.itemId, "--runner-version",expectedRunnerVersion,"--runner-digest",expectedRunnerDigest,"--max-runtime-ms","0", - "--reconnect-grace-ms","15000","--reconnect-delay-ms","25","--acpx-launch-authority-digest","sha256:"+"d".repeat(64), + "--reconnect-grace-ms",String(reconnectGraceMs),"--reconnect-delay-ms","25","--acpx-launch-authority-digest","sha256:"+"d".repeat(64), "--acpx-sidecar-command",nodeCommand,"--acpx-sidecar-command-sha256",digest(nodeCommand), "--acpx-sidecar-script",script,"--acpx-sidecar-script-sha256",digest(script), ...(candidateAdmission ? ["--acpx-candidate-admission-json", JSON.stringify(candidateAdmission)] : [])]; @@ -131,7 +135,7 @@ async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidat const bounded = maxLifetimeMs === undefined ? undefined : spawnRunner({ connectUrl: controller.connectUrl, stateDirectory: resolve(root, "runner"), identity, ticket, maxOutboxBytes: 16 * 1024 * 1024, p0ReserveBytes: 1024 * 1024, - maxRuntimeMs: 0, maxLifetimeMs: deadlineAt! - Date.now(), reconnectGraceMs: 15_000, + maxRuntimeMs: 0, maxLifetimeMs: deadlineAt! - Date.now(), reconnectGraceMs, runnerBinaryPath: binary, runnerVersion: expectedRunnerVersion, runnerDigest: expectedRunnerDigest, acpxLaunchProfile: { ...(candidateAdmission ? { candidateAdmission } : {}), authorityDigest: `sha256:${"d".repeat(64)}`, command: nodeCommand, commandSha256: digest(nodeCommand), sidecarScript: script, sidecarScriptSha256: digest(script) }, @@ -150,12 +154,14 @@ async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidat expect(child.exitCode,"runner exit; "+diagnostics).toBeNull(); expect(child.signalCode,"runner signal; "+diagnostics).toBeNull(); expect(["completed","failed"]).toContain(controller.getCommand(id)?.status); - },{timeout:8000}); + },{timeout:type === "session.open" ? 35_000 : 8000}); } catch (error) { let runnerDiagnostics: unknown; try { runnerDiagnostics = JSON.parse(readFileSync(resolve(root, "runner/runner-state.json"), "utf8")).diagnostics?.slice(-8); } catch { /* Startup may not have created state. */ } throw new Error("Hermetic Runner command did not settle: " + JSON.stringify({ id, type, authenticatedConnections: controller.activeRunnerConnectionCount(), command: controller.getCommand(id), + malformedFrames: controller.store.state.malformedFrames, acceptedConnections: controller.store.state.connectionCount, + lastCommittedEvents: controller.store.state.committedEvents.slice(-4).map(event => ({ sourceSeq: event.sourceSeq, eventType: event.eventType })), remainingLifetimeMs: deadlineAt === undefined ? null : deadlineAt - Date.now(), exitCode: child.exitCode, signalCode: child.signalCode, stderr: diagnostics, runnerDiagnostics }).slice(-8192), { cause: error }); } @@ -211,9 +217,9 @@ async function fixture(maxLifetimeMs?: number, candidateAdmission?: AcpxCandidat } async function assertActiveExpiry(candidate = false) { - // One fixed grant covers the six-second cold delay, four existing eight- - // second command budgets, and seven seconds of active-turn headroom. Loaded - // CI exhausted the earlier fifteen-second grant before session.open settled. + // One fixed grant covers the six-second cold delay, the native thirty-second + // startup bound, and active-turn headroom. The fixture must not time out + // session.open before the provider admission deadline it exercises. // Never renew this authority after setup or the active-turn witness. const lifetimeMs = 45_000; const candidateAdmission = candidate ? bindAcpxCandidateAdmission({ schema: "paperclip.acpx-candidate-admission.v1", companyId: "company-expiry", @@ -271,7 +277,7 @@ it("keeps the original native ACPX process and active turn through controller re writeFileSync(resolve(run.root,"action"),"complete"); await vi.waitFor(()=>expect(run.controller().store.state.committedEvents.some(event=>event.eventType==="turn.completed")).toBe(true),{timeout:5000}); }finally{await run.cleanup();} -},20000); +},60000); it("adopts the live ACPX execution through the production native transport",async()=>{ const run=await fixture(); @@ -312,24 +318,95 @@ it("adopts the live ACPX execution through the production native transport",asyn await adopted?.transport.close().catch(()=>undefined); await run.cleanup(); } -},25000); +},60000); -it("recovers a turn that completed while its controller was offline",async()=>{ +it.each(["legacy", "stale"])("reconnects before requesting a %s sidecar terminal goal",async(mode)=>{ const run=await fixture(); try{ await run.ready(); expect(await run.command("turn.start",{turnId:"provider-turn-1",text:"Finish while disconnected"})).toMatchObject({status:"completed"}); const pid=run.launches()[0].pid; + const port=Number(new URL(run.controller().connectUrl).port); await run.stopController(); + writeFileSync(resolve(run.root,"hold-goal"),"1"); + if(mode==="stale") writeFileSync(resolve(run.root,"stale-goal"),"1"); writeFileSync(resolve(run.root,"action"),"complete"); await vi.waitFor(()=>expect(readFileSync(resolve(run.root,"complete-receipt"),"utf8")).toBe(String(pid)),{timeout:3000}); + // Observe two actual dial attempts after the terminal arrived, including a backoff cycle. + let attempts=0; + const probe=createNetServer(socket=>{attempts++;socket.destroy();}); + await new Promise((ready,reject)=>{probe.once("error",reject);probe.listen(port,"127.0.0.1",ready);}); + try { + await vi.waitFor(()=>expect(attempts).toBeGreaterThanOrEqual(2),{timeout:5000}); + expect(readFileSync(resolve(run.root,"commands.jsonl"),"utf8").split("\n").filter(command=>command==="session.goal.get")).toHaveLength(1); + } finally { + await new Promise(done=>probe.close(()=>done())); + } await run.replaceController(); + // The legacy terminal needs a fresh goal. Holding its reply must not hold off authentication. + await vi.waitFor(()=>expect(run.controller().store.state.connectionCount).toBeGreaterThan(1),{timeout:5000}); + writeFileSync(resolve(run.root,"release-goal"),"1"); await vi.waitFor(()=>expect(run.controller().store.state.committedEvents.some(event=>event.eventType==="turn.completed")).toBe(true),{timeout:5000}); expect(await run.command("session.snapshot",{includePendingRuntimeRequests:true})).toMatchObject({status:"completed",result:{result:{runtimeRequestsLive:true,activeProviderTurnId:null}}}); expect(run.launches()).toHaveLength(1); expect(()=>process.kill(pid,0)).not.toThrow(); }finally{await run.cleanup();} -},20000); +},60000); + +it("durably drains noisy active output and one offline terminal before its controller returns", async () => { + const run = await fixture(undefined, undefined, 0, 60_000); + const runnerState = () => JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8")); + try { + await run.ready(); + expect(await run.command("turn.start", { turnId: "provider-turn-1", text: "Finish the admitted work offline" })).toMatchObject({ status: "completed" }); + const pid = run.launches()[0].pid; + await run.stopController(); + writeFileSync(resolve(run.root, "offline-action"), "burst"); + await vi.waitFor(() => expect(runnerState().outbox.some((event: any) => event.envelope.payload.payload.text?.startsWith("offline-287:"))).toBe(true), { timeout: 20_000 }); + writeFileSync(resolve(run.root, "action"), "complete"); + await vi.waitFor(() => expect(runnerState().outbox.filter((event: any) => event.eventType === "turn.completed")).toHaveLength(1), { timeout: 5000 }); + const offline = runnerState(); + expect(offline.peakOutboxBytes).toBeLessThanOrEqual(offline.maxOutboxBytes); + expect(() => process.kill(pid, 0)).not.toThrow(); + await run.replaceController(); + await vi.waitFor(() => expect(run.controller().store.state.committedEvents.some(event => event.sourceSeq === offline.nextSourceSeq - 1)).toBe(true), { timeout: 10_000 }); + const replay = new Map(run.controller().store.state.committedEvents.map(event => [event.sourceSeq, event.sourceEventId])); + for (const event of offline.outbox) expect(replay.get(event.sourceSeq)).toBe(event.envelope.payload.sourceEventId); + expect(run.controller().store.state.committedEvents.filter(event => event.eventType === "turn.completed")).toHaveLength(1); + expect(run.launches()).toHaveLength(1); + expect(readFileSync(resolve(run.root, "commands.jsonl"), "utf8").split("\n").filter(command => command === "turn.start")).toHaveLength(1); + } finally { await run.cleanup(); } +}, 90_000); + +it("remains silent at an offline approval and resolves the original request exactly once after adoption", async () => { + const run = await fixture(); + const runnerState = () => JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8")); + try { + await run.ready(); + expect(await run.command("turn.start", { turnId: "provider-turn-1", text: "Wait for approval" })).toMatchObject({ status: "completed" }); + await run.stopController(); + writeFileSync(resolve(run.root, "offline-action"), "approval"); + await vi.waitFor(() => expect(runnerState().outbox.some((event: any) => event.eventType === "runtime_request.created")).toBe(true), { timeout: 5000 }); + const waiting = runnerState(); + await new Promise(resolveSilence => setTimeout(resolveSilence, 250)); + expect(runnerState().nextSourceSeq).toBe(waiting.nextSourceSeq); + expect(existsSync(resolve(run.root, "approval-effects"))).toBe(false); + await run.replaceController(); + const snapshot = await run.command("session.snapshot", { includePendingRuntimeRequests: true }); + expect(snapshot).toMatchObject({ status: "completed", result: { result: { + activeProviderTurnId: "provider-turn-1", runtimeRequestsLive: true, + pendingRuntimeRequests: [{ requestId: "permission-1", type: "permission", status: "pending" }], + } } }); + expect(existsSync(resolve(run.root, "approval-effects"))).toBe(false); + const resolution = await run.command("request.resolve", { requestId: "permission-1", resolution: { action: "accept" } }); + expect(resolution).toMatchObject({ status: "completed" }); + expect(await run.command("request.resolve", { requestId: "permission-1", resolution: { action: "accept" } })).toMatchObject({ status: "failed" }); + expect(readFileSync(resolve(run.root, "approval-effects"), "utf8")).toBe("approved\n"); + writeFileSync(resolve(run.root, "action"), "complete"); + await vi.waitFor(() => expect(run.controller().store.state.committedEvents.filter(event => event.eventType === "turn.completed")).toHaveLength(1), { timeout: 5000 }); + expect(run.launches()).toHaveLength(1); + } finally { await run.cleanup(); } +}, 60_000); it("distinguishes a provider-process crash from controller replacement",async()=>{ const run=await fixture(); @@ -346,21 +423,20 @@ it("distinguishes a provider-process crash from controller replacement",async()= probe.listen(port,"127.0.0.1",()=>probe.close(error=>error?reject(error):resolveFree())); }); },{timeout:3000}); + await vi.waitFor(() => { + const saved = JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8")); + expect(saved.outbox.filter((event: any) => event.eventType === "turn.failed")).toHaveLength(1); + }, { timeout: 5000 }); await run.replaceController(); const result=await run.command("session.snapshot",{includePendingRuntimeRequests:true}); - expect(result,JSON.stringify(result)).toMatchObject({status:"failed",result:{result:{code:"command_execution_failed"}}}); - const message = (result.result as {result:{message:string}}).result.message; - const lossReasons = [ - "ACPX request snapshot requires a live provider", - "ACPX sidecar request transport failed at session.snapshot: command serialization failed: Broken pipe (os error 32)", - "ACPX sidecar request transport failed at session.snapshot: failed to write process command: Broken pipe (os error 32)", - "ACPX sidecar exited at session.snapshot: exitCode=Some(9) signal=None", - "ACPX sidecar closed stdout at session.snapshot", - ]; - expect(lossReasons.map(reason => `ACPX live request snapshot failed: ${reason}`)).toContain(message); + expect(result,JSON.stringify(result)).toMatchObject({status:"failed",result:{result:{ + code:"command_execution_failed", message:"ACPX request snapshot requires the surviving provider", + }}}); + await vi.waitFor(() => expect(run.controller().store.state.committedEvents.filter(event => event.eventType === "turn.failed")).toHaveLength(1), { timeout: 5000 }); + await vi.waitFor(() => expect(run.controller().store.state.committedEvents.find(event => event.eventType === "run.terminal")?.envelope).toMatchObject({ payload: { payload: { reportedWorkDisposition: "needs_review" } } }), { timeout: 5000 }); expect(run.launches()).toHaveLength(1); }finally{await run.cleanup();} -},20000); +},60000); it("requires a surviving provider when requesting a live snapshot after explicit close",async()=>{ const run=await fixture(); @@ -374,4 +450,4 @@ it("requires a surviving provider when requesting a live snapshot after explicit expect(result).toMatchObject({status:"failed",result:{result:{message:"ACPX request snapshot requires the surviving provider"}}}); expect(run.launches()).toHaveLength(1); }finally{await run.cleanup();} -},20000); +},60000); diff --git a/packages/paperclip-runner/src/test-support/acpx-blocked-output-host.mjs b/packages/paperclip-runner/src/test-support/acpx-blocked-output-host.mjs new file mode 100644 index 00000000000..d610cd0c673 --- /dev/null +++ b/packages/paperclip-runner/src/test-support/acpx-blocked-output-host.mjs @@ -0,0 +1,67 @@ +// Synthetic provider boundary for the real sidecar's pipe-pressure test. +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { writeFileSync } from "node:fs"; +import { join } from "node:path"; + +export class AcpxRuntimeHost { + static async open(options) { + const host = new AcpxRuntimeHost(options); + await once(host.provider, "message"); + writeFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT, "provider-pid"), String(host.provider.pid)); + return host; + } + + constructor(options) { + this.options = options; + this.closed = false; + this.closeCount = 0; + this.provider = spawn(process.execPath, ["-e", ` + process.on("SIGTERM", () => process.exit(0)); + process.on("disconnect", () => process.exit(0)); + process.send("ready"); + setInterval(() => {}, 1000); + `], { stdio: ["ignore", "ignore", "ignore", "ipc"], env: { PATH: process.env.PATH } }); + this.result = new Promise(resolve => { this.settle = resolve; }); + } + + binding() { + return { normalizedSessionId: this.options.normalizedSessionId, + commandDigest: `sha256:${"a".repeat(64)}`, profileDigest: `sha256:${"b".repeat(64)}`, + workspaceDigest: `sha256:${"c".repeat(64)}`, requestedModel: this.options.model, + effectiveModel: this.options.model, permissionMode: this.options.permissionMode }; + } + + identity() { + const { commandDigest: _command, ...binding } = this.binding(); + return { schema: "paperclip.runner.acpx-identity.v2", ...binding, + acpxRecordId: "blocked-output-record", backendSessionId: "blocked-output-backend", + agentSessionId: "blocked-output-agent", providerLifetimeFenceCandidates: [49152, 49153, 49154] }; + } + + async status() { return {}; } + goalCapability() { return null; } + steeringCapability() { return null; } + + startTurn({ requestId }) { + const host = this; + return { requestId, promptStarted: Promise.resolve(), result: this.result, + events: (async function* () { + for (let index = 0; index < 4096 && !host.closed; index++) { + yield { type: "text_delta", text: `blocked-${index}:${"x".repeat(4096)}` }; + } + })() }; + } + + async close({ reason }) { + this.closeCount++; + this.closed = true; + const bufferedBytes = process.stdout.writableLength; + const exited = once(this.provider, "exit"); + this.provider.kill("SIGTERM"); + const [code, signal] = await exited; + this.settle({ status: "cancelled" }); + writeFileSync(join(process.env.PAPERCLIP_TEST_OUTPUT_ROOT, "provider-cleanup.json"), + JSON.stringify({ reason, bufferedBytes, closeCount: this.closeCount, pid: this.provider.pid, code, signal })); + } +}