fix(accounting): close a remote-Runner turn completed across a session restore as a lower bound (#20524)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native runs settle their charge once the final usage receipt is
complete.
> - #20517 fixed one case for the local Runner. If a restart stops the
provider before a turn's final usage report, and the restored Runner
then reports that turn completed, the turn now closes as a lower bound
instead of waiting 10 minutes.
> - Remote Runners deliver usage through a separate producer-owned
receiver. That receiver dropped `session.resumed`, so the same turn
there still waited for the 10-minute closure timeout.
> - This pull request applies the #20517 rule to that receiver.
> - The benefit is that remote ACPX and OpenCode runs settle promptly
with an honest lower-bound label after a mid-turn restore.

## Linked Issues or Issue Description

Refs #20517.

**What happened?**

After #20517, `session.resumed` reached turn accounting only through the
local Runner's fallback recorder. The remote producer receiver
(`native-usage-delivery.ts`) and the session-authority filter
(`isNativeUsageAccountingEvent`) both dropped it. A remote ACPX or
OpenCode turn that was open across a session restore, and then reported
terminal without a complete report, left the receipt incomplete until
the closure timeout.

**Expected behavior**

Same rule as the local Runner. Such a turn closes as a lower bound right
away. A complete report that arrives after the restore is still used,
and a turn that starts after the restore stays pending until its own
report is complete.

**Steps to reproduce**

`native-usage-delivery.integration.test.ts > a turn restored mid-turn on
a remote Runner > closes the turn as a lower bound, also after the
controller restarts between restore and terminal` fails on master: the
receipt stays `accountingReceiptReady: false`.

## What Changed

- **Receiver (`createNativeUsageReceiver`).** It now accepts
`session.resumed` for turn-reporting providers (ACPX, OpenCode) while a
turn is open. It journals the restore as a receipt like any other
producer event, so turn accounting marks the open turn restored. A
replacement controller rebuilds the same state from that journal.
- **Restores never fail delivery.** A restore from another run or
source, or one with no open turn, is ignored, not rejected. An
unacknowledged lifecycle event would otherwise replay without end.
- **Session-authority filter.** `acceptAccountingEvent` now uses a new
`isNativeUsageDeliveryEvent`: the usage events plus `session.resumed`. A
restore for another run is ignored there too.
- **Recovery routing is unchanged.** `acceptRecoveryAccountingEvent` and
`onRecoveryAccountingEvent` keep the usage-only
`isNativeUsageAccountingEvent`, so a source run's restore never requires
a proven recovery source.
- **Turn accounting** gets a `hasOpenTurn()` helper.
- **Docs:** `doc/SPEC-implementation.md` now covers both Runners.

## Verification

- `native-usage-delivery.integration.test.ts`, describe `a turn restored
mid-turn on a remote Runner` (real PostgreSQL):

  | Test | On master | Here |
  |---|---|---|
| `closes the turn as a lower bound, also after the controller restarts
between restore and terminal`: the restore is journaled by one receiver
and the terminal arrives at a rebuilt one; replaying everything again is
idempotent | fails | passes |
| `keeps the turn's own final report when it arrives after the restore`
(guard) | passes | passes |
| `keeps a turn started after the restore pending, and journals nothing
for a restore with no open turn` (guard) | passes | passes |

- `native-session-executor.test.ts > binds retained warm transport usage
to the successor run authority` covers routing through the
session-authority filter:
  - a current-run restore reaches the successor's capture;
  - a draining-run restore reaches the drained capture;
  - an unrelated-run restore resolves with no capture;
  - recovery routing ignores a source restore.

It fails when the filter is reverted to `isNativeUsageAccountingEvent`.
- Related suites pass: `native-usage-delivery.integration` and
`native-usage-accounting` (46 tests), and the usage, accounting, receipt
and authority cases in `native-session-executor.test.ts` (224 tests).
Server typecheck passes.
- **The live remote path is unproven.** No Daytona wave ran because the
Daytona Runner image has not been rebuilt yet. The local Runner
counterpart (#20517) passed restart-chaos L1, seed 20386, 4/4
(`w-20261110T104332-s20386`).

## Risks

- **Upgrade safety.** Usage state saved by current master has no
journaled restore and rebuilds as before. A receiver that predates this
change replays a journaled restore as an ignored event, which is the
behaviour on master. A restore that an older controller already
acknowledged is not replayed, so that run waits for the closure timeout
as it does today.
- **One extra receipt per mid-turn restore**, only for ACPX and OpenCode
and only while a turn is open. Nothing is added before `turn.start`: a
restore with no open turn returns immediately.
- **Codex and other providers are unaffected.** They do not use turn
accounting.
- No schema or persisted-format change.

## Model Used

- Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through
Claude Code with shell, file-edit and code-execution tools.

## 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
- [x] All Paperclip CI gates are green
- [x] 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 <noreply@anthropic.com>

server/src/services/native-runtime/native-session-executor.test.ts

@@ -8698,12 +8698,25 @@ describe("native terminal-turn accounting", () => {
       expect(firstCapture).toHaveBeenCalledTimes(2);
       expect(secondCapture).toHaveBeenCalledTimes(1);
       await expect(retainedHook({ ...firstReport, runId: "unrelated-run" })).rejects.toThrow("native_usage_run_binding_mismatch");
+      // A session restore reaches the producer that owns its run, so an open
+      // turn can close as a lower bound; another run's restore is ignored.
+      const currentResume = { ...secondReport, eventType: "session.resumed" } as PrpEvent;
+      const drainingResume = { ...firstReport, eventType: "session.resumed" } as PrpEvent;
+      await retainedHook(currentResume);
+      expect(secondCapture).toHaveBeenLastCalledWith(currentResume);
+      await retainedHook(drainingResume);
+      expect(firstCapture).toHaveBeenLastCalledWith(drainingResume);
+      const [firstCalls, secondCalls] = [firstCapture.mock.calls.length, secondCapture.mock.calls.length];
+      await expect(retainedHook({ ...currentResume, runId: "unrelated-run" })).resolves.toBeUndefined();
+      expect([firstCapture.mock.calls.length, secondCapture.mock.calls.length]).toEqual([firstCalls, secondCalls]);
       const recoveryHook = state.createTransport.mock.calls.at(-1)![0].onRecoveryAccountingEvent!;
       const source = { runnerInstanceId: "warm-producer", environmentLeaseId: "warm-usage-workspace",
         runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId!, turnId: "source-turn", itemId: "source-item" };
       await recoveryHook(firstReport, source);
-      expect(firstCapture).toHaveBeenCalledTimes(3);
-      expect(secondCapture).toHaveBeenCalledTimes(1);
+      expect(firstCapture).toHaveBeenCalledTimes(firstCalls + 1);
+      expect(secondCapture).toHaveBeenCalledTimes(secondCalls);
+      // Recovery routing stays usage-only: a source restore needs no proven source.
+      await expect(recoveryHook({ ...drainingResume, runId: "foreign-source" }, source)).resolves.toBeUndefined();
       await expect(recoveryHook({ ...firstReport, runId: "foreign-source" }, source)).rejects.toThrow("native_recovery_usage_source_unproven");
     } finally {
       if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;

server/src/services/native-runtime/native-session-executor.ts

@@ -26,7 +26,7 @@ import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./
 import { rehydratePinnedSkillAssets } from "./runtime-context.js";
 import { createNativeUsageDelivery } from "./native-usage-delivery.js";
 import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js";
-import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js";
+import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent, isNativeUsageDeliveryEvent } from "./native-usage-accounting.js";
 import { crashPoint } from "@paperclipai/shared/crash-points";
 export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js";
 import { measureRunnerOperation, recordRunnerTimingReceipt } from "./runner-operation-timing.js";
@@ -1968,7 +1968,9 @@ class SessionAuthorityEpoch {
   }
 
   async acceptAccountingEvent(event: PrpEvent) {
-    if (!isNativeUsageAccountingEvent(event)) return;
+    if (!isNativeUsageDeliveryEvent(event)) return;
+    // A restore carries no usage; it never fails delivery for another run.
+    if (!isNativeUsageAccountingEvent(event) && event.runId !== this.runId && event.runId !== this.#drainingRun?.runId) return;
     this.#assertCurrent();
     if (event.runId !== this.runId && event.runId !== this.#drainingRun?.runId) throw new Error("native_usage_run_binding_mismatch");
     await (event.runId === this.runId ? this.onAccountingEvent : this.#drainingRun?.capture)?.(event);

server/src/services/native-runtime/native-usage-accounting.ts

@@ -143,6 +143,11 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi
       return true;
     },
     snapshot,
+    /** A restore matters only while a turn is open. */
+    hasOpenTurn() {
+      for (const turn of turns.values()) if (!turn.terminal) return true;
+      return false;
+    },
     finish(turnId: string | null) {
       const turn = turnId ? turns.get(turnId) : undefined;
       if (turn) turn.terminal = true;
@@ -215,3 +220,11 @@ export function usesNativeUsageDelivery(input: {
 export function isNativeUsageAccountingEvent(event: Pick<PrpEvent, "eventType">) {
   return ["usage.reported", "turn.started", "turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"].includes(event.eventType);
 }
+
+/** Producer delivery for the run itself also needs the session restore: a
+ * turn open across it can close as a lower bound (see createNativeTurnAccounting).
+ * Recovery routing keeps isNativeUsageAccountingEvent: a source run's restore
+ * carries no usage and must not require a proven recovery source. */
+export function isNativeUsageDeliveryEvent(event: Pick<PrpEvent, "eventType">) {
+  return isNativeUsageAccountingEvent(event) || event.eventType === "session.resumed";
+}

server/src/services/native-runtime/native-usage-delivery.integration.test.ts

@@ -559,6 +559,54 @@ describe("producer-owned native usage", () => {
     expect((await read()).usageJson).toMatchObject({ accountingReceiptReady: true });
   });
 
+  describe("a turn restored mid-turn on a remote Runner", () => {
+    const restoredSeed = async () => ({ ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } });
+    const partial = (input: Awaited<ReturnType<typeof restoredSeed>>, sequence: number) => {
+      const value = event(input, sequence, "usage.reported");
+      (value.payload as Record<string, unknown>).runDeltaAvailable = false;
+      return value;
+    };
+    const read = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0];
+
+    // The local Runner rule from #20517: K1 stopped the provider before the
+    // turn's final report; the restored Runner reported the turn completed.
+    it("closes the turn as a lower bound, also after the controller restarts between restore and terminal", async () => {
+      const input = await restoredSeed();
+      const history = [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never)];
+      const done = event(input, 4, "turn.completed");
+      let delivery = await createNativeUsageDelivery(db, input);
+      for (const value of history) await delivery.capture(value);
+      // A replacement controller rebuilds the receiver from the saved journal.
+      delivery = await createNativeUsageDelivery(db, input);
+      await delivery.capture(done);
+      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true,
+        pricingProvenance: { evidence: expect.stringContaining("lower bound") } });
+      // Replaying every event into another receiver is idempotent.
+      delivery = await createNativeUsageDelivery(db, input);
+      for (const value of [...history, done]) await delivery.capture(value);
+      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true });
+    });
+
+    it("keeps the turn's own final report when it arrives after the restore", async () => {
+      const input = await restoredSeed();
+      const delivery = await createNativeUsageDelivery(db, input);
+      for (const value of [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never),
+        event(input, 4, "usage.reported"), event(input, 5, "turn.completed")]) await delivery.capture(value);
+      const usage = (await read(input.runId)).usageJson;
+      expect(usage).toMatchObject({ accountingReceiptReady: true, outputTokens: 10 });
+      expect(JSON.stringify(usage?.pricingProvenance ?? {})).not.toContain("lower bound");
+    });
+
+    it("keeps a turn started after the restore pending, and journals nothing for a restore with no open turn", async () => {
+      const input = await restoredSeed();
+      const delivery = await createNativeUsageDelivery(db, input);
+      for (const value of [event(input, 1, "session.resumed", undefined as never), event(input, 2, "turn.started"), partial(input, 3),
+        event(input, 4, "turn.completed")]) await delivery.capture(value);
+      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: false });
+      expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(3);
+    });
+  });
+
   it("replays partial and final usage once after a lost commit response and restores turn aggregation", async () => {
     const input = { ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } };
     let loseResponse = false;

server/src/services/native-runtime/native-usage-delivery.ts

@@ -155,7 +155,15 @@ function createNativeUsageReceiver(db: Db, input: NativeUsageDeliveryInput, sour
   let pending: UsageReceiptEnvelope | undefined;
   let chain = Promise.resolve();
   const capture = async (event: PrpEvent) => {
-    if (event.eventType !== "usage.reported" && event.eventType !== "turn.started" && !terminal.has(event.eventType)) return;
+    // A restore is journaled only for turn-reporting providers with an open
+    // turn, so a rebuilt receiver can close that turn as a lower bound.
+    // Receivers that predate it replay the receipt as an ignored event.
+    if (event.eventType === "session.resumed") {
+      // Never fail delivery for a lifecycle event: an unacknowledged event
+      // would replay without end. Another source's restore is not ours.
+      if (!turnAccounting || event.runId !== input.runId || event.sourceInstanceId !== input.runnerInstanceId
+        || (!turnAccounting.hasOpenTurn() && !accepted.has(event.sourceEventId))) return;
+    } else if (event.eventType !== "usage.reported" && event.eventType !== "turn.started" && !terminal.has(event.eventType)) return;
     if (event.runId !== input.runId || event.sourceInstanceId !== input.runnerInstanceId) throw new Error("native_usage_source_mismatch");
     const payload = event.payload as Record<string, unknown>;
     if (event.eventType === "usage.reported" && await captureLateUsageForOwner(db, input, event, payload)) return;