From ef83d566cacf1d5f740d701560597dad2bfa4619 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:22:37 -0600 Subject: [PATCH] fix(native): ask for the result once after a restart continuation ends without one (#19771) 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 > - Native runs continue a turn that a control-plane restart lost, in the same conversation (#19501, #19699) > - In restart storms 5, 6b and 8, ACPX Claude runs crossed restarts, ran the continuation, and then failed with `native_session_retry_exhausted` > - The continuation completed, but Claude ended it with "I'll wait for the notification when it completes" for a background command, and did not call `paperclip_finish` > - The next same-run attempt then sent no turn at all and failed with "session returned no semantic result" > - This pull request makes the continuation ask for the result, and asks once more in the same attempt if the continuation still ends without one > - The benefit is that a restart-interrupted task settles without a Retry, and the restarts that the run survived cannot block the request ## Linked Issues or Issue Description Refs #19501, #19699, #19720 **What happened?** Storm 8, run 84c532f7 and its Retry 0faead66 (ACPX Claude, task CHA-5). The steps are the same in both runs: 1. `resume_dead_runner`, then `session.resumed`, then `turn.failed` with `provider_turn_lost_on_restore` (recoverable). 2. The restart continuation runs. Claude starts the bench in the background and ends the turn with "I'll wait for the notification when it completes". The turn completes with no `paperclip_finish`. 3. Lifecycle: "provider turn completed without a semantic result; same-run disposition recovery persisted" (attempt 2). 4. 30 s later, attempt 3 restores the session. It sends no turn. After about 2 s it fails with `native_finalization_missing: session returned no semantic result`, and the attempt count makes it `native_session_retry_exhausted`. **Expected behavior** The continuation tells the agent to finish and report its result. If the continuation still completes without a result, Paperclip asks for the disposition one time. Restarts that the run already survived do not block that request. An agent that never reports a result still fails closed. **Root cause** - `packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts` `startTurn` (lines 224 and 248): the restart continuation uses the driver's one-shot disposition marker (`dispositionOnlyRecoveryAvailable`). #19699 uses the marker on purpose, so that a crash can adopt the accepted continuation turn. The marker then stays consumed with the continuation turn id. - `packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts` (line 226): only a *lost* continuation gives the marker back. A *completed* continuation with no result does not. - `packages/paperclip-runner/src/native-session-runtime.ts` (`dispositionRecoveryWasSubmitted`, about line 2761): on the next attempt, the consumed marker looks like a disposition request that already ran. Recovery replays the continuation's terminal and starts no turn. So the disposition request never happens. - `server/src/services/native-runtime/native-session-executor.ts` (lines 9755-9761 and `nativeSessionFailureDisposition` at 7768): the `semantic_result_missing` same-run retry charges a provider attempt on the same counter as restart resumes. In this run, attempt 3 was the last allowed attempt. - The continuation prompt (`restartContinuation`) did not tell the agent to report a result, or not to stop and wait for a background notification. ## What Changed - Codex-protocol driver (used by Codex and by ACPX through runnerd): when a restart continuation completes without a semantic result, the driver gives back the one-shot disposition marker. The disposition request is not a continuation, so it consumes the marker for good. - Native session runtime: after that, the same attempt sends one disposition request (`Paperclip semantic-result recovery after a restart continuation ... invoke exactly one paperclip_finish or paperclip_block`). It does this only when the driver shows that the continuation borrowed the marker and gave it back. Other drivers keep the current behavior. - Restart continuation prompt: it now says to finish in this turn, not to end the turn to wait for a background command or a notification, and to report one `paperclip_finish` or `paperclip_block`. - `doc/execution-semantics.md`: one additive paragraph. - No server change. The disposition request runs in the same attempt, so it uses no provider attempt. This avoids `native-restart-recovery.ts` (#19747). ## Verification Red before the fix, green after: - `codex-app-server-driver.recovery.test.ts` "keeps one disposition request after a restart continuation completes without a result". Red: marker stays consumed on `turn-2`. Green: the marker is given back, one disposition turn starts, and a resultless disposition turn does not re-arm. - `native-session-runtime.test.ts` "settles a restart continuation that ended without a result through one disposition turn (Codex driver)". Real Codex-protocol driver with a fake transport. Red: only one `turn/start`, then `native_finalization_missing`. Green: the disposition turn `turn-3` reports the result and the run completes. - `native-session-runtime.test.ts` "asks once for the disposition when a restart continuation ends without a result" and "fails closed when the driver keeps the continuation's disposition marker". Also run: - `npx vitest run src/native-session-runtime.test.ts src/drivers/codex/` in `packages/paperclip-runner`: 593 passed. - `src/live/runnerd-restart-continuation.test.ts` (real runnerd and fake Codex, killed runner): 5 passed. - `tsc --noEmit` for `packages/paperclip-runner`. - `pnpm verify` exits 2 because the doc change is a global CI input. Required CI supplies the broad coverage. Not run: a live replay of storm 8 t05 with Claude. The chaos harness needs provider keys from the operator's shell. ## Risks - Low to medium. A run whose continuation ends without a result now gets one more provider turn in the same attempt. The bound is one disposition request per continuation. Each continuation still needs an admitted restart, and the provider-attempt budget bounds those. - If the controller crashes after the continuation completes but before its checkpoint, the checkpoint keeps the stricter consumed marker. That case fails closed as before. - The prompt text changes for restart continuations. It does not change a protocol schema or a profile digest. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (shell, file edits, code 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 --- doc/execution-semantics.md | 10 + .../src/contracts/harness-driver.ts | 6 + .../src/contracts/native-session-backend.ts | 6 + .../codex-app-server-driver.recovery.test.ts | 34 +++ .../drivers/codex/codex-harness-session.ts | 3 + .../src/drivers/codex/codex-session-state.ts | 8 + .../drivers/codex/codex-session-terminal.ts | 12 + .../src/native-session-runtime.test.ts | 205 ++++++++++++++++++ .../src/native-session-runtime.ts | 69 +++++- ...unner-restart-recovery.integration.test.ts | 27 ++- 10 files changed, 370 insertions(+), 10 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 105d0abc60..7ee6c0e10d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2641,6 +2641,16 @@ 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. +The continuation tells the agent to finish in that turn, not to end it while +waiting for a background command or a notification, and to report one +`paperclip_finish` or `paperclip_block`. The continuation only borrows the +one-shot disposition marker so a crash can adopt its accepted turn. If the +continuation completes without a semantic result, the driver returns the marker +and the same attempt sends one disposition request. That request does not use +another provider attempt, so restarts the run already survived cannot block it. +A disposition request that also ends without a result consumes the marker and +fails closed as before. + 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/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 6e8e6ee4cc..22dc49246d 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -582,6 +582,12 @@ export interface HarnessSession { }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan"; + /** + * This restart continuation borrowed the driver's one-shot disposition + * marker. If it completes without a result the driver returns the marker, + * and orchestration may send one disposition request in the same attempt. + */ + borrowedDispositionMarker?: true; }>; steer?(input: { /** Queued follow-ups remain distinct from active-turn steering. */ diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 0fee816ce9..f994ece9d0 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -224,6 +224,12 @@ export interface NativeSession { }): Promise<{ turnId: string; effectiveCollaborationMode?: "default" | "plan"; + /** + * This restart continuation borrowed the driver's one-shot disposition + * marker. If it completes without a result the driver returns the marker, + * and orchestration may send one disposition request in the same attempt. + */ + borrowedDispositionMarker?: true; }>; steer?(input: { mode?: "steer" | "follow_up"; 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 df682da37e..4fc33ec527 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 @@ -63,6 +63,40 @@ function cursorProviderIdentity(mode: unknown) { } describe("Codex app-server Codex driver", () => { + it("keeps one disposition request after a restart continuation completes without a result", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-resultless-continuation", normalizedSessionId: "session-resultless-continuation", 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()); + transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + await session.startTurn({ message: { role: "user", text: "Continue." }, continuation: true }); + // The provider ends the continuation without paperclip_finish, for + // example while it says it will wait for a background command. + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } }); + await collectUntilTerminal(session.events()); + const afterContinuation = await session.snapshot(); + // The continuation borrowed the one-shot marker for crash adoption; it was + // not the disposition request, so the allowance is still available. + expect(afterContinuation).toMatchObject({ activeTurnId: null, semanticResult: null, + dispositionOnlyRecoveryConsumed: false, dispositionOnlyRecoveryTurnId: null }); + expect(nativeRestartInterruptedTurnId({ ...afterContinuation, semanticResult: null })).toBeNull(); + transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-3", status: "inProgress", items: [] } }); + await expect(session.startTurn({ message: { role: "user", text: "Report the disposition." } })) + .resolves.toMatchObject({ turnId: "turn-3" }); + expect(await session.snapshot()).toMatchObject({ dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-3" }); + // The disposition request itself is the bound: a resultless answer to it + // does not re-arm another request. + transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-3", status: "completed", items: [] } }); + await collectUntilTerminal(session.events()); + expect(await session.snapshot()).toMatchObject({ dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-3" }); + await expect(session.startTurn({ message: { role: "user", text: "Again." } })).rejects.toThrow(); + await session.close({ reason: "fixture complete" }); + }); + 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({ 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 9026c93ef9..5a4285942d 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -152,6 +152,7 @@ export class CodexHarnessSession this.resultTurnId = null; this.dispositionOnlyRecoveryConsumed = false; this.dispositionOnlyRecoveryTurnId = null; + this.restartContinuationTurnId = null; this.terminal = false; this.terminalTurns.clear(); this.turnStarted = false; @@ -368,6 +369,7 @@ export class CodexHarnessSession if (dispositionOnlyRecovery) { this.dispositionOnlyRecoveryTurnId = turnId; } + this.restartContinuationTurnId = dispositionOnlyRecovery && continuationTurn ? turnId : null; this.emit("turn.accepted", { turnId }, { turnId }); if (this.interruptQueued) { this.interruptQueued = false; @@ -376,6 +378,7 @@ export class CodexHarnessSession return { turnId, effectiveCollaborationMode, + ...(this.restartContinuationTurnId === turnId ? { borrowedDispositionMarker: true as const } : {}), }; } 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 b2f797a573..7e9e9cf862 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -153,6 +153,14 @@ export class CodexSessionState { dispositionOnlyRecoveryAvailable = false; dispositionOnlyRecoveryConsumed = false; dispositionOnlyRecoveryTurnId: string | null = null; + /** + * A restart continuation borrows the one-shot disposition marker so a + * crash during it can adopt the accepted turn. It is not the disposition + * request: if it completes without a result, the allowance is returned. + * In memory only; a checkpoint taken during the continuation keeps the + * stricter consumed marker. + */ + restartContinuationTurnId: string | null = null; turnStarted = false; readonly terminalTurns = new Map(); /** 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 7332e9ad02..f344b5af03 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-terminal.ts @@ -236,6 +236,18 @@ export function mapTerminalTurn( state.dispositionOnlyRecoveryTurnId = null; state.terminal = false; state.dispositionOnlyRecoveryAvailable = true; + } else if (eventType === "turn.completed" && state.result === null && state.conversationMode === "task" && + state.currentGoal?.status !== "active" && state.restartContinuationTurnId === turnId && + state.dispositionOnlyRecoveryConsumed && state.dispositionOnlyRecoveryTurnId === turnId) { + // The restart continuation only borrowed the marker. It ended without a + // result (for example, the agent said it would wait for a background + // command), so the one disposition request it displaced is still owed. + // That request is not a continuation and consumes the marker for good. + state.restartContinuationTurnId = null; + state.dispositionOnlyRecoveryConsumed = false; + state.dispositionOnlyRecoveryTurnId = null; + state.terminal = false; + state.dispositionOnlyRecoveryAvailable = true; } } diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 0f2cc6840d..2f86ee4308 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -337,6 +337,57 @@ describe("executeNativeSession recovery", () => { expect(executionInput.workspace).toEqual(originalWorkspace); }); + it("settles a restart continuation that ended without a result through one disposition turn (Codex driver)", async () => { + const lost = { code: "provider_turn_lost_on_restore", recoverable: true }; + const originalTransport = new FakeCodexTransport(); + const recoveredTransport = new FakeCodexTransport(); + const transports = [originalTransport, recoveredTransport]; + const executionInput = { ...input, workspace: { ...input.workspace, cwd: WORKSPACE } }; + const backend = createCodexNativeSessionBackend(executionInput, { + runnerInstanceId: "runner-recovery", + transportFactory: () => transports.shift()!, + }); + const original = await backend.openSession({ identity, workingDirectory: WORKSPACE }); + await original.startTurn({ message: { role: "user", text: "Finish the assigned review." } }); + const checkpoint = await original.snapshot(); + expect(checkpoint).toMatchObject({ activeTurnId: "turn-1", semanticResult: null }); + await original.close({ reason: "controller lost during the turn" }); + recoveredTransport.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "turn-1", status: "failed", error: lost, items: [] }] } }; + const turnStarts = () => recoveredTransport.calls.filter(call => call.method === "turn/start"); + Object.defineProperty(recoveredTransport, "turnStartResponse", { + get: () => Promise.resolve({ turn: { id: `turn-${turnStarts().length + 1}`, status: "inProgress", items: [] } }), + }); + const checkpoints: PersistedNativeSession[] = []; + const completeRun = vi.fn(async () => undefined); + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession(value) { checkpoints.push(structuredClone(value)); }, + async appendEvent(event: PrpEvent) { return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + completeRun, + }; + const execution = executeNativeSession({ + input: executionInput, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", resumeInterruptedTurn: true, + }); + void execution.catch(() => undefined); + await vi.waitFor(() => expect(turnStarts()).toHaveLength(1)); + // Like the RC-60 Claude turns: the continuation ends saying it will wait + // for a background command, with no paperclip_finish. + recoveredTransport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } }); + await vi.waitFor(() => expect(turnStarts()).toHaveLength(2)); + const disposition = JSON.stringify(turnStarts()[1]!.params); + expect(disposition).toContain("Paperclip semantic-result recovery after a restart continuation"); + recoveredTransport.push("paperclip/runResult", { threadId: "thread-1", turnId: "turn-3", result }); + recoveredTransport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-3", status: "completed", items: [] } }); + await expect(execution).resolves.toMatchObject({ result, turnId: "turn-3" }); + expect(completeRun).toHaveBeenCalledOnce(); + expect(turnStarts()).toHaveLength(2); + expect(checkpoints.at(-1)).toMatchObject({ semanticResult: result, dispositionOnlyRecoveryConsumed: true, + dispositionOnlyRecoveryTurnId: "turn-3" }); + }); + it.each([ "completed", "pending", "duplicate", "conflicting", "other turn", "newer turn", "wrong session", "invalid", "restore-lost live", "restore-lost read", @@ -584,6 +635,160 @@ describe("executeNativeSession recovery", () => { expect(nativeRestartInterruptedTurnId(interrupted)).toBe("turn-recovery"); }); + it.each([ + { label: "asks once for the disposition when a restart continuation ends without a result", driverReArms: true }, + { label: "fails closed when the driver keeps the continuation's disposition marker", driverReArms: false }, + ])("$label", async ({ driverReArms }) => { + 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 lostTerminal = { turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }; + const completedTerminal = (turnId: string) => ({ turnId, fingerprint: JSON.stringify({ terminalState: "completed", error: null, result: null }) }); + let starts = 0; + const startTurn = vi.fn(async () => ++starts === 1 + ? { turnId: "turn-continuation", borrowedDispositionMarker: true as const } : { turnId: "turn-disposition" }); + let continuationDone = false; + 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; + // The agent ends the continuation saying it will wait for a + // background command, without paperclip_finish. + continuationDone = true; + yield { ...runnerEvent(2, "turn.completed", { status: "completed", error: null }), turnId: "turn-continuation" }; + if (startTurn.mock.calls.length < 2) return; + yield { ...runnerEvent(3, "turn.completed", { status: "completed", error: null }), turnId: "turn-disposition" }; + }, + startTurn, + async result() { return startTurn.mock.calls.length >= 2 ? { result, terminal, turnId: "turn-disposition" } : null; }, + async snapshot() { + const calls = startTurn.mock.calls.length; + if (calls === 0) return { ...checkpoint, activeTurnId: null, terminalTurns: [lostTerminal] }; + // The continuation borrows the driver's one-shot marker while it runs. + if (calls === 1 && !continuationDone) return { ...checkpoint, activeTurnId: "turn-continuation", terminalTurns: [lostTerminal], + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-continuation" }; + if (calls === 1) return { ...checkpoint, activeTurnId: null, terminalTurns: [lostTerminal, completedTerminal("turn-continuation")], + dispositionOnlyRecoveryConsumed: !driverReArms, dispositionOnlyRecoveryTurnId: driverReArms ? null : "turn-continuation" }; + return { ...checkpoint, activeTurnId: null, terminalTurns: [lostTerminal, completedTerminal("turn-continuation"), completedTerminal("turn-disposition")], + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-disposition" }; + }, + 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() {}, + }; + const execution = executeNativeSession({ + input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + }); + if (!driverReArms) { + await expect(execution).rejects.toThrow("native_finalization_missing: session returned no semantic result"); + expect(startTurn).toHaveBeenCalledOnce(); + return; + } + await expect(execution).resolves.toMatchObject({ result, turnId: "turn-disposition" }); + expect(startTurn).toHaveBeenCalledTimes(2); + const continuation = startTurn.mock.calls[0]![0] as Parameters[0]; + expect(continuation.message.text).toContain("paperclip_finish"); + expect(continuation.message.text).toContain("background"); + const disposition = startTurn.mock.calls[1]![0] as Parameters[0]; + expect(disposition.continuation).toBeUndefined(); + expect(disposition.message.text).toContain("Paperclip semantic-result recovery"); + expect(disposition.message.text).toContain("paperclip_finish or paperclip_block"); + expect(disposition.message.text).not.toContain(input.task.prompt); + }); + + it("asks for the disposition when the continuation completes before its start checkpoint settles", async () => { + // The checkpoint already records the lost turn, so the continuation starts + // while the event consumer is running. The provider completes it at once: + // the driver has returned the marker before any post-start snapshot, and + // the completion reaches the consumer while the start checkpoint is held. + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const lostTerminal = { turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }; + const completedTerminal = (turnId: string) => ({ turnId, fingerprint: JSON.stringify({ terminalState: "completed", error: null, result: null }) }); + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "acpx_runtime", sessionId: "record-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: null, + terminalTurns: [lostTerminal], semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + let releaseStartCheckpoint!: () => void; + const startCheckpointHeld = new Promise(resolve => { releaseStartCheckpoint = resolve; }); + let continuationCompletionSeen = false; + let starts = 0; + const startTurn = vi.fn(async () => ++starts === 1 + ? { turnId: "turn-continuation", borrowedDispositionMarker: true as const } : { turnId: "turn-disposition" }); + let wakeEvents!: () => void; + let eventsWoken = new Promise(resolve => { wakeEvents = resolve; }); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + while (startTurn.mock.calls.length < 1) { await eventsWoken; eventsWoken = new Promise(resolve => { wakeEvents = resolve; }); } + continuationCompletionSeen = true; + yield { ...runnerEvent(2, "turn.completed", { status: "completed", error: null }), turnId: "turn-continuation" }; + while (startTurn.mock.calls.length < 2) { await eventsWoken; eventsWoken = new Promise(resolve => { wakeEvents = resolve; }); } + yield { ...runnerEvent(3, "turn.completed", { status: "completed", error: null }), turnId: "turn-disposition" }; + }, + startTurn: async (value) => { const started = await startTurn(value); wakeEvents(); return started; }, + async result() { return startTurn.mock.calls.length >= 2 ? { result, terminal, turnId: "turn-disposition" } : null; }, + async snapshot() { + const calls = startTurn.mock.calls.length; + if (calls === 0) return structuredClone(checkpoint); + if (calls === 1) return { ...checkpoint, terminalTurns: [lostTerminal, completedTerminal("turn-continuation")], + dispositionOnlyRecoveryConsumed: false, dispositionOnlyRecoveryTurnId: null }; + return { ...checkpoint, terminalTurns: [lostTerminal, completedTerminal("turn-continuation"), completedTerminal("turn-disposition")], + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-disposition" }; + }, + 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 persisted: PersistedNativeSession[] = []; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession(snapshot) { + // Hold the continuation's start checkpoint until the consumer has + // already received the continuation's completion. + if (startTurn.mock.calls.length === 1 && persisted.every(value => value.terminalTurns?.length !== 2)) { + await vi.waitFor(() => expect(continuationCompletionSeen).toBe(true)); + await new Promise(resolve => setTimeout(resolve, 20)); + releaseStartCheckpoint(); + } + persisted.push(structuredClone(snapshot)); + }, + 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-disposition" }); + await startCheckpointHeld; + expect(startTurn).toHaveBeenCalledTimes(2); + expect((startTurn.mock.calls[0]![0] as { continuation?: true }).continuation).toBe(true); + expect(startTurn.mock.calls[1]![0].message.text).toContain("Paperclip semantic-result recovery after a restart continuation"); + }); + it("never continues task work when a restart loses a compaction turn", async () => { const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; const checkpoint: PersistedNativeSession = { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 471236db58..a0897f4031 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2642,6 +2642,7 @@ export async function executeNativeSession( const snapshot = await session.snapshot(signal ? { signal } : undefined); signal?.throwIfAborted(); await persistCheckpoint(snapshot, signal); + return snapshot; }; const recoveredSnapshot = await session.snapshot(); const restartContinuation = () => ({ @@ -2652,6 +2653,8 @@ export async function executeNativeSession( "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.", + "Finish in this turn. Nothing resumes this conversation after the turn ends, so do not end it to wait for a background command or a notification; wait for or check that work first.", + "Then invoke exactly one paperclip_finish or paperclip_block with the accurate disposition.", ].join("\n"), completion: { revision: input.completionContract.contract.revision, @@ -2661,8 +2664,67 @@ export async function executeNativeSession( continuation: true as const, requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const, }); + // A restart continuation borrows the driver's one-shot disposition + // marker so a crash during it can adopt the accepted turn. A driver that + // hands the marker back when the continuation completes without a result + // still owes that one disposition request; ask it in this same attempt so + // a restart-recovery budget already spent cannot block it. + const continuationDispositionTurn = () => { + const envelope = buildNativeModelEnvelope(input); + envelope.completionContract = structuredClone(input.completionContract.contract); + if ("requestedSkills" in envelope) envelope.requestedSkills = []; + envelope.task.prompt = [ + "Paperclip semantic-result recovery after a restart continuation.", + "The previous turn ended without paperclip_finish or paperclip_block, so Paperclip cannot settle this task.", + "Do not repeat completed work. If a command you started is still running, wait for or check it in this turn; nothing resumes this conversation after the turn ends.", + "Then invoke exactly one paperclip_finish or paperclip_block with the accurate current disposition, and stop.", + ].join("\n"); + if ("workspace" in envelope && envelope.workspace && recoveredSnapshot.workingDirectory) { + envelope.workspace.cwd = recoveredSnapshot.workingDirectory; + } + return { + message: { role: "user" as const, text: JSON.stringify(envelope) }, + requestedCollaborationMode: "executionMode" in input ? input.executionMode : "default" as const, + }; + }; let restartContinuationStarted = false; + let borrowedContinuationTurnId: string | null = null; + let continuationDispositionRequested = false; + // The consumer may already be running when a continuation starts, and the + // continuation can complete before its start checkpoint settles. A + // terminal waits for this setup so it neither misses the borrowed turn nor + // persists its own checkpoint ahead of the continuation's. + let continuationSetup: Promise | null = null; + const startRestartContinuation = (signal?: AbortSignal) => { + restartContinuationStarted = true; + continuationSetup = (async () => { + // The driver reports the borrowed marker with the accepted turn; a + // later snapshot may already show the marker returned. + const started = await session.startTurn(restartContinuation()); + if (started.borrowedDispositionMarker === true) borrowedContinuationTurnId = started.turnId; + await checkpoint(signal); + })(); + return continuationSetup; + }; + const requestContinuationDisposition = async (event: PrpEvent, signal: AbortSignal) => { + if (continuationDispositionRequested || event.eventType !== "turn.completed") return false; + await continuationSetup?.catch(() => undefined); + signal.throwIfAborted(); + if (borrowedContinuationTurnId === null || event.turnId !== borrowedContinuationTurnId) return false; + const snapshot = await session.snapshot(); + signal.throwIfAborted(); + if (snapshot.semanticResult || snapshot.activeTurnId || snapshot.goal || snapshot.pendingRuntimeRequests?.length || + snapshot.dispositionOnlyRecoveryConsumed !== false || + !snapshot.terminalTurns?.some(terminal => terminal.turnId === borrowedContinuationTurnId)) return false; + continuationDispositionRequested = true; + await persistCheckpoint(snapshot, signal); + signal.throwIfAborted(); + await session.startTurn(continuationDispositionTurn()); + await checkpoint(signal); + return true; + }; const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => { + if (await requestContinuationDisposition(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 @@ -2679,8 +2741,7 @@ export async function executeNativeSession( // history to adopt an accepted continuation if its checkpoint was lost. await persistCheckpoint(snapshot, signal); signal.throwIfAborted(); - await session.startTurn(restartContinuation()); - await checkpoint(signal); + await startRestartContinuation(signal); return true; }; const recoveredActiveTurnId = recovered @@ -2911,9 +2972,7 @@ export async function executeNativeSession( } await checkpoint(); } else if (shouldStartFreshTurn && recovered && checkpointedInterruption) { - restartContinuationStarted = true; - await session.startTurn(restartContinuation()); - await checkpoint(); + await startRestartContinuation(); } else if (shouldStartFreshTurn) { let modelEnvelope = recovered ? buildNativeModelEnvelope(input, { resumedSession: 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 c88ac3d35f..b2f50dfc52 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 @@ -7,9 +7,9 @@ import { createHash, randomUUID } from "node:crypto"; import { createRequire } from "node:module"; import { execFile, spawn } from "node:child_process"; import { existsSync } from "node:fs"; -import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; +import { chmod, copyFile, mkdir, mkdtemp, readFile, readdir, rename, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { join, resolve } from "node:path"; +import { dirname, join, resolve } from "node:path"; import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, QUALIFIED_ACPX_PROFILES } from "@paperclipai/paperclip-runner"; import { createCapabilityRunnerdCodexTransport, defaultCapabilityRunnerdBinary, readRunnerdArtifactBinding, type CapabilityRunnerdCodexTransportOptions } from "@paperclipai/paperclip-runner/live"; import { and, eq, inArray, sql } from "drizzle-orm"; @@ -152,7 +152,24 @@ child.on("exit", code => process.exit(code ?? 1)); process.on("SIGTERM", () => child.kill("SIGTERM")); `; await writeFile(shim, script); - const nodeDigest = digest(await readFile(process.execPath)); + // Runnerd refuses a group- or world-writable qualified ACPX launch. A + // hosted CI image can ship such a Node, or run under umask 0002. + await chmod(shim, 0o600); + let sidecarNode = process.execPath; + if ((await stat(sidecarNode)).mode & 0o022) { + // Same approach as the runner's copyOwnedNodeRuntime test helper. + const owned = join(root, "sidecar-node"); + await mkdir(owned); + sidecarNode = join(owned, "runtime"); + await copyFile(process.execPath, sidecarNode); + await chmod(sidecarNode, 0o700); + // Homebrew Node resolves its libnode dylib next to the relocated executable. + const libraries = join(dirname(process.execPath), "..", "lib"); + for (const name of (await readdir(libraries).catch(() => [] as string[])).filter(name => /^libnode\..*\.dylib$/.test(name))) { + await copyFile(join(libraries, name), join(owned, name)); + } + } + const nodeDigest = digest(await readFile(sidecarNode)); const identity = { runId: "fresh-resume-run", normalizedSessionId: "checkpointed-session", runnerInstanceId: "fresh-runner", environmentLeaseId: "same-lease", turnId: "same-turn", itemId: "same-item" }; const stateDirectory = join(root, ".paperclip-runtime", "paperclip-runner", "sessions", @@ -172,11 +189,11 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); }, prepareLaunch: async input => ({ configurationId: "qualified-fixture", requirementsDigest: runnerLaunchRequirementsDigest(input), provider: createRunnerProviderLaunch(input, { options: {}, root: directory, cwd: root, codexHome: root, - codexCommand: "", providerNodeCommand: process.execPath, opencodeProxyPath: "", acpxSidecarPath: shim }) }), + codexCommand: "", providerNodeCommand: sidecarNode, opencodeProxyPath: "", acpxSidecarPath: shim }) }), runnerProcessLauncher: spec => { launches(adoptExistingRunner); const child = spawn(binary, [...spec.args, "--acpx-launch-authority-digest", digest("fixture authority"), - "--acpx-sidecar-command", process.execPath, "--acpx-sidecar-command-sha256", nodeDigest, + "--acpx-sidecar-command", sidecarNode, "--acpx-sidecar-command-sha256", nodeDigest, "--acpx-sidecar-script", shim, "--acpx-sidecar-script-sha256", digest(script)], { env: { PATH: process.env.PATH, ...spec.environment }, stdio: ["ignore", "ignore", "pipe"], detached: true }); children.push(child);