From ca472d543a5a51a82b855ec773c72fd6c6b62388 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 02:19:21 -0600 Subject: [PATCH] fix(runner): settle a saved governed wait whose shutdown drain left sequence gaps (#20494) 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 > - A native agent run can ask the board a question and park on that saved (governed) wait > - After the question is saved, the runner interrupts the provider and drains its usage and terminal for a bounded time > - The drain keeps only usage, the terminal and cancellation facts of the parked turn. It discards every other numbered provider event > - If the control plane restarts before the drain settles, recovery replays the saved journal. That replay required strictly contiguous `sourceSeq`, so the discarded events made a valid saved wait look like the wrong conversation > - This pull request accepts a gap after the saved trigger only when every event after the gap is a drain-retained fact of the same turn. Identity checks do not change > - The benefit is that the asking run settles its saved wait after a restart instead of recording a failure ## Linked Issues or Issue Description **What happened?** Restart-chaos wave `w-20261110T063253-s208`, tasks t02 (Codex Q1, PV2: provider terminal dropped) and t03 (Codex Q1, PV1: provider terminal held), each with K1 (`kill -9` of the control plane about 3-4 s after the fault). The question, the board answer and the final brief were all correct. But the asking run was recorded `failed: native_session_interrupted`, and the grader failed `T:asking_run_settled`. **Cause** The run log of the asking run shows `failureDetail.message: native_governed_wait_replay_mismatch`. It is thrown at `packages/paperclip-runner/src/native-session-runtime.ts:2344` (master). The persisted runner events of the asking run have `sourceSeq` 33..40 and then 43 (t02), and 41, 42 and then 45 (t03). The two missing numbers are events that the accounting drain (`consumeTurn` → `settleDurableResult`, around line 1206 on master) read and discarded, because they are not usage, a terminal or a cancellation fact. Runs without a restart have the same gap (for example `w-20261110T032350-s201/t01`: 34, then 37). Those runs settle without replay, so the gap was not seen before. With a held or dropped terminal and a kill before the shutdown settlement receipt, recovery must replay. Its strict contiguity check then fails on a gap that the runner made itself. **Expected behavior** The question was saved before the provider was interrupted. Recovery must settle that same saved wait, as #20336 does after verified shutdown. Recovery must not start a new model turn. **Related** Refs #20336 (settle saved handoffs after verified shutdown). Not related in cause or code: #20477, #20482, #20483. ## What Changed - `isGovernedDrainRetainedEvent` (new, in `native-session-runtime.ts`) names the facts that the governed-wait drain persists: usage, the turn terminal, the interrupt acknowledgement, cancelled or expired runtime requests, and a failed or cancelled `run.terminal`, all for the parked turn. The drain now uses it in place of the same inline condition. Drain behavior does not change. - Governed-wait recovery replay accepts a `sourceSeq` gap only when: - the trigger is the first replayed event (a gap before it still fails), - sequence numbers strictly increase, - every event after the first gap is a drain-retained fact of the trigger's turn. - The run, normalized session and source instance identity checks do not change. Any mismatch still fails with `native_governed_wait_replay_mismatch`. - `doc/execution-semantics.md`: the governed-wait text said that shutdown keeps a contiguous journal. It now describes the gaps and the replay rule. Note for the sibling Claude/ACPX `turn.cancel` work: this change touches `settleDurableResult` only in the drain filter line (same condition, shared helper). It does not change cancel, the drain timeout or `settleCancelledTurn`. ## Verification - Red/green: `packages/paperclip-runner/src/native-session-runtime.test.ts` - `settles a saved governed wait whose drain left sequence gaps before a restart (terminal-after-restart)` - `settles a saved governed wait whose drain left sequence gaps before a restart (late-terminal-saved)` The first run uses the real drain. It discards two numbered events, so the history is trigger, acknowledgement, then usage with a gap. The stream then ends without a terminal, as with PV1/PV2 plus K1. On master, recovery rejects with `native_governed_wait_replay_mismatch` (the live error). With this change, recovery settles the saved wait as yielded. It does not call `resolveGovernedWait` again. It starts no second turn. The terminal is appended once. When the terminal arrived late and is already saved, recovery does not cancel the provider again. A further recovery of the settled checkpoint does not submit or settle again. - Fail closed: `keeps rejecting an unproven governed-wait replay with a sequence gap: …` covers a gap before the trigger, a foreign run, a foreign session, a foreign source instance, usage of another turn, a non-accounting event after a gap, and an accounting event followed by a non-accounting event. - `cd packages/paperclip-runner && npx vitest run src/native-session-runtime.test.ts`: 261 passed. `npx tsc --noEmit -p .`: clean. - `cd server && npx vitest run src/services/native-runtime/native-session-executor.test.ts -t governed`: 12 passed. - `pnpm verify` returned exit 2 (test file outside resolved graphs). Required CI supplies the broad coverage. - Live restart-chaos wave: see the comment below with the wave id and result. ## Risks - Low. The rule loosens only the sequence check, and only after the saved trigger, and only for facts that the drain itself retains and that carry no work authority. Identity checks stay strict. A non-accounting event after a gap still fails closed. - No change to the normal start path or turn startup (fast-turn rule). - No schema, protocol or env change. The pack `contractDigest` and provider `commandDigest` do not change. No Daytona image rebuild is necessary. ## Model Used - Anthropic Claude, model ID `claude-opus-5-5`, used through Claude Code with tool use (shell, file edits, test runs). ## 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 --- doc/execution-semantics.md | 8 +- .../src/native-session-runtime.test.ts | 114 ++++++++++++++++++ .../src/native-session-runtime.ts | 38 ++++-- 3 files changed, 149 insertions(+), 11 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 304f312b917..fb76d65e902 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4197,8 +4197,12 @@ through the ordinary control-plane observers without starting another model turn When a later heartbeat run reuses the provider session, it clears that prior run's wait marker along with its terminal, result and pending-request authority. Already-numbered cancellation acknowledgements and cancelled/expired runtime requests -are retained during shutdown so recovery can replay a contiguous journal. They -carry no permission to start work. Stream closure, stream failure or drain timeout +are retained during shutdown with usage and the terminal. They carry no permission +to start work. The drain discards the turn's other numbered events, so the retained +journal can have sequence gaps after the trigger. Recovery accepts a gap only when +the trigger is the first replayed event and every event after the gap is one of +those retained facts of the same turn. Any run, session or source mismatch still +fails closed. Stream closure, stream failure or drain timeout after complete usage leaves the saved wait unfinalized until execution is proven stopped; complete token counters alone cannot certify that boundary. Codex allows up to 30 seconds for its interruption terminal, matching the transport command diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index c8386aaf6d5..f6eca1ad66d 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -9611,6 +9611,120 @@ describe("executeNativeSession recovery", () => { expect(start).not.toHaveBeenCalled(); expect(complete).not.toHaveBeenCalled(); }); + // Restart chaos (Codex Q1 with PV1/PV2 + K1): the provider held or dropped + // the parked turn's terminal and the controller died during the accounting + // drain. The drain had already discarded numbered non-accounting events, so + // the saved replay has gaps after the trigger. + it.each(["terminal-after-restart", "late-terminal-saved"])("settles a saved governed wait whose drain left sequence gaps before a restart (%s)", async variant => { + const lateTerminalSaved = variant === "late-terminal-saved"; + const turn = (event: PrpEvent) => ({ ...event, turnId: "turn-waiting" }); + const trigger = turn(runnerEvent(1, "item.completed", { kind: "dynamicToolCall", item: { id: "ask-1", name: "request_human_input" } })); + const acknowledgement = turn(runnerEvent(2, "item.completed", { kind: "interrupt_acknowledgement", status: "acknowledged" })); + const discardedMessage = turn(runnerEvent(3, "item.completed", { kind: "agentMessage", text: "I asked the board." })); + const discardedToolEnd = turn(runnerEvent(4, "tool.execution.completed", { name: "request_human_input", status: "completed" })); + const usage = turn(runnerEvent(5, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } })); + const stopped = turn(runnerEvent(6, "turn.interrupted", { providerTerminalObserved: true })); + const history: PrpEvent[] = []; + let checkpoint: PersistedNativeSession | undefined; + const start = vi.fn(async () => ({ turnId: trigger.turnId })); + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const complete = vi.fn(async () => undefined); + const snapshot = (): PersistedNativeSession => ({ backendKind: "mock", sessionId: identity.sessionId, identity, + providerSessionId: "same-provider", activeTurnId: trigger.turnId, cursor: String(history.at(-1)?.sourceSeq ?? 0) }); + const makeSession = (recovering: boolean): NativeSession => ({ identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + startTurn: start, revokeTurnPublication: vi.fn(), cancel, + async snapshot() { return snapshot(); }, async result() { return null; }, async close() {}, + async *events() { + if (!recovering) { + // The terminal is held (or dropped) and the process then dies. + yield trigger; yield acknowledgement; yield discardedMessage; yield discardedToolEnd; yield usage; + return; + } + if (!lateTerminalSaved) yield stopped; + }, + }); + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "drain-gap", version: "1", capabilities: await makeSession(false).capabilities() }; }, + async openSession() { return makeSession(false); }, + async recoverSession() { return { recovered: true, session: makeSession(true) }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, completeRun: complete, + async checkpointSession(value) { checkpoint = structuredClone(value); }, + async appendEvent(event: PrpEvent) { + const duplicate = history.some(row => row.sourceEventId === event.sourceEventId); + if (!duplicate) history.push(event); + return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: duplicate ? "duplicate" : "committed" }; + }, + async replayEvents(request) { + const events = history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq) + .sort((left, right) => left.sourceSeq - right.sourceSeq); + return { events, highestContiguousSourceSeq: events.at(-1)?.sourceSeq ?? request.afterSourceSeq }; + }, + }; + const options = { input, backend, controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" }; + await expect(executeNativeSession({ ...options, resolveGovernedWait: ({ event }) => event.sourceSeq === 1 ? yieldedResult : null })) + .rejects.toThrow(); + // The question is saved, and the drain kept only accounting facts. + expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult }); + expect(history).toEqual([trigger, acknowledgement, usage]); + expect(complete).not.toHaveBeenCalled(); + cancel.mockClear(); + if (lateTerminalSaved) history.push(stopped); + + const answerAlreadySaved = vi.fn(() => null); + await expect(executeNativeSession({ ...options, persistedSession: checkpoint, resolveGovernedWait: answerAlreadySaved })) + .resolves.toMatchObject({ result: yieldedResult, turnId: trigger.turnId, terminal: { runTerminalState: "succeeded" } }); + expect(start).toHaveBeenCalledOnce(); + // A saved terminal settles without interrupting provider work again. + expect(cancel).toHaveBeenCalledTimes(lateTerminalSaved ? 0 : 1); + expect(answerAlreadySaved).not.toHaveBeenCalled(); + expect(complete).toHaveBeenCalledOnce(); + expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]); + + // A further recovery of the settled wait neither submits nor settles again. + const settledCheckpoint = structuredClone(checkpoint); + await expect(executeNativeSession({ ...options, persistedSession: settledCheckpoint })) + .resolves.toMatchObject({ result: yieldedResult, turnId: trigger.turnId, terminal: { runTerminalState: "succeeded" } }); + expect(start).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledTimes(lateTerminalSaved ? 0 : 1); + expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]); + }); + + it.each([ + "gap-before-trigger", "gap-foreign-run", "gap-foreign-session", "gap-foreign-instance", + "gap-other-turn-usage", "gap-unaccounted-event", "unaccounted-event-after-gap", + ])("keeps rejecting an unproven governed-wait replay with a sequence gap: %s", async fault => { + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" }; + const afterGap = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record, overrides: Partial = {}) => + ({ ...runnerEvent(sourceSeq, eventType, payload), turnId: trigger.turnId, ...overrides }) as PrpEvent; + const usagePayload = { kind: "usage", usage: { runDelta: { inputTokens: 1, outputTokens: 1 }, runDeltaComplete: true } }; + const history: PrpEvent[] = fault === "gap-before-trigger" ? [{ ...trigger, sourceSeq: 2, sourceEventId: "late-trigger" }] : [trigger]; + if (fault === "gap-foreign-run") history.push(afterGap(3, "item.completed", usagePayload, { runId: "another-run" })); + if (fault === "gap-foreign-session") history.push(afterGap(3, "item.completed", usagePayload, { normalizedSessionId: "another-session" })); + if (fault === "gap-foreign-instance") history.push(afterGap(3, "item.completed", usagePayload, { sourceInstanceId: "another-runner" })); + if (fault === "gap-other-turn-usage") history.push(afterGap(3, "item.completed", usagePayload, { turnId: "another-turn" })); + if (fault === "gap-unaccounted-event") history.push(afterGap(3, "item.completed", { kind: "agentMessage", text: "more work" })); + if (fault === "unaccounted-event-after-gap") { + history.push(afterGap(3, "item.completed", usagePayload), afterGap(4, "item.delta", { kind: "agentMessage", delta: "more" })); + } + const checkpoint: PersistedNativeSession = { backendKind: "mock", sessionId: identity.sessionId, identity, + activeTurnId: trigger.turnId, governedWait: { sourceEvent: trigger, result: yieldedResult } }; + const start = vi.fn(); + const complete = vi.fn(); + const session: NativeSession = { identity: () => identity, startTurn: start, async result() { return null; }, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async snapshot() { return checkpoint; }, async *events() {}, async close() {} }; + const backend: NativeSessionBackend = { async descriptor() { return { kind: "mock", name: "unsafe-gap", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { return session; }, async recoverSession() { return { recovered: true, session }; } }; + const port: ControlPlanePort = { async openRun() {}, async checkpointSession() {}, async appendEvent() { throw new Error("unexpected append"); }, + completeRun: complete, async replayEvents(request) { return { events: history.filter(row => row.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: request.afterSourceSeq }; } }; + await expect(executeNativeSession({ input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" })).rejects.toThrow("native_governed_wait_replay_mismatch"); + expect(start).not.toHaveBeenCalled(); expect(complete).not.toHaveBeenCalled(); + }); + it.each([0, 6_000, 12_000])("retains actual shutdown usage after %s ms before finalizing a governed wait", async (receiptDelayMs) => { vi.useFakeTimers(); try { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 433b8fbb8ec..f06e80af74c 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -418,6 +418,22 @@ function isTurnTerminal(event: PrpEvent): boolean { ].includes(event.eventType); } +/** + * The only provider facts a governed-wait accounting drain persists after its + * trigger: usage, the turn terminal and cancellation acknowledgements of the + * parked turn. The drain discards every other numbered event, so the saved + * replay has sequence gaps exactly where those discarded events were. + */ +function isGovernedDrainRetainedEvent(event: PrpEvent, turnId: string | undefined): boolean { + if (event.turnId !== turnId) return false; + return isTurnTerminal(event) + || (event.eventType === "item.completed" + && (event.payload.kind === "usage" || event.payload.kind === "interrupt_acknowledgement")) + || event.eventType === "runtime_request.cancelled" + || event.eventType === "runtime_request.expired" + || (event.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(event.payload.runTerminalState))); +} + function terminalFromEvent( event: PrpEvent, disposition: PrpTerminalState["reportedWorkDisposition"], @@ -1206,14 +1222,10 @@ async function consumeTurn( if (next.done || drainAbort.signal.aborted) return; const trailing = next.value; const terminal = isTurnTerminal(trailing); - const usage = trailing.eventType === "item.completed" && trailing.payload.kind === "usage"; - // These already-numbered cancellation facts carry no new work - // authority. Retain them so later receipts have contiguous replay. - const cancellation = trailing.eventType === "runtime_request.cancelled" - || trailing.eventType === "runtime_request.expired" - || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement") - || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState))); - if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) { + // Usage, the terminal and already-numbered cancellation facts carry + // no new work authority. Everything else is discarded, leaving gaps + // that governed-wait recovery replay accepts only for these facts. + if (isGovernedDrainRetainedEvent(trailing, event.turnId)) { const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal, ...(terminal && trailing.turnId && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false @@ -2333,14 +2345,22 @@ export async function executeNativeSession( timeoutMessage: "native governed-wait recovery replay timed out", operation: async signal => { let afterSourceSeq = event.sourceSeq - 1; + let gapObserved = false; while (true) { const page = await options.controlPlane.replayEvents({ runId: identity.runId, sourceInstanceId: options.runnerInstanceId, afterSourceSeq, limit: 1_000 }, { signal }); signal.throwIfAborted(); if (!page.events.length) break; for (const entry of page.events) { + // The trigger must be the first replayed event. After it, the + // accounting drain persisted only facts of the parked turn and + // discarded the rest, so a gap is legitimate only when every later + // event is one of those facts. Identity never relaxes. + if (entry.sourceSeq !== afterSourceSeq + 1) gapObserved = true; if (entry.runId !== identity.runId || entry.normalizedSessionId !== identity.sessionId - || entry.sourceInstanceId !== event.sourceInstanceId || entry.sourceSeq !== afterSourceSeq + 1 + || entry.sourceInstanceId !== event.sourceInstanceId || !Number.isSafeInteger(entry.sourceSeq) + || entry.sourceSeq <= afterSourceSeq || (gapObserved && (governedReplay.length === 0 || !event.turnId + || !isGovernedDrainRetainedEvent(entry, event.turnId))) || governedReplay.length >= 10_000) throw new Error("native_governed_wait_replay_mismatch"); governedReplay.push(entry); afterSourceSeq = entry.sourceSeq;