diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 37f0d4605d1..73bff682ae6 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -586,6 +586,62 @@ describe("executeNativeSession recovery", () => { expect(checkpoints.some(snapshot => snapshot.terminalTurns?.some(turn => turn.turnId === "turn-recovery"))).toBe(true); }); + it.each([ + { label: "after the continuation starts", staleFirst: false }, + { label: "before the continuation starts", staleFirst: true }, + ])("keeps the restart continuation when the restored runner reports an earlier unrecorded continuation lost $label", async ({ staleFirst }) => { + let staleAppended!: () => void; + const staleTerminal = new Promise(resolve => { staleAppended = resolve; }); + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const interrupted: PersistedNativeSession = { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: null, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + const startTurn = vi.fn(async () => { + if (staleFirst) await staleTerminal; + return { 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 }), turnId: "turn-unrecorded-continuation" }; + 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 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 events: PrpEvent[] = []; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event: PrpEvent) { + events.push(event); + if (event.eventType === "turn.failed") staleAppended(); + return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + await expect(executeNativeSession({ + input, backend, controlPlane: port, persistedSession: interrupted, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + })).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } }); + expect(startTurn).toHaveBeenCalledOnce(); + expect((startTurn.mock.calls[0]![0] as Parameters[0]).continuation).toBe(true); + expect(events.filter(event => event.eventType === "turn.failed").map(event => event.turnId)).toEqual(["turn-unrecorded-continuation"]); + }); + it.each(["acpx_runtime", "opencode_server"] as const)("continues a restart-lost %s turn once in the same conversation", async (driverKind) => { const qualified = resolveQualifiedAcpxProfile("claude", "claude-sonnet-5"); const acpxInput: NativeExecutionInputV1 = (driverKind === "acpx_runtime" ? { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 397c1d69f70..180521404b6 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2704,6 +2704,9 @@ export async function executeNativeSession( }; let restartContinuationStarted = false; let borrowedContinuationTurnId: string | null = null; + let restartContinuationTurnId: string | null = null; + let settleStartupDecision: () => void = () => undefined; + const startupDecision = new Promise(resolve => { settleStartupDecision = resolve; }); let continuationDispositionRequested = false; // The consumer may already be running when a continuation starts, and the // continuation can complete before its start checkpoint settles. A @@ -2718,6 +2721,7 @@ export async function executeNativeSession( const steeredMessages = await options.getUnreadSteeredMessages?.() ?? []; signal?.throwIfAborted(); const started = await session.startTurn(restartContinuation(steeredMessages)); + restartContinuationTurnId = started.turnId; if (started.borrowedDispositionMarker === true) borrowedContinuationTurnId = started.turnId; await checkpoint(signal); })(); @@ -2740,8 +2744,20 @@ export async function executeNativeSession( await checkpoint(signal); return true; }; + const supersededByRestartContinuation = async (event: PrpEvent, signal: AbortSignal) => { + if (!recovered || !checkpointedInterruption || event.eventType !== "turn.failed" || + !isNativeRestartInterruption(event.payload.error) || !event.turnId || + event.turnId === checkpointedInterruption) return false; + await startupDecision; + signal.throwIfAborted(); + if (!restartContinuationStarted) return false; + await continuationSetup?.catch(() => undefined); + signal.throwIfAborted(); + return restartContinuationTurnId !== null && event.turnId !== restartContinuationTurnId; + }; const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => { if (await requestContinuationDisposition(event, signal)) return true; + if (await supersededByRestartContinuation(event, signal)) return true; if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn || // A compaction-only run never continues task work after a restart. // Runnerd records a restart-lost turn only for these providers, after @@ -3044,7 +3060,9 @@ export async function executeNativeSession( }); await checkpoint(); } + settleStartupDecision(); } catch (error) { + settleStartupDecision(); // Consumption starts before provider launch so eager events cannot be // lost. If launch or its checkpoint fails, abort any in-flight append // before joining cleanup so the failed turn cannot commit late or