fix(runner): ignore replayed history for restored terminal turns (#19035)

## 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 <noreply@anthropic.com>
This commit is contained in:
Devin FoleyandClaude Opus 5.5 authored and GitHub committed 2026-11-04 18:43:37 -08:00
1 parent c3c2a680ee
commit 6452e964a3
3 files changed
+229 -12

No files matched your search

@@ -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<string, unknown>) => 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" });
});
});
@@ -37,6 +37,28 @@ import {
text,
} from "./codex-driver-values.js";
const DELTA_KINDS: Record<string, string> = {
"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<void> {
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<string, string> = {
"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<string, unknown>,
@@ -137,6 +137,12 @@ export class CodexSessionState {
dispositionOnlyRecoveryTurnId: string | null = null;
turnStarted = false;
readonly terminalTurns = new Map<string, string>();
/**
* 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<string>();
readonly workspaceChangesByTurn = new Map<string, Record<string, unknown>>();
readonly emittedFileReferences = new Set<string>();
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 ||