diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f94909d6c8f..701852da70b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1000,6 +1000,19 @@ recording the proof and successor lineage. Retained provider files are not edite Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Automatic attempts retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +An exact local Codex restart can restore the conversation after losing its active +turn. Runnerd records this as `provider_turn_lost_on_restore`, with no invented +task result. The admitted restart attempt may send one continuation in that same +conversation. It preserves the workspace and asks the agent to reconcile unfinished +commands and external actions before proceeding. It does not resend the task or +tool calls. An outcome that cannot be reconciled remains a blocker. The continuation +uses the existing durable one-shot recovery marker; recovery adopts an accepted +turn if the controller dies before checkpointing its ID. The interrupted terminal +is checkpointed before submission. Real provider failures, accepted semantic +results, intentional stops, and historical unmarked failures keep their existing +terminal behavior. This continuation uses the already charged restart attempt and +does not reset the provider-attempt budget. + Provider execution and control-plane finalization have different clocks. A healthy provider can think or execute a long tool without output. Once execution settles, recovery and finalization control steps have a 60-second deadline, checked on startup and every 15 seconds. With a healthy database and scheduler, an abandoned transition must be repaired or surfaced within 90 seconds. Terminal persistence must not wait on provider cleanup or publication; a late finalizer cannot change a reassigned or closed task or release another run's locks. Historical ambiguous runs are never automatically replayed after an upgrade. Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete. diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 1757c04fe18..ed8f723652c 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -110,6 +110,14 @@ proofs, encryption keys, environment variables, provider credentials, command arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics remain in the bounded `native_run_finalizations.recovery_history` ledger. +For a local Codex turn lost while runnerd was stopped, the canonical `turn.failed` +event retains `error.code: "provider_turn_lost_on_restore"` and +`error.recoverable: true`. It records the interrupted turn, not a final task +outcome. There is no synthesized result for that interruption. An admitted +same-session continuation emits its own `turn.submitted`, accepted turn ID, and +terminal events; only the final outcome closes the run. Reconstructed thread +history retains the original error so reconciliation and live delivery agree. + ## Native Local Process Stop Evidence The server writes `native.local_process_stopped` in the same transaction that 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 ce3b9e3e4b6..b1ed069b092 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 @@ -652,6 +652,9 @@ fn run() -> Result<(), Box> { .iter() .any(|value| value == "--require-codex-home-auth"); let durable_turn_ids = args.iter().any(|value| value == "--durable-turn-ids"); + let hold_first_durable_turn = args + .iter() + .any(|value| value == "--hold-first-durable-turn"); let durable_tool_ids = args.iter().any(|value| value == "--durable-tool-ids"); let expected_canonical_task_context_file = argument(&args, "--expected-canonical-task-context-file"); @@ -1348,6 +1351,7 @@ fn run() -> Result<(), Box> { .checked_add(1) .ok_or("fake provider turn sequence exhausted")?; } + let hold_turn = hold_turn || (hold_first_durable_turn && state.next_turn == 1); if reject_second_turn_start && turn_start_count == 2 { send(json!({ "method": "warning", diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index 545544c73eb..8d038f5fe17 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -2103,17 +2103,29 @@ impl CodexCommandExecutor { "providerTurnId": previous_active_turn_id, "status": "failed", "providerTerminalObserved": false, + "error": if provider_label == "codex" { json!({ + "code": "provider_turn_lost_on_restore", + "recoverable": true, + "message": "The runner restored the conversation after process loss, but the previous turn is no longer active.", + }) } else { Value::Null }, }), }; - let outcome = terminal_events( - state, - "turn.failed", - state.goal.as_ref().map(|goal| goal.status.as_str()), - ); - state.extend_terminal_events(with_terminal_outcome( - vec![provider_terminal], - outcome, - ))?; + if provider_label == "codex" { + // No provider result was observed. Preserve the lost + // turn as a fact, but leave the run open for the + // controller's bounded same-conversation recovery. + state.push_terminal_event(provider_terminal)?; + } else { + let outcome = terminal_events( + state, + "turn.failed", + state.goal.as_ref().map(|goal| goal.status.as_str()), + ); + state.extend_terminal_events(with_terminal_outcome( + vec![provider_terminal], + outcome, + ))?; + } } } else { state.push_event(reconciled)?; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 9550fb2e95f..34e3b46d0c7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -3484,6 +3484,65 @@ fn durable_backend_replays_pending_tool_calls_without_mutating_the_event_queue() fs::remove_dir_all(directory).expect("remove Codex integration-test directory"); } +#[test] +fn restart_lost_turn_retains_a_recoverable_cause_without_inventing_a_result() { + let directory = temporary_directory("restart-lost-turn"); + let config = provider_config( + &directory, + &["--durable-turn-ids", "--linger-after-turn-start"], + ); + let mut first = CodexCommandExecutor::new(&directory); + first + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": config}), + )) + .unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Watch the existing check run."}), + )) + .unwrap(); + // The process disappeared while its command was in flight. Codex restores + // the conversation, but cannot restore the running turn. + drop(first); + fs::write( + directory.join("fake-state.json"), + serde_json::to_vec(&json!({ + "threadId": "codex-thread-1", "activeTurnId": null, "nextTurn": 1, + })) + .unwrap(), + ) + .unwrap(); + let mut restored = CodexCommandExecutor::new(&directory); + restored + .execute(&command("snapshot", 4, "session.snapshot", json!({}))) + .unwrap(); + let events = poll_and_ack(&mut restored).unwrap(); + let interrupted = events + .iter() + .find(|event| event.event_type == "turn.failed") + .unwrap(); + assert_eq!( + interrupted.payload["error"]["code"], + "provider_turn_lost_on_restore" + ); + assert_eq!(interrupted.payload["error"]["recoverable"], true); + assert!(!events + .iter() + .any(|event| event.event_type == "run.result.proposed")); + restored.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + #[test] fn durable_backend_settles_pending_tools_when_recovery_finds_the_turn_ended() { let directory = temporary_directory("durable-tool-ended-offline"); diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index fd280b5df81..143dd877beb 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -673,6 +673,7 @@ class HarnessNativeSession implements NativeSession { // The provider has no active turn to interrupt yet. Do not launch work // whose events cancellation would suppress and leave the owner waiting. if (this.#explicitlyCancelled) throw new Error("native_session_cancelled"); + this.#terminal = null; try { const started = await this.#session.startTurn(input); this.#assertProtocolIntegrity(); diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 7a84e9577d2..87af9e2ad05 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -63,7 +63,7 @@ export interface PersistedNativeSession { terminal?: PrpTerminalState | null; activeTurnId?: string | null; terminalTurns?: PersistedHarnessTurnTerminal[]; - /** Durable at-most-once marker for a resultless terminal recovery turn. */ + /** At-most-once marker for resultless disposition repair or restart continuation. */ dispositionOnlyRecoveryConsumed?: boolean; dispositionOnlyRecoveryTurnId?: string | null; pendingRuntimeRequests?: HarnessRuntimeRequest[]; @@ -77,6 +77,31 @@ export interface NativeSessionRecoveryResult { reason?: string; } +/** Runner-owned cause: the process was restored, but its active turn was lost. */ +export const NATIVE_RESTART_INTERRUPTION_CODE = "provider_turn_lost_on_restore"; + +export function isNativeRestartInterruption(error: unknown): boolean { + if (!error || typeof error !== "object" || Array.isArray(error)) return false; + const failure = error as Record; + return failure.code === NATIVE_RESTART_INTERRUPTION_CODE && failure.recoverable === true; +} + +/** The terminal fingerprint survives a crash between interruption and continuation. */ +export function nativeRestartInterruptedTurnId(snapshot: Pick): string | null { + if (snapshot.driverKind !== "codex_app_server" || snapshot.activeTurnId != null || + snapshot.semanticResult != null || snapshot.dispositionOnlyRecoveryConsumed || + snapshot.goal != null || snapshot.pendingRuntimeRequests?.length) return null; + const terminal = Array.isArray(snapshot.terminalTurns) ? snapshot.terminalTurns.at(-1) : null; + if (!terminal || typeof terminal.fingerprint !== "string" || !terminal.turnId) return null; + try { + const fingerprint = JSON.parse(terminal.fingerprint); + return fingerprint?.terminalState === "failed" && fingerprint.result === null && + isNativeRestartInterruption(fingerprint.error) ? terminal.turnId : null; + } catch { return null; } +} + /** * Cancellation is a synchronous authority transition followed by passive * provider cleanup. Once `cancel` returns, the provider session must no longer 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 8be04491732..6f3b5ad8a76 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 @@ -17,7 +17,7 @@ import type { PersistedHarnessSession, PersistedHarnessTurnTerminal, } from "../../contracts/harness-driver.js"; -import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; +import { NativeSessionProtocolIntegrityError, nativeRestartInterruptedTurnId } from "../../contracts/native-session-backend.js"; import { HarnessReconciliationError } from "../../contracts/harness-driver.js"; import { CODEX_CODEX_PROTOCOL_VERSION, @@ -528,6 +528,12 @@ export class CodexAppServerDriver implements HarnessDriver { turns.forEach((turn, index) => { if (terminalIds.has(text(turn.id))) lastKnownTerminalIndex = index; }); + if (nativeRestartInterruptedTurnId({ + ...snapshot, semanticResult: null, + }) && (!providerHistoryIsArray || lastKnownTerminalIndex < 0)) { + await cancellation.wait(cancellation.close()); + return { recovered: false, reason: "restart interruption history is incomplete" }; + } // Releasing a consumed marker requires both an actual history array // and a checkpointed terminal that anchors its ordering. An array that // omits every durable terminal may be truncated or inconsistent, so diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts index dc4e03e33cd..7c76d0edffa 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts @@ -45,6 +45,29 @@ import { import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; describe("Codex app-server Codex driver", () => { + it("does not resend a restart continuation when its history anchor is missing", async () => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-restart-anchor", normalizedSessionId: "session-restart-anchor", workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Finish the request." } }); + first.push("turn/completed", { threadId: "thread-1", turn: { + id: "turn-1", status: "failed", error: { code: "provider_turn_lost_on_restore", recoverable: true }, items: [], + } }); + await collectUntilTerminal(original.events()); + const snapshot = await original.snapshot(); + await original.close({ reason: "controller lost before continuation checkpoint" }); + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "uncheckpointed-turn", status: "completed", items: [] }], + } }; + await expect(driver.recoverSession?.(snapshot)).resolves.toMatchObject({ + recovered: false, reason: "restart interruption history is incomplete", + }); + expect(second.calls.some(call => call.method === "turn/start")).toBe(false); + }); + it.each([null, "checkpointed-prior-turn"])("recovers an autonomous goal turn beyond checkpoint %s", async (checkpointTurnId) => { const first = new FakeCodexTransport(); const second = new FakeCodexTransport(); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts index f2c5d70f29d..47612b9267b 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts @@ -1,4 +1,5 @@ import { HarnessReconciliationError } from "../../contracts/harness-driver.js"; +import { isNativeRestartInterruption } from "../../contracts/native-session-backend.js"; import type { PrpStructuredRunResult } from "../../protocol/replay-contract.js"; import { boundedCodexPayload as boundedPayload, isRetainableCodexPayload } from "./codex-boundaries.js"; import type { CodexSessionState } from "./codex-session-state.js"; @@ -168,6 +169,13 @@ export function mapTerminalTurn( { turnId }, ); finalize(state, status); + if (status === "failed" && isNativeRestartInterruption(turn.error) && state.result === null && + !state.dispositionOnlyRecoveryConsumed) { + // Orchestration may use the paid restart attempt for one same-thread + // continuation. Its accepted turn id uses the existing durable marker. + state.terminal = false; + state.dispositionOnlyRecoveryAvailable = true; + } } function terminalFingerprint( diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 5fbca7d8fd4..68a7c84cc8f 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3581,20 +3581,31 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (method === "thread/turns/list") { data = turns.map(turn => ({ ...turn, items: [], itemsView: "notLoaded" })); } else { - if (params.turnId !== this.#turnId) throw new Error("codex_history_unavailable: requested turn is outside the retained runner event window"); + if (!turns.some(turn => turn.id === params.turnId)) throw new Error("codex_history_unavailable: requested turn is outside the retained runner event window"); const items = new Map>(); + let semanticResultItem: Record | null = null; let observedTurn = ""; let observedStart = false; for (const event of this.#core?.store.state.committedEvents ?? []) { + if (event.envelope.runId !== this.#core?.store.state.identity.runId) continue; const payload = record(record(event.envelope.payload).payload); if (event.eventType === "turn.started") observedTurn = String(payload.providerTurnId ?? payload.turnId ?? record(payload.turn).id ?? ""); if (event.eventType === "turn.started" && observedTurn === params.turnId) observedStart = true; + if (event.eventType === "run.result.proposed" && observedTurn === params.turnId) { + // Reconciliation can precede notification delivery. Recover the + // runner's authoritative result with the exact retained turn, + // rather than launching work again just to obtain a disposition. + const id = `runner-result-${event.sourceSeq}`; + semanticResultItem = { turnId: observedTurn, item: { id, type: "agentMessage", text: JSON.stringify(payload) } }; + } if (event.eventType !== "item.completed" || observedTurn !== params.turnId) continue; const item = record(rehydrateRunnerdItemNotification(payload, this.#threadId, observedTurn).item); if (typeof item.id === "string") items.set(item.id, { turnId: observedTurn, item }); } if (!observedStart) throw new Error("codex_history_incomplete: requested turn start is outside the retained runner event window"); data = [...items.values()]; + // Runner authority wins over schema-shaped prose in an assistant item. + if (semanticResultItem) data.push(semanticResultItem); } if (params.sortDirection === "desc") data.reverse(); const offset = params.cursor == null ? 0 : Number(params.cursor); @@ -3655,8 +3666,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await new Promise((resolveWait) => setTimeout(resolveWait, 10)); } if (terminal !== undefined) { + const payload = record(record(terminal.envelope.payload).payload); recoveredTurns.push({ id: this.#turnId, + // Reconciliation must retain the cause, including the runner's + // explicit process-loss marker, rather than inventing error:null. + error: payload.error ?? record(payload.turn).error ?? null, status: terminal.eventType === "turn.completed" ? "completed" @@ -3668,6 +3683,24 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }); } } + // A controller can lose the checkpoint after a continuation is accepted. + // Keep its prior terminal as the history anchor, so driver recovery can + // adopt the later accepted turn instead of submitting it again. Items + // remain lazy and fail closed if their start left the retained window. + const priorTerminals = new Map>(); + for (const event of this.#core?.store.state.committedEvents ?? []) { + if (event.envelope.runId !== this.#core?.store.state.identity.runId || + !["turn.completed", "turn.failed", "turn.interrupted", "turn.cancelled"].includes(event.eventType)) continue; + const payload = record(record(event.envelope.payload).payload); + const turnId = payload.providerTurnId ?? payload.turnId ?? record(payload.turn).id; + if (typeof turnId !== "string" || !turnId || turnId === this.#turnId) continue; + priorTerminals.set(turnId, { + id: turnId, + status: event.eventType.slice("turn.".length), + error: payload.error ?? record(payload.turn).error ?? null, + }); + } + recoveredTurns.unshift(...priorTerminals.values()); this.#recoveryTurnBindingPending = false; this.#pumpEvents(); return { diff --git a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts new file mode 100644 index 00000000000..e1e10951657 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -0,0 +1,125 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createRunnerdNativeSessionBackend } from "../backends/codex-native-backend.js"; +import type { NativeExecutionInputV1 } from "../contracts/native-execution.js"; +import type { PersistedNativeSession } from "../contracts/native-session-backend.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; +import { executeNativeSession } from "../native-session-runtime.js"; +import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; + +it.each(["none", "before-start", "after-start"] as const)("completes after the real runner and provider are killed (second crash: %s)", async (crash) => { + const root = await mkdtemp(join(tmpdir(), "runner-restart-continuation-")); + const identity = { runId: "run-restart", sessionId: "session-restart", companyId: "company-restart", issueId: "issue-restart", agentId: "agent-restart" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-restart" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1 }, + task: { identifier: "TEST-1", 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-restart", 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"], + stateDirectory: root, + prpIdentity: { runnerInstanceId: "runner-restart", environmentLeaseId: "lease-restart", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-restart", itemId: "item-restart" }, + }; + 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-restart", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + await session.startTurn({ message: { role: "user", text: execution.task.prompt } }); + const persisted = await session.snapshot(); + expect(persisted.activeTurnId).toBe("provider-turn-1"); + const durableEvents: PrpEvent[] = []; + const initialEvents = session.events()[Symbol.asyncIterator](); + while ((durableEvents.at(-1)?.sourceSeq ?? 0) < Number(persisted.cursor)) { + const event = await initialEvents.next(); + if (event.done) throw new Error("initial event history ended early"); + durableEvents.push(event.value); + } + const owner = first.evidence(); + await first.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + // Simulate Codex's process-loss history: the thread remains, its in-memory + // turn does not. Keep the turn counter to prevent provider identity reuse. + 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: persisted.activeTurnId }); + let recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-restart", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const checkpoints: PersistedNativeSession[] = []; + let recoveryCheckpoint = persisted; + let failCheckpoint = crash !== "none"; + const interruptedController = new Error("controller lost before continuation submission"); + const execute = () => executeNativeSession({ + input: execution, backend: recovering, persistedSession: recoveryCheckpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-restart", controlPlaneInstanceId: "controller-restart", timeoutMs: 15_000, + controlPlane: { + async openRun() {}, + async checkpointSession(snapshot) { + if (failCheckpoint && crash === "after-start" && snapshot.dispositionOnlyRecoveryTurnId === "provider-turn-2") { + failCheckpoint = false; + throw interruptedController; + } + checkpoints.push(structuredClone(snapshot)); + recoveryCheckpoint = structuredClone(snapshot); + if (failCheckpoint && crash === "before-start" && snapshot.terminal?.runTerminalState === "failed") { + failCheckpoint = false; + throw interruptedController; + } + }, + async appendEvent(event) { events.push(event); durableEvents.push(event); return { cursor: durableEvents.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = durableEvents.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + if (crash !== "none") { + await expect(execute()).rejects.toBe(interruptedController); + await restored.transport.close().catch(() => undefined); + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: recoveryCheckpoint.activeTurnId }); + recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-restart", transportFactory: () => restored!.transport }); + } + const completion = await execute(); + expect(completion).toMatchObject({ providerSessionId: persisted.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } }); + expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(1); + expect(events.filter(event => event.eventType === "run.terminal")).toHaveLength(1); + if (crash !== "after-start") expect(checkpoints.some(snapshot => snapshot.activeTurnId === "provider-turn-2" && snapshot.terminal === null)).toBe(true); + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2); + const journal = JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")); + const submissions = journal.commands.filter((command: { type: string }) => command.type === "turn.start"); + expect(submissions).toHaveLength(2); + expect(submissions[1].payload.text).toContain("Reconcile any unfinished tool or command"); + } 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 }); + } +}, 30_000); diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 5d3418e1cb8..db0344becc7 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -225,6 +225,71 @@ function highestContiguous(events: PrpEvent[]): number { } describe("executeNativeSession recovery", () => { + it.each([ + { label: "continues a restart-lost Codex turn", allowed: true, error: { code: "provider_turn_lost_on_restore", recoverable: true } }, + { label: "requires the control plane restart claim", allowed: false, error: { code: "provider_turn_lost_on_restore", recoverable: true } }, + { label: "preserves real provider errors", allowed: true, error: { code: "model_not_found", recoverable: false } }, + { label: "does not reopen historical error-null failures", allowed: true, error: null }, + ])("$label without replaying the task", async ({ label, allowed, error }) => { + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-recovery", + terminalTurns: [], semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + const failure = error; + const interrupted: PersistedNativeSession = { + ...checkpoint, activeTurnId: null, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + }; + const startTurn = vi.fn(async () => ({ turnId: "turn-continuation" })); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + yield runnerEvent(1, "turn.failed", { status: "failed", error: failure }); + if (!startTurn.mock.calls.length) return; + yield { ...runnerEvent(2, "turn.completed", { status: "completed" }), turnId: "turn-continuation" }; + }, + startTurn, + async result() { return startTurn.mock.calls.length ? { result, terminal, turnId: "turn-continuation" } : null; }, + async snapshot() { return structuredClone(interrupted); }, + async close() {}, + }; + const events: PrpEvent[] = []; + const checkpoints: PersistedNativeSession[] = []; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { throw new Error("must preserve the provider session"); }, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession(snapshot) { checkpoints.push(structuredClone(snapshot)); }, + async appendEvent(event) { events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + const execution = executeNativeSession({ + input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: allowed, + }); + if (label !== "continues a restart-lost Codex turn") { + await expect(execution).rejects.toMatchObject({ code: "native_provider_terminal_failed" }); + expect(startTurn).not.toHaveBeenCalled(); + return; + } + await expect(execution).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } }); + expect(startTurn).toHaveBeenCalledOnce(); + const submitted = startTurn.mock.calls[0]![0] as Parameters[0]; + expect(submitted.continuation).toBe(true); + expect(submitted.message.text).toContain("Reconcile any unfinished tool or command"); + expect(submitted.message.text).not.toContain(input.task.prompt); + expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(1); + expect(events.filter(event => event.eventType === "run.terminal")).toHaveLength(1); + expect(checkpoints.some(snapshot => snapshot.terminalTurns?.some(turn => turn.turnId === "turn-recovery"))).toBe(true); + }); + it.each([undefined, 0, 7 * 24 * 60 * 60 * 1000, 30 * 24 * 60 * 60 * 1000])( "honors long-lived turn duration independently of operation bounds (%s)", async (turnTimeoutMs) => { @@ -6234,7 +6299,12 @@ describe("executeNativeSession recovery", () => { expect(openRun).not.toHaveBeenCalled(); }); - it.each([false, true])("replaces a provider session that already ended with a failed terminal (prepared: %s)", async (preparedMode) => { + it.each([ + { preparedMode: false, interrupted: false }, + { preparedMode: true, interrupted: false }, + { preparedMode: false, interrupted: true }, + { preparedMode: true, interrupted: true }, + ])("replaces a failed provider with full task context (prepared: $preparedMode, interrupted: $interrupted)", async ({ preparedMode, interrupted }) => { const digest = "0".repeat(64); const skill = { key: "company/recovery-skill", @@ -6286,6 +6356,7 @@ describe("executeNativeSession recovery", () => { } as const; const checkpoint: PersistedNativeSession = { backendKind: "mock", + driverKind: "codex_app_server", sessionId: "driver-failed", identity, providerSessionId: "provider-failed", @@ -6299,7 +6370,10 @@ describe("executeNativeSession recovery", () => { runTerminalState: "failed", reportedWorkDisposition: "yielded", }, - terminalTurns: [{ turnId: "turn-failed", fingerprint: "failed" }], + terminalTurns: [{ turnId: "turn-failed", fingerprint: interrupted ? JSON.stringify({ + terminalState: "failed", result: null, + error: { code: "provider_turn_lost_on_restore", recoverable: true }, + }) : "failed" }], pendingRuntimeRequests: [], lineage: [], }; @@ -6335,8 +6409,8 @@ describe("executeNativeSession recovery", () => { async close() {}, }; const recoverSession = vi.fn(async () => ({ - recovered: true as const, - session: replacementSession, + recovered: false as const, + reason: "provider thread is unavailable", })); const openReplacementSession = vi.fn(async () => replacementSession); const onContinuityBreak = vi.fn(async () => undefined); @@ -6397,7 +6471,7 @@ describe("executeNativeSession recovery", () => { ).resolves.toMatchObject({ providerSessionId: "provider-replacement" }); expect(getFreshSessionHandoff).toHaveBeenCalledOnce(); - expect(recoverSession).not.toHaveBeenCalled(); + expect(recoverSession).toHaveBeenCalledTimes(interrupted ? 1 : 0); expect(openReplacementSession).toHaveBeenCalledOnce(); const replacementEnvelope = JSON.parse( startTurn.mock.calls[0]![0].message.text, @@ -6419,7 +6493,7 @@ describe("executeNativeSession recovery", () => { expect(JSON.stringify(replacementEnvelope)).not.toContain("ONLY_NEW_COMMENT"); expect(startTurn.mock.calls[0]![0]).not.toHaveProperty("continuation"); expect(onContinuityBreak).toHaveBeenCalledWith({ - reason: "provider session ended with a failed terminal", + reason: interrupted ? "provider thread is unavailable" : "provider session ended with a failed terminal", previousDriverSessionId: "driver-failed", previousProviderSessionId: "provider-failed", replacementDriverSessionId: "driver-replacement", diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 2ccfd7a4c1b..e76409e7abc 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -23,6 +23,8 @@ import { NativeSessionCloseUnrecoverableError, NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, + isNativeRestartInterruption, + nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { validatePrpStructuredRunResult, @@ -184,6 +186,8 @@ export interface NativeSessionGoalControl { } export interface ExecuteNativeSessionOptions { + /** The control plane proved the old local runner stopped and charged a recovery attempt. */ + resumeInterruptedTurn?: boolean; /** Read bounded task history only after recovery requires a fresh conversation. */ getFreshSessionHandoff?: () => Promise; /** Durable launch intent, after cleanup admission and before provider calls. */ @@ -865,6 +869,7 @@ async function consumeTurn( requestId: string; }, initialGoal?: HarnessThreadGoal | null, + resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise, ) { let timer: ReturnType | undefined; const appendAbort = new AbortController(); @@ -1174,6 +1179,8 @@ async function consumeTurn( } if (isTurnTerminal(event)) { if (providerFailure) throw providerFailure; + if (semanticResultProposal === null && governedResult === null && + inputTimers.size === 0 && await resumeInterruptedTurn?.(event, appendAbort.signal)) continue; return { event, eventCount, @@ -1871,6 +1878,12 @@ export async function executeNativeSession( previousProviderSessionId: string | null; } | null = null; let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null; + const checkpointedInterruption = persistedSession && nativeRestartInterruptedTurnId(persistedSession); + if (checkpointedInterruption && persistedSession) { + // The turn failed because its process vanished; the conversation is still + // usable. Keep its terminal fingerprint so replay cannot execute it twice. + persistedSession = { ...persistedSession, terminal: null }; + } await options.onSessionAdmission?.(); if (options.existingSession) { if (options.existingSession.attachRun === undefined) { @@ -2157,6 +2170,42 @@ export async function executeNativeSession( await persistCheckpoint(snapshot, signal); }; const recoveredSnapshot = await session.snapshot(); + const restartContinuation = () => ({ + message: { role: "user" as const, text: JSON.stringify({ + schema: "paperclip.native-continuation.v1", + events: [ + "Paperclip restarted and restored this same conversation. The previous turn was interrupted.", + "Continue the current user request from the recorded progress and preserved workspace.", + "Reconcile any unfinished tool or command before proceeding: inspect its current state and recorded outcomes.", + "Do not repeat completed work or blindly rerun an action whose outcome is unknown. If an external action cannot be reconciled, report the specific blocker.", + ].join("\n"), + completion: { + revision: input.completionContract.contract.revision, + criterionIds: input.completionContract.contract.criteria.map(criterion => criterion.id), + }, + }) }, + continuation: true as const, + requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const, + }); + let restartContinuationStarted = false; + const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => { + if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn || + input.provider.kind !== "codex" || options.sessionGoalControl || options.resumeSessionGoalHeartbeat || + recoveredSnapshot.goal || persistedSession?.semanticResult || persistedSession?.pendingRuntimeRequests?.length || + event.eventType !== "turn.failed" || !isNativeRestartInterruption(event.payload.error) || + event.turnId !== persistedSession?.activeTurnId) return false; + const snapshot = await session.snapshot(); + signal.throwIfAborted(); + if (nativeRestartInterruptedTurnId(snapshot) !== event.turnId || snapshot.pendingRuntimeRequests?.length) return false; + restartContinuationStarted = true; + // Persist the old terminal before launching. Recovery inspects provider + // history to adopt an accepted continuation if its checkpoint was lost. + await persistCheckpoint(snapshot, signal); + signal.throwIfAborted(); + await session.startTurn(restartContinuation()); + await checkpoint(signal); + return true; + }; const recoveredActiveTurnId = recovered ? (recoveredSnapshot.activeTurnId ?? null) : (persistedSession?.activeTurnId ?? null); @@ -2294,6 +2343,7 @@ export async function executeNativeSession( } : undefined, recoveredSnapshot.goal, + resumeInterruptedTurn, ) : Promise.resolve({ event: recoveryTerminal, @@ -2333,6 +2383,10 @@ export async function executeNativeSession( await session.goal({ action: "get", requestId }); } await checkpoint(); + } else if (shouldStartFreshTurn && recovered && checkpointedInterruption) { + restartContinuationStarted = true; + await session.startTurn(restartContinuation()); + await checkpoint(); } else if (shouldStartFreshTurn) { let modelEnvelope = recovered ? buildNativeModelEnvelope(input, { resumedSession: true }) diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 2fc54b97a25..96378b4bc2d 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -16,6 +16,7 @@ import { redactSensitiveText } from "../../redaction.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { reportRunFailure } from "../run-failure-report.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; +import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; export type NativeControllerIdentity = { bootId: string; @@ -764,7 +765,8 @@ export async function claimNativeRestartRecoveries(input: { hasProviderEvidence, checkpointFailed: (checkpointRecord.terminal as Record | undefined) - ?.runTerminalState === "failed", + ?.runTerminalState === "failed" && + nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, providerAttempt: row.coordinator.attempt, }); const ownershipChanged = diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 820661f8893..c73db0b848e 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -1018,6 +1018,30 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( ]); }); + it.each([2, 3])("recovers a checkpointed restart interruption within provider budget %s", async (attempt) => { + const fixture = await seedRun(`INTERRUPTED-${attempt}`); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { + driverKind: "codex_app_server", + sessionId: fixture.native.normalizedSessionId, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "interrupted-provider", + activeTurnId: null, + semanticResult: null, + terminal: { runTerminalState: "failed", turnTerminalState: "failed" }, + terminalTurns: [{ turnId: "interrupted-turn", fingerprint: JSON.stringify({ + terminalState: "failed", result: null, + error: { code: "provider_turn_lost_on_restore", recoverable: true }, + }) }], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const dispositions = await claimNativeRestartRecoveries({ + db: fixture.db, controller: successor, restartKind: "hot", runIds: [fixture.runId], + }); + expect(dispositions).toEqual([expect.objectContaining(attempt < 3 + ? { kind: "resume_dead_runner", runId: fixture.runId, providerAttempt: attempt } + : { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]); + }); + it("terminalizes the recorded failed-checkpoint incident atomically instead of resuming it on upgrade", async () => { const fixture = await seedRun("FAILED-CHECKPOINT"); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index ec28b68800e..981169652ec 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -8338,6 +8338,7 @@ async function executePaperclipNativeSessionWithinScope( trace.activate(runnerSessionStartupScope); const result = await trace.run(runnerSessionStartupScope, () => executeNativeSession({ + resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner", getFreshSessionHandoff: input.getFreshSessionHandoff, onSessionAdmission: async () => { // Invalidate prior stop evidence before a backend can spawn. diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index 80f98ee6f0f..e113899dfa4 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -128,6 +128,7 @@ export const validatePrpStructuredRunResult = runner.validatePrpStructuredRunResult; export const NativeProviderTerminalFailure = runner.NativeProviderTerminalFailure; +export const nativeRestartInterruptedTurnId = runner.nativeRestartInterruptedTurnId; export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTerminatedRemoteNativeSessionCleanup; export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup;