diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7f74393e1f9..2f38000c3f7 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2572,6 +2572,15 @@ result that exact turn proposed, recovery also settles the turn with that result accepted result instead of failing and rerunning finished work. A missing turn without such a result still fails closed, because its outcome is unknown. +Each admitted restart attempt that loses a turn this way gets its own continuation. +When a later restart loses the continuation turn itself, its outcome is also +unknown, so the marker re-arms and the next admitted attempt continues again. The +provider-attempt budget bounds the repeats; an exhausted budget still blocks with +Retry. A semantic result that reaches the driver before its turn start settles +waits for that start, like a terminal, instead of failing the new turn. If this +session's own protocol failure ends a pending turn start, the run reports +`codex_turn_start_protocol_failure` with the cause instead of a runner exit. + 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. In particular, a done report from a run whose task now belongs to another agent, a user or nobody is kept as an accepted result but records `completion_owner_superseded` and preserves the task status for the current owner. Historical ambiguous runs are never automatically replayed after an upgrade. When that deadline fails a native run with a prepared workspace, the same diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index fdb0f7461b7..333062e4703 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -103,6 +103,8 @@ export function nativeRestartInterruptedTurnId(snapshot: Pick): string | null { // Runnerd records this cause for Codex app-server and ACPX sessions after // proving the old provider lifetime ended and restoring the conversation. + // A lost continuation re-arms the driver's marker before it is checkpointed, + // so a consumed marker here still means the allowance is used. if (!RESTART_CONTINUATION_DRIVER_KINDS.has(snapshot.driverKind ?? "") || snapshot.activeTurnId != null || snapshot.semanticResult != null || snapshot.dispositionOnlyRecoveryConsumed || snapshot.goal != null || snapshot.pendingRuntimeRequests?.length) return null; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts index 19e218768af..dadb9e93ae2 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts @@ -46,6 +46,51 @@ import { import { RUNNERD_RUNTIME_ACTIVITY } from "./codex-driver-values.js"; describe("Codex app-server Codex driver", () => { + it("binds a semantic result that arrives before its pending turn start settles", async () => { + // Runnerd can deliver the new turn's semantic result before turn/start's + // response binds the turn (as after a restart continuation). Like a + // terminal, the result waits for the start to settle instead of failing + // the session as an unbound result. + const transport = new FakeCodexTransport(); + let releaseStart!: (value: Record) => void; + transport.turnStartResponse = new Promise(resolve => { releaseStart = resolve; }); + const session = await makeDriver([transport]).openSession({ + runId: "early-result", normalizedSessionId: "early-result-session", workingDirectory: WORKSPACE, + }); + const starting = session.startTurn({ message: { role: "user", text: "Finish." } }); + await new Promise(resolve => setTimeout(resolve, 0)); + transport.push("paperclip/runResult", { threadId: "thread-1", turnId: "turn-1", result }); + await new Promise(resolve => setTimeout(resolve, 0)); + releaseStart({ turn: { id: "turn-1", status: "inProgress", items: [] } }); + await expect(starting).resolves.toMatchObject({ turnId: "turn-1" }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } }); + const events = await collectUntilTerminal(session.events()); + expect(events.some(event => event.eventType === "session.failed")).toBe(false); + expect(events.at(-1)?.eventType).toBe("turn.completed"); + expect(events.filter(event => event.eventType === "run.result.proposed")) + .toEqual([expect.objectContaining({ turnId: "turn-1" })]); + await session.close({ reason: "fixture complete" }); + }); + + it("names its own protocol failure when that failure ends a pending turn start", async () => { + // The protocol failure closes the transport, so runnerd exits and the + // pending start reports a runner exit. The run must keep the real cause. + const transport = new FakeCodexTransport(); + let rejectStart!: (error: Error) => void; + transport.turnStartResponse = new Promise((_resolve, reject) => { rejectStart = reject; }); + const session = await makeDriver([transport]).openSession({ + runId: "start-protocol-failure", normalizedSessionId: "start-protocol-failure-session", workingDirectory: WORKSPACE, + }); + const starting = session.startTurn({ message: { role: "user", text: "Finish." } }); + await new Promise(resolve => setTimeout(resolve, 0)); + transport.push("item/started", { threadId: "thread-1", turnId: "unbound-turn", item: { id: "item-1", type: "agentMessage" } }); + await new Promise(resolve => setTimeout(resolve, 0)); + rejectStart(new Error("runnerd exited before provider turn startup")); + await expect(starting).rejects.toThrow(/codex_turn_start_protocol_failure: turn_binding_mismatch/); + await expect(starting).rejects.not.toThrow(/runnerd exited/); + await session.close({ reason: "fixture complete" }).catch(() => undefined); + }); + it("rejects provider-forged and previous-run activity notifications", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport]).openSession({ 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 8b1aca6caef..2ca4e79dabf 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 @@ -42,7 +42,7 @@ import { type PrpEvent, type PrpStructuredRunResult, } from "./codex-app-server-driver.test-support.js"; -import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; +import { NativeSessionProtocolIntegrityError, nativeRestartInterruptedTurnId } from "../../contracts/native-session-backend.js"; function cursorProviderIdentity(mode: unknown) { return { @@ -62,6 +62,66 @@ function cursorProviderIdentity(mode: unknown) { } describe("Codex app-server Codex driver", () => { + it("offers another restart continuation when a later restart loses the continuation turn itself", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-two-restarts", normalizedSessionId: "session-two-restarts", workingDirectory: WORKSPACE, + }); + const lost = { code: "provider_turn_lost_on_restore", recoverable: true }; + await session.startTurn({ message: { role: "user", text: "Finish the request." } }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", error: lost, items: [] } }); + await collectUntilTerminal(session.events()); + expect(nativeRestartInterruptedTurnId({ ...await session.snapshot(), semanticResult: null })).toBe("turn-1"); + // The first restart's continuation is accepted, then a second restart + // loses it before any result. + transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + await session.startTurn({ message: { role: "user", text: "Continue." }, continuation: true }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "failed", error: lost, items: [] } }); + await collectUntilTerminal(session.events()); + const snapshot = await session.snapshot(); + // The lost continuation re-arms the marker exactly like the first loss. + expect(snapshot.dispositionOnlyRecoveryConsumed).toBe(false); + expect(nativeRestartInterruptedTurnId({ ...snapshot, semanticResult: null })).toBe("turn-2"); + // A consumed marker still means the allowance is used. + expect(nativeRestartInterruptedTurnId({ ...snapshot, semanticResult: null, dispositionOnlyRecoveryConsumed: true })).toBeNull(); + await session.close({ reason: "fixture complete" }); + }); + + it("re-arms the continuation when recovery finds the checkpointed continuation turn lost", async () => { + // Controller lost during the first continuation: the checkpoint names the + // continuation as the active turn and as the consumed marker's turn. + const lost = { code: "provider_turn_lost_on_restore", recoverable: true }; + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-lost-continuation", normalizedSessionId: "session-lost-continuation", 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: lost, items: [] } }); + await collectUntilTerminal(original.events()); + first.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + await original.startTurn({ message: { role: "user", text: "Continue." }, continuation: true }); + const checkpoint = await original.snapshot(); + expect(checkpoint).toMatchObject({ activeTurnId: "turn-2", dispositionOnlyRecoveryConsumed: true, + dispositionOnlyRecoveryTurnId: "turn-2" }); + await original.close({ reason: "controller lost during the continuation" }); + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [ + { id: "turn-1", status: "failed", error: lost, items: [] }, + { id: "turn-2", status: "failed", error: lost, items: [] }, + ] } }; + const recovery = await driver.recoverSession!(checkpoint); + expect(recovery).toMatchObject({ recovered: true }); + const recovered = recovery.session!; + const restored = await recovered.snapshot(); + expect(restored.dispositionOnlyRecoveryConsumed).toBe(false); + expect(nativeRestartInterruptedTurnId({ ...restored, semanticResult: null })).toBe("turn-2"); + second.turnStartResponse = Promise.resolve({ turn: { id: "turn-3", status: "inProgress", items: [] } }); + await expect(recovered.startTurn({ message: { role: "user", text: "Continue." }, continuation: true })) + .resolves.toMatchObject({ turnId: "turn-3" }); + await recovered.close({ reason: "fixture complete" }); + }); + it.each(["agent", "plan", "ask"])("retains native Cursor %s mode through checkpoint and recovery", async (mode) => { const identity = cursorProviderIdentity(mode); const first = new FakeCodexTransport("thread-1", "provider-session-1", identity); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 813b69452d0..eb60ed356b5 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -1,7 +1,7 @@ import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from "./codex-history.js"; import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js"; import { randomUUID } from "node:crypto"; -import { isNativeRestartInterruption, NativeProviderTerminalFailure } from "../../contracts/native-session-backend.js"; +import { isNativeRestartInterruption, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; import type { HarnessGoalOperation, @@ -325,6 +325,14 @@ export class CodexHarnessSession this.terminal = true; } } + if (this.protocolFailed && !(error instanceof CodexRpcError) && !(error instanceof NativeSessionProtocolIntegrityError)) { + // This session's own protocol failure closed the transport, so the + // start then reports a lost runner. Keep the cause that ended it. + throw new Error( + `codex_turn_start_protocol_failure: ${this.protocolFailureCode}: ${this.protocolFailureMessage}`, + { cause: error }, + ); + } throw error; } finally { this.turnStartPending = false; 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 26c79d213cf..3e35bef3d68 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -449,6 +449,10 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex return; } if (notification.method === "paperclip/runResult") { + // Like a terminal, a semantic result can reach the notification channel + // before turn/start's response binds its turn (for example, a restart + // continuation). Bind it only after that start settles. + await state.turnStartSettled; if ( threadId !== state.opened.threadId || !state.notificationNamesActiveTurn(turnId, "semantic result") 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 777a061a9ac..f2e38af1e4c 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts @@ -175,9 +175,16 @@ export function mapTerminalTurn( state.emit(eventType, payload, { turnId }); finalize(state, status); if (status === "failed" && isNativeRestartInterruption(turn.error) && state.result === null && - !state.dispositionOnlyRecoveryConsumed) { + (!state.dispositionOnlyRecoveryConsumed || state.dispositionOnlyRecoveryTurnId === turnId)) { // Orchestration may use the paid restart attempt for one same-thread // continuation. Its accepted turn id uses the existing durable marker. + // A later restart that loses that continuation turn itself leaves its + // outcome unknown too, so the next admitted restart attempt may + // continue again; the provider attempt budget bounds the repeats. + // Reset the marker so that continuation is submitted, checkpointed and + // adopted after a crash exactly like the first one. + state.dispositionOnlyRecoveryConsumed = false; + state.dispositionOnlyRecoveryTurnId = null; state.terminal = false; state.dispositionOnlyRecoveryAvailable = true; } 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 92e8c5517a2..4744f0523dc 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 @@ -2509,8 +2509,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( { attempt: 3, driverKind: "codex_app_server" }, { attempt: 2, driverKind: "acpx_runtime" }, { attempt: 2, driverKind: "opencode_server" }, - ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind }) => { - const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}`); + // A second restart that loses the first restart's continuation turn. + { attempt: 2, driverKind: "codex_app_server", lostContinuation: true }, + ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation }) => { + const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}`); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { driverKind, sessionId: fixture.native.normalizedSessionId, @@ -2523,6 +2525,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( terminalState: "failed", result: null, error: { code: "provider_turn_lost_on_restore", recoverable: true }, }) }], + // A restart during the first continuation: its turn is still the + // checkpointed active turn and the consumed marker names it. + ...(lostContinuation ? { activeTurnId: "continuation-turn", terminal: null, + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "continuation-turn" } : {}), } } }).where(eq(heartbeatRuns.id, fixture.runId)); await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId)); const dispositions = await claimNativeRestartRecoveries({