diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts new file mode 100644 index 00000000000..9837fa2003d --- /dev/null +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts @@ -0,0 +1,160 @@ +import { + FakeCodexTransport, + WORKSPACE, + collectUntilTerminal, + describe, + expect, + it, + makeDriver, + vi, + type PrpEvent, +} from "./codex-app-server-driver.test-support.js"; + +describe("Codex driver terminal-turn history replay", () => { + it.each(["before", "during"] as const)("ignores replayed history for a checkpointed terminal turn %s disposition submission", async (order) => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + second.readResponse = { + thread: { + id: "thread-1", + sessionId: "provider-session-1", + cwd: WORKSPACE, + turns: [{ id: "turn-1", status: "completed", items: [] }], + }, + }; + let acceptTurnStart!: (value: Record) => void; + second.turnStartResponse = new Promise((resolve) => { acceptTurnStart = resolve; }); + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-terminal-history-replay", + normalizedSessionId: "normalized-terminal-history-replay", + workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + first.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + await collectUntilTerminal(original.events()); + const snapshot = await original.snapshot(); + expect(snapshot).toMatchObject({ + activeTurnId: null, + semanticResult: null, + terminalTurns: [{ turnId: "turn-1" }], + }); + await original.close({ reason: "sandbox replaced after result-less terminal" }); + + const recovery = await driver.recoverSession?.(snapshot); + const recovered = recovery?.session; + expect(recovered).toBeDefined(); + await recovered!.reconcile?.(); + const events: PrpEvent[] = []; + const consume = (async () => { + for await (const event of recovered!.events()) events.push(event); + })(); + // A replacement runner can re-deliver the checkpointed turn's committed + // history before or while the disposition turn is submitted. + const started = order === "during" + ? recovered!.startTurn({ message: { role: "user", text: "Report disposition only." } }) + : null; + if (started) await vi.waitFor(() => expect(second.calls.some((call) => call.method === "turn/start")).toBe(true)); + second.push("turn/started", { + threadId: "thread-1", + turn: { id: "turn-1", status: "inProgress", items: [] }, + }); + second.push("item/started", { + threadId: "thread-1", + turnId: "turn-1", + item: { id: "msg-1", type: "agentMessage", text: "" }, + }); + second.push("item/agentMessage/delta", { + threadId: "thread-1", + turnId: "turn-1", + itemId: "msg-1", + delta: "Working", + }); + second.push("item/completed", { + threadId: "thread-1", + turnId: "turn-1", + item: { id: "msg-1", type: "agentMessage", text: "Working on it." }, + }); + const command = { id: "cmd-1", type: "commandExecution", command: "ls", status: "completed" }; + second.push("item/started", { threadId: "thread-1", turnId: "turn-1", item: command }); + second.push("item/commandExecution/outputDelta", { + threadId: "thread-1", + turnId: "turn-1", + itemId: "cmd-1", + delta: "file.txt", + }); + second.push("item/completed", { threadId: "thread-1", turnId: "turn-1", item: command }); + second.push("turn/plan/updated", { + threadId: "thread-1", + turnId: "turn-1", + plan: [{ step: "Inspect", status: "completed" }], + }); + second.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + await vi.waitFor(() => expect(second.queue.values).toHaveLength(0)); + await new Promise((resolve) => setTimeout(resolve, 0)); + const submission = started + ?? recovered!.startTurn({ message: { role: "user", text: "Report disposition only." } }); + await vi.waitFor(() => expect(second.calls.some((call) => call.method === "turn/start")).toBe(true)); + acceptTurnStart({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + await expect(submission).resolves.toMatchObject({ turnId: "turn-2" }); + second.push("turn/started", { + threadId: "thread-1", + turn: { id: "turn-2", status: "inProgress", items: [] }, + }); + second.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-2", status: "completed", items: [] }, + }); + await vi.waitFor(() => + expect(events.some((event) => event.eventType === "turn.completed")).toBe(true), + ); + await recovered!.close({ reason: "test complete" }); + await consume; + expect(events.some((event) => event.eventType === "session.failed")).toBe(false); + expect(events.filter((event) => event.turnId === "turn-1")).toEqual([]); + expect( + events.filter((event) => event.eventType.startsWith("tool.execution.") || event.eventType === "plan.updated"), + ).toEqual([]); + expect( + events.filter((event) => + event.eventType === "harness.diagnostic" + && event.payload.classification === "restored_terminal_replay" + && event.payload.receivedTurnId === "turn-1").length, + ).toBeGreaterThan(0); + expect( + events + .filter((event) => ["turn.started", "turn.completed", "turn.failed"].includes(event.eventType)) + .map((event) => [event.eventType, event.turnId]), + ).toEqual([["turn.started", "turn-2"], ["turn.completed", "turn-2"]]); + expect(second.calls.filter((call) => call.method === "turn/start")).toHaveLength(1); + }); + + it("still rejects history for a turn that became terminal in the live session", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-live-terminal-replay", + normalizedSessionId: "normalized-live-terminal-replay", + workingDirectory: WORKSPACE, + }); + await session.startTurn({ message: { role: "user", text: "Complete." } }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + transport.push("turn/started", { + threadId: "thread-1", + turn: { id: "turn-1", status: "inProgress", items: [] }, + }); + const events: PrpEvent[] = []; + for await (const event of session.events()) events.push(event); + expect( + events.find((event) => event.eventType === "session.failed")?.payload, + ).toMatchObject({ code: "turn_binding_mismatch" }); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index e6c1a415866..326d249e929 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -37,6 +37,28 @@ import { text, } from "./codex-driver-values.js"; +const DELTA_KINDS: Record = { + "item/agentMessage/delta": "agentMessage", + "item/plan/delta": "plan", + "item/reasoning/summaryTextDelta": "reasoning", + "item/reasoning/textDelta": "reasoning", + "item/commandExecution/outputDelta": "commandExecution", + "item/fileChange/outputDelta": "fileChange", + "item/fileChange/patchUpdated": "fileChange", + "turn/diff/updated": "diff", + "turn/plan/updated": "plan", +}; + +// Turn-scoped activity a resumed runner can replay from committed history. +// turn/completed is excluded: its replay still passes the fingerprint check. +const REPLAYABLE_TURN_METHODS = new Set([ + "turn/started", + "item/started", + "item/completed", + "item/mcpToolCall/progress", + ...Object.keys(DELTA_KINDS), +]); + export async function pumpNotifications(state: CodexSessionState): Promise { try { for await (const notification of state.transport.notifications()) { @@ -159,6 +181,13 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex const threadId = text(params.threadId); const turnId = text(params.turnId, text(turn.id)); const itemId = text(item.id, text(params.itemId)); + // Check before canonical mapping, which would otherwise emit tool and plan + // events for the replayed turn while no turn is active. + if ( + threadId === state.opened.threadId + && REPLAYABLE_TURN_METHODS.has(notification.method) + && ignoreRestoredTerminalReplay(state, notification.method, turnId) + ) return; if (notification.method === "paperclip/runtimeRequestExpired") { const payload = record(params.payload); const requestId = text(payload.requestId); @@ -530,18 +559,7 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex recordWorkspaceChanges(state, turnId, item.changes, true); return; } - const deltaKinds: Record = { - "item/agentMessage/delta": "agentMessage", - "item/plan/delta": "plan", - "item/reasoning/summaryTextDelta": "reasoning", - "item/reasoning/textDelta": "reasoning", - "item/commandExecution/outputDelta": "commandExecution", - "item/fileChange/outputDelta": "fileChange", - "item/fileChange/patchUpdated": "fileChange", - "turn/diff/updated": "diff", - "turn/plan/updated": "plan", - }; - const deltaKind = deltaKinds[notification.method]; + const deltaKind = DELTA_KINDS[notification.method]; if (deltaKind !== undefined) { if (!state.notificationNamesActiveTurn(turnId, "item update")) return; const methodChannel = channelForDelta(state, notification.method); @@ -602,6 +620,26 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex } } +// Resume can re-deliver committed history for a turn the checkpoint already +// settled. Between turns the identity classifier sees no active turn, so the +// replay would otherwise reach the live-turn guards and fail the session. +function ignoreRestoredTerminalReplay( + state: CodexSessionState, + method: string, + turnId: string, +): boolean { + if (!state.isRestoredTerminalReplay(turnId)) return false; + if (state.notificationIdentityDiagnostics < 32) { + state.notificationIdentityDiagnostics += 1; + state.emit("harness.diagnostic", { + code: "provider_notification_identity", method: method.slice(0, 128), + classification: "restored_terminal_replay", + expectedTurnId: state.activeTurnId, receivedTurnId: turnId.slice(0, 256), + }); + } + return true; +} + function channelForStartedItem( state: CodexSessionState, item: Record, diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts index 382e43c0c33..f2bc252abe0 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -137,6 +137,12 @@ export class CodexSessionState { dispositionOnlyRecoveryTurnId: string | null = null; turnStarted = false; readonly terminalTurns = new Map(); + /** + * Turns already terminal in the checkpoint this session was restored from. + * A replacement runner can re-deliver their committed history as live + * notifications; those replays are not new provider activity. + */ + readonly restoredTerminalTurns = new Set(); readonly workspaceChangesByTurn = new Map>(); readonly emittedFileReferences = new Set(); readonly itemChannels = new Map< @@ -250,6 +256,7 @@ export class CodexSessionState { ); } this.terminalTurns.set(terminal.turnId, terminal.fingerprint); + this.restoredTerminalTurns.add(terminal.turnId); } if (this.activeTurnId && this.terminalTurns.has(this.activeTurnId)) { this.activeTurnId = null; @@ -332,6 +339,18 @@ export class CodexSessionState { } } + /** + * True for a notification that only replays a turn restored as terminal. + * The caller drops it without emitting turn or item events; its terminal + * fingerprint was already committed, and turn/completed still checks it. + */ + isRestoredTerminalReplay(turnId: string): boolean { + return turnId.length > 0 + && !this.protocolFailed + && turnId !== this.activeTurnId + && this.restoredTerminalTurns.has(turnId); + } + notificationNamesActiveTurn(turnId: string, kind: string): boolean { if ( this.protocolFailed ||