diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 82a056efbff..c772fb40098 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1352,6 +1352,52 @@ it.each([false, true])("refuses a terminal call decided before earlier results w } }, 20_000); +it("refuses a terminal call that arrives before a held call's result (restart chaos t05)", async () => { + // Restart chaos t05: Codex called write_document during a K4 hold, then + // re-sampled until it called paperclip_block. After the reattach the + // controller recovered the held write while the Runner delivered the block, + // and the block arrived a few milliseconds before the write's result. + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-held-terminal-")); + const OriginalCore = durableControlPlane.DurablePrpControlPlane; + let callbacks!: ConstructorParameters[0]; + let core!: DurablePrpControlPlane; + const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane").mockImplementation(function (options) { + callbacks = options; + return core = new OriginalCore(options); + }); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, + }); + const handler = vi.fn(async () => ({ success: true, contentItems: [{ type: "inputText" as const, text: "{\"blocked\":true}" }] })); + bundle.transport.setServerRequestHandler(handler); + try { + await bundle.transport.request("thread/start", { cwd: tmpdir() }); + const identity = core.store.state.identity; + let seq = 9100; + const commit = (eventType: string, payload: Record) => { + seq += 1; + core.store.state.committedEvents.push({ sourceSeq: seq, sourceEventId: `held-terminal-${seq}`, eventType, priority: 1, + envelope: { payload: { eventType, turnId: identity.turnId, payload } }, deliveryCount: 1, logicalEffectCount: 1 }); + return `held-terminal-${seq}`; + }; + commit("tool.execution.started", { executionId: "exec-held-write", transport: "dynamic", name: "write_document", status: "running" }); + const sourceEventId = commit("tool.execution.started", { executionId: "exec-held-block", transport: "dynamic", name: "paperclip_block", status: "running" }); + // The recovered write returns just after the block reaches the transport. + setTimeout(() => core.queueCommand("semantic_tool.result", { callId: "exec-held-write", operationId: "write_document", + inputDigest: "sha256:write", result: { created: false }, isError: false }, "command_tool_held_terminal_write", true), 100); + const outcome = await callbacks.onSemanticToolInput!({ callId: "held-block-call", operationId: "paperclip_block", + input: { summary: "Paperclip has not confirmed the save" }, sourceEventId, sourceEventType: "semantic_tool.input", correlation: identity }); + expect(handler).not.toHaveBeenCalled(); + expect(outcome).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results", retryable: true } } }); + expect(JSON.stringify(outcome.result)).toContain("Those calls have now returned"); + } finally { + coreSpy.mockRestore(); + await bundle.transport.close().catch(() => undefined); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 20_000); + it.each(["after_budget", "within_budget", "interrupted_within_budget", "persistence_failure"] as const)( "fences reusable suspension against late semantic completion (%s)", async (mode) => { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index d657075010b..493b5e8cf91 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -105,7 +105,7 @@ import { import { resolvePackagedRunnerBinary } from "./runner-binary.js"; import { RUNNERD_CANONICAL_ITEM, RUNNERD_RUNTIME_ACTIVITY } from "../drivers/codex/codex-driver-values.js"; import { crashPoint } from "../crash-point.js"; -import { isTerminalOperation, staleTerminalCallResult, unseenReturnedToolCalls } from "./terminal-call-freshness.js"; +import { isTerminalOperation, settleEarlierToolCalls, staleTerminalCallResult } from "./terminal-call-freshness.js"; // URL directory conversion preserves a trailing separator while path-derived // build artifacts do not. Normalize once so a source build cannot be @@ -7125,9 +7125,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // had returned but not yet reached the agent (it was held across a // controller restart) is stale: refuse it so the agent continues with // those results instead of ending the task on a wait that is over. - const unseen = unseenReturnedToolCalls(core.store.state.committedEvents, core.store.state.commands, - { terminalSourceEventId: call.sourceEventId, turnId: call.correlation.turnId }); - if (unseen.length) return staleTerminalCallResult(call.operationId, unseen); + // After a reattach the held calls are recovered while this call is + // delivered, so wait for earlier calls that have not returned yet + // (restart chaos t05) before judging what the agent had seen. + const settled = await settleEarlierToolCalls( + () => this.#closed || core !== this.#core ? null : { events: core.store.state.committedEvents, commands: core.store.state.commands }, + { terminalSourceEventId: call.sourceEventId, turnId: call.correlation.turnId }, + ); + if (!settled || this.#closed || core !== this.#core || epoch !== this.#turnStartResponseEpoch || threadId !== this.#threadId) { + throw new SemanticToolNotDispatchedError(); + } + if (settled.unseen.length || settled.pending.length) { + return staleTerminalCallResult(call.operationId, settled.unseen, settled.pending); + } } const outcome = unwrapToolResponse( await this.#handler({ diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts index cb2b4ce26de..6fe1cb72eda 100644 --- a/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { isTerminalOperation, staleTerminalCallResult, unseenReturnedToolCalls } from "./terminal-call-freshness.js"; +import { isTerminalOperation, pendingEarlierToolCalls, settleEarlierToolCalls, staleTerminalCallResult, unseenReturnedToolCalls } from "./terminal-call-freshness.js"; let seq = 0; const event = (eventType: string, payload: Record, turnId = "turn-1") => ({ @@ -25,10 +25,11 @@ describe("unseenReturnedToolCalls", () => { expect(unseenReturnedToolCalls(events, [], { terminalSourceEventId: block.sourceEventId })).toEqual([]); }); - it("leaves a long-running call with no result to the agent's judgement", () => { + it("reports a call with no result yet as pending, not as returned", () => { const block = started("b", "paperclip_block"); - expect(unseenReturnedToolCalls([started("w", "write_document"), block], [], { terminalSourceEventId: block.sourceEventId })) - .toEqual([]); + const events = [started("w", "write_document"), block]; + expect(unseenReturnedToolCalls(events, [], { terminalSourceEventId: block.sourceEventId })).toEqual([]); + expect(pendingEarlierToolCalls(events, [], { terminalSourceEventId: block.sourceEventId })).toEqual(["write_document"]); }); it("ignores other turns, process tools and the terminal calls themselves", () => { @@ -68,3 +69,36 @@ describe("staleTerminalCallResult", () => { expect(outcome.result.error.message).toContain("write_document"); }); }); + +describe("settleEarlierToolCalls", () => { + it("waits for a held call that returns after the terminal call arrived (restart chaos t05)", async () => { + const block = started("b", "paperclip_block"); + const events = [started("w", "write_document"), block]; + const commands: { type: string; payload: Record }[] = []; + setTimeout(() => commands.push({ type: "semantic_tool.result", payload: { callId: "w", operationId: "write_document" } }), 30); + const settled = await settleEarlierToolCalls(() => ({ events, commands }), + { terminalSourceEventId: block.sourceEventId }, { timeoutMs: 5_000, pollMs: 5 }); + expect(settled).toEqual({ unseen: ["write_document"], pending: [] }); + }); + + it("reports a call still pending after the bound", async () => { + const block = started("b", "paperclip_block"); + const events = [started("w", "write_document"), block]; + const settled = await settleEarlierToolCalls(() => ({ events, commands: [] }), + { terminalSourceEventId: block.sourceEventId }, { timeoutMs: 20, pollMs: 5 }); + expect(settled).toEqual({ unseen: [], pending: ["write_document"] }); + const outcome = staleTerminalCallResult("paperclip_block", settled!.unseen, settled!.pending); + expect(outcome).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results", retryable: true } } }); + expect(outcome.result.error.message).toContain("has not returned yet"); + }); + + it("does not wait when every earlier call was seen, and stops when the session is gone", async () => { + const block = started("b", "paperclip_block"); + const events = [started("w", "write_document"), result("w"), completed("w", "write_document"), block]; + const startedAt = Date.now(); + expect(await settleEarlierToolCalls(() => ({ events, commands: [] }), { terminalSourceEventId: block.sourceEventId })) + .toEqual({ unseen: [], pending: [] }); + expect(Date.now() - startedAt).toBeLessThan(1_000); + expect(await settleEarlierToolCalls(() => null, {})).toBeNull(); + }); +}); diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.ts index ef19cc97724..0d8ce22cf9d 100644 --- a/packages/paperclip-runner/src/live/terminal-call-freshness.ts +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.ts @@ -19,8 +19,15 @@ * which some drivers commit, counts the same); * - committed `tool.execution.completed` for that call: the provider received it. * A call with a result but no provider completion before the terminal call was - * still pending from the agent's point of view. A call with no result yet (a - * long-running tool) does not count: the agent's decision is still its own. + * still pending from the agent's point of view. + * + * The backlog is not strictly ordered: after a reattach the controller + * recovers the held calls while the Runner delivers the agent's later + * terminal call, so the terminal call can arrive before the held calls have + * returned (restart chaos t05). A terminal call therefore first waits, for + * a bounded time, for earlier calls of its turn that have not returned. If + * any is still pending after the wait the terminal call is refused too: the + * agent decided without it, and its outcome is not known yet. */ const TERMINAL_OPERATIONS = new Set(["paperclip_finish", "paperclip_block"]); @@ -31,15 +38,45 @@ type Command = { type: string; payload: Record }; const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +type TerminalCallScope = { terminalSourceEventId?: string | null; turnId?: string | null }; +type ToolCallState = { name: string; returned: boolean; seen: boolean }; + /** Names of calls that returned before the terminal call but that the agent had not yet seen. */ export function unseenReturnedToolCalls( events: readonly CommittedEvent[], commands: readonly Command[], - input: { terminalSourceEventId?: string | null; turnId?: string | null }, + input: TerminalCallScope, ): string[] { + return callNames(earlierToolCalls(events, commands, input), call => call.returned && !call.seen); +} + +/** + * Names of calls made before the terminal call that have neither returned nor + * reached the agent. After a reattach the Runner hands the controller its + * held calls and the agent's later terminal call together, so these are + * usually about to return (restart chaos t05: a held write_document returned + * a few milliseconds after the paperclip_block decided without it). + */ +export function pendingEarlierToolCalls( + events: readonly CommittedEvent[], + commands: readonly Command[], + input: TerminalCallScope, +): string[] { + return callNames(earlierToolCalls(events, commands, input), call => !call.returned && !call.seen); +} + +function callNames(calls: Map, keep: (call: ToolCallState) => boolean): string[] { + return [...new Set([...calls.values()].filter(call => keep(call) && !TERMINAL_OPERATIONS.has(call.name)).map(call => call.name))]; +} + +function earlierToolCalls( + events: readonly CommittedEvent[], + commands: readonly Command[], + input: TerminalCallScope, +): Map { const returnedCallIds = new Set(commands.flatMap(command => command.type === "semantic_tool.result" && typeof command.payload.callId === "string" ? [command.payload.callId] : [])); - const calls = new Map(); + const calls = new Map(); for (const saved of events) { if (input.terminalSourceEventId && saved.sourceEventId === input.terminalSourceEventId) break; const event = record(saved.envelope.payload); @@ -66,25 +103,54 @@ export function unseenReturnedToolCalls( } } for (const [id, call] of calls) if (returnedCallIds.has(id)) call.returned = true; - return [...new Set([...calls.values()] - .filter(call => call.returned && !call.seen && !TERMINAL_OPERATIONS.has(call.name)) - .map(call => call.name))]; + return calls; +} + +/** How long a terminal call waits for earlier calls of its turn to return. */ +export const TERMINAL_CALL_SETTLE_TIMEOUT_MS = 30_000; + +/** + * Waits (bounded) until no call made before the terminal call is still + * pending, then reports which earlier calls returned unseen and which are + * still pending. `read` returns null once the session is gone. + */ +export async function settleEarlierToolCalls( + read: () => { events: readonly CommittedEvent[]; commands: readonly Command[] } | null, + input: TerminalCallScope, + options: { timeoutMs?: number; pollMs?: number } = {}, +): Promise<{ unseen: string[]; pending: string[] } | null> { + const deadline = Date.now() + (options.timeoutMs ?? TERMINAL_CALL_SETTLE_TIMEOUT_MS); + for (;;) { + const state = read(); + if (!state) return null; + const pending = pendingEarlierToolCalls(state.events, state.commands, input); + if (pending.length === 0 || Date.now() >= deadline) { + return { unseen: unseenReturnedToolCalls(state.events, state.commands, input), pending }; + } + await new Promise(resolve => setTimeout(resolve, options.pollMs ?? 50)); + } } export function isTerminalOperation(operationId: string): boolean { return TERMINAL_OPERATIONS.has(operationId); } -/** The retryable answer to a terminal call decided before the agent saw `unseen`. */ -export function staleTerminalCallResult(operationId: string, unseen: readonly string[]) { - const names = unseen.join(", "); +/** + * The retryable answer to a terminal call decided before the agent saw the + * results of `unseen` (returned) or `pending` (not returned yet). + */ +export function staleTerminalCallResult(operationId: string, unseen: readonly string[], pending: readonly string[] = []) { + const names = [...unseen, ...pending].join(", "); + const now = pending.length === 0 + ? "Those calls have now returned. Read their results and continue the task; " + : `${pending.join(", ")} has not returned yet. Wait for its result (do not repeat a write until it returns), then continue the task; `; return { result: { error: { code: "terminal_call_before_results", retryable: true, message: `Paperclip did not record ${operationId}. It was decided while earlier Paperclip calls (${names}) had not returned yet: ` - + "Paperclip was briefly unreachable. Those calls have now returned. Read their results and continue the task; " + + "Paperclip was briefly unreachable. " + now + `call ${operationId} again only if it is still the right outcome.`, }, },