fix(runner): settle a saved governed wait whose shutdown drain left sequence gaps (#20494)

## 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 <noreply@anthropic.com>

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

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<string, unknown>, overrides: Partial<PrpEvent> = {}) =>
+      ({ ...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 {

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;