From d1ba17eeca65ffaf7cf7c5d8daa448e3331bae17 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 10 Sep 2026 15:37:05 -0700 Subject: [PATCH] fix(adapter-utils): fail fast when the sandbox control channel is lost mid-turn (#13158) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Adapter utilities run agent turns and report their results to the control plane > - A lost sandbox control channel can leave an agent turn without a result > - The host then waits for the full adapter timeout instead of reporting the loss > - This pull request adds a push loss signal and a bounded host wait > - The benefit is a prompt failure terminal when the agent stops answering ## Linked Issues or Issue Description **What happened?** A sandbox control channel loss during an Agent Client Protocol turn left the host waiting for the four-hour adapter execution timeout. **Expected behavior** The host should detect the terminal channel loss, stop the turn, and report a safe failure without waiting for the agent. **Steps to reproduce** 1. Start an Agent Client Protocol turn through a sandbox adapter. 2. Close the duplex control channel while the turn remains active. 3. Observe the host response before the adapter timeout expires. **Paperclip version or commit** Test the pull request commit set at `10b6bbc5525a79fd575298607dd5a25ae448fc8a`. **Deployment mode** The change applies to sandbox-backed adapter execution. ## What Changed - Add `onLoss(listener)` to the duplex bridge handle. - Register the loss listener at turn start and read losses latched before turn start. - Cancel the turn on loss and arm a 30-second host deadline. - Close the stream locally when the deadline wins and create a host terminal. - Derive the public error from the closed `DuplexLossReason` enum. - Add tests for loss order, cancellation, timeout, and safe error output. ## Verification - Run `pnpm --filter @paperclipai/adapter-utils exec tsc --noEmit`. - Run `pnpm --filter @paperclipai/adapter-utils exec vitest run src/acpx-engine/execute.test.ts -t "run-disposition seam"`. - Confirm that the full pull request workflow passes. ## Risks The new deadline changes a lost-channel path from a long wait to a host-built failure after 30 seconds. Orderly completion keeps its existing behavior. The deadline race against a pending `turn.result` has no direct test. ## Model Used OpenAI Codex, GPT-5, with tool use and code execution. The runtime does not expose a more specific deployment version or context-window size. ## 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/acpx-engine/constants.ts | 8 + .../src/acpx-engine/execute.test.ts | 273 ++++++++++++++++++ .../adapter-utils/src/acpx-engine/execute.ts | 232 ++++++++++----- .../adapter-utils/src/execution-target.ts | 30 ++ 4 files changed, 470 insertions(+), 73 deletions(-) diff --git a/packages/adapter-utils/src/acpx-engine/constants.ts b/packages/adapter-utils/src/acpx-engine/constants.ts index bb515b21b7..663d19ec24 100644 --- a/packages/adapter-utils/src/acpx-engine/constants.ts +++ b/packages/adapter-utils/src/acpx-engine/constants.ts @@ -18,6 +18,14 @@ export const ACPX_HANDSHAKE_TIMEOUT_MS = 60_000; // of a channel loss. export const ACPX_HANDSHAKE_TRANSPORT_POLL_MS = 250; +// The bound on how long the host waits, after a latched terminal sandbox +// duplex-channel loss, for the agent to answer the `turn.cancel()` request. +// `cancel()` only asks the agent to end the turn; it does not end the turn by +// itself. An agent that stopped answering never honors it, so this deadline +// is the host-side bound that ends the run without the agent's help. It is +// much smaller than the whole-adapter execution timeout. +export const ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS = 30_000; + export const ACPX_ADAPTER_AGENT_IDS = { claude_local: "claude", codex_local: "codex", diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index ae0ff1aa15..e9c4f0a576 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -6129,6 +6129,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => let lossOrdered = false; let lossReason: string | null = null; let completionOrdered = false; + let lossListener: ((reason: string) => void) | null = null; const readDisposition = () => ({ failed: lossOrdered, lossReason }); const markOrderlyCompletion = vi.fn(() => { if (completionOrdered || lossOrdered) return; @@ -6138,6 +6139,12 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => markOrderlyCompletion(); return readDisposition(); }); + const onLoss = vi.fn((listener: (reason: string) => void) => { + lossListener = listener; + return () => { + if (lossListener === listener) lossListener = null; + }; + }); const stop = vi.fn(async () => {}); const handle = { env: { @@ -6148,19 +6155,25 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => readRunDisposition: () => readDisposition(), settleRunDisposition, markOrderlyCompletion, + onLoss, stop, }; return { handle, markOrderlyCompletion, settleRunDisposition, + onLoss, readDisposition, // Record the first ordered loss. A loss ordered after a completion, or a // second loss, is a no-op — the same rule the real transport applies. + // A loss that latches here (the first ordered call) also pushes the + // reason to the one registered listener, the same way the real HTTP/2 + // transport's disposition latch does. emitLoss: (reason: string) => { if (lossOrdered || completionOrdered) return; lossOrdered = true; lossReason = reason; + lossListener?.(reason); }, }; } @@ -6215,6 +6228,81 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => }; } + // A runtime whose one turn never resolves on its own — it hangs exactly + // like a turn whose sandbox duplex channel died mid-turn produces no + // terminal result. The turn only ends when something calls `cancel()`, the + // same mechanism the push seam calls. `onCancel` observes each call. + function hangingTurnRuntime(onCancel: (reason: string | undefined) => void) { + let release: (() => void) | null = null; + const released = new Promise((resolve) => { + release = resolve; + }); + return { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + await released; + })(), + result: (async () => { + await released; + return { status: "cancelled" as const, stopReason: "cancelled" }; + })(), + cancel: async (input?: { reason?: string }) => { + onCancel(input?.reason); + release?.(); + }, + }), + setConfigOption: async () => {}, + close: async () => {}, + }; + } + + // A runtime whose one turn hangs exactly like the real `acpx` shape: its + // `cancel()` only sends the cancel request and returns. It does NOT settle + // the turn — neither `events` nor `result` ever resolves on its own. + // `closeStream()` ends the event drain locally, with no agent cooperation, + // the same way the real runtime's does; it still leaves `result` pending. + // This is the sensitivity control for the fail-fast deadline: only the + // deadline, not the cancel request, can end this turn. + function unresponsiveCancelTurnRuntime(input: { + onCancel: (reason: string | undefined) => void; + onCloseStream: (reason: string | undefined) => void; + }) { + let endEvents: (() => void) | null = null; + const eventsEnded = new Promise((resolve) => { + endEvents = resolve; + }); + return { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + await eventsEnded; + })(), + // Never settles on its own. The real acpx result settles only when + // the provider process returns or rejects, bounded by the adapter + // execution timeout — not by a `session/cancel` request. + result: new Promise(() => {}), + cancel: async (reasonInput?: { reason?: string }) => { + input.onCancel(reasonInput?.reason); + }, + closeStream: async (reasonInput?: { reason?: string }) => { + input.onCloseStream(reasonInput?.reason); + endEvents?.(); + }, + }), + setConfigOption: async () => {}, + close: async () => {}, + }; + } + async function setupRemoteSandbox() { const root = await makeTempRoot(); const stateDir = path.join(root, "state"); @@ -6238,6 +6326,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => handle: unknown, runtime: unknown, sandbox: Awaited>, + deps: Partial = {}, ) { vi.mocked(startAdapterExecutionTargetPaperclipBridge).mockImplementationOnce( async () => handle as never, @@ -6247,6 +6336,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => ); const execute = createAcpxEngineExecutor({ createRuntime: () => runtime as never, + ...deps, }); return await execute({ runId: "run-duplex-seam", @@ -6342,6 +6432,25 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => expect(fake.readDisposition().failed).toBe(false); }); + it("keeps duplex_channel_lost precedence when the loss latches before a failed terminal", async () => { + const sandbox = await setupRemoteSandbox(); + const fake = createFakeBridgeHandle(); + // Latch the loss before the ACP terminal resolves, and the terminal + // itself also reports a provider failure. + const runtime = runtimeWithFailedResult(() => fake.emitLoss("provider_exit")); + + const result = await runRemote(fake.handle, runtime, sandbox); + + expect(result.exitCode).not.toBe(0); + // The duplex loss reason wins over the provider's own failed terminal. + expect(result.errorCode).toBe("duplex_channel_lost"); + // The message carries only the typed loss reason, not the raw provider + // failure text. + expect(result.errorMessage).toContain("provider_exit"); + expect(result.errorMessage).not.toContain("agent failed"); + expect(result.resultJson).toMatchObject({ status: "failed" }); + }); + it("releases the runtime locally and places no remote close call once the duplex channel is lost", async () => { const sandbox = await setupRemoteSandbox(); const fake = createFakeBridgeHandle(); @@ -6435,6 +6544,170 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => expect(result.errorCode).not.toBe("acpx_session_init_failed"); expect(result.errorCode).not.toBe("acpx_handshake_timeout"); }, 10000); + + it("aborts an in-flight turn and fails the run when the duplex channel latches a loss mid-turn", async () => { + const sandbox = await setupRemoteSandbox(); + const fake = createFakeBridgeHandle(); + const cancelReasons: (string | undefined)[] = []; + // This turn never returns a terminal result on its own: without a push + // seam it would wait for the wall-clock adapter execution timeout. It + // ends only once something calls `cancel()`. + const runtime = hangingTurnRuntime((reason) => cancelReasons.push(reason)); + + const resultPromise = runRemote(fake.handle, runtime, sandbox); + // Wait until the turn registers its loss listener, then latch the loss — + // the same order a real mid-turn channel death follows: the turn starts, + // then later the channel is lost. + await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled()); + fake.emitLoss("provider_exit"); + + const result = await resultPromise; + + // The push seam cancelled the hanging turn instead of waiting for the + // turn to return a terminal result on its own, so the run ends promptly + // instead of waiting for the adapter execution timeout. + expect(cancelReasons).toHaveLength(1); + expect(result.exitCode).not.toBe(0); + expect(result.errorCode).toBe("duplex_channel_lost"); + // The failure message carries only the closed loss-reason enum, never + // raw provider text. + expect(result.errorMessage).toContain("provider_exit"); + expect(result.resultJson).toMatchObject({ status: "failed" }); + }, 5000); + + it("bounds the wait with a deadline when a latched loss cancel gets no agent cooperation", async () => { + const sandbox = await setupRemoteSandbox(); + const fake = createFakeBridgeHandle(); + const cancelReasons: (string | undefined)[] = []; + const closeStreamReasons: (string | undefined)[] = []; + // The real acpx shape: `cancel()` only requests cancellation and returns. + // It settles neither `events` nor `result`. Only the fail-fast deadline + // can end this turn. + const runtime = unresponsiveCancelTurnRuntime({ + onCancel: (reason) => cancelReasons.push(reason), + onCloseStream: (reason) => closeStreamReasons.push(reason), + }); + + const resultPromise = runRemote(fake.handle, runtime, sandbox, { + // Small and fake-time-free: real-timer test, so the deadline must stay + // short enough to run fast without waiting 60 real seconds. + duplexLossCancelDeadlineMs: 25, + }); + await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled()); + fake.emitLoss("provider_exit"); + + const result = await resultPromise; + + // The seam still tried the cooperative cancel first. + expect(cancelReasons).toHaveLength(1); + // The agent never answered the cancel, so the deadline ended the event + // drain locally instead of waiting for it. + expect(closeStreamReasons).toHaveLength(1); + // The run reached a failure terminal within the deadline, even though + // neither `events` nor `result` ever settled on their own. + expect(result.exitCode).not.toBe(0); + expect(result.errorCode).toBe("duplex_channel_lost"); + // The failure message carries only the closed loss-reason enum, never + // raw provider text. + expect(result.errorMessage).toContain("provider_exit"); + expect(result.resultJson).toMatchObject({ status: "failed" }); + }, 5000); + + it("awaits stream closure and the event drain before finalizing a duplex loss deadline", async () => { + const sandbox = await setupRemoteSandbox(); + const fake = createFakeBridgeHandle(); + let endEvents: (() => void) | null = null; + const eventsEnded = new Promise((resolve) => { + endEvents = resolve; + }); + let releaseCloseStream!: () => void; + const closeStreamGate = new Promise((resolve) => { + releaseCloseStream = resolve; + }); + let closeStreamCalls = 0; + // `closeStream()` stays pending on a gate the test controls, and only + // ends the event drain once the test releases that gate. If the run + // finalizes before the gate opens, the seam did not wait for the close + // call, so a late event on this drain could still land after the result. + const runtime = { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + await eventsEnded; + })(), + result: new Promise(() => {}), + cancel: async () => {}, + closeStream: async () => { + closeStreamCalls += 1; + await closeStreamGate; + endEvents?.(); + }, + }), + setConfigOption: async () => {}, + close: async () => {}, + }; + + const resultPromise = runRemote(fake.handle, runtime, sandbox, { + duplexLossCancelDeadlineMs: 25, + }); + await vi.waitFor(() => expect(fake.onLoss).toHaveBeenCalled()); + fake.emitLoss("provider_exit"); + + await vi.waitFor(() => expect(closeStreamCalls).toBe(1)); + // The close call has not resolved yet, so the run must still be pending. + let settled = false; + void resultPromise.then(() => { + settled = true; + }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(settled).toBe(false); + + releaseCloseStream(); + const result = await resultPromise; + + expect(result.errorCode).toBe("duplex_channel_lost"); + }, 5000); + + it("does not abort or fail an already-completed run when the duplex channel loses after an orderly completion", async () => { + const sandbox = await setupRemoteSandbox(); + const fake = createFakeBridgeHandle(); + const cancelReasons: (string | undefined)[] = []; + const runtime = { + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed" as const, stopReason: "end_turn" }), + cancel: async (input?: { reason?: string }) => { + cancelReasons.push(input?.reason); + }, + }), + setConfigOption: async () => {}, + close: async () => {}, + }; + + const result = await runRemote(fake.handle, runtime, sandbox); + expect(result.exitCode).toBe(0); + expect(result.errorCode ?? null).toBeNull(); + + // The channel dies only after the turn already completed cleanly. The + // loss listener the turn registered is still live at this point, but the + // latch already marked the orderly completion, so the loss cannot relatch + // and must never reach a cancel call on the (already-finished) turn. + fake.emitLoss("provider_exit"); + + expect(cancelReasons).toHaveLength(0); + expect(fake.readDisposition().failed).toBe(false); + }); }); describe("ACPX startup handshake guard and late-completion fence", () => { diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index ab615987f0..3f537fedd1 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -92,6 +92,7 @@ import { type AcpSessionStore, } from "acpx/runtime"; import { + ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS, ACPX_HANDSHAKE_TIMEOUT_MS, ACPX_HANDSHAKE_TRANSPORT_POLL_MS, DEFAULT_ACP_ENGINE_AGENT, @@ -342,6 +343,14 @@ export interface AcpxRemoteManagedHomeResult { export interface AcpxEngineExecutorOptions { createRuntime?: AcpxRuntimeFactory; now?: () => number; + /** + * The bound on how long the fail-fast seam waits for a cooperative + * `turn.cancel()` after a latched terminal sandbox duplex-channel loss, + * before it ends the turn without the agent's help. Defaults to + * {@link ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS}. Tests inject a small value + * to drive the deadline without real time. + */ + duplexLossCancelDeadlineMs?: number; warmHandles?: Map; /** * Per-session staged-runtime cache for the remote runner-backed lane (PR 3). @@ -3704,6 +3713,7 @@ function openTurnSpan( export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const createRuntime = deps.createRuntime ?? createAcpRuntime; const now = deps.now ?? (() => Date.now()); + const duplexLossCancelDeadlineMs = deps.duplexLossCancelDeadlineMs ?? ACPX_DUPLEX_LOSS_CANCEL_DEADLINE_MS; const warmHandles = deps.warmHandles ?? defaultWarmHandles; const stagedRuntimes = deps.stagedRuntimes ?? defaultStagedRuntimes; const stagingLocks = deps.stagingLocks ?? defaultStagingLocks; @@ -3800,6 +3810,24 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { let releaseStagingLease: (() => void) | null = null; let stopTimer: ReturnType | undefined; let removeStopListener: (() => void) | undefined; + // Unregisters the sandbox duplex bridge's loss listener (below, in + // `stepTurnStart`). Set only on a sandbox target whose bridge exposes + // `onLoss`; stays undefined everywhere else, so the cleanup call is a + // no-op there. + let removeLossListener: (() => void) | undefined; + // Bounds the wait after a latched terminal duplex loss so a silent agent + // cannot hold the run open on the cooperative `turn.cancel()` request + // alone. `stepTurnStart` arms `lossDeadlineTimer` the moment a loss + // latches; it stays undefined everywhere else, so the cleanup call below + // is a no-op there. `stepEventRelay` races the turn against + // `lossDeadline` and, once it fires, ends the event drain and hands + // `turnFinalize` a host-built terminal instead of the agent's. + let lossDeadlineTimer: ReturnType | undefined; + let lossDeadlineTripped = false; + let resolveLossDeadline: (() => void) | undefined; + const lossDeadline = new Promise((resolve) => { + resolveLossDeadline = resolve; + }); let forcedStop = false; let runtimeStopConfirmed = false; let safeInterruptedSession = false; @@ -4606,6 +4634,36 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { signal, }); activeTurn = turn; + // A latched sandbox duplex-channel loss otherwise has no way to reach + // this turn: the bridge only exposes a pull read, and the engine + // pulls it at the terminal-finalization boundary, which runs only + // after the turn already returned a terminal result. A channel that + // dies mid-turn then leaves the turn with no terminal result to + // return, so it waits for the wall-clock adapter execution timeout + // instead of failing fast. Cancel the turn the moment a terminal loss + // latches — whether it latches from here on, or already latched + // before this turn started — so the turn returns a terminal result + // right away. `turnFinalize` reads the same latch and builds the + // failure from the typed loss reason alone. + const bridge = prepared.paperclipBridge; + if (bridge?.onLoss) { + const cancelForLoss = (reason: DuplexLossReason) => { + void turn.cancel({ reason: `paperclip sandbox duplex channel lost (${reason})` }).catch(() => {}); + // `cancel()` only asks the agent to end the turn; it does not end + // the turn by itself. Start the fail-fast deadline the moment the + // loss latches, so the run does not wait past this bound for an + // agent that stopped answering. + if (!lossDeadlineTimer && !lossDeadlineTripped) { + lossDeadlineTimer = setTimeout(() => { + lossDeadlineTripped = true; + resolveLossDeadline?.(); + }, duplexLossCancelDeadlineMs); + } + }; + removeLossListener = bridge.onLoss(cancelForLoss); + const alreadyLatched = bridge.readRunDisposition?.(); + if (alreadyLatched?.failed) cancelForLoss(alreadyLatched.lossReason ?? "other"); + } // ACP can resolve the turn before its provider exits. Keep the Stop // deadline armed through settlement, including provider cleanup. const armStopDeadline = () => { @@ -4631,40 +4689,74 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }, }; }; + // The host-built terminal `stepEventRelay` hands to `turnFinalize` once + // the fail-fast deadline fires with no agent-supplied terminal. Its + // `status` mirrors the shape a real cooperative cancel already + // produces, so `turnFinalize` needs no change: it reads the latched + // loss disposition, not this `stopReason`, to build the reported + // failure and its message. + const LOSS_DEADLINE_TERMINAL: AcpRuntimeTurnResult = { + status: "cancelled", + stopReason: "paperclip_duplex_loss_deadline", + }; const stepEventRelay = async (): Promise => { const turn = activeTurn as AcpRuntimeTurn; const toolTitles = new Map(); - for await (const event of turn.events) { - // ACPX currently flattens client-side filesystem/terminal receipts - // into status text. They cannot establish complete action outcomes. - if (event.type === "status" && /^(fs|terminal)\//.test(event.text)) incompleteToolInventory = true; - if (event.type === "tool_call") { - if (!event.toolCallId) incompleteToolInventory = true; - else { - const previous = interruptionTools.get(event.toolCallId); - interruptionTools.set(event.toolCallId, { - kind: event.kind ?? previous?.kind, - status: event.status ?? previous?.status, - }); + const drainEvents = (async (): Promise => { + for await (const event of turn.events) { + // ACPX currently flattens client-side filesystem/terminal receipts + // into status text. They cannot establish complete action outcomes. + if (event.type === "status" && /^(fs|terminal)\//.test(event.text)) incompleteToolInventory = true; + if (event.type === "tool_call") { + if (!event.toolCallId) incompleteToolInventory = true; + else { + const previous = interruptionTools.get(event.toolCallId); + interruptionTools.set(event.toolCallId, { + kind: event.kind ?? previous?.kind, + status: event.status ?? previous?.status, + }); + } } + if (event.type === "text_delta" && event.stream !== "thought") { + currentOutputChunk.push(event.text); + } else if (event.type === "tool_call" && event.tag !== "tool_call_update") { + // ACP makes tool-call status optional. The normalized event tag is + // the reliable boundary between an initial call and its updates, + // so a statusless initial call must still end the preceding output + // segment while updates must not create extra boundaries. + flushOutputSegment(); + } + if (event.type === "status" && event.tag === "usage_update") { + eventBreakdown = event.breakdown ?? eventBreakdown; + eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd; + } + await emitRuntimeEvent(ctx, event, toolTitles, prepared.coalescePlaceholderToolUpdates); } - if (event.type === "text_delta" && event.stream !== "thought") { - currentOutputChunk.push(event.text); - } else if (event.type === "tool_call" && event.tag !== "tool_call_update") { - // ACP makes tool-call status optional. The normalized event tag is - // the reliable boundary between an initial call and its updates, - // so a statusless initial call must still end the preceding output - // segment while updates must not create extra boundaries. - flushOutputSegment(); - } - if (event.type === "status" && event.tag === "usage_update") { - eventBreakdown = event.breakdown ?? eventBreakdown; - eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd; - } - await emitRuntimeEvent(ctx, event, toolTitles, prepared.coalescePlaceholderToolUpdates); + })(); + // A latched loss already asked the agent to cancel (above, in + // `cancelForLoss`); that request settles neither `turn.events` nor + // `turn.result` by itself. Race the event drain against the fail-fast + // deadline so a silent agent cannot hold this wait open. + const eventsEnded = await Promise.race([ + drainEvents.then(() => true as const), + lossDeadline.then(() => false as const), + ]); + if (!eventsEnded) { + // The deadline won: stop waiting on the agent. `closeStream` ends + // the event drain locally, with no agent cooperation required. Await + // both the close call and the drain it unblocks before this step + // returns, so no late runtime event can still mutate shared state + // (output segments, tool inventory) after finalization reads it. + await turn.closeStream({ reason: "paperclip duplex loss cancel deadline" }).catch(() => {}); + await drainEvents.catch(() => {}); + flushOutputSegment(); + return LOSS_DEADLINE_TERMINAL; } flushOutputSegment(); - return await turn.result; + // `turn.result` settles only when the agent's provider process + // returns or rejects; a latched loss that armed the deadline after + // the event drain already ended must still bound this wait. + return await Promise.race([turn.result, lossDeadline.then(() => LOSS_DEADLINE_TERMINAL)]); }; const stepTurnFinalize = async ( input: TurnFinalizeInput, @@ -4673,33 +4765,23 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const terminal = input.terminal; const timedOut = input.timedOut; // Read the sandbox duplex control-channel disposition at the ACP - // terminal-finalization boundary, before the bridge teardown. A control - // channel that died mid-turn latches a failure with a typed loss reason; - // a healthy channel or a normal-teardown loss reports a success. Only a - // nominally completed, non-timed-out terminal is success-eligible, so the - // seam reads the disposition only there. For that success-eligible - // terminal the seam marks the host-observed orderly completion, so a later - // teardown loss cannot flip the run to a failure. The file bridge path - // never sets these methods, so the optional calls no-op there. + // terminal-finalization boundary, before the bridge teardown, on every + // terminal outcome. A control channel that died before this point + // latches a failure with a typed loss reason; a healthy channel or a + // normal-teardown loss reports a success. The read and the mark of the + // host-observed orderly completion happen atomically in one broker + // step, with no `await` between them, so a teardown loss cannot slip + // in between. This stops a later teardown `channel_exit` from latching + // a false loss. The mark no-ops once a loss already latched, so a real + // mid-turn loss still fails the run — including a loss that arrived + // through the in-flight-turn cancel this seam issues, which surfaces + // here as a `cancelled` (not `completed`) terminal, not just through a + // nominally completed terminal. The file bridge path never sets this + // method, so the optional call no-ops there. let duplexLossReason: DuplexLossReason | null = null; - if (terminal.status === "completed" && !timedOut) { - // Success-eligible terminal. Atomically read the disposition and mark - // the orderly completion in one broker step. No `await` separates the - // read from the mark, so a teardown loss cannot slip in between them. A - // latched loss fails the run closed; a healthy channel marks its - // orderly completion, so a later teardown loss stays a normal teardown. - const disposition = prepared.paperclipBridge?.settleRunDisposition?.() ?? null; - if (disposition?.failed) { - duplexLossReason = disposition.lossReason ?? "other"; - } - } else { - // Non-success-eligible terminal (failed, cancelled, or timed out). A - // deliberate host teardown follows, so mark the orderly completion now. - // This stops the teardown `channel_exit` from latching `lossSeq`, from - // emitting a false loss event, and from incrementing the loss counters. - // The mark no-ops once a loss latched, so a real mid-run loss still - // fails the run. - prepared.paperclipBridge?.markOrderlyCompletion?.(); + const disposition = prepared.paperclipBridge?.settleRunDisposition?.() ?? null; + if (disposition?.failed) { + duplexLossReason = disposition.lossReason ?? "other"; } // A terminal that reports "completed" but whose duplex control channel // died before the completion is not a success. The seam fails it closed. @@ -4778,12 +4860,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { signal: timedOut ? "SIGTERM" : null, timedOut, errorMessage, - errorCode: terminal.status === "failed" - ? "acpx_turn_failed" - : timedOut - ? "acpx_timeout" - : channelLost - ? DUPLEX_CHANNEL_LOST_ERROR_CODE + errorCode: timedOut + ? "acpx_timeout" + : channelLost + ? DUPLEX_CHANNEL_LOST_ERROR_CODE + : terminal.status === "failed" + ? "acpx_turn_failed" : null, sessionId: sessionHandle.backendSessionId ?? sessionHandle.runtimeSessionName, sessionParams: buildSessionParams({ prepared, handle: sessionHandle }), @@ -4831,16 +4913,6 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { resources: emptyConsumed, }; } - if (terminal.status === "failed") { - return { - kind: "failed", - cause: { - kind: "turn_failed", - error: terminal.error instanceof Error ? terminal.error : new Error(String(terminal.error)), - }, - resources: emptyConsumed, - }; - } if (terminal.status === "cancelled") { return { kind: "cancelled", @@ -4848,10 +4920,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { resources: emptyConsumed, }; } - // A completed terminal whose duplex control channel died mid-turn returns - // a failed completion, so the coordinator settles for a failure and the - // reuse decision forbids a save. The message carries only the typed loss - // reason, so no raw provider text rides the cause. + // A duplex control-channel loss outranks a provider-reported failure or + // completion: the loss reason explains why the provider terminal reads + // the way it does, not the other way round. This also covers a + // "completed" terminal whose channel died mid-turn. The message carries + // only the typed loss reason, so no raw provider text rides the cause, + // even when the provider terminal itself reports `failed`. if (channelLost) { return { kind: "failed", @@ -4862,6 +4936,16 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { resources: emptyConsumed, }; } + if (terminal.status === "failed") { + return { + kind: "failed", + cause: { + kind: "turn_failed", + error: terminal.error instanceof Error ? terminal.error : new Error(String(terminal.error)), + }, + resources: emptyConsumed, + }; + } return { kind: "finalized" }; } const err = input.error; @@ -5195,6 +5279,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { } finally { clearTimeout(stopTimer); removeStopListener?.(); + removeLossListener?.(); + clearTimeout(lossDeadlineTimer); // End the run root span exactly once, on every return and on a throw. runRootSpan.end(runFailed); // Release the per-session staging lease as the run's final act, AFTER the diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 06b8ff781e..4ed16aa0de 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -344,6 +344,20 @@ export interface AdapterExecutionTargetPaperclipBridgeHandle { * bridge path never sets it, so the method is absent there. */ markOrderlyCompletion?(): void; + /** + * Register a listener for a newly latched terminal loss. The listener + * fires at most once, and only for a loss that flips the disposition to + * failed — never for a clean channel end that orders after a + * host-observed orderly completion. Returns a function that unregisters + * the listener. + * + * The caller uses this to abort an in-flight Agent Client Protocol turn + * the moment the channel dies, instead of waiting for the turn to return + * a terminal result on its own (a dead channel can leave a turn with + * nothing to return). The file bridge path never sets it, so the method + * is absent there. + */ + onLoss?(listener: (reason: DuplexLossReason) => void): () => void; stop(): Promise; } @@ -3661,12 +3675,19 @@ interface Http2RunDispositionLatch { markOrderlyCompletion(): void; /** Atomically mark the orderly completion and read the disposition. */ settleRunDisposition(): DuplexBrokerRunDisposition; + /** + * Register a listener that fires once, only on the call to `recordLoss` + * that actually latches a new terminal loss. Returns a function that + * unregisters the listener. + */ + onLoss(listener: (reason: DuplexLossReason) => void): () => void; } function createHttp2RunDispositionLatch(): Http2RunDispositionLatch { let lossOrdered = false; let lossReason: DuplexLossReason | null = null; let completionOrdered = false; + let lossListener: ((reason: DuplexLossReason) => void) | null = null; const markOrderlyCompletion = (): void => { if (completionOrdered || lossOrdered) return; completionOrdered = true; @@ -3679,6 +3700,7 @@ function createHttp2RunDispositionLatch(): Http2RunDispositionLatch { if (lossOrdered || completionOrdered) return false; lossOrdered = true; lossReason = reason; + lossListener?.(reason); return true; }, markOrderlyCompletion, @@ -3686,6 +3708,12 @@ function createHttp2RunDispositionLatch(): Http2RunDispositionLatch { markOrderlyCompletion(); return { failed: lossOrdered, lossReason }; }, + onLoss(listener: (reason: DuplexLossReason) => void): () => void { + lossListener = listener; + return () => { + if (lossListener === listener) lossListener = null; + }; + }, }; } @@ -4672,6 +4700,8 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: { // the mark and a teardown loss cannot slip in between. settleRunDisposition: (): DuplexBrokerRunDisposition => dispositionLatch.settleRunDisposition(), markOrderlyCompletion: (): void => dispositionLatch.markOrderlyCompletion(), + onLoss: (listener: (reason: DuplexLossReason) => void): (() => void) => + dispositionLatch.onLoss(listener), stop: async () => { // Close the HTTP/2 server's sessions, then the channel, before // lease release, so no live provider session remains when the