From 42bc238aa5a1208cabecba76773e0500fcd33959 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 01:08:19 -0600 Subject: [PATCH] fix(runner): keep a restart continuation when the lost turn's terminal is delivered again (#20196) 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 > - When a restart loses a native run's active turn, the run records it as `provider_turn_lost_on_restore` and continues in a same-conversation continuation turn > - #20118 taught the event consumer to skip a stale restart-lost terminal of another turn once a continuation is running. It applied only when the checkpoint already recorded the interruption > - When the checkpoint instead names the lost turn as active (the drain hit right after the first turn was submitted), the consumer starts the continuation on the first lost terminal. If the same lost terminal is then delivered again, it was taken as the run's terminal and failed the run > - This pull request applies the skip to any recovered run once its continuation has started. For a checkpointed interruption it still waits for the startup decision first > - The benefit is that a turn lost on restore always continues, however often its lost terminal is delivered ## Linked Issues or Issue Description Refs #20118 **What happened?** Local restart-chaos K13 regression storm w-20261109T055902-s9402 (master 9b71d03215): 94 of 95 passed. In t90 (claude M1 parent), run `e5eab3cb`: 1. The run submitted its first turn at 06:28:25, and a graceful drain followed at 06:28:28. 2. After restore at 06:28:41, the ACPX runner replayed the turn's accept and start, then reported `turn_lab_1abbeca2…` as `provider_turn_lost_on_restore`. 3. The controller started continuation `turn_lab_a3d009c6…`; the saved checkpoint names it as the active turn. 4. The run failed at 06:28:45 with `native_provider_terminal_failed: The runner restored the conversation after process loss, but the previous turn is no longer active.` and needed a Retry. **Expected behavior** Once the continuation is running, a later delivery of the lost turn's terminal is history, not the run's terminal. The run finishes on the continuation. **Steps to reproduce** 1. Recover a run whose checkpoint names the interrupted turn as active (not yet recorded as terminal). 2. Deliver that turn's restart-lost terminal, which starts the continuation, then deliver it again before the continuation completes. 3. Before this change, the second delivery ends the run as failed. **Paperclip version or commit** 9b71d03215 (wave). Branch rebased on current master. ## What Changed - `native-session-runtime.ts`: `supersededByRestartContinuation` applies to every recovered run. - Once a continuation has started, a restart-lost `turn.failed` for any turn other than that continuation keeps the consumer going. - With a checkpointed interruption, it still waits for the startup decision first. Before this change it excluded the checkpointed interrupted turn itself, which could also end a run if that turn's replayed terminal arrived before the startup continuation began. - `native-session-runtime.test.ts`: - a lost terminal delivered again after the continuation starts; - the checkpointed interrupted turn's own terminal arriving before the continuation starts. ## Verification - Red/green: both new cases fail on master (the run ends failed or on the stale terminal) and pass with the change. The #20118 cases still pass. - `vitest run` over the runner restart, recovery, continuation and terminal test files: 9 files, 389 tests passed. - `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm run typecheck:typescript` (`packages/paperclip-runner`): passed. ## Risks - Low risk. The skip still requires a recovered run, a recoverable `provider_turn_lost_on_restore` failure, a continuation that actually started, and a terminal for a different turn than that continuation. A loss of the continuation itself, real provider failures, and runs that started no continuation keep their current handling. ## 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/native-session-runtime.test.ts | 61 +++++++++++++++++-- .../src/native-session-runtime.ts | 15 +++-- 2 files changed, 65 insertions(+), 11 deletions(-) diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index fd5d7939cab..e09f3a35dad 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -587,9 +587,10 @@ describe("executeNativeSession recovery", () => { }); 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 }) => { + { label: "after the continuation starts", staleFirst: false, staleTurnId: "turn-unrecorded-continuation" }, + { label: "before the continuation starts", staleFirst: true, staleTurnId: "turn-unrecorded-continuation" }, + { label: "as the checkpointed interrupted turn before the continuation starts", staleFirst: true, staleTurnId: "turn-recovery" }, + ])("keeps the restart continuation when the restored runner reports an earlier unrecorded continuation lost $label", async ({ staleFirst, staleTurnId }) => { let staleAppended!: () => void; const staleTerminal = new Promise(resolve => { staleAppended = resolve; }); const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; @@ -607,7 +608,7 @@ describe("executeNativeSession recovery", () => { 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(1, "turn.failed", { status: "failed", error: failure }), turnId: staleTurnId }; yield { ...runnerEvent(2, "turn.completed", { status: "completed" }), turnId: "turn-continuation" }; }, startTurn, @@ -639,7 +640,7 @@ describe("executeNativeSession recovery", () => { })).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"]); + expect(events.filter(event => event.eventType === "turn.failed").map(event => event.turnId)).toEqual([staleTurnId]); }); it("recovers a restart continuation whose checkpoint still carries the lost turn's failed terminal", async () => { @@ -689,6 +690,56 @@ describe("executeNativeSession recovery", () => { .toBe(false); }); + it.each([ + { label: "after it starts", proposedFirst: false }, + { label: "after the continuation proposed its result", proposedFirst: true }, + ])("keeps the restart continuation when the lost turn's terminal is delivered again $label", async ({ proposedFirst }) => { + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "acpx_runtime", sessionId: "record-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-recovery", + terminalTurns: [], semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + 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; + if (proposedFirst) yield { ...runnerEvent(2, "run.result.proposed", result), turnId: "turn-continuation" }; + yield runnerEvent(3, "turn.failed", { status: "failed", error: failure }); + yield { ...runnerEvent(4, "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 port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event: PrpEvent) { 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: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + })).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } }); + expect(startTurn).toHaveBeenCalledOnce(); + }); + 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 6246376bee9..995b143d4f2 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1111,6 +1111,7 @@ async function consumeTurn( initialGovernedWait?: PersistedNativeSession["governedWait"], /** Settles a compaction-only turn, which reports no semantic result. */ compactionResult?: PrpStructuredRunResult, + supersededTerminal?: (event: PrpEvent, signal: AbortSignal) => Promise, ) { let timer: ReturnType | undefined; const appendAbort = new AbortController(); @@ -1504,6 +1505,7 @@ async function consumeTurn( } } if (isTurnTerminal(event)) { + if (await supersededTerminal?.(event, appendAbort.signal)) continue; if (providerFailure) throw providerFailure; if (compactionResult && event.eventType === "turn.completed") { return { @@ -2752,11 +2754,12 @@ export async function executeNativeSession( 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 (!recovered || event.eventType !== "turn.failed" || + !isNativeRestartInterruption(event.payload.error) || !event.turnId) return false; + if (checkpointedInterruption) { + await startupDecision; + signal.throwIfAborted(); + } if (!restartContinuationStarted) return false; await continuationSetup?.catch(() => undefined); signal.throwIfAborted(); @@ -2764,7 +2767,6 @@ export async function executeNativeSession( }; 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 @@ -2948,6 +2950,7 @@ export async function executeNativeSession( options.sessionCompaction ? sessionCompactionResult(input, options.sessionCompaction.requestId) : undefined, + supersededByRestartContinuation, ) : Promise.resolve({ event: recoveryTerminal,