From 6452e964a363695a101e3efe4d6c82e48c8c57c5 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 4 Nov 2026 18:43:37 -0800 Subject: [PATCH] fix(runner): ignore replayed history for restored terminal turns (#19035) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Paperclip Runner executes native provider sessions (Codex app-server and ACPX) and can resume a run from an exact persisted checkpoint on a replacement sandbox. > - When a turn completes without a semantic result, the server schedules one same-run disposition recovery from that checkpoint. > - On the replacement sandbox, the restored runner can deliver the committed history of the checkpointed turn again as live notifications, before the disposition turn starts. > - Between turns there is no active turn. The identity classifier only marks settled-turn notifications as `stale_turn` while a different turn is active, so the replayed `turn/started` reaches the live-turn guard and fails the session with `turn_binding_mismatch` ("Provider started an unexpected turn."). Recovery then stops with `native_event_replay_conflict`. > - This pull request makes the driver drop replayed history for turns that were already terminal in the restored checkpoint. > - The benefit is that exact-checkpoint disposition recovery can continue instead of failing closed on its own history. ## Linked Issues or Issue Description **What happened?** A hosted ACPX run (Claude model) completed its first turn without a semantic result. The server persisted an exact-checkpoint disposition recovery. The second attempt restored the harness state from the verified failover backup on a new sandbox. About 3.7 seconds after the runner transport was ready, and before any `turn.submitted` event, the driver failed with `turn_binding_mismatch: Provider started an unexpected turn.` The run then stopped with `native_event_replay_conflict` and required workspace repair. **Expected behavior** Notifications that only replay a turn that the checkpoint already recorded as terminal must not fail the resumed session. The disposition turn must start normally. **Steps to reproduce** 1. Open a session, start turn `turn-1`, and complete it without a semantic result. 2. Take the snapshot and recover the session on a new transport. 3. Deliver `turn/started`, `item/*` and `turn/completed` notifications for `turn-1` again, before or during the disposition `turn/start`. 4. Before this change, the session fails with `turn_binding_mismatch`. If the replay arrives while `turn/start` is pending, the driver binds `turn-1` as the new turn and fails with "Codex turn identity changed during start". **Paperclip version or commit** Observed on a staging release built from `c2163b3ecf08`. Reproduced on `master` at `89703c3`. ## What Changed - `CodexSessionState` records the turn IDs that were terminal in the restored checkpoint (`restoredTerminalTurns`). It also adds `isRestoredTerminalReplay(turnId)`. This is true only when the turn was restored as terminal, is not the active turn, and the session has not failed. - `codex-session-notifications.ts` drops `turn/started`, `item/started`, `item/completed` and item or turn delta notifications for such turns. It emits no turn or item events for them. It emits a bounded `provider_notification_identity` diagnostic with `classification: "restored_terminal_replay"`, and shares the existing limit of 32 diagnostics. - A replayed `turn/completed` is unchanged. It still goes through `mapTerminalTurn`, which compares the terminal fingerprint and fails on a conflict. - New regression file `codex-app-server-driver.terminal-replay.test.ts`. It replays history before and during the disposition submission. It also checks that a live session still fails closed on `turn/started` for a turn that ended in that same session. ## Verification - Red before the fix, on `master` with only the test file added: - `npx vitest run src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts` gives 2 failed and 1 passed. The "before" case fails with `NativeProviderTerminalFailure: Provider started an unexpected turn.` (`providerCode: turn_binding_mismatch`). This is the production error. The "during" case fails with `Codex turn identity changed during start`. - Green after the fix: the same command gives 3 passed. - Related suites: `npx vitest run src/drivers/codex src/drivers/acpx/codex-acpx-driver.test.ts src/conformance/harness-driver.test.ts src/backends/harness-driver-backend.test.ts src/live/live-session.test.ts` gives 567 passed and 1 failed. The failure is `codex-boundaries.test.ts > defers provider-owned workspace existence without weakening its assignment`. It also fails on unchanged `master` in my sandbox, because the sandbox workspace path triggers the sensitive-path check. It is not related to this change. - `tsc -p tsconfig.json --noEmit` shows no errors in the changed files. The only errors come from the unbuilt `@paperclipai/paperclip-eval-kernel` package in `src/eval/workflow-harness.ts`, because I did not run the eval build prep step locally. ## Risks - The change relaxes the guard only for turn IDs restored as terminal from the checkpoint. Notifications for those IDs are now ignored and do not fail the session. A provider cannot use a settled turn ID for new work, so these notifications cannot be new provider activity. - Live post-terminal notifications, pre-turn notifications, foreign turn IDs, `paperclip/runResult`, and tool calls keep their current fail-closed behavior. The new test covers the live post-terminal case. - Limitation: this change makes the driver tolerant of the replay. It does not change why the runnerd transport cursor replays the committed window on exact-authority recovery. The transport-level reason was not reproduced here. The new diagnostic makes the event visible in run logs, so later evidence can confirm or reject that path. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used as an agent with tool use, shell and test execution, and repository search. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- ...-app-server-driver.terminal-replay.test.ts | 160 ++++++++++++++++++ .../codex/codex-session-notifications.ts | 62 +++++-- .../src/drivers/codex/codex-session-state.ts | 19 +++ 3 files changed, 229 insertions(+), 12 deletions(-) create mode 100644 packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts 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 ||