From c55af26821a3eb1d060a0bc33938d5d892fefb8a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 11:14:24 -0600 Subject: [PATCH] fix(runner): treat a replayed start of a turn lost on restore as history (t48) (#20074) 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 > - A native Codex run must survive a control-plane restart and continue its work > - #20057 made recovery adopt an in-flight turn, or start a new turn when the restored runner reports the old turn lost > - One case was not covered. A graceful drain can detach the controller after the runner accepted `turn.start` but before the provider started the turn > - The runner then records the provider's `turn.started` while no controller is attached. After a restart it replays that start, and the Codex session rejects it as an unexpected turn > - This pull request treats that replayed start as history from the earlier generation > - The benefit is that the run starts its continuation turn instead of stopping with `native_event_replay_conflict` ## Linked Issues or Issue Description Refs #20057 **What happened?** A restart chaos wave ended with one failed task (t48, Codex). A graceful SIGTERM suspended the controller 300 ms after the runner accepted `turn.start`. The provider's `turn.started` came after the controller detached. Recovery classified the run as `resume_dead_runner`. The restored runner reported the turn as `provider_turn_lost_on_restore`, which is correct. But it also replayed the undelivered `turn.started`. The Codex session failed with "Provider started an unexpected turn." (`turn_binding_mismatch`), and recovery stopped with `native_event_replay_conflict`. **Expected behavior** Recovery starts a new continuation turn. The replayed start of the lost turn is recorded as history, and the run completes. **Steps to reproduce** 1. Start a Codex native run. 2. Detach the controller gracefully after the runner accepts `turn.start` and before the provider emits `turn/started`. 3. Let the runner record `turn.started` while it is unattached, then kill the runner and the provider. 4. Restore the run. Before this change, the session fails with `turn_binding_mismatch`. **Paperclip version or commit** master at a9cf4ee731 (the wave ran on 66a2cbab98). ## What Changed - `codex-app-server-driver-impl.ts`: recovery identifies the turn that this run's latest `turn.start` had the runner accept, when recovery did not adopt it and provider history does not show it live. Recovery passes it to the session as `lostAcceptedTurnId`. - `codex-session-state.ts`: the session keeps `lostAcceptedTurnId` as an unconfirmed lost turn. The stale-event rule treats its replayed start and items as history. - `codex-session-notifications.ts`: `turn/started` applies the stale-event rule to earlier-generation and unconfirmed lost turns. The runner's restart-loss terminal confirms the lost turn. Any other terminal, or a semantic result, from that turn fails closed, so a real outcome is never dropped. A start for a turn that this generation settled, or for a turn the session never saw, still fails closed. - `runnerd-restart-continuation.test.ts`: a new connected test with the real runner and the fake Codex server reproduces the t48 shape. A new fake option, `--turn-started-gate`, holds the provider's start until the test detaches the controller, so the test does not depend on timing. - `codex-app-server-driver.terminal-replay.test.ts`: driver tests for the replayed start (history), and for an ordinary failure or a result from the lost turn (fail closed). ## Verification - Red: before the fix, the new test failed with `NativeProviderTerminalFailure: Provider started an unexpected turn.` at `codex-harness-session.ts:211`. This is the same error and stack as the chaos run. - Green: `npx vitest run src/drivers/codex src/live/runnerd-restart-continuation.test.ts` (in `packages/paperclip-runner`) passed 392 of 392. - The wider Codex, restart and recovery test files (45 files, 1153 tests) passed on the final revision. The two fail-closed driver tests fail on the first revision, where the lost turn's outcome was dropped as history. The one failure in an earlier run ("still rejects history for a turn that became terminal in the live session") showed that the first version was too broad. The rule now applies only to earlier-generation turns, and that test passes. - `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` in `packages/paperclip-runner` passed. ## Risks - Low risk. The change only applies after a restore, to a turn that the runner accepted for this run's latest `turn.start`, and only when recovery did not adopt that turn and provider history does not show it in progress. - A turn that the session never saw still fails closed. A re-start of a turn this generation settled still fails closed. A real failure or a result from the lost turn fails closed instead of being dropped. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test execution). ## 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 --- .../src/bin/fake-codex-app-server.rs | 7 ++ .../codex/codex-app-server-driver-impl.ts | 15 +++ ...-app-server-driver.terminal-replay.test.ts | 73 ++++++++++++++ .../codex/codex-session-notifications.ts | 27 ++++++ .../src/drivers/codex/codex-session-state.ts | 13 ++- .../live/runnerd-restart-continuation.test.ts | 97 +++++++++++++++++++ 6 files changed, 231 insertions(+), 1 deletion(-) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index e068f36da61..69603bfcf8d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -1862,6 +1862,13 @@ fn run() -> Result<(), Box> { if let Some(delay) = argument(&args, "--turn-started-delay-ms") { thread::sleep(Duration::from_millis(delay.parse()?)); } + // A test can hold the start until it has detached the controller. + if let Some(gate) = argument(&args, "--turn-started-gate").map(PathBuf::from) { + let deadline = std::time::Instant::now() + Duration::from_secs(30); + while !gate.is_file() && std::time::Instant::now() < deadline { + thread::sleep(Duration::from_millis(10)); + } + } // An accepted turn can be lost before it starts. if args .iter() diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index 4e77e387541..7086713a4dc 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -641,6 +641,11 @@ export class CodexAppServerDriver implements HarnessDriver { // that turn and no other turn is active; any other active turn stays // foreign and fails closed when its notifications arrive. let adoptedAcceptedTurnId: string | null = null; + // A turn this run's latest turn.start had the runner accept, which the + // restore neither adopted nor finds live (the provider lost it). Its + // undelivered reports, such as a turn/started the runner recorded after + // the previous controller detached, are earlier-generation history. + let lostAcceptedTurnId: string | null = null; if ( !recoveringAutonomousGoal && recoveredActiveTurnId === null && @@ -669,6 +674,14 @@ export class CodexAppServerDriver implements HarnessDriver { ) { recoveredActiveTurnId = acceptedTurnId; adoptedAcceptedTurnId = acceptedTurnId; + } else if ( + completedTurnId.length > 0 && + completedTurnId !== acceptedTurnId && + !active.some((turn) => text(turn.id) === completedTurnId) && + // A checkpointed terminal turn keeps the existing history behavior. + !(snapshot.terminalTurns ?? []).some((turn) => turn.turnId === completedTurnId) + ) { + lostAcceptedTurnId = completedTurnId; } } if (opened.context.liveConsole) @@ -691,6 +704,7 @@ export class CodexAppServerDriver implements HarnessDriver { dispositionOnlyRecoveryConsumed, dispositionOnlyRecoveryTurnId, adoptedAcceptedTurnId, + lostAcceptedTurnId, stalePendingRuntimeRequests: snapshot.pendingRuntimeRequests ?? [], lineage: snapshot.lineage, sourceSequence: snapshot.lastSourceSequence ?? 0, @@ -1018,6 +1032,7 @@ export class CodexAppServerDriver implements HarnessDriver { dispositionOnlyRecoveryConsumed?: boolean; dispositionOnlyRecoveryTurnId?: string | null; adoptedAcceptedTurnId?: string | null; + lostAcceptedTurnId?: string | null; stalePendingRuntimeRequests?: HarnessRuntimeRequest[]; lineage?: HarnessThreadLineageEntry[]; sourceSequence: number; 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 index f2b0ef4fda2..b644ed0ebc2 100644 --- 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 @@ -158,6 +158,79 @@ describe("Codex driver terminal-turn history replay", () => { ).toMatchObject({ code: "turn_binding_mismatch" }); }); + it("keeps a checkpointed failed turn as history even when the runner's accepted-turn record names it", async () => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport() as FakeCodexTransport & { controllerAcceptedCompletedTurnId?: () => string | null }; + const failed = { id: "turn-1", status: "failed", items: [], error: { code: "provider_failed", message: "failed", recoverable: false } }; + second.controllerAcceptedCompletedTurnId = () => "turn-1"; + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [failed] } }; + second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + const driver = makeDriver([first, second]); + const original = await driver.openSession({ runId: "run-saved-failure", normalizedSessionId: "normalized-saved-failure", workingDirectory: WORKSPACE }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + first.push("turn/completed", { threadId: "thread-1", turn: failed }); + await collectUntilTerminal(original.events()); + const snapshot = await original.snapshot(); + expect(snapshot.terminalTurns).toMatchObject([{ turnId: "turn-1" }]); + await original.close({ reason: "controller detached" }); + const recovery = await driver.recoverSession!(snapshot); + expect(recovery.recovered).toBe(true); + const session = recovery.session!; + await session.reconcile?.(); + // The replacement runner re-delivers the checkpointed failure. + second.push("turn/completed", { threadId: "thread-1", turn: failed }); + await new Promise(resolve => setTimeout(resolve, 20)); + await session.startTurn({ message: { role: "user", text: "Report disposition only." } }); + second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } }); + const events = await collectUntilTerminal(session.events()); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + expect(events.find(event => event.eventType === "turn.completed")?.turnId).toBe("turn-2"); + }); + + async function recoverWithUnadoptedAcceptedTurn(runId: string) { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport() as FakeCodexTransport & { controllerAcceptedCompletedTurnId?: () => string | null }; + // The runner accepted "turn-lost" for the previous controller; the restored provider has no live turn. + second.controllerAcceptedCompletedTurnId = () => "turn-lost"; + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [] } }; + second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + const driver = makeDriver([first, second]); + const original = await driver.openSession({ runId, normalizedSessionId: `normalized-${runId}`, workingDirectory: WORKSPACE }); + const snapshot = await original.snapshot(); + await original.close({ reason: "controller detached" }); + const recovery = await driver.recoverSession!(snapshot); + expect(recovery.recovered).toBe(true); + // The runner replays the start it recorded after the previous controller detached. + second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-lost", status: "inProgress", items: [] } }); + return { session: recovery.session!, transport: second }; + } + + it("treats the replayed start of an accepted turn the restore lost as history, then starts its own turn (t48)", async () => { + const { session, transport } = await recoverWithUnadoptedAcceptedTurn("run-late-start"); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-lost", status: "failed", items: [], + error: { code: "provider_turn_lost_on_restore", message: "lost", recoverable: true } } }); + await new Promise(resolve => setTimeout(resolve, 20)); + await session.startTurn({ message: { role: "user", text: "Continue." } }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } }); + const events = await collectUntilTerminal(session.events()); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + expect(events.find(event => event.eventType === "turn.completed")?.turnId).toBe("turn-2"); + expect(events.some(event => event.eventType === "harness.diagnostic" + && event.payload.classification === "stale_generation_event" && event.payload.receivedTurnId === "turn-lost")).toBe(true); + }); + + it.each([ + ["an ordinary failure", { method: "turn/completed", params: { threadId: "thread-1", turn: { id: "turn-lost", status: "failed", items: [], + error: { code: "provider_failed", message: "real failure", recoverable: false } } } }], + ["a semantic result", { method: "paperclip/runResult", params: { threadId: "thread-1", turnId: "turn-lost", result: { status: "completed" } } }], + ])("fails closed on %s from an accepted turn the restore did not find live", async (_label, report) => { + const { session, transport } = await recoverWithUnadoptedAcceptedTurn("run-lost-outcome"); + transport.push(report.method, report.params); + 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" }); + }); + it("records a restore-lost turn this session never bound without failing its own turn", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport]).openSession({ 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 fa37cf7037e..5a2ba36ad28 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -205,6 +205,14 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex // items and plans never re-enter the stream as fresh work. Its terminal // keeps its own conflict check below, a provider that starts it again // still fails, and canonical provider diagnostics keep flowing. + // A lost turn that reports a result was not lost: never drop that outcome as history. + if (identity.classification === "root" && identity.turnId !== null + && notification.method === "paperclip/runResult" + && state.unconfirmedLostTurns.has(identity.turnId) && identity.turnId !== state.activeTurnId) { + state.failProtocol("turn_binding_mismatch", + "Provider reported a result for a turn the restore did not find live."); + return; + } if (identity.classification === "root" && identity.turnId !== null && identity.threadId === state.opened.threadId && state.isHistoryTurn(identity.turnId) @@ -521,6 +529,15 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex return; } if (notification.method === "turn/started") { + // A start replayed for an earlier generation's turn (for example, one + // the runner recorded after the previous controller detached and then + // lost on restore) is history, never a start of this session's turn. A + // restart of a turn this generation settled still fails closed. + if ((state.priorGenerationTurns.has(turnId) || state.unconfirmedLostTurns.has(turnId)) + && state.isHistoryTurn(turnId)) { + state.recordHistoryEvent(notification.method, turnId); + return; + } const autonomousGoalTurn = state.currentGoal?.status === "active" && state.activeTurnId === null && !state.turnStartPending @@ -565,6 +582,16 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex // channel. Wait for the pending turn/start to settle first, so // turn.accepted always precedes the terminal event for the same turn. await state.turnStartSettled; + // The restore did not find this accepted turn live. Only the runner's restart-loss + // terminal confirms that; any other outcome is real and must not become history. + if (state.unconfirmedLostTurns.has(turnId) && turnId !== state.activeTurnId) { + state.unconfirmedLostTurns.delete(turnId); + if (!ignoreUnboundRestoredTurnLoss(state, turn, turnId)) { + state.failProtocol("turn_binding_mismatch", + "Provider reported a non-restart outcome for a turn the restore did not find live."); + } + return; + } // A terminal for a turn an earlier generation accepted is history, even // when it disagrees with the checkpointed terminal (for example, a // restored provider that reports the turn lost on restore). A terminal 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 649684e8a26..5314dd5ae66 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -175,6 +175,12 @@ export class CodexSessionState { * generation they form the history set (see isHistoryTurn). */ readonly priorGenerationTurns = new Set(); + /** + * A turn the runner accepted for an earlier generation that the restore did not find live. + * Its replayed start and items are history; only a restart-loss terminal confirms it lost. + * Any other outcome it reports (a terminal or a result) fails closed rather than being dropped. + */ + readonly unconfirmedLostTurns = new Set(); readonly workspaceChangesByTurn = new Map>(); readonly emittedFileReferences = new Set(); #workspaceRoots: string[] | null = null; @@ -213,6 +219,8 @@ export class CodexSessionState { dispositionOnlyRecoveryTurnId?: string | null; /** A recovered turn whose turn.accepted the previous controller never emitted. */ adoptedAcceptedTurnId?: string | null; + /** A turn the runner accepted for an earlier generation that the restore lost. */ + lostAcceptedTurnId?: string | null; stalePendingRuntimeRequests?: HarnessRuntimeRequest[]; lineage?: HarnessThreadLineageEntry[]; goal?: HarnessThreadGoal | null; @@ -342,6 +350,9 @@ export class CodexSessionState { // accepted) also belongs to an earlier generation. isHistoryTurn keeps it // live while it stays active; once reconcile settles it, its late reports // are history too. + if (input.lostAcceptedTurnId && input.lostAcceptedTurnId !== this.activeTurnId + && !this.terminalTurns.has(input.lostAcceptedTurnId)) + this.unconfirmedLostTurns.add(input.lostAcceptedTurnId); for (const turnId of [input.activeTurnId, input.adoptedAcceptedTurnId]) if (turnId) this.priorGenerationTurns.add(turnId); if (dispositionOnlyRecoveryTurnId && dispositionOnlyRecoveryTurnId !== this.activeTurnId) @@ -410,7 +421,7 @@ export class CodexSessionState { && !this.protocolFailed && turnId !== this.activeTurnId && (this.priorGenerationTurns.has(turnId) || this.restoredTerminalTurns.has(turnId) - || this.terminalTurns.has(turnId)); + || this.terminalTurns.has(turnId) || this.unconfirmedLostTurns.has(turnId)); } /** Every turn the history rule covers, for classifiers that take a set. */ diff --git a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts index 7a117c991e8..269f5c68d9b 100644 --- a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -500,3 +500,100 @@ it("adopts the in-flight turn when a restart lands just after turn.start (t77)", await rm(root, { recursive: true, force: true }); } }, 40_000); + +it("starts a new turn when the lost turn's start reached the runner only after the controller detached (t48)", async () => { + // A graceful drain suspended the controller right after the runner accepted + // turn.start but before the provider started the turn. The runner recorded + // the provider's turn.started while no controller was attached, then died. + // The restored runner replays that undelivered turn.started and reports the + // turn lost; recovery must start a new turn instead of failing on the replay. + const root = await mkdtemp(join(tmpdir(), "runner-restart-late-start-")); + const identity = { runId: "run-late-start", sessionId: "session-late-start", companyId: "company-late-start", issueId: "issue-late-start", agentId: "agent-late-start" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-late-start" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-5", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-late-start", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn", "--turn-started-gate", join(root, "release-turn-started")], + stateDirectory: root, + turnStartTimeoutMs: 8_000, + prpIdentity: { runnerInstanceId: "runner-late-start", environmentLeaseId: "lease-late-start", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-late-start", itemId: "item-late-start" }, + }; + const first = createRunnerdCodexTransport(options); + let restored: ReturnType | undefined; + const alive = (pid: number | null) => { + if (!pid) return false; + try { process.kill(pid, 0); return true; } catch { return false; } + }; + const kill = (pid: number | null) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + }; + try { + const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-late-start", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + // The last checkpoint the controller persisted: no active turn yet. + const checkpoint = await session.snapshot(); + expect(checkpoint.activeTurnId ?? null).toBeNull(); + const starting = session.startTurn({ message: { role: "user", text: execution.task.prompt } }).catch((error: unknown) => error); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + await vi.waitFor(async () => { + const control = JSON.parse(await readFile(controlPath, "utf8")); + expect(control.commands.find((command: { type: string }) => command.type === "turn.start")).toMatchObject({ status: "completed" }); + }, { timeout: 8_000, interval: 25 }); + const owner = first.evidence(); + // Graceful drain: the controller detaches while the runner and provider live on. + await first.detachControllerForRestart(); + await starting; + await writeFile(join(root, "release-turn-started"), ""); + // The provider starts the accepted turn while no controller is attached. + await vi.waitFor(async () => { + expect(await readFile(join(root, "runner", "runner-state.json"), "utf8")).toContain("turn.started"); + }, { timeout: 8_000, interval: 25 }); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")); + providerState.activeTurnId = null; + await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState)); + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-late-start", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const completion = await executeNativeSession({ + input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-late-start", controlPlaneInstanceId: "controller-late-start", timeoutMs: 15_000, + controlPlane: { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } }); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2); + } finally { + await restored?.transport.close().catch(() => undefined); + for (const bundle of [first, restored]) { + if (!bundle) continue; + const evidence = bundle.evidence(); + kill(evidence.runnerPid); + kill(evidence.providerPid); + } + await rm(root, { recursive: true, force: true }); + } +}, 45_000);