From ff1e497334c8c33f6e747ad339a6d59b3d6321da Mon Sep 17 00:00:00 2001 From: Dev Agent Date: Thu, 20 Aug 2026 04:26:37 +0000 Subject: [PATCH] fix(runner): replay durable duplicate after interrupted restore Co-Authored-By: Paperclip --- .../src/live/live-session.test.ts | 93 +++++++++++++++++++ .../paperclip-runner/src/live/live-session.ts | 25 ++++- 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/packages/paperclip-runner/src/live/live-session.test.ts b/packages/paperclip-runner/src/live/live-session.test.ts index 706d77561e0..c70ccd0def9 100644 --- a/packages/paperclip-runner/src/live/live-session.test.ts +++ b/packages/paperclip-runner/src/live/live-session.test.ts @@ -132,6 +132,10 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { this.#handler = handler; } + invokeServerRequest(request: CodexRpcServerRequest): Promise> { + return this.#handler(request); + } + async close(): Promise { this.#closed = true; this.notificationsQueue.close(); @@ -462,6 +466,95 @@ describe("Capability live runnerd and Codex session", () => { }); }); + it("replays only a durable duplicate after restore reconciles the provider turn as interrupted", async () => { + const state = providerState(); + const store = new InMemoryCapabilityLiveSessionStore(); + const firstService = new CapabilityLiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + const first = await firstService.create({ + runId: "run-terminal-replay", + sessionId: "session-terminal-replay", + attemptId: "attempt-terminal-killed", + turnTimeoutMs: 50, + }); + state.holdAfterTool = true; + const killedTurn = first.sendMessage("Apply idempotent progress once.").then( + () => null, + (error: unknown) => error, + ); + await vi.waitFor(async () => { + expect((await store.load(first.id))?.mockState).toContain("progress-governed-once"); + }); + await state.transports[0]!.close(); + await expect(killedTurn).resolves.toMatchObject({ message: expect.stringContaining("timed out") }); + state.turns.set("turn-1", "interrupted"); + + const resumedService = new CapabilityLiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + const resumed = await resumedService.resume({ + sessionId: first.id, + attemptId: "attempt-terminal-resumed", + resumeOf: "attempt-terminal-killed", + }); + expect(resumed.snapshot()).toMatchObject({ + activeTurnId: null, + providerThreadId: state.threadId, + terminalTurns: [ + { + turnId: "turn-1", + status: "interrupted", + reconciledByAttemptId: "attempt-terminal-resumed", + }, + ], + }); + + const replay = await state.transports[1]!.invokeServerRequest({ + id: "request-turn-1:deterministic-resume-replay", + method: "item/tool/call", + params: { + threadId: state.threadId, + turnId: "turn-1", + callId: "call-turn-1:deterministic-resume-replay", + tool: "report_progress", + arguments: { + idempotencyKey: "progress-governed-once", + body: "Progress persisted through the live Codex tool loop.", + }, + }, + }); + expect(replay.success).toBe(true); + expect(String((replay.contentItems as Array>)[0]?.text)).toContain( + '\"disposition\":\"duplicate\"', + ); + expect(resumed.mockState().comments).toHaveLength(1); + expect(resumed.mockState().idempotency).toHaveLength(1); + expect(resumed.snapshot().evidence.filter((entry) => entry.kind === "tool_result")).toHaveLength(2); + + const rejected = await state.transports[1]!.invokeServerRequest({ + id: "request-turn-1:new-key", + method: "item/tool/call", + params: { + threadId: state.threadId, + turnId: "turn-1", + callId: "call-turn-1:new-key", + tool: "report_progress", + arguments: { + idempotencyKey: "progress-new-effect", + body: "This effect must not be applied.", + }, + }, + }); + expect(rejected.success).toBe(false); + expect(resumed.mockState().comments).toHaveLength(1); + expect(resumed.mockState().idempotency).toHaveLength(1); + expect(resumed.snapshot().evidence.filter((entry) => entry.kind === "tool_result")).toHaveLength(2); + await resumedService.shutdown(resumed.id, "terminal replay test complete"); + }); + it("fails closed for missing, corrupt, mismatched, and provider-drifted checkpoints", async () => { const directory = await mkdtemp(join(tmpdir(), "capability-live-fail-closed-")); const binding = { diff --git a/packages/paperclip-runner/src/live/live-session.ts b/packages/paperclip-runner/src/live/live-session.ts index a19bfe3963f..a5d07bfb9be 100644 --- a/packages/paperclip-runner/src/live/live-session.ts +++ b/packages/paperclip-runner/src/live/live-session.ts @@ -1346,10 +1346,11 @@ export class CapabilityLiveSession { ) { this.#activeTurnId = turnId; } + const terminalReplay = this.#isDurableTerminalReplay(turnId, request.params.arguments); if ( threadId !== this.#providerThreadId || turnId.length === 0 || - turnId !== this.#activeTurnId || + (turnId !== this.#activeTurnId && !terminalReplay) || callId.length === 0 ) { return { @@ -1371,6 +1372,7 @@ export class CapabilityLiveSession { operationId, input: request.params.arguments, }); + const replayDisposition = result.ok ? text(record(result.result).disposition) : ""; this.#appendEvidence("tool_result", turnId, { callId, operationId, @@ -1380,9 +1382,30 @@ export class CapabilityLiveSession { }); this.#emit({ turnId, kind: "activity", reason: "tool_result" }); await this.#persist(); + if (terminalReplay && replayDisposition !== "duplicate") { + return { + success: false, + contentItems: [{ type: "inputText", text: "Terminal-turn replay did not resolve to a durable duplicate." }], + }; + } return this.#codexToolResponse(result); } + #isDurableTerminalReplay(turnId: string, input: unknown): boolean { + if (turnId.length === 0 || this.#activeTurnId !== null) return false; + const terminal = this.#terminalTurns.find((candidate) => candidate.turnId === turnId); + if ( + terminal?.status !== "interrupted" || + terminal.reconciledByAttemptId !== this.#currentAttemptId + ) return false; + const idempotencyKey = text(record(input).idempotencyKey); + if (idempotencyKey.length === 0) return false; + const scope = `${this.#authority.runId}:semantic-command`; + return this.#port.snapshot().idempotency.some( + (entry) => entry.scope === scope && entry.key === idempotencyKey, + ); + } + #codexToolResponse(result: CapabilitySemanticToolResult): Record { return { success: result.ok,