diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index 60dd9a190ff..eaa91d2006e 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -4374,6 +4374,7 @@ export class DurablePrpControlPlane { recordFailure("persist_result", error); } }; + crashPoint("runner.semantic_tool.after_input_before_dispatch"); const outcome = this.#onSemanticToolInput(call); dispatched = true; void outcome diff --git a/packages/paperclip-runner/src/crash-point.ts b/packages/paperclip-runner/src/crash-point.ts index 97117dcb507..f3ca07cfe41 100644 --- a/packages/paperclip-runner/src/crash-point.ts +++ b/packages/paperclip-runner/src/crash-point.ts @@ -13,7 +13,8 @@ import { appendFileSync, existsSync, readFileSync } from "node:fs"; export type RunnerCrashPointName = | "runner.process.after_spawn_before_identity" | "runner.turn_start.after_runner_accept_before_controller_checkpoint" - | "runner.turn_start.after_controller_journal_before_db_checkpoint"; + | "runner.turn_start.after_controller_journal_before_db_checkpoint" + | "runner.semantic_tool.after_input_before_dispatch"; type Armed = { name: string; nth: number; hits: number; receiptPath: string | null; gatePath: string | null; 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 accfce00ee9..6ee022d89e9 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1283,6 +1283,62 @@ it.each([false, true])("uses the same API result envelope and fences handler pub } }, 20_000); +it.each([false, true])("refuses a terminal call decided before earlier results were seen (seen: %s)", async seen => { + // Restart chaos t03: Codex called write_document, the controller was held, + // and Codex re-sampled the model until it called paperclip_block. After the + // reattach the Runner delivered the backlog in order: write_document ran and + // returned, then the stale block arrived before Codex had seen that result. + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-stale-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 = 9000; + const commit = (eventType: string, payload: Record) => { + seq += 1; + core.store.state.committedEvents.push({ sourceSeq: seq, sourceEventId: `stale-terminal-${seq}`, eventType, priority: 1, + envelope: { payload: { eventType, turnId: identity.turnId, payload } }, deliveryCount: 1, logicalEffectCount: 1 }); + return `stale-terminal-${seq}`; + }; + const write = { executionId: "exec-write", transport: "dynamic", name: "write_document" }; + commit("tool.execution.started", { ...write, status: "running" }); + // Paperclip's result, recorded as the controller records it for the Runner. + core.queueCommand("semantic_tool.result", { callId: "exec-write", operationId: "write_document", + inputDigest: "sha256:write", result: { created: true }, isError: false }, "command_tool_stale_terminal_write", true); + if (seen) commit("tool.execution.completed", { ...write, status: "completed" }); + // The terminal call's own event bounds the history (in production its + // semantic_tool.input event; any later event works the same way). + const sourceEventId = commit("tool.execution.started", { executionId: "exec-block", transport: "dynamic", name: "paperclip_block", status: "running" }); + const blockInput = { summary: "Paperclip has not returned the save" }; + const outcome = await callbacks.onSemanticToolInput!({ callId: "block-call", operationId: "paperclip_block", + input: blockInput, sourceEventId, sourceEventType: "semantic_tool.input", correlation: identity }); + if (seen) { + expect(handler).toHaveBeenCalledTimes(1); + expect(outcome.isError).not.toBe(true); + } else { + expect(handler).not.toHaveBeenCalled(); + expect(outcome).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results", retryable: true } } }); + expect(JSON.stringify(outcome.result)).toContain("write_document"); + } + } 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 3b261b0e65d..d8ac94c8ef6 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -105,6 +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"; // URL directory conversion preserves a trailing separator while path-derived // build artifacts do not. Normalize once so a source build cannot be @@ -7109,6 +7110,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ) { throw new SemanticToolNotDispatchedError(); } + if (isTerminalOperation(call.operationId)) { + // A finish or block decided while earlier Paperclip calls of this turn + // 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); + } const outcome = unwrapToolResponse( await this.#handler({ id: call.callId, diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts new file mode 100644 index 00000000000..53b3066b94d --- /dev/null +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest"; +import { isTerminalOperation, staleTerminalCallResult, unseenReturnedToolCalls } from "./terminal-call-freshness.js"; + +let seq = 0; +const event = (eventType: string, payload: Record, turnId = "turn-1") => ({ + sourceEventId: `e${++seq}`, envelope: { payload: { eventType, turnId, payload } }, +}); +const started = (id: string, name: string, turnId?: string) => + event("tool.execution.started", { executionId: id, transport: "dynamic", name, status: "running" }, turnId); +const completed = (id: string, name: string) => + event("tool.execution.completed", { executionId: id, transport: "dynamic", name, status: "completed" }); +const result = (id: string) => event("item.completed", { item: { type: "tool_result", id, tool_use_id: id, result: {} } }); + +describe("unseenReturnedToolCalls", () => { + it("names a call that returned but that the agent had not seen when it decided (restart chaos t03)", () => { + const block = started("b", "paperclip_block"); + const events = [started("w", "write_document"), result("w"), block, completed("w", "write_document")]; + expect(unseenReturnedToolCalls(events, [], { terminalSourceEventId: block.sourceEventId, turnId: "turn-1" })) + .toEqual(["write_document"]); + }); + + it("is empty once the agent saw the result before deciding", () => { + const block = started("b", "paperclip_block"); + const events = [started("w", "write_document"), result("w"), completed("w", "write_document"), block]; + expect(unseenReturnedToolCalls(events, [], { terminalSourceEventId: block.sourceEventId })).toEqual([]); + }); + + it("leaves a long-running call with no result to the agent's judgement", () => { + const block = started("b", "paperclip_block"); + expect(unseenReturnedToolCalls([started("w", "write_document"), block], [], { terminalSourceEventId: block.sourceEventId })) + .toEqual([]); + }); + + it("ignores other turns, process tools and the terminal calls themselves", () => { + const events = [ + started("old", "write_document", "turn-0"), result("old"), + event("tool.execution.started", { executionId: "sh", transport: "process", name: "/bin/sh" }), result("sh"), + started("f", "paperclip_finish"), result("f"), + ]; + expect(unseenReturnedToolCalls(events, [], { turnId: "turn-1" })).toEqual([]); + }); +}); + +it("counts the controller's semantic_tool.result command as the returned result", () => { + const block = started("b", "paperclip_block"); + const commands = [{ type: "semantic_tool.result", payload: { callId: "w", operationId: "write_document" } }]; + expect(unseenReturnedToolCalls([started("w", "write_document"), block], commands, { terminalSourceEventId: block.sourceEventId })) + .toEqual(["write_document"]); +}); + +describe("staleTerminalCallResult", () => { + it("is a retryable error that names the calls", () => { + expect(isTerminalOperation("paperclip_block")).toBe(true); + expect(isTerminalOperation("write_document")).toBe(false); + const outcome = staleTerminalCallResult("paperclip_block", ["write_document"]); + expect(outcome).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results", retryable: true } } }); + expect(outcome.result.error.message).toContain("write_document"); + }); +}); diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.ts new file mode 100644 index 00000000000..b64827dc6ca --- /dev/null +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.ts @@ -0,0 +1,89 @@ +/** + * Whether a terminal Paperclip call (paperclip_finish or paperclip_block) was + * decided before the agent saw results that had already returned. + * + * Codex keeps a dynamic tool call pending while its client is silent and + * samples the model again about once a minute. When the controller is down + * (a restart or a long hold) the Runner keeps Codex's tool requests until + * Paperclip is back, so the model can conclude "Paperclip is not answering" + * and call paperclip_block while its earlier calls are still pending. After + * the reattach the Runner delivers that backlog in order: the earlier calls + * run and return (for example the document is saved), then the stale block + * arrives and would end a task that is no longer blocked (restart chaos t03). + * + * The evidence: + * - committed `tool.execution.started` with transport `dynamic`: the provider + * made a call (its `executionId` is the provider call id); + * - the controller's `semantic_tool.result` command for that `callId`: + * Paperclip returned a result to the Runner (an `item.completed` tool_result, + * 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. + */ + +const TERMINAL_OPERATIONS = new Set(["paperclip_finish", "paperclip_block"]); + +type CommittedEvent = { sourceEventId: string; envelope: Record }; +type Command = { type: string; payload: Record }; + +const record = (value: unknown): Record => + value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +/** 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 }, +): string[] { + const returnedCallIds = new Set(commands.flatMap(command => command.type === "semantic_tool.result" + && typeof command.payload.callId === "string" ? [command.payload.callId] : [])); + const calls = new Map(); + for (const saved of events) { + if (input.terminalSourceEventId && saved.sourceEventId === input.terminalSourceEventId) break; + const event = record(saved.envelope.payload); + if (input.turnId && typeof event.turnId === "string" && event.turnId !== input.turnId) continue; + const payload = record(event.payload); + const type = event.eventType; + if ((type === "tool.execution.started" || type === "tool.execution.completed") && payload.transport === "dynamic") { + const id = typeof payload.executionId === "string" ? payload.executionId : null; + if (!id) continue; + const call = calls.get(id) ?? { name: typeof payload.name === "string" ? payload.name : "tool", returned: false, seen: false }; + if (type === "tool.execution.completed") call.seen = true; + calls.set(id, call); + continue; + } + const item = record(payload.item); + if (type === "item.completed" && item.type === "tool_result") { + const id = typeof item.tool_use_id === "string" ? item.tool_use_id : typeof item.id === "string" ? item.id : null; + const call = id ? calls.get(id) : undefined; + if (call) call.returned = true; + } + } + 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))]; +} + +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(", "); + 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; " + + `call ${operationId} again only if it is still the right outcome.`, + }, + }, + isError: true, + }; +} diff --git a/packages/shared/src/crash-point-registry.ts b/packages/shared/src/crash-point-registry.ts index 65d2cacd0ec..708b1e43103 100644 --- a/packages/shared/src/crash-point-registry.ts +++ b/packages/shared/src/crash-point-registry.ts @@ -64,6 +64,12 @@ export const CRASH_POINTS = { kinds: ["native_restart_recovery_claim"], invariants: ["I4", "I6"], }, + "runner.semantic_tool.after_input_before_dispatch": { + description: "The controller received a dynamic tool call (for example write_document) from the Runner and has not dispatched it. On Daytona the Runner and Codex outlive the controller, so with --hold-ms Codex keeps the call pending through the outage (restart chaos t03); a local restart also stops the Runner.", + environments: ["local", "daytona"], + kinds: ["native_executor_claim"], + invariants: ["I4", "I7"], + }, "native.result.after_accept_before_finalize": { description: "The run's result and its workspace_finalizing finalization row are committed; finalization has not started. The accepted result must be finalized exactly once.", environments: ["local", "daytona"], diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index ff4196ba3c8..69900e93a4e 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -90,13 +90,14 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC lifecycle: { type: "string" }, "k4-hold-max-seconds": { type: "string" }, "crash-point": { type: "string" }, + "crash-point-hold-ms": { type: "string" }, }, }); const keep = values.keep === true; const pollMs = integer(values["poll-ms"], "poll-ms", 3_000, 2_000, 5_000); const budgetCents = integer(values["budget-cents"], "budget-cents", 3_000, 1, 100_000); if (values.replay !== undefined) { - for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point"] as const) { + for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point", "crash-point-hold-ms"] as const) { if (values[key] !== undefined) throw new Error(`--replay reproduces the saved plan; do not combine it with --${key}`); } return { mode: "replay", ref: parseReplayRef(values.replay), keep, @@ -149,7 +150,10 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC ...(values["k4-hold-max-seconds"] !== undefined ? { k4HoldMaxMs: integer(values["k4-hold-max-seconds"], "k4-hold-max-seconds", 300, 120, 600) * 1000 } : {}), ...(crashPoint ? { crashPoint } : {}), + ...(crashPoint && values["crash-point-hold-ms"] !== undefined + ? { crashPointHoldMs: integer(values["crash-point-hold-ms"], "crash-point-hold-ms", 0, 0, 600_000) } : {}), }; + if (!crashPoint && values["crash-point-hold-ms"] !== undefined) throw new Error("--crash-point-hold-ms applies only to --crash-point"); if (env !== "daytona" && values["daytona-cpu"] !== undefined) throw new Error("--daytona-cpu applies only to --env daytona"); if (env !== "daytona" && values["daytona-reuse-lease"]) throw new Error("--daytona-reuse-lease applies only to --env daytona"); if (options.daytonaCpu !== undefined) { diff --git a/tests/restart-chaos/crash-points.ts b/tests/restart-chaos/crash-points.ts index fdbb846105a..98a052e1f8e 100644 --- a/tests/restart-chaos/crash-points.ts +++ b/tests/restart-chaos/crash-points.ts @@ -83,7 +83,7 @@ export function parseSuiteArgs(argv: readonly string[]): SuiteOptions { points: { type: "string" }, kinds: { type: "string" }, adapters: { type: "string" }, workloads: { type: "string" }, env: { type: "string" }, nth: { type: "string" }, seed: { type: "string" }, list: { type: "boolean" }, "daytona-image": { type: "string" }, "daytona-cpu": { type: "string" }, "budget-cents": { type: "string" }, - "poll-ms": { type: "string" }, keep: { type: "boolean" }, lifecycle: { type: "string" }, + "poll-ms": { type: "string" }, keep: { type: "boolean" }, lifecycle: { type: "string" }, "hold-ms": { type: "string" }, "daytona-reuse-lease": { type: "boolean" }, } }); const env = values.env ?? "local"; @@ -100,6 +100,7 @@ export function parseSuiteArgs(argv: readonly string[]): SuiteOptions { for (const key of ["daytona-image", "daytona-cpu", "budget-cents", "poll-ms", "lifecycle"] as const) { if (values[key] !== undefined) passthrough.push(`--${key}`, values[key]!); } + if (values["hold-ms"] !== undefined) passthrough.push("--crash-point-hold-ms", values["hold-ms"]); if (values.keep) passthrough.push("--keep"); if (values["daytona-reuse-lease"]) passthrough.push("--daytona-reuse-lease"); return { points: csv(values.points), kinds: csv(values.kinds), adapters, workloads, env, nth, seed, list: values.list === true, passthrough }; diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 30ca3c0a045..73e06d45538 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -160,6 +160,7 @@ async function main(): Promise { const violationKey = (violation: { rule: string; kind: string; recordId: string }) => `${violation.rule}:${violation.kind}:${violation.recordId}`; if (crashPoint) { if (plan.tasks.length !== batchSize * crashPoints.length) throw new Error(`--tasks must be a multiple of the ${crashPoints.length} crash point(s)`); + supervisor.crashPointHoldMs = plan.options.crashPointHoldMs ?? 0; supervisor.useCrashPointControl({ controlPath: path.join(supervisor.paths.temporaryRoot, "crash-point.control"), receiptPath: path.join(waveDir, "crash-point-receipts.jsonl") }); } diff --git a/tests/restart-chaos/supervisor.ts b/tests/restart-chaos/supervisor.ts index eadf5919bee..b1c218d4a3a 100644 --- a/tests/restart-chaos/supervisor.ts +++ b/tests/restart-chaos/supervisor.ts @@ -115,6 +115,8 @@ export class ChaosSupervisor { private crashPointControl: CrashPointControl | null = null; /** Called with the receipt of a crash-point self-kill once the replacement is ready (or failed). */ onCrashPoint: ((receipt: FaultReceipt) => void) | null = null; + /** Crash-point suite: keep the control plane down this long after a self-kill (a K4-style hold). */ + crashPointHoldMs = 0; private constructor(readonly repositoryRoot: string, readonly port: number, readonly paths: SupervisorPaths, private readonly timeline: Timeline) { @@ -302,7 +304,8 @@ export class ChaosSupervisor { this.faultCount += 1; const now = Date.now(); const receipt: FaultReceipt = { - faultId: `f${String(this.faultCount).padStart(2, "0")}`, kind: "K1", requestedBy: "crash-point", signal: "SIGKILL", holdMs: 0, + faultId: `f${String(this.faultCount).padStart(2, "0")}`, kind: this.crashPointHoldMs ? "K4" : "K1", requestedBy: "crash-point", + signal: "SIGKILL", holdMs: this.crashPointHoldMs, loadAvg1m: Math.round(os.loadavg()[0]! * 100) / 100, crashPoint: hit, requestedAt: now, startedAt: now, stoppedAt: null, restartedAt: null, healthyAt: null, readyAt: null, phases: [], serverLogOffsetAtStart: this.serverLogOffset(), serverLogOffsetAtReady: null, error: null, @@ -322,6 +325,10 @@ export class ChaosSupervisor { if (this.child === previous) this.child = null; await this.waitForHealthToStop(); receipt.stoppedAt = Date.now(); + if (this.crashPointHoldMs) { + this.timeline.add("crash_point_hold", { faultId: receipt.faultId, holdMs: this.crashPointHoldMs }); + await delay(this.crashPointHoldMs); + } receipt.restartedAt = Date.now(); const child = await this.spawnServer(); receipt.healthyAt = await this.waitForHealth(child, receipt.phases); diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 826882763a3..179acc04ee6 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -62,6 +62,8 @@ export interface WaveOptions { * SIGKILLs itself at its nth hit and is restarted like K1. */ crashPoint?: string; + /** Crash-point suite: hold the control plane down this long after each self-kill (K4-style). */ + crashPointHoldMs?: number; } export interface StormFault {