fix(native): resume connection tasks after approval decisions (#15471)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native tasks can pause while a human decides whether to allow a connection action. > - The next turn needs both the saved decision and complete accounting for the previous turn. > - Cancellation could discard final usage, and complete direct Claude API receipts could remain unpriced. > - A stale blocked or review report could also request approval again after the original card was declined. > - This pull request retains shutdown accounting and rejects approval waits bound to an already resolved action. > - The benefit is reliable continuation with the existing budget and approval controls. ## Linked Issues or Issue Description Refs #15420. Related: #15312 addresses requester ownership during dispatch. This change addresses receipt capture and final-response validation. ## What Changed - Retain usage events after native cancellation. Continue to reject late provider messages and work. - Drain same-turn accounting and terminal events for at most fifteen seconds after a durable governed wait. Keep incomplete accounting blocked. - Estimate complete, unpriced, direct Anthropic API receipts for the exact `claude-sonnet-5` model. Record the rate version and assumptions. Use the one-hour cache-write rate when the receipt lacks cache TTL. - Bind stale approval reports to exact interaction, action-request, or invocation IDs in the same company, task, agent, and run. Cover blocked, review, and response-wake reports. Keep independent reviews valid. - Fence checkpoint and result writes after a controller detaches for restart, including operations waiting for a database lock. Reject stale successful returns before certifying accounting. - Allow bounded subscription teardown only after retaining an actual provider terminal. - Journal the exact governed-wait trigger and disposition before provider interruption. Recover that wait independently of a later saved answer, replay retained accounting, and reject mismatched or unproven terminal evidence. - Give settling governed turns a bounded window before shutdown detaches their controller. - Keep fuzzy external app matches alongside installed capability matches instead of forcing an unrelated provider question for a generic query. - Return up to twenty exact active catalog tool names after an invalid request, after eligibility checks; still reject the request without granting access or creating an approval. - Require retained provider terminal proof before settling a governed wait, including when complete usage arrives before stream closure/error/timeout. Retain harmless numbered cancellation events so restart replay stays contiguous. - Isolate accounting-test OpenCode config from the host plugin directory. - Add regression coverage and document the accounting, restart and connection-search behavior. ## Verification Current PR source: `0cf08efd75f8fb23f7989beda6dbda92587088bf`. The live matrix below measured frozen `09a776bcb1f77422156f8be11e13f8c29f43e7f7`; later review fixes are verified separately and do not relabel those runs. - Repository typecheck and build pass. - Current runner runtime and cancellation suites: 191 tests pass. Six new regressions cover stream end/error/timeout without provider-stop proof and contiguous cancellation acknowledgement/request replay; all six failed before the fix. The existing bounded cleanup case now explicitly supplies terminal proof. All 31 adapter accounting tests pass with isolated fixture config. - New checkpoint-rebinding and approval-criterion suites: 65 tests pass; runner HTTP integration: 29 tests pass. Unchanged executor/control-plane suites: 640 tests pass; database-backed connection suites: 68 tests pass. - Current retained evidence verification covers 222 file hashes across all fifteen original result artifacts. The three-case [published report](https://d1p6rlowie26tp.cloudfront.net/runner-e2e/campaigns/gha-37651896525-1/index.html) and all eight screenshot hashes verify. The [three-case recovery report](https://d1p6rlowie26tp.cloudfront.net/runner-e2e/campaigns/gha-37662587356-1/index.html) and all six screenshots also verify; the nine-result campaign did not publish. - Evaluation support: 1,805 Vitest tests pass, one skipped; 128 Node tests pass. Eval typecheck and catalog discovery pass. - Full local repository unit run was interrupted before the follow-up edits after three tool-access failures and one runner HTTP failure. Those failures pass in isolation; the 09a full run was interrupted after one rapid Slack callback-ordering failure and seven skill-service failures. All eight pass both isolated and with full-runner environment settings, and all 74 skill-service tests pass together; the subsequent full run reported two 15-second OpenCode accounting timeouts and was stopped with exit 130 to apply review fixes. The timeouts reproduce while copying this host’s 61 MB OpenCode config. All 31 tests pass after isolating config inside each fixture without increasing timeouts or changing assertions. A complete local full-suite pass is not claimed. Repository-wide CI also passes on the final review-fix head in [run 37669185952](https://github.com/paperclipai/paperclip/actions/runs/37669185952). Earlier database-skipped diagnostics and the older ENFILE run are retained and are not full-suite passing evidence. - Three-case live campaign [37651896525](https://github.com/paperclipai/paperclip/actions/runs/37651896525) passes all three original grades on frozen source 09a: 55/55 checks, eight succeeded run records, complete accounting receipts, matching checkpoint identities and no pending approvals. Campaign [37653533353](https://github.com/paperclipai/paperclip/actions/runs/37653533353) adds nine original passes (173/173 checks, eighteen succeeded records) on the identical source. Its other three jobs failed before runner assignment or any step while GitHub could not load the paid environment; those original infrastructure failures are retained. Campaign [37662587356](https://github.com/paperclipai/paperclip/actions/runs/37662587356) completes only those unstarted cells: all three original grades pass (57/57 checks, six succeeded run records). All fifteen exact cases now pass on source 09a: eight FAIL → PASS, seven PASS → PASS, zero new overall failures and zero pending pairs. Total current evidence: 285/285 checks and thirty-two succeeded run records, complete accounting receipts, matching checkpoint identities, no pending approvals or retry records. Earlier campaigns retain forty-seven additional run records and two known same-run recovery attempts; actual provider-call counts and invoices remain unknown. The nine-result campaign skipped publication and its public URL returns 403; original artifacts remain retained. Previous ba9 campaign [37645656840](https://github.com/paperclipai/paperclip/actions/runs/37645656840) completed 2 PASS / 1 FAIL: Claude restart/approval and Codex decline pass, while OpenCode resumes but times out searching for exact tool names and never creates the access card. That failure and incomplete cancelled-run accounting remain preserved; the new catalog error guidance targets this observed dead end. Campaign [37642957312](https://github.com/paperclipai/paperclip/actions/runs/37642957312) remains 0 PASS / 3 FAIL and exposed the now-corrected cross-run marker leak and unknown-criterion approval gap. The original baseline remains 7 PASS / 8 FAIL, first repair 2 PASS / 3 FAIL, and second repair 0 PASS / 3 FAIL. All fifteen selected cases are qualified by their original grades in this bounded trial. These live grades belong to 09a. Its twelve governed-wait checkpoints retain matching same-turn terminal fingerprints, but passing artifacts omit detailed event journals; the later six adversarial regressions qualify the new terminal-proof and replay guards separately. Final-head repository CI passes. [Fresh Greptile review](https://github.com/paperclipai/paperclip/pull/15471#issuecomment-6044369780) is 5/5, confirms both findings are fixed, and reports no new actionable issues. All review threads are resolved and the PR has no merge conflicts. ## Risks - Governed cancellation drains accounting for up to fifteen seconds. Restart detachment gives a settling batch up to twenty seconds to finish. An incomplete receipt or unproven provider terminal still prevents successful qualification. - Claude prices are estimates, not invoices. The estimate assumes standard global API pricing and uses a conservative cache-write rate. Unsupported models, billers, and billing modes remain unpriced. - Approval identity matching must remain scoped to the current run and the requested approval. It does not authorize execution of a declined call. - Original baseline and final candidate have different merged master context. Exact-case outcomes are before/after observations, not isolated causal attribution to this repair. - No schema migration, fixture, oracle or grader change. Search-result guidance now treats fuzzy external matches as suggestions. Existing app authorization and provider-consent checks remain required. ## Model Used - OpenAI GPT-6 through Codex, with code editing, terminal tools, and test execution. The exact deployment identifier and context window are not exposed in this session. ## 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 #` 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 (191 current runtime tests and 31 accounting tests; interrupted full-suite history and CI coverage are disclosed above) - [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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-24 10:13:11 +02:00 · 2026-10-07 14:43:07 -05:00
1 parent 5405b1f46f
commit fd8c6b920a
@@ -1554,6 +1554,20 @@ for contracts, recovery behavior, Storybook, and acceptance workflows.

 Board may override by raising budget or explicitly resuming agent.

+Native runs retain final usage receipts during a bounded accounting-only drain
+after a governed wait cancels provider work. This does not accept late messages,
+tool calls, or new completion proposals. Missing or incomplete receipts continue
+to block budget admission.
+A controller that detaches for server restart loses checkpoint and completion
+write authority. A closing event stream is not proof that the governed run
+finished; the replacement controller must adopt and settle the original run.
+
+Complete direct Anthropic API receipts for `claude-sonnet-5` can use a versioned
+list-price estimate when the provider supplies no run price. The receipt records
+the rates and assumptions. Aggregate cache writes use the one-hour rate because
+their TTL is unknown. Estimates are not invoices. Other models, billers, unknown
+billing types, and incomplete receipts remain unpriced.
+
 ## 13.3 Cost Event Ingestion

 `POST /companies/:companyId/cost-events` body:
@@ -57,6 +57,12 @@ grants consent. A provider connection is not proof that its underlying app is
 authorized. Multiple matching aggregator apps are returned as candidates; the
 agent searches the selected `aggregator.targetService` to get its provider question.

+A fuzzy or partial external app match remains a suggestion alongside authorized
+installed capability matches. It cannot replace those matches with a mandatory
+provider question. For example, a query for a page service can return the
+installed page tools alongside a Page X suggestion; selecting the exact Page X
+service is required before that external route asks for provider consent.
+
 ## Regression coverage

 ```sh
@@ -69,3 +75,10 @@ Circleback/Attio/ClickUp, indexed Executor tools, and provider consent. The nati
 tool authority test searches a paragraph longer than the old 200-character limit
 and then creates a real connection interaction. Existing tests retain checks for
 private metadata, administrative denials, stale identities, and saved declines.
+
+When `connection_request` rejects an unknown `toolNames` entry, its error includes
+up to twenty exact names from that eligible connection's active tool catalog.
+This is discovery metadata: the request still fails with 422, grants no access,
+and creates no approval card. The agent must select the needed exact names and
+submit a new request; normal user approval, catalog-version and policy checks
+still apply. Ineligible connections and inactive catalog entries are excluded.
@@ -1382,6 +1382,23 @@ controller must recover it through the authenticated ownership checks. This
 preserves active work and queued messages without treating a server restart as
 user cancellation.

+When a native turn creates a governed approval wait, its checkpoint retains the
+exact committed triggering event and yielded disposition before interrupting the
+provider. Shutdown gives these turns a shared, bounded 20-second window to finish
+settling their usage and terminal evidence before detaching. Recovery verifies
+the trigger against the retained event ledger and settles that same turn, even
+if the user has already answered the interaction. It replays retained accounting
+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
+after complete usage leaves the saved wait unfinalized until a provider terminal
+is retained; complete token counters alone cannot certify a stopped provider.
+A timeout, missing accounting, or an unproven provider terminal remains a failure;
+the saved wait does not certify successful provider execution.
+
 Before either shutdown path exits, idle warm sessions close through their
 normal suspend-and-checkpoint path. Remote sessions therefore leave verified
 backup authority for the next controller even though their last run is already
@@ -974,6 +974,28 @@ describe("HarnessDriverBackend", () => {
    await expect(iterator.next()).rejects.toThrow("provider transport lost after resolution");
  });

+  it.each([false, true])("retains final usage without accepting late output (journal boundary only: %s)", async journalOnly => {
+    class AccountingAfterCancelSession extends FakeHarnessSession {
+      async interrupt() {}
+      override async *events() {
+        yield prpEvent(1, "item.completed", { kind: "agentMessage", text: "late answer" });
+        yield prpEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } });
+        yield prpEvent(3, "turn.completed", { status: "completed" });
+      }
+    }
+    const backend = new HarnessDriverBackend({ ...driver, async openSession() { return new AccountingAfterCancelSession(); } });
+    const session = await backend.openSession({
+      identity: { runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1" }, workingDirectory: "/workspace",
+    });
+    if (journalOnly) session.revokeTurnPublication();
+    else await session.cancel({ reason: "durable governed wait", signal: new AbortController().signal }).cleanup;
+    const events = [];
+    for await (const event of session.events()) events.push(event);
+    expect(events.filter(event => event.payload.kind === "agentMessage")).toHaveLength(0);
+    expect(events.find(event => event.payload.kind === "usage")?.payload.usage).toMatchObject({ runDelta: { inputTokens: 7, outputTokens: 3 } });
+    expect(events.some(event => event.eventType === "turn.cancelled")).toBe(true);
+  });
+
  it("does not start provider work when Stop arrives before the first turn", async () => {
    const provider = new FakeHarnessSession();
    const start = vi.spyOn(provider, "startTurn");
@@ -549,7 +549,10 @@ class HarnessNativeSession implements NativeSession {
          (event.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(event.payload.runTerminalState))) ||
          (event.eventType === "item.completed" &&
            event.payload.kind === "interrupt_acknowledgement");
-        if (this.#explicitlyCancelled && !isCancellationEvent) continue;
+        // Accounting for work already performed survives cancellation. It grants
+        // no tool, message, semantic-result, or continuation authority.
+        const isUsageReceipt = event.eventType === "item.completed" && event.payload.kind === "usage";
+        if (this.#explicitlyCancelled && !isCancellationEvent && !isUsageReceipt) continue;
        sourceInstanceId = event.sourceInstanceId;
        lastSourceSequence = Math.max(lastSourceSequence, event.sourceSeq);
        if (event.eventType === "runtime_request.created") {
@@ -735,6 +738,10 @@ class HarnessNativeSession implements NativeSession {
    return this.#session.interrupt(input);
  }

+  revokeTurnPublication() {
+    this.#explicitlyCancelled = true;
+  }
+
  cancel(input: { reason: string; signal: AbortSignal }) {
    if (input.signal.aborted) {
      throw (
@@ -744,7 +751,7 @@ class HarnessNativeSession implements NativeSession {
    // This flag is the adapter's synchronous publication boundary. Provider
    // interruption happens afterward as passive cleanup, so a slow or broken
    // transport cannot synthesize or publish new accepted output for the turn.
-    this.#explicitlyCancelled = true;
+    this.revokeTurnPublication();
    const interrupt = this.#session.interrupt;
    return {
      cleanup:
@@ -63,6 +63,10 @@ export interface PersistedNativeSession {
  terminal?: PrpTerminalState | null;
  activeTurnId?: string | null;
  terminalTurns?: PersistedHarnessTurnTerminal[];
+  /** Control-plane disposition bound to the exact committed tool event. It is
+   * independent of the interaction's later answer and proves no provider terminal. */
+  governedWait?: { sourceEvent: PrpEvent; result: PrpStructuredRunResult };
+
  /** At-most-once marker for resultless disposition repair or restart continuation. */
  dispositionOnlyRecoveryConsumed?: boolean;
  dispositionOnlyRecoveryTurnId?: string | null;
@@ -200,6 +204,9 @@ export interface NativeSession {
    correlationId?: string;
  }): Promise<void>;
  interrupt?(input: { turnId?: string; reason?: string }): Promise<void>;
+  /** Revoke publication synchronously while the control plane journals a wait.
+   * Does not interrupt the provider; cancel owns the subsequent passive cleanup. */
+  revokeTurnPublication?(): void;
  /** Commit cancellation synchronously; the returned promise owns cleanup only. */
  cancel?(input: {
    reason: string;
@@ -4403,7 +4403,9 @@ describe("executeNativeSession recovery", () => {
    expect(lifecycle).toEqual(["closed"]);
  });

-  it("commits a governed wait without waiting for abort-insensitive provider cleanup", async () => {
+  it("commits a proved governed wait without waiting for abort-insensitive provider cleanup", async () => {
+    vi.useFakeTimers();
+    try {
    const lifecycle: string[] = [];
    let cancellationSignal: AbortSignal | undefined;
    const cancel = vi.fn(({ signal }: { signal: AbortSignal }) => {
@@ -4429,6 +4431,8 @@ describe("executeNativeSession recovery", () => {
      },
      async *events() {
        yield runnerEvent(1, "item.completed");
+        expect(cancel).toHaveBeenCalledOnce();
+        yield runnerEvent(2, "turn.interrupted", { providerTerminalObserved: true });
      },
      async startTurn() {
        return { turnId: "turn-recovery" };
@@ -4498,7 +4502,13 @@ describe("executeNativeSession recovery", () => {
      keepSessionOpen: true,
      onSession: (current) => retainedSessions.push(current),
    });
-    await expect(execution).resolves.toMatchObject({ result: yieldedResult });
+    let settled = false;
+    const completed = execution.then(value => { settled = true; return value; });
+    void completed.catch(() => undefined);
+    await vi.advanceTimersByTimeAsync(9_900);
+    expect(settled).toBe(false);
+    await vi.advanceTimersByTimeAsync(200);
+    await expect(completed).resolves.toMatchObject({ result: yieldedResult });
    expect(cancel).toHaveBeenCalledOnce();
    expect(cancellationSignal?.aborted).toBe(true);
    expect(close).toHaveBeenCalledOnce();
@@ -4506,9 +4516,11 @@ describe("executeNativeSession recovery", () => {
    expect(retainedSessions.at(-1)).toBeNull();
    expect(events.map((event) => event.eventType)).toEqual([
      "item.completed",
+      "turn.interrupted",
      "run.result.accepted",
      "run.terminal",
    ]);
+    } finally { vi.useRealTimers(); }
  });

  it("uses the execution timeout when a completion report has no provider terminal", async () => {
@@ -4637,7 +4649,7 @@ describe("executeNativeSession recovery", () => {
    expect(retainedSessions.at(-1)).toBeNull();
  });

-  it("retains a reusable session while a semantic terminal releases its remote subscription", async () => {
+  it.each([false, true])("retains a reusable session while terminal teardown settles (governed=%s)", async (governed) => {
    const close = vi.fn(async () => undefined);
    const retainedSessions: Array<NativeSession | null> = [];
    const session: NativeSession = {
@@ -4653,9 +4665,9 @@ describe("executeNativeSession recovery", () => {
      },
      async *events() {
        try {
-          yield runnerEvent(1, "run.result.proposed", result);
+          yield { ...runnerEvent(1, governed ? "item.completed" : "run.result.proposed", governed ? { kind: "dynamicToolCall" } : result), turnId: "turn-recovery" };
          yield {
-            ...runnerEvent(2, "turn.completed"),
+            ...runnerEvent(2, governed ? "turn.interrupted" : "turn.completed"),
            turnId: "turn-recovery",
          };
        } finally {
@@ -4680,6 +4692,7 @@ describe("executeNativeSession recovery", () => {
          lineage: [],
        };
      },
+      cancel: () => ({ cleanup: Promise.resolve() }),
      close,
    };
    const backend: NativeSessionBackend = {
@@ -4720,10 +4733,11 @@ describe("executeNativeSession recovery", () => {
        controlPlane: port,
        runnerInstanceId: "runner-recovery",
        controlPlaneInstanceId: "control-recovery",
+        resolveGovernedWait: governed ? () => yieldedResult : undefined,
        keepSessionOpen: true,
        onSession: (current) => retainedSessions.push(current),
      }),
-    ).resolves.toMatchObject({ result, terminal });
+    ).resolves.toMatchObject({ result: governed ? yieldedResult : result });

    expect(close).not.toHaveBeenCalled();
    expect(retainedSessions).toEqual([session]);
@@ -7641,7 +7655,192 @@ describe("executeNativeSession recovery", () => {
    ]);
  });

-  it("parks a provider turn immediately after a durable governed wait appears", async () => {
+  it.each(["end", "throw", "timeout"])("keeps a governed wait unfinalized when complete usage is followed by %s without a provider terminal", async fault => {
+    vi.useFakeTimers();
+    try {
+      const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" };
+      const usage = { ...runnerEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId };
+      const history: PrpEvent[] = [];
+      let checkpoint: PersistedNativeSession | undefined;
+      let releaseStream!: () => void;
+      const streamClosed = new Promise<void>(resolve => { releaseStream = resolve; });
+      const complete = vi.fn(async () => undefined);
+      const session: NativeSession = {
+        identity: () => identity,
+        async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; },
+        async startTurn() { return { turnId: trigger.turnId }; },
+        cancel() { return { cleanup: Promise.resolve() }; },
+        async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity, activeTurnId: trigger.turnId }; },
+        async result() { return null; }, async close() { releaseStream(); },
+        async *events() {
+          yield trigger; yield usage;
+          if (fault === "throw") throw new Error("provider stream lost");
+          if (fault === "timeout") await streamClosed;
+        },
+      };
+      const backend: NativeSessionBackend = {
+        async descriptor() { return { kind: "mock", name: "missing-stop", version: "1", capabilities: await session.capabilities() }; },
+        async openSession() { return session; },
+      };
+      const port: ControlPlanePort = {
+        async openRun() {}, completeRun: complete,
+        async checkpointSession(value) { checkpoint = structuredClone(value); },
+        async appendEvent(event) { history.push(event); return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
+        async replayEvents(request) { return { events: history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: 0 }; },
+      };
+      const outcome = expect(executeNativeSession({ input, backend, controlPlane: port, timeoutMs: 1_000,
+        runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", resolveGovernedWait: () => yieldedResult }))
+        .rejects.toMatchObject({ providerCode: "governed_wait_provider_terminal_unproven" });
+      void outcome.catch(() => undefined);
+      await vi.advanceTimersByTimeAsync(600);
+      await outcome;
+      expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult });
+      expect(complete).not.toHaveBeenCalled();
+      expect(history).toEqual([trigger, usage]);
+    } finally { vi.useRealTimers(); }
+  });
+
+  it.each(["interrupt_acknowledgement", "runtime_request.cancelled", "runtime_request.expired"])("retains %s in contiguous governed-wait history for recovery", async cancellationKind => {
+    const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" };
+    const acknowledgement: PrpEvent = { ...runnerEvent(2, cancellationKind === "interrupt_acknowledgement" ? "item.completed" : cancellationKind as "runtime_request.cancelled" | "runtime_request.expired",
+      cancellationKind === "interrupt_acknowledgement" ? { kind: cancellationKind } : { requestId: "request-1" }), turnId: trigger.turnId };
+    const usage = { ...runnerEvent(3, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId };
+    const stopped = { ...runnerEvent(4, "turn.interrupted", { providerTerminalObserved: true }), turnId: trigger.turnId };
+    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 session: NativeSession = {
+      identity: () => identity,
+      async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; },
+      startTurn: start, cancel,
+      async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity, activeTurnId: trigger.turnId }; },
+      async result() { return null; }, async close() {},
+      async *events() { yield trigger; yield acknowledgement; yield usage; yield stopped; },
+    };
+    const backend: NativeSessionBackend = {
+      async descriptor() { return { kind: "mock", name: "cancel-history", version: "1", capabilities: await session.capabilities() }; },
+      async openSession() { return session; }, async recoverSession() { return { recovered: true, session }; },
+    };
+    const port: ControlPlanePort = {
+      async openRun() {}, completeRun: complete,
+      async checkpointSession(value) { checkpoint = structuredClone(value); },
+      async appendEvent(event) {
+        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);
+        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: () => yieldedResult }))
+      .resolves.toMatchObject({ result: yieldedResult });
+    expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]);
+    await expect(executeNativeSession({ ...options, persistedSession: checkpoint })).resolves.toMatchObject({ result: yieldedResult });
+    expect(start).toHaveBeenCalledOnce();
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it.each(["trigger", "usage", "terminal"])("recovers a journaled governed wait after its answer is saved (last committed: %s)", async lastCommitted => {
+    const terminalCommitted = lastCommitted === "terminal";
+    const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall", item: { id: "ask-1", name: "request_human_input" } }), turnId: "turn-waiting" };
+    const usage = { ...runnerEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId };
+    const stopped = { ...runnerEvent(3, "turn.interrupted", { providerTerminalObserved: true }), turnId: trigger.turnId };
+    const history: PrpEvent[] = [];
+    let checkpoint: PersistedNativeSession | undefined;
+    let crashed = false;
+    let interrupted = false;
+    const revoke = vi.fn();
+    const cancel = vi.fn(() => { interrupted = true; return { cleanup: Promise.resolve() }; });
+    const start = vi.fn(async () => ({ turnId: trigger.turnId }));
+    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: revoke, cancel,
+      async snapshot() { return snapshot(); }, async result() { return null; }, async close() {},
+      async *events() {
+        if (!recovering) { yield trigger; return; }
+        // The restart resumes settlement only; it must not submit a model turn.
+        expect(interrupted).toBe(true);
+        yield usage;
+        yield stopped;
+      },
+    });
+    const backend: NativeSessionBackend = {
+      async descriptor() { return { kind: "mock", name: "durable-wait", 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);
+        if (value.governedWait && !crashed) {
+          expect(revoke).toHaveBeenCalledOnce();
+          expect(cancel).not.toHaveBeenCalled();
+          crashed = true;
+          throw new Error("controller lost after governed checkpoint");
+        }
+      },
+      async appendEvent(event) {
+        if (!history.some(row => row.sourceEventId === event.sourceEventId)) history.push(event);
+        return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" };
+      },
+      async replayEvents(request) {
+        const events = history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq);
+        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: () => yieldedResult }))
+      .rejects.toThrow("controller lost after governed checkpoint");
+    expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult });
+    expect(complete).not.toHaveBeenCalled();
+    // The failed controller's ordinary cleanup also invokes cancel. Its
+    // checkpoint assertion above proves interruption followed persistence.
+    cancel.mockClear(); interrupted = false;
+    if (lastCommitted !== "trigger") history.push(usage);
+    if (terminalCommitted) 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();
+    expect(cancel).toHaveBeenCalledTimes(terminalCommitted ? 0 : 1);
+    expect(answerAlreadySaved).not.toHaveBeenCalled();
+    expect(complete).toHaveBeenCalledOnce();
+    expect(history.filter(event => event.payload.kind === "usage")).toEqual([usage]);
+  });
+
+  it.each(["missing-trigger", "changed-trigger", "unproven-terminal", "different-turn"])("rejects unsafe governed-wait recovery: %s", async fault => {
+    const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" };
+    const checkpoint: PersistedNativeSession = { backendKind: "mock", sessionId: identity.sessionId, identity,
+      activeTurnId: trigger.turnId, governedWait: { sourceEvent: trigger, result: yieldedResult } };
+    const history = fault === "missing-trigger" ? [] : [fault === "changed-trigger" ? { ...trigger, payload: {} } : trigger];
+    if (fault === "unproven-terminal") history.push({ ...trigger, sourceSeq: 2, sourceEventId: "unproven", eventType: "turn.failed", payload: { providerTerminalObserved: false } });
+    if (fault === "different-turn") history.push({ ...trigger, sourceSeq: 2, sourceEventId: "new-turn", turnId: "different", eventType: "turn.started" });
+    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-wait", 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: history.at(-1)?.sourceSeq ?? 0 }; } };
+    await expect(executeNativeSession({ input, backend, controlPlane: port, persistedSession: checkpoint,
+      runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" })).rejects.toThrow(/governed_wait|failed terminal/);
+    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 {
    const yielded: PrpStructuredRunResult = {
      schema: "paperclip.run_result.v1",
      reportedWorkDisposition: "yielded",
@@ -7681,8 +7880,8 @@ describe("executeNativeSession recovery", () => {
      turnId: "turn-waiting",
    };
    const turnInterrupted: PrpEvent = {
-      ...controlEvent(2, "turn.interrupted", { reason: "governed_wait" }),
-      sourceEventId: "provider-recovery:2",
+      ...controlEvent(3, "turn.interrupted", { reason: "governed_wait" }),
+      sourceEventId: "provider-recovery:3",
      sourceInstanceId: "provider-recovery",
      sourceKind: "provider",
      turnId: "turn-waiting",
@@ -7711,6 +7910,9 @@ describe("executeNativeSession recovery", () => {
      async *events() {
        yield itemCompleted;
        await cancelled;
+        await new Promise<void>(resolve => setTimeout(resolve, receiptDelayMs));
+        yield { ...itemCompleted, sourceSeq: 2, sourceEventId: "provider-recovery:2",
+          payload: { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } } };
        yield turnInterrupted;
      },
      async startTurn() {
@@ -7778,7 +7980,7 @@ describe("executeNativeSession recovery", () => {
      completeRun,
    };

-    const completed = await executeNativeSession({
+    const execution = executeNativeSession({
      input,
      backend,
      controlPlane: port,
@@ -7788,7 +7990,10 @@ describe("executeNativeSession recovery", () => {
        event.eventType === "item.completed" ? yielded : null,
    });

+    await vi.advanceTimersByTimeAsync(receiptDelayMs + 1);
+    const completed = await execution;
    expect(cancel).toHaveBeenCalledOnce();
+    expect(events.find(event => event.payload.kind === "usage")?.payload.usage).toMatchObject({ runDelta: { inputTokens: 7, outputTokens: 3 } });
    expect(completed).toMatchObject({
      result: yielded,
      terminal: {
@@ -7799,9 +8004,12 @@ describe("executeNativeSession recovery", () => {
    });
    expect(events.map((event) => event.eventType)).toEqual([
      "item.completed",
+      "item.completed",
+      "turn.interrupted",
      "run.result.accepted",
      "run.terminal",
    ]);
+     } finally { vi.useRealTimers(); }
  });

  it("hands a committed structured input to the durable wait after its live window", async () => {
@@ -47,6 +47,11 @@ const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100;
 // bounded by the transport. Other iterator and handoff cleanup keeps the short
 // fail-closed quarantine boundary below.
 const REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS = 10_000;
+// Governed cancellation can traverse the ACPX turn cancellation (2s), host
+// cancellation (2s), protocol/TERM/KILL shutdown (7s), and final status read
+// (1s). Allow that bounded settlement plus transport scheduling; never treat
+// elapsed time or a cancellation acknowledgement as usage/terminal evidence.
+const GOVERNED_ACCOUNTING_DRAIN_MS = 15_000;
 const DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS = 30_000;
 type NativeSessionCleanupDomain = string;

@@ -870,12 +875,15 @@ async function consumeTurn(
  },
  initialGoal?: HarnessThreadGoal | null,
  resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise<boolean>,
+  checkpointGovernedWait?: (wait: NonNullable<PersistedNativeSession["governedWait"]>, signal: AbortSignal) => Promise<void>,
+  initialGovernedWait?: PersistedNativeSession["governedWait"],
 ) {
  let timer: ReturnType<typeof setTimeout> | undefined;
  const appendAbort = new AbortController();
  const governedCleanupOperations = new Set<Promise<unknown>>();
  let governedCancellationCommitted = false;
  let semanticResultObserved = false;
+  let governedProviderTerminalObserved = false;
  let deferredGovernedCleanupSettlement: Promise<unknown> | null = null;
  let deferredSessionCancellationSettlement: Promise<unknown> | null = null;
  const inputTimers = new Map<string, ReturnType<typeof setTimeout>>();
@@ -922,7 +930,7 @@ async function consumeTurn(
    let latestSessionGoal: HarnessThreadGoal | null = null;
    let resultSource: "semantic_result" | "governed_wait" | null = null;
    let providerFailure: NativeProviderTerminalFailure | null = null;
-    const settleDurableResult = (
+    const settleDurableResult = async (
      event: PrpEvent,
      result: PrpStructuredRunResult,
      reason: string,
@@ -930,16 +938,77 @@ async function consumeTurn(
      if (session.cancel === undefined) {
        throw new Error("native_governed_wait_cancellation_unavailable");
      }
-      const cancellation = session.cancel({
-        reason,
-        signal: appendAbort.signal,
-      });
+      // Revoke new publication before any asynchronous journal write. Persist
+      // the disposition before provider interruption, retaining usage and terminal
+      // evidence even when the interaction is answered during the restart.
+      session.revokeTurnPublication?.();
+      await checkpointGovernedWait?.({ sourceEvent: event, result }, appendAbort.signal);
+      appendAbort.signal.throwIfAborted();
+      const cancellation = session.cancel({ reason, signal: appendAbort.signal });
      governedCancellationCommitted = true;
      const cleanup = cancellation.cleanup;
      governedCleanupOperations.add(cleanup);
      void cleanup
        .catch(() => quarantineSession("governed_cleanup_failed"))
        .finally(() => governedCleanupOperations.delete(cleanup));
+      // Cancellation revokes provider work immediately, but shutdown can still
+      // supply its final usage. Keep a bounded accounting-only drain before
+      // acknowledging the run; otherwise the next wake sees permanent debt.
+      // Never let a stuck stream or cleanup revoke the durable governed wait.
+      const drainAbort = new AbortController();
+      const stopDrain = () => drainAbort.abort(appendAbort.signal.reason);
+      appendAbort.signal.addEventListener("abort", stopDrain, { once: true });
+      const drainMs = Math.min(GOVERNED_ACCOUNTING_DRAIN_MS, timeoutMs > 0 ? Math.max(1, Math.floor(timeoutMs / 4)) : GOVERNED_ACCOUNTING_DRAIN_MS);
+      let drainTimer: ReturnType<typeof setTimeout> | undefined;
+      const drained = (async () => {
+        while (!drainAbort.signal.aborted) {
+          const next = await eventIterator.next();
+          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)) {
+            const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal });
+            if (drainAbort.signal.aborted) return;
+            eventCount += receipt.disposition === "committed" ? 1 : 0;
+            highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq);
+            if (terminal) {
+              if (trailing.eventType === "turn.failed" || trailing.payload.providerTerminalObserved === false) {
+                throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
+              }
+              governedProviderTerminalObserved = true;
+              return;
+            }
+          }
+        }
+      })();
+      await Promise.race([
+        drained.catch(error => {
+          quarantineSession("governed_accounting_drain_failed");
+          if (error instanceof NativeProviderTerminalFailure) throw error;
+        }),
+        new Promise<void>(resolve => { drainTimer = setTimeout(() => {
+          drainAbort.abort(new Error("governed accounting drain timed out"));
+          quarantineSession("governed_accounting_drain_timed_out");
+          resolve();
+        }, drainMs); }),
+      ]).finally(() => {
+        clearTimeout(drainTimer);
+        appendAbort.signal.removeEventListener("abort", stopDrain);
+        drainAbort.abort();
+      });
+      // A durable wait and complete usage cannot prove that the provider
+      // stopped. Leave the journaled wait recoverable, without a successful
+      // result or an accounting-completion boundary, if shutdown is unproven.
+      if (!governedProviderTerminalObserved) {
+        throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
+      }
      return {
        event,
        eventCount,
@@ -947,6 +1016,10 @@ async function consumeTurn(
        governedResult: result,
      };
    };
+    if (initialGovernedWait) {
+      return settleDurableResult(initialGovernedWait.sourceEvent, initialGovernedWait.result,
+        "Paperclip resumed settlement of this durable governed interaction.");
+    }
    while (true) {
      const next = await eventIterator.next().catch((error) => {
        throw providerFailure ?? error;
@@ -1309,15 +1382,14 @@ async function consumeTurn(
      );
    } else {
      // Iterator and provider cleanup own no control-plane mutation authority.
-      // A reusable session with a semantic result needs a longer bounded
-      // window for the remote event subscription to release. This applies
-      // after the provider terminal, which still crosses the remote PRP
-      // acknowledgement boundary and routinely takes longer than
-      // the generic local cleanup grace. Governed waits and unrelated stalled
-      // cleanup retain the short fail-closed boundary.
+      // A reusable session needs a bounded window for its remote terminal
+      // subscription to release. A governed wait qualifies only after its
+      // actual provider terminal was retained during the accounting drain.
+      // Missing terminal evidence and unrelated stalled cleanup retain the
+      // short fail-closed boundary.
      const teardownSettled = await settlesWithin(
        passiveTeardownSettlement,
-        semanticResultObserved
+        semanticResultObserved || governedProviderTerminalObserved
          ? reusableSessionCancellationGraceMs
          : FAILED_OPERATION_SETTLEMENT_GRACE_MS,
      );
@@ -1871,6 +1943,42 @@ export async function executeNativeSession(
    issueId: input.binding.issueId,
    agentId: input.binding.agentId,
  };
+  let governedWait = persistedSession?.governedWait;
+  const governedReplay: PrpEvent[] = [];
+  if (governedWait) {
+    const event = governedWait.sourceEvent;
+    const validation = validatePrpStructuredRunResult(governedWait.result);
+    if (!event?.turnId || event.runId !== identity.runId || event.normalizedSessionId !== identity.sessionId
+      || event.sourceInstanceId !== options.runnerInstanceId || !Number.isSafeInteger(event.sourceSeq) || event.sourceSeq < 1
+      || !validation.ok || validation.result.reportedWorkDisposition !== "yielded"
+      || validation.result.completionClaim?.contractRevision !== input.completionContract.contract.revision) {
+      throw new Error("native_governed_wait_checkpoint_mismatch");
+    }
+    await runAbortableOperationWithin({
+      timeoutMs: options.timeoutMs ?? 900_000,
+      timeoutMessage: "native governed-wait recovery replay timed out",
+      operation: async signal => {
+        let afterSourceSeq = event.sourceSeq - 1;
+        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) {
+            if (entry.runId !== identity.runId || entry.normalizedSessionId !== identity.sessionId
+              || entry.sourceInstanceId !== event.sourceInstanceId || entry.sourceSeq !== afterSourceSeq + 1
+              || governedReplay.length >= 10_000) throw new Error("native_governed_wait_replay_mismatch");
+            governedReplay.push(entry);
+            afterSourceSeq = entry.sourceSeq;
+          }
+        }
+      },
+    });
+    if (canonicalJson(governedReplay[0]) !== canonicalJson(event)) throw new Error("native_governed_wait_trigger_unproven");
+    if (governedReplay.some(entry => entry.eventType === "turn.started" && entry.turnId !== event.turnId)) {
+      throw new Error("native_governed_wait_turn_mismatch");
+    }
+  }
  let recovered = false;
  let session: NativeSession | null = null;
  let continuityBreak: {
@@ -2152,7 +2260,7 @@ export async function executeNativeSession(
      externalSignal?: AbortSignal,
    ) =>
      persistCheckpointWithin({
-        snapshot,
+        snapshot: governedWait ? { ...snapshot, governedWait } : snapshot,
        controlPlane: options.controlPlane,
        onCheckpoint: options.onCheckpoint,
        timeoutMs: checkpointTimeoutMs,
@@ -2240,6 +2348,21 @@ export async function executeNativeSession(
      await persistCheckpoint(recoveredSnapshot);
    }

+    const governedTerminal = governedWait ? governedReplay.find(event =>
+      event.turnId === governedWait!.sourceEvent.turnId && isTurnTerminal(event)) : undefined;
+    if (governedTerminal?.eventType === "turn.failed" || governedTerminal?.payload.providerTerminalObserved === false) {
+      throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false);
+    }
+    if (governedWait && recoveredActiveTurnId && recoveredActiveTurnId !== governedWait.sourceEvent.turnId) {
+      throw new Error("native_governed_wait_turn_mismatch");
+    }
+    if (governedWait) {
+      for (const event of governedReplay) {
+        if (event.turnId === governedWait.sourceEvent.turnId && (event.payload.kind === "usage" || isTurnTerminal(event))) {
+          await options.controlPlane.appendEvent(event);
+        }
+      }
+    }
    let consumed = {
      event: null as PrpEvent | null,
      eventCount: 0,
@@ -2322,7 +2445,10 @@ export async function executeNativeSession(
        replayedDisposition?.terminal ?? dispositionFallback;
      const consumptionAbort = new AbortController();
      const consuming =
-        recoveryTerminal === null
+        governedTerminal && governedWait
+          ? Promise.resolve({ event: governedWait.sourceEvent, governedResult: governedWait.result,
+              eventCount: 0, highestContiguousSourceSeq: governedTerminal.sourceSeq })
+          : recoveryTerminal === null
          ? consumeTurn(
              session,
              options.controlPlane,
@@ -2345,6 +2471,11 @@ export async function executeNativeSession(
                : undefined,
              recoveredSnapshot.goal,
              resumeInterruptedTurn,
+              async (wait, signal) => {
+                governedWait = structuredClone(wait);
+                await checkpoint(signal);
+              },
+              governedWait,
            )
          : Promise.resolve({
              event: recoveryTerminal,
@@ -2360,12 +2491,12 @@ export async function executeNativeSession(
      // that later rejection becomes process-fatal under Node's strict policy.
      void consuming.catch(() => undefined);
      try {
-        const shouldStartFreshTurn =
+        const shouldStartFreshTurn = !governedWait && (
          !recovered ||
          (!recoveredActiveTurnId &&
            !adoptedProviderTerminal &&
            !checkpointedDispositionTerminal &&
-            !dispositionRecoveryStillOwned);
+            !dispositionRecoveryStillOwned));
        if (options.sessionGoalControl) {
          // Explicit controls remain authoritative after controller loss. In
          // particular, pause/clear/edit must reach a recovered session even
@@ -29,6 +29,8 @@ const cases: Array<{ name: string; execute: () => Promise<Execute>; event: unkno
 ];

 describe("CLI adapter accounting on timeout", () => {
+  // Keep OpenCode runtime config copies inside each fixture; host config may
+  // include plugins and large dependency trees unrelated to accounting.
  const directories: string[] = [];
  afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); });
  it.each(cases.filter(f => ["codex", "cursor", "opencode"].includes(f.name)))("$name preserves malformed optional counters through the checkpoint wrapper", async fixture => {
@@ -49,7 +51,7 @@ describe("CLI adapter accounting on timeout", () => {
        runId: "invalid-counter-run", agent: { id: "agent", companyId: "company", name: "Worker", adapterType: `${fixture.name}_local`, adapterConfig: {} },
        runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
        config: { engine: "cli", command, cwd: dir, model: fixture.name === "opencode" ? "openai/test" : "gpt-6-astra", paperclipRuntimeSkills: [],
-          env: { OPENAI_API_KEY: "fixture", OPENCODE_ALLOW_ALL_MODELS: "1" } },
+          env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "fixture", OPENCODE_ALLOW_ALL_MODELS: "1" } },
        context: {}, onLog: async () => {}, onUsage,
      });
      const valid = cached === 0 || cached === undefined;
@@ -77,7 +79,7 @@ describe("CLI adapter accounting on timeout", () => {
      runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
      runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
      config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [],
-        env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
+        env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
      context: {}, onLog: async () => {}, onUsage,
    };
    const execute = await fixture.execute();
@@ -211,7 +213,7 @@ describe("CLI adapter accounting on timeout", () => {
      const result = await execute({
        runId: "tail-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
        runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
-        config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } },
+        config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { XDG_CONFIG_HOME: dir, ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } },
        context: {}, onLog: async () => {}, onUsage,
      });
      expect(result.usage).toMatchObject({ inputTokens: 40, outputTokens: 20, cachedInputTokens: 200 });
@@ -231,7 +233,7 @@ describe("CLI adapter accounting on timeout", () => {
      runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
      runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
      config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [],
-        env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", MOONSHOT_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
+        env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", MOONSHOT_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
      context: {}, onLog: async () => {},
    });
    expect(result.timedOut).toBe(true);
@@ -317,9 +317,34 @@ const support = await getEmbeddedPostgresTestSupport();
      expect(result.results.some(item => item.service === "google-sheets")).toBe(true);
      expect(result.instruction).not.toContain("search its exact name");
    });
+    it("keeps installed capability matches when a generic query resembles an external app name", async () => {
+      await resetQuestions();
+      const installed = await seedProvider("Studio library", "notion:list_pages");
+      await db.update(toolCatalogEntries).set({ description: "Read recent pages and return their titles and verification code" })
+        .where(eq(toolCatalogEntries.connectionId, installed.id));
+      const service = connectionIntentService(db);
+      const result = await service.search(claims,
+        "connected page service that can find or list recent pages and return page titles plus a verification code");
+      expect(result.results).toEqual(expect.arrayContaining([expect.objectContaining({
+        service: `connection:${installed.id}`, connectionId: installed.id, state: "ready",
+      })]));
+      expect(result.providerQuestion).toBeUndefined();
+      expect(result.instruction).not.toContain("Ask the responsible user with providerQuestion");
+      // A deliberate app name still reaches the existing governed provider choice.
+      const named = await service.search(claims, "Page X");
+      expect(named.providerQuestion?.id).toBe("connection-provider:page-x");
+      expect(named.results[0]?.service).toBe("via:composio:page-x");
+    });
+    it("keeps an external typo as a discovery suggestion until its app is selected", async () => {
+      await resetQuestions();
+      const result = await connectionIntentService(db).search(claims,
+        "Find Circlebak meeting transcripts and action items from yesterday");
+      expect(result.results.some(item => item.service === "via:composio:circleback-mcp")).toBe(true);
+      expect(result.providerQuestion).toBeUndefined();
+      expect(result.instruction).toContain("aggregator.targetService");
+    });
    it.each([
      ["help me find tools for circle back", "circleback-mcp"],
-      ["Find Circlebak meeting transcripts and action items from yesterday", "circleback-mcp"],
      ["Can you connect Circleback MCP to get all our meeting notes?", "circleback-mcp"],
      ["Find Attio tools to review all our customer contacts before next week's meeting", "attio"],
      ["Help me find a ClickUp connection to organize our team's projects", "clickup"],
@@ -583,6 +608,51 @@ const support = await getEmbeddedPostgresTestSupport();
      );
      expect(result.instruction).toContain("Arcade");
    });
+    it("returns exact eligible catalog names after a guessed tool without granting access", async () => {
+      await resetQuestions();
+      const connection = await seedProvider("arcade", "Hubspot_ListContacts", "responsible-user", false);
+      await db.delete(toolConnectionInstalls).where(eq(toolConnectionInstalls.connectionId, connection.id));
+      await seedProvider("composio", "Unrelated_PrivateTool", "other-user", false);
+      await db.insert(toolCatalogEntries).values({ companyId: claims.company_id, connectionId: connection.id,
+        toolName: "Hubspot_RemovedTool", name: "Removed", versionHash: "old", entryKind: "tool", status: "removed" });
+      const answer = await selectProvider("via:arcade:hubspot");
+      const service = connectionIntentService(db);
+      const error = await service.request(claims, "via:arcade:hubspot", {
+        selectionInteractionId: answer.id, toolNames: ["hubspot_list_contacts"],
+      }).catch(error => error);
+      expect(error).toMatchObject({ status: 422 });
+      expect(error.message).toContain('"Hubspot_ListContacts"');
+      expect(error.message).toContain("Request only the needed exact names");
+      expect(error.message).not.toMatch(/Unrelated_PrivateTool|Hubspot_RemovedTool/);
+      expect(await db.select().from(toolProfiles).where(eq(toolProfiles.companyId, claims.company_id))).toEqual([]);
+      const interactions = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.companyId, claims.company_id));
+      expect(interactions.map(row => [row.id, row.status])).toEqual([[answer.id, "answered"]]);
+      const requested = await service.request(claims, "via:arcade:hubspot", {
+        selectionInteractionId: answer.id, toolNames: ["Hubspot_ListContacts"],
+      });
+      expect(requested.state).toBe("needs_user_action");
+      const loaded = await service.loadIntent(requested.interactionId!);
+      expect(loaded.interaction.payload.accessRequest?.tools.map(tool => tool.toolName)).toEqual(["Hubspot_ListContacts"]);
+      expect(loaded.interaction.status).toBe("pending");
+      expect(await db.select().from(toolProfiles).where(eq(toolProfiles.companyId, claims.company_id))).toEqual([]);
+    });
+    it("bounds catalog-name recovery and does not disclose an ineligible connection", async () => {
+      await resetQuestions();
+      const connection = await seedProvider("arcade", "Read_00", "responsible-user", false);
+      await db.insert(toolCatalogEntries).values(Array.from({ length: 24 }, (_, index) => ({
+        companyId: claims.company_id, connectionId: connection.id, toolName: `Read_${String(index + 1).padStart(2, "0")}`,
+        name: `Read_${String(index + 1).padStart(2, "0")}`, versionHash: "v1", entryKind: "tool" as const, status: "active" as const,
+      })));
+      const service = connectionIntentService(db);
+      const error = await service.request(claims, "arcade", { connectionId: connection.id, toolNames: ["guessed"] }).catch(error => error);
+      expect(error).toMatchObject({ status: 422 });
+      expect(error.message).toContain("first 20 of 25");
+      expect(error.message.match(/Read_\d{2}/g)).toHaveLength(20);
+      await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.connectionId, connection.id));
+      const denied = await service.request(claims, "arcade", { connectionId: connection.id, toolNames: ["guessed"] }).catch(error => error);
+      expect(denied.message).not.toContain("Read_");
+      expect(denied.message).toContain("not eligible");
+    });
    it("does not switch providers when the chosen route loses permission", async () => {
      await resetQuestions();
      const answer = await selectProvider("via:arcade:hubspot");
@@ -0,0 +1,29 @@
+import { describe, expect, it } from "vitest";
+import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils";
+import { priceAnthropicReceipt } from "./anthropic-pricing.js";
+
+const receipt: AdapterUsageCheckpoint = { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic",
+  billingType: "metered_api", model: "claude-sonnet-5", costUsd: null, costStatus: "unpriced",
+  usage: { inputTokens: 22462, cachedInputTokens: 208218, cacheWriteTokens: 22450, outputTokens: 2450 } };
+describe("complete Anthropic receipt pricing", () => {
+  it("prices complete counters conservatively and labels their provenance as an estimate", () => {
+    const priced = priceAnthropicReceipt(receipt);
+    expect(priced.costUsdExact).toBe("0.155967600");
+    expect(priced.costStatus).toBe("estimated");
+    expect(priced.pricingProvenance).toMatchObject({ source: "rate_card", version: "anthropic-standard-2026-10-07" });
+    expect(priced.pricingProvenance?.evidence).toContain("one-hour upper rate");
+    expect(priced.usage).toBe(receipt.usage);
+    expect(receipt.costStatus).toBe("unpriced");
+  });
+  it.each([
+    { complete: false }, { billingType: "unknown" }, { billingType: "subscription_included" },
+    { provider: "unknown" }, { biller: "aws_bedrock" }, { model: "claude-sonnet-latest" },
+    { usageBasis: "session_cumulative" }, { costUsd: 0.12 }, { costUsdExact: "0.12" }, { cacheAdjustedCostUsd: 0.12 },
+    { pricingContext: { serviceTier: "priority" } }, { usageByModel: [{ model: "other", usage: { inputTokens: 1, outputTokens: 1 }, costUsd: 1 }] },
+    { usage: undefined }, { usage: { inputTokens: 1, outputTokens: 1 } },
+    { usage: { inputTokens: 1, outputTokens: 1, cacheWriteTokens: 2, cachedInputTokens: 0 } },
+  ] as Partial<AdapterUsageCheckpoint>[])("preserves incomplete, incompatible or already priced receipts: %j", override => {
+    const original = { ...receipt, ...override };
+    expect(priceAnthropicReceipt(original)).toBe(original);
+  });
+});
@@ -0,0 +1,31 @@
+import { centsToUsd, unitsToCents, usdToUnits } from "@paperclipai/shared";
+import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils";
+
+// Direct Anthropic API list prices, verified 2026-10-07:
+// https://platform.claude.com/docs/en/about-claude/pricing
+// Keep this bounded to the qualified model; aliases and other billers stay unknown.
+const RATES = ["2", "0.2", "4", "10"] as const;
+
+/** A complete token receipt can receive an explicit estimate, never an invoice
+ * claim. ACP aggregates cache writes without TTL; conservatively price them at
+ * the documented one-hour rate rather than silently assume five-minute writes. */
+export function priceAnthropicReceipt(receipt: AdapterUsageCheckpoint): AdapterUsageCheckpoint {
+  if (!receipt.complete || !receipt.usage || receipt.usageBasis !== "per_run"
+    || receipt.provider !== "anthropic" || receipt.biller !== "anthropic"
+    || !["api", "metered_api"].includes(receipt.billingType ?? "")
+    || receipt.model !== "claude-sonnet-5" || receipt.usageByModel?.length
+    || receipt.costUsd != null || receipt.costUsdExact != null || receipt.cacheAdjustedCostUsd != null
+    || (receipt.pricingContext?.serviceTier && !["standard", "default"].includes(receipt.pricingContext.serviceTier))) return receipt;
+  const usage = receipt.usage;
+  if (usage.cacheWriteTokens === undefined || usage.cachedInputTokens === undefined) return receipt;
+  const counts = [usage.inputTokens - usage.cacheWriteTokens, usage.cachedInputTokens, usage.cacheWriteTokens, usage.outputTokens];
+  if (counts.some(count => !Number.isSafeInteger(count) || count < 0)) return receipt;
+  const numerator = counts.reduce((total, count, index) => total + BigInt(count) * usdToUnits(RATES[index]), 0n);
+  return { ...receipt, costUsdExact: centsToUsd(unitsToCents((numerator + 500_000n) / 1_000_000n)), costStatus: "estimated",
+    pricingProvenance: {
+      source: "rate_card", version: "anthropic-standard-2026-10-07",
+      evidence: "https://platform.claude.com/docs/en/about-claude/pricing; standard global API pricing assumed; cache-write TTL unavailable, one-hour upper rate used; not an invoice",
+      inputCentsPerMillion: "200", cachedInputCentsPerMillion: "20", cacheWriteCentsPerMillion: "400", outputCentsPerMillion: "1000", serviceTier: "standard",
+    },
+  };
+}
@@ -466,7 +466,8 @@ export function connectionIntentService(db: Db) {
    }
    const publicMatches = searchAggregatorServices(preparedQuery);
    const bestMatches = publicMatches.filter(match => Math.floor(match.nameScore / 100) === Math.floor((publicMatches[0]?.nameScore ?? 0) / 100));
-    const publicService = findAggregatorService(serviceQuery, publicMatches) ?? (bestMatches.length === 1 ? bestMatches[0]!.service : undefined);
+    const namedPublicService = findAggregatorService(serviceQuery, publicMatches);
+    const publicService = namedPublicService ?? (bestMatches.length === 1 ? bestMatches[0]!.service : undefined);
    // Extract service namespaces only from the current identity's indexed tools.
    // Generic provider search/execute descriptions do not prove app support.
    const indexedServices = new Set<string>();
@@ -553,6 +554,14 @@ export function connectionIntentService(db: Db) {
      return alternatives;
    }
    const alternatives = await aggregatorAlternatives(targetService, targetName, publicService);
+    // A typo/prefix match is a discovery hint, not a selected app. Generic
+    // capabilities (e.g. "pages") can resemble a catalog name ("Page X").
+    // Retain authorized installed matches instead of prescribing provider consent.
+    const namedTarget = namedPublicService !== undefined
+      || indexedMatches.some(match => match.slug === targetService && match.nameScore >= 500);
+    if (alternatives.length && !namedTarget) {
+      return discoverySuggestions([...ranked(), ...alternatives]);
+    }
    if (explicit && explicitConsent) {
      const selected = alternatives.find(item => item.aggregator?.provider === explicit.provider);
      if (!selected) return { version: 1, query, results: [], instruction: "The explicitly requested external provider is unavailable or its app support could not be verified. Explain the limitation. Do not switch providers automatically." };
@@ -586,7 +595,7 @@ export function connectionIntentService(db: Db) {

    function discoverySuggestions(results: ConnectionSearchResultItem[]): ConnectionsSearchResult {
      return { version: 1, query, results: results.slice(0, 40),
-        instruction: "Multiple apps match this query. Choose the relevant service and method using descriptions and purposes. For available or needs_user_action results, tool methods and AgentMail use connection_request to show the inline setup card; other channel or AI methods use setupPath. Use ready tools as installed; ready AI authentication applies to the next execution without reconnection. For an aggregator result, search its aggregator.targetService to obtain the provider-choice question and follow that instruction before requesting a connection. Respect unavailable states. Do not treat a search match as provider consent or app authorization." };
+        instruction: "These are possible connection matches. Choose the relevant service and method using descriptions and purposes. For available or needs_user_action results, tool methods and AgentMail use connection_request to show the inline setup card; other channel or AI methods use setupPath. Use ready tools as installed; ready AI authentication applies to the next execution without reconnection. For an aggregator result, search its aggregator.targetService to obtain the provider-choice question and follow that instruction before requesting a connection. Respect unavailable states. Do not treat a search match as provider consent or app authorization." };
    }
  }

@@ -730,7 +739,12 @@ export function connectionIntentService(db: Db) {
        const catalog = (await indexedCatalog(connection.id, context.run.companyId)).filter(tool => tool.entryKind === "tool");
        const names = options.toolNames ?? (app.slug === "composio" ? ["COMPOSIO_SEARCH_TOOLS", "COMPOSIO_MANAGE_CONNECTIONS"] : catalog.map(tool => tool.toolName));
        const tools = names.map(name => catalog.find(tool => tool.toolName === name));
-        if (options.toolNames && tools.some(tool => !tool)) throw unprocessable("A requested tool is not in this connection's active catalog");
+        if (options.toolNames && tools.some(tool => !tool)) {
+          // Return discovery metadata only after validating this connection's
+          // eligibility. Never substitute guessed names or grant access here.
+          const availableNames = catalog.map(tool => tool.toolName).sort();
+          throw unprocessable(`A requested tool is not in this connection's active catalog. Available indexed tool names (first ${Math.min(20, availableNames.length)} of ${availableNames.length}): ${JSON.stringify(availableNames.slice(0, 20))}. Request only the needed exact names with connection_request. No access was granted.`);
+        }
        const effective = await access.getEffectiveProfilesForAgent(context.run.companyId, context.agent.id);
        const installed = effective.installedConnections.some(item => item.id === connection.id);
        const missing = !installed || tools.some(tool => tool && !effective.allowedTools.some(allowed => allowed.id === tool.id));
@@ -54,6 +54,47 @@ describe("native final-response feedback", () => {
    await db.update(issueThreadInteractions).set({ sourceRunId: null }).where(eq(issueThreadInteractions.id, interactionId));
    await expect(nativeCompletionFeedback(db, value.runId, waiting)).resolves.toContain("Completion report accepted");
  });
+  it.each(["blocked", "needs_review", "yielded"] as const)("rejects a %s approval report bound to an already declined invocation", async (disposition) => {
+    const value = await fixture(), interactionId = randomUUID(), invocationId = randomUUID(), actionRequestId = randomUUID();
+    await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: value.companyId, issueId: value.issueId,
+      sourceRunId: value.runId, createdByAgentId: value.agentId, kind: "request_confirmation", status: "rejected",
+      continuationPolicy: "wake_assignee", result: { version: 1, outcome: "rejected" },
+      payload: { version: 1, prompt: "Approve service read", toolAction: { version: 1, actionRequestId, invocationId, toolName: "pages.read", toolDisplayName: "Read pages", connectionId: null, applicationId: null, appDisplayName: null, risk: "read", previewMarkdown: "Read pages", argumentsSummaryJson: "{}", argumentsHash: "test-hash", expiresAt: "2026-10-07T12:00:00Z" } },
+    });
+    // The provider can name its exact invocation in the summary without citing
+    // an interaction evidence ref, as in the retained OpenCode failure.
+    const report: PrpStructuredRunResult = { ...done, reportedWorkDisposition: disposition,
+      summary: disposition === "blocked" ? "Waiting on the page-service approval." : `Waiting on the page-service approval (invocationId ${invocationId}).`,
+      ...(disposition === "blocked" ? { blocker: { reasonCode: "approval_required", owner: { kind: "user" as const, name: "Release Owner" }, scope: "task_wide" as const, unblockAction: `Resolve invocation ${invocationId}` } } : {}),
+      completionClaim: { ...done.completionClaim, objectiveSatisfied: false, remainingWork: [{ description: "Wait for the decision", blocksCompletion: true }] },
+      attentionRequests: [{ kind: "approval", ownerClass: "human", summary: "Approve the page-service call" }],
+      ...(disposition === "yielded" ? { continuation: { kind: "response_wake" as const, idempotencyKey: "wait", summary: "Wait for approval" } } : {}),
+    };
+    await expect(nativeCompletionFeedback(db, value.runId, report)).rejects.toThrow("already resolved");
+    // A blocked report may name the exact action only in an unmet criterion.
+    const criterionOnly: PrpStructuredRunResult = { ...report, blocker: undefined,
+      summary: "Await the requested approval", evidence: [{ ref: `approval:${actionRequestId}` }],
+      completionClaim: { ...report.completionClaim, criteria: [{ criterionId: "objective", status: "not_satisfied", evidenceRefs: [`approval:${actionRequestId}`] }] },
+    };
+    await expect(nativeCompletionFeedback(db, value.runId, criterionOnly)).rejects.toThrow("already resolved");
+    // A provider can mark the approval criterion unknown while citing the exact
+    // declined interaction; that still is not a new review target.
+    await expect(nativeCompletionFeedback(db, value.runId, {
+      ...criterionOnly, completionClaim: { ...criterionOnly.completionClaim, criteria: [{ criterionId: "objective", status: "unknown", evidenceRefs: [`interaction:${interactionId}`] }] },
+    })).rejects.toThrow("already resolved");
+    // Satisfied historical evidence does not bind the independent review target.
+    if (disposition !== "yielded") await expect(nativeCompletionFeedback(db, value.runId, {
+      ...criterionOnly, completionClaim: { ...criterionOnly.completionClaim, criteria: [{ criterionId: "past-action", status: "satisfied", evidenceRefs: [`approval:${actionRequestId}`] }] },
+    })).resolves.toContain("report accepted");
+    // Completed-action evidence is not the target of an independent review.
+    if (disposition !== "yielded") await expect(nativeCompletionFeedback(db, value.runId, {
+      ...report, blocker: undefined, summary: "Review a separate deliverable", evidence: [{ ref: `interaction:${interactionId}` }],
+    })).resolves.toContain("report accepted");
+    // Neither a similar description nor another run's card is identity proof.
+    await expect(nativeCompletionFeedback(db, value.runId, { ...report, blocker: undefined, summary: "Wait for a separate approval" })).resolves.toContain("report accepted");
+    await db.update(issueThreadInteractions).set({ sourceRunId: null }).where(eq(issueThreadInteractions.id, interactionId));
+    await expect(nativeCompletionFeedback(db, value.runId, report)).resolves.toContain("report accepted");
+  });
  it.each(["accepted", "rejected"] as const)("allows a new question after a %s tool approval cited as completed evidence", async (status) => {
    const value = await fixture(), interactionId = randomUUID(), questionId = randomUUID();
    await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: value.companyId, issueId: value.issueId,
@@ -176,21 +176,31 @@ export async function nativeCompletionFeedback(
      .limit(1)
      .then((rows) => rows[0]),
  ]);
-  // A response wake for this run's tool action already has a durable approval
-  // surface. Reject a repeated approval request before it becomes a new review.
-  // General evidence can also cite completed actions while waiting on something
-  // else, so a resolved card alone is not a stale wait target.
-  if (result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "response_wake") {
-    const referencedIds = result.evidence.flatMap(({ ref }) => {
-      const match = typeof ref === "string" ? /^interaction:([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})$/i.exec(ref) : null;
-      return match ? [match[1]!] : [];
-    });
-    const referenced = referencedIds.length ? await db.select().from(issueThreadInteractions).where(and(
+  // Match an existing action by its exact durable identity, regardless of how
+  // the provider labels its final disposition. A summary can cite the invocation
+  // ID instead of an interaction evidence ref; prose similarity is not identity.
+  if (signals.actionableAttentionRequests.some(request => request.kind === "approval" && request.ownerClass === "human")) {
+    const approvalTarget = {
+      summary: result.summary,
+      attentionRequests: signals.actionableAttentionRequests.filter(request => request.kind === "approval" && request.ownerClass === "human"),
+      blocker: result.blocker,
+      // An unmet criterion can name the action still being requested. Completed
+      // evidence elsewhere in the report is not authority to suppress a review.
+      unmetCriteria: result.completionClaim.criteria.filter(criterion => criterion.status !== "satisfied"),
+      continuation: result.continuation,
+      // The existing response-wake contract also names its card through evidence.
+      evidence: result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "response_wake" ? result.evidence : [],
+    };
+    const referencedIds = new Set(JSON.stringify(approvalTarget).match(/\b[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}\b/gi)?.map(id => id.toLowerCase()) ?? []);
+    const cards = referencedIds.size ? await db.select().from(issueThreadInteractions).where(and(
      eq(issueThreadInteractions.companyId, run.companyId), eq(issueThreadInteractions.issueId, issue.id),
      eq(issueThreadInteractions.sourceRunId, run.id), eq(issueThreadInteractions.createdByAgentId, run.agentId),
      eq(issueThreadInteractions.kind, "request_confirmation"), eq(issueThreadInteractions.continuationPolicy, "wake_assignee"),
-      inArray(issueThreadInteractions.id, referencedIds),
    )) : [];
+    const referenced = cards.filter(card => {
+      const action = record(record(card.payload).toolAction);
+      return [card.id, action.actionRequestId, action.invocationId].some(id => typeof id === "string" && referencedIds.has(id.toLowerCase()));
+    });
    for (const card of referenced) {
      const action = record(record(card.payload).toolAction);
      if (action.version !== 1 || typeof action.actionRequestId !== "string") continue;
@@ -5141,6 +5141,62 @@ describe("native startup cancellation fence", () => {
 });

 describe("native startup restart detachment", () => {
+  it("waits for a checkpointed governed settlement before relinquishing its controller", async () => {
+    const restarting = structuredClone(execution);
+    restarting.binding.runId = "restart-governed-settlement";
+    let release!: () => void, announce!: () => void;
+    const accounting = new Promise<void>(resolve => { release = resolve; });
+    const checkpointed = new Promise<void>(resolve => { announce = resolve; });
+    const detach = vi.fn(async () => undefined);
+    state.execute.mockReset().mockImplementationOnce(async options => {
+      await options.onSession({ detachControllerForRestart: detach });
+      await options.onCheckpoint({ governedWait: { sourceEvent: { turnId: "settling-turn" } } });
+      announce();
+      await accounting;
+      await options.onSession(null);
+      throw new Error("settlement test complete");
+    });
+    const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" }).catch(error => error);
+    await checkpointed;
+    const detached = detachNativeSessionsForRestart([restarting.binding.runId]);
+    try {
+      await new Promise(resolve => setImmediate(resolve));
+      expect(detach).not.toHaveBeenCalled();
+      release();
+      await outcome;
+      expect(await detached).toMatchObject({ inactiveRunIds: [restarting.binding.runId] });
+      expect(detach).not.toHaveBeenCalled();
+    } finally { release(); await outcome; await detached; }
+  });
+  it("bounds governed settlement without allowing the detached controller to publish success", async () => {
+    const restarting = structuredClone(execution);
+    restarting.binding.runId = "restart-governed-settlement-timeout";
+    let release!: () => void, announce!: () => void;
+    const accounting = new Promise<void>(resolve => { release = resolve; });
+    const checkpointed = new Promise<void>(resolve => { announce = resolve; });
+    const detach = vi.fn(async () => undefined);
+    const onUsage = vi.fn(async () => undefined);
+    state.execute.mockReset().mockImplementationOnce(async options => {
+      await options.onSession({ detachControllerForRestart: detach });
+      await options.onCheckpoint({ governedWait: { sourceEvent: { turnId: "settling-turn" } } });
+      announce(); await accounting;
+      await options.onSession(null);
+      return { result: {}, terminal: { runTerminalState: "succeeded" }, usage: null };
+    });
+    const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner", onUsage }).catch(error => error);
+    await checkpointed;
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
+    const detaching = detachNativeSessionsForRestart([restarting.binding.runId]);
+    try {
+      await vi.advanceTimersByTimeAsync(19_999);
+      expect(detach).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(1);
+      expect(await detaching).toMatchObject({ detachedRunIds: [restarting.binding.runId] });
+      release();
+      expect(await outcome).toBeInstanceOf(NativeControllerDetachedForRestartError);
+      expect(onUsage).not.toHaveBeenCalled();
+    } finally { vi.useRealTimers(); release(); await outcome; await detaching; }
+  });
  it("waits for in-flight runner startup and its detach acknowledgement before shutdown returns", async () => {
    const root = await mkdtemp(join(tmpdir(), "native-startup-detach-"));
    const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
@@ -5186,6 +5242,24 @@ describe("native startup restart detachment", () => {
    }
  });

+  it("rejects a governed completion returned successfully by an already detached controller", async () => {
+    const restarting = structuredClone(execution);
+    restarting.binding.runId = "restart-successful-old-consumer";
+    const detach = vi.fn(async () => undefined);
+    const onUsage = vi.fn(async () => undefined);
+    state.execute.mockReset().mockImplementationOnce(async options => {
+      await options.onSession({ detachControllerForRestart: detach });
+      await detachNativeSessionsForRestart([restarting.binding.runId]);
+      await options.onSession(null);
+      return { result: {},
+        terminal: { runTerminalState: "succeeded" }, usage: null };
+    });
+    await expect(executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting,
+      runnerInstanceId: "runner", onUsage })).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError);
+    expect(detach).toHaveBeenCalledOnce();
+    expect(onUsage).not.toHaveBeenCalled();
+  });
+
  it.each([false, true])("settles failed startup without claiming detachment (deadline exceeded: %s)", async (exceedDeadline) => {
    const root = await mkdtemp(join(tmpdir(), "native-startup-failure-"));
    const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
@@ -5240,6 +5314,23 @@ describe("native startup restart detachment", () => {
 });

 describe("native terminal-turn accounting", () => {
+  it.each([false, true])("prices only complete direct Claude API turn accounting (%s)", async complete => {
+    const { rehydrateRunnerdUsageNotification } = await import("@paperclipai/paperclip-runner/live");
+    const { priceAnthropicReceipt } = await import("../anthropic-pricing.js");
+    const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {});
+    const counts = { inputTokens: 12, outputTokens: 4, cacheReadTokens: 30, cacheWriteTokens: 20, providerCostUsd: 0 };
+    state.execute.mockReset().mockImplementationOnce(async () => {
+      const notification = rehydrateRunnerdUsageNotification({ provider: "acpx", cumulative: { ...counts, providerCostUsd: 10 }, runDelta: counts, runDeltaAvailable: complete }, "session", "turn");
+      await accountingEvents.committed!({ eventType: "item.completed", turnId: "turn", payload: { kind: "usage", usage: notification.tokenUsage } } as unknown as PrpEvent);
+      await accountingEvents.committed!({ eventType: "turn.interrupted", turnId: "turn", payload: {} } as unknown as PrpEvent);
+      return { result: { summary: "Waiting for approval" }, terminal: { runTerminalState: "succeeded" }, turnId: "turn", normalizedSessionId: "session", providerSessionId: null, driverKind: "acpx_runtime", driverVersion: "1", nativeEventCount: 2, highestContiguousSourceSeq: 2, usage: null };
+    });
+    await executePaperclipNativeSession({ db: leaseDb(), execution: { ...execution, provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-5" } } as NativeExecutionInput, runnerInstanceId: "runner", runnerEnvironment: { ANTHROPIC_API_KEY: "fixture" }, onUsage });
+    const receipt = onUsage.mock.calls.at(-1)![0];
+    const priced = priceAnthropicReceipt(receipt);
+    if (complete) expect(priced).toMatchObject({ complete: true, costStatus: "estimated", costUsdExact: "0.000150000" });
+    else expect(priced).toMatchObject({ complete: false, costStatus: "unpriced", costUsd: null });
+  });
  it.each(["claude", "codex"].flatMap(agent => [false, true].flatMap(restart =>
    [false, true].flatMap(partialFirst => [false, true].map(zeroFirst => ({ agent, restart, partialFirst, zeroFirst }))),
  )))("accumulates ACPX turns for $agent (restart: $restart, partial: $partialFirst, zero first: $zeroFirst)", async ({ agent, restart, partialFirst, zeroFirst }) => {
@@ -242,6 +242,9 @@ type NativeSessionStartup = {
  resolve: (session: ActiveNativeSession | null) => void;
  stopRequested?: boolean;
  cancellationSettled?: Promise<void>;
+  governedWait?: boolean;
+  settled: Promise<void>;
+  settle: () => void;
 };
 const nativeSessionStartups = new Map<string, NativeSessionStartup>();

@@ -279,7 +282,20 @@ export async function detachNativeSessionsForRestart(
  const detachedRunIds: string[] = [];
  const inactiveRunIds: string[] = [];
  const unsupportedRunIds: string[] = [];
+  const settlementDeadline = Date.now() + 20_000;
  for (const runId of new Set(runIds)) {
+    // A governed stop already revoked new work and is collecting terminal
+    // accounting. Give that exact owner a bounded chance to persist it before
+    // fencing callbacks and relinquishing the runner. Timeout is not success.
+    const settling = nativeSessionStartups.get(runId);
+    if (settling?.governedWait) {
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      try {
+        await Promise.race([settling.settled, new Promise<void>(resolve => {
+          timer = setTimeout(resolve, Math.max(0, settlementDeadline - Date.now()));
+        })]);
+      } finally { clearTimeout(timer); }
+    }
    nativeRunsDetachingForRestart.add(runId);
    const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId);
    if (!active) {
@@ -6002,6 +6018,7 @@ function loadWarmNativeCheckpoint(
        activeTurnId: null,
        terminalTurns: [],
        pendingRuntimeRequests: [],
+        governedWait: undefined,
      };
  if (path !== scopedPath || envelope.configDigest !== configDigest) {
    // Upgrade the validated checkpoint atomically. When moving from a legacy
@@ -7381,7 +7398,10 @@ export async function executePaperclipNativeSession(input: {
  // Register before the first asynchronous operation on either backend path.
  // A duplicate execution must not replace the original startup handoff.
  let resolveStartup!: (session: ActiveNativeSession | null) => void;
+  let settle!: () => void;
  const startup: NativeSessionStartup = {
+    settled: new Promise<void>(resolve => { settle = resolve; }),
+    settle: () => settle(),
    promise: new Promise<ActiveNativeSession | null>(resolve => { resolveStartup = resolve; }),
    resolve: session => resolveStartup(session),
  };
@@ -7478,6 +7498,7 @@ export async function executePaperclipNativeSession(input: {
      executingNativeOwnerScopes.delete(ownerScope);
    }
    startup.resolve(null);
+    startup.settle();
    if (nativeSessionStartups.get(runId) === startup) {
      nativeSessionStartups.delete(runId);
    }
@@ -7993,6 +8014,11 @@ async function executePaperclipNativeSessionWithinScope(
    }
  };
  let completedConversationReply: PrpEvent | null = null;
+  const assertControllerActive = () => {
+    if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
+      throw new NativeControllerDetachedForRestartError();
+    }
+  };
  const controlPlane = new PaperclipControlPlanePort(
    input.db,
    {
@@ -8008,6 +8034,7 @@ async function executePaperclipNativeSessionWithinScope(
    },
    {
      privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY,
+      assertControllerActive,
      onCommittedEvent: async (event) => {
        await observeAccountingEvent(event);
        await toolTrace.observe(event);
@@ -8541,8 +8568,15 @@ async function executePaperclipNativeSessionWithinScope(
            controlPlane,
            runnerInstanceId: effectiveRunnerInstanceId,
            controlPlaneInstanceId,
-            resolveGovernedWait: ({ event }) =>
-              governedWaitObservation.consume(event),
+            resolveGovernedWait: ({ event }) => {
+              assertControllerActive();
+              const result = governedWaitObservation.consume(event);
+              if (result) {
+                const startup = nativeSessionStartups.get(input.execution.binding.runId);
+                if (startup) startup.governedWait = true;
+              }
+              return result;
+            },
            resolveMissingResult: async ({ terminalEvent }) => {
              // Governed waits take precedence over an ordinary chat reply.
              // Execution tasks still require their normal semantic finish.
@@ -8581,6 +8615,10 @@ async function executePaperclipNativeSessionWithinScope(
            requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined,
            onSessionClosed: input.instructionWorkingCopy?.collectStopped,
            onCheckpoint: async (snapshot) => {
+              if (snapshot.governedWait) {
+                const startup = nativeSessionStartups.get(input.execution.binding.runId);
+                if (startup) startup.governedWait = true;
+              }
              snapshot = identityRedactor.redact(snapshot);
              if (warmSessionId !== null && warmConfigDigest !== null) {
                await persistWarmNativeCheckpoint(
@@ -8774,6 +8812,9 @@ async function executePaperclipNativeSessionWithinScope(
      },
      { parentName: "task.run" },
    );
+    // A detached consumer can resolve successfully after its stream closes.
+    // Only the replacement controller may settle the run or certify accounting.
+    assertControllerActive();
    // Persist provider accounting before any workspace/issue finalization. A
    // detached controller or failed finalizer must not lose a completed turn.
    // session.usage() may be an attachment baseline, a partial report, or a
@@ -2125,10 +2125,17 @@ describe("rebindNativeSessionCheckpoint", () => {
  });

  it("retains provider identity but clears prior turn and event state", () => {
+    const prior = previousRun();
+    const profile = prior.runnerProfileJson as Record<string, unknown>;
+    const checkpoint = profile.sessionCheckpoint as Record<string, unknown>;
+    checkpoint.governedWait = { sourceEvent: { runId: previousRunId, turnId: "old-turn" }, result: { reportedWorkDisposition: "yielded" } };
+    const before = structuredClone(prior);
    const rebound = rebindNativeSessionCheckpoint({
-      previousRun: previousRun(),
+      previousRun: prior,
      currentExecution: execution(currentRunId),
    });
+    expect(rebound?.governedWait).toBeUndefined();
+    expect(prior).toEqual(before);
    expect(rebound).toMatchObject({
      sessionId: "provider-thread-123",
      providerSessionId: "provider-thread-123",
@@ -568,6 +568,9 @@ export function rebindNativeSessionCheckpoint(input: {
    activeTurnId: null,
    terminalTurns: [],
    pendingRuntimeRequests: [],
+    // Settlement evidence belongs to the prior heartbeat run, even when the
+    // provider session is reused for the user's response.
+    governedWait: undefined,
    providerRecoveryPolicy,
  };
 }
@@ -299,6 +299,67 @@ describe("PaperclipControlPlanePort conformance", () => {
    }
  });

+  it.each([
+    ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"],
+    ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"],
+  ] as const)("revokes an in-flight %s on %s", async (operation, reason) => {
+    const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
+    const runnerId = randomUUID();
+    await db.insert(heartbeatRuns).values({
+      id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
+      status: "running", runtimeMode: "native", nativeIssueId: identity.issueId,
+      nativeSessionId: identity.sessionId, runnerInstanceId: runnerId,
+      completionContractId: contractId, completionContractSha256: contractSha,
+      contextSnapshot: { issueId: identity.issueId },
+    });
+    const binding = { ...identity, completionContractId: contractId, completionContractSha256: contractSha,
+      sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `detached-${identity.runId}` };
+    let detached = false;
+    const controller = new AbortController();
+    const port = new PaperclipControlPlanePort(db, binding, {
+      assertControllerActive: () => { if (detached) throw new Error("controller_detached"); },
+    });
+    await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
+    const mutate = () => operation === "event" ? port.appendEvent({
+      schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceSeq: 1,
+      sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId,
+      normalizedSessionId: identity.sessionId, eventType: "turn.interrupted", schemaVersion: 1,
+      priority: 0, emittedAt: new Date().toISOString(), payload: { reason: "governed_wait" },
+    }, { signal: controller.signal }) : operation === "checkpoint"
+      ? port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity }, { signal: controller.signal })
+      : port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }, { signal: controller.signal });
+    let pending: Promise<unknown> | undefined;
+    await db.transaction(async tx => {
+      await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)).for("update");
+      const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>;
+      pending = mutate().then(() => null, error => error);
+      let waiting = false;
+      for (let attempt = 0; attempt < 100; attempt++) {
+        const [state] = await db.execute(sql`select exists (
+          select 1 from pg_stat_activity where ${backend.pid} = any(pg_blocking_pids(pid))
+        ) as waiting`) as unknown as Array<{ waiting: boolean }>;
+        if (state.waiting) { waiting = true; break; }
+        await new Promise(resolve => setTimeout(resolve, 10));
+      }
+      expect(waiting).toBe(true);
+      if (reason === "detach") detached = true;
+      else controller.abort(new Error("controller_detached"));
+    });
+    expect(await pending).toMatchObject({ message: "controller_detached" });
+    await expect(mutate()).rejects.toThrow("controller_detached");
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
+    expect(run.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
+    expect(run.nextEventSeq).toBe(1);
+    expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId))).toEqual([]);
+    expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]);
+    expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, identity.runId))).toEqual([]);
+    // A new controller's independently authorized port can still settle it.
+    const replacement = new PaperclipControlPlanePort(db, binding);
+    await replacement.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
+    await expect(replacement.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT,
+      terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })).resolves.toBeUndefined();
+  });
+
  it("redacts identity private material before persisting native events and checkpoints", async () => {
    const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
    const runnerId = randomUUID();
@@ -80,6 +80,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
  #redactedDeltaBytes = 0;
  readonly #binding: PaperclipControlPlaneBinding;
  #sessionId: string | null = null;
+  readonly #assertControllerActive?: () => void;
  readonly #onCommittedEvent?: (event: PrpEvent) => Promise<void>;
  readonly #onDuplicateEvent?: (event: PrpEvent) => Promise<void>;

@@ -89,17 +90,25 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
    options: {
      /** Runtime-only secret; never part of the persisted binding. */
      privateKeyPem?: string;
+      /** Synchronous revocation fence for a controller handing off on restart. */
+      assertControllerActive?: () => void;
      onCommittedEvent?: (event: PrpEvent) => Promise<void>;
      onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
    } = {},
  ) {
    this.#db = db;
+    this.#assertControllerActive = options.assertControllerActive;
    this.#identityRedactor = createAgentIdentityRedactor(options.privateKeyPem);
    this.#binding = structuredClone(binding);
    this.#onCommittedEvent = options.onCommittedEvent;
    this.#onDuplicateEvent = options.onDuplicateEvent;
  }

+  #assertActive(options?: { signal: AbortSignal }): void {
+    this.#assertControllerActive?.();
+    options?.signal.throwIfAborted();
+  }
+
  #matchesPersistedBinding(run: typeof heartbeatRuns.$inferSelect): boolean {
    return run.companyId === this.#binding.companyId
      && run.agentId === this.#binding.agentId
@@ -112,6 +121,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
  }

  async openRun(input: OpenControlPlaneRunInput): Promise<void> {
+    this.#assertControllerActive?.();
    const identity = input.identity;
    if (
      identity.companyId !== this.#binding.companyId
@@ -134,6 +144,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
    if (!run || !this.#matchesPersistedBinding(run)) {
      throw new Error("native_open_run_not_authorized");
    }
+    this.#assertControllerActive?.();
    this.#sessionId = identity.sessionId;
  }

@@ -160,7 +171,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
    return structuredClone(snapshot as PersistedNativeSession);
  }

-  async checkpointSession(snapshot: PersistedNativeSession): Promise<void> {
+  async checkpointSession(snapshot: PersistedNativeSession, options?: { signal: AbortSignal }): Promise<void> {
+    this.#assertActive(options);
    snapshot = this.#identityRedactor.redact(snapshot);
    const identity = snapshot.identity;
    if (
@@ -174,6 +186,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
      const run = await tx.select().from(heartbeatRuns)
        .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1)
        .then((rows) => rows[0] ?? null);
+      this.#assertActive(options);
      if (!run || !this.#matchesPersistedBinding(run)) {
        throw new Error("native_session_checkpoint_binding_mismatch");
      }
@@ -187,10 +200,12 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
        nativePhaseUpdatedAt: new Date(),
        updatedAt: new Date(),
      }).where(eq(heartbeatRuns.id, this.#binding.runId));
+      this.#assertActive(options);
    });
  }

-  async appendEvent(value: NativeRunEvent | PrpEvent) {
+  async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) {
+    this.#assertActive(options);
    if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported");
    const validated = validatePrpEvent(value);
    if (!validated.ok) throw new Error(`native_event_schema_invalid:${validated.issues[0]?.message ?? "unknown"}`);
@@ -235,22 +250,30 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
        }
      }
    }
-    const persisted = await appendHeartbeatRunEvent(this.#db, {
-      companyId: this.#binding.companyId,
-      runId: this.#binding.runId,
-      agentId: this.#binding.agentId,
-      eventType: event.eventType,
-      stream: "system",
-      level: event.eventType.includes("failed") ? "error" : "info",
-      payload: { prpEvent: event as unknown as Record<string, unknown> },
-      nativeSource: {
-        sourceInstanceId: event.sourceInstanceId,
-        sourceEventId: event.sourceEventId,
-        sourceSeq: event.sourceSeq,
-        protocolSchemaVersion: event.schemaVersion,
-        canonicalPayload: event as unknown as Record<string, unknown>,
-      },
+    const persisted = await this.#db.transaction(async tx => {
+      this.#assertActive(options);
+      const receipt = await appendHeartbeatRunEvent(tx as unknown as Db, {
+        companyId: this.#binding.companyId,
+        runId: this.#binding.runId,
+        agentId: this.#binding.agentId,
+        eventType: event.eventType,
+        stream: "system",
+        level: event.eventType.includes("failed") ? "error" : "info",
+        payload: { prpEvent: event as unknown as Record<string, unknown> },
+        nativeSource: {
+          sourceInstanceId: event.sourceInstanceId,
+          sourceEventId: event.sourceEventId,
+          sourceSeq: event.sourceSeq,
+          protocolSchemaVersion: event.schemaVersion,
+          canonicalPayload: event as unknown as Record<string, unknown>,
+        },
+      });
+      // The nested append has no publication side effects. Revocation rolls
+      // back its event and sequence allocation before the outer commit.
+      this.#assertActive(options);
+      return receipt;
    });
+    this.#assertActive(options);
    if (persisted.disposition === "committed") {
      publishChatPublicationCommitSignal({
        companyId: this.#binding.companyId,
@@ -305,7 +328,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
    return { events, highestContiguousSourceSeq: cursor };
  }

-  async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput): Promise<void> {
+  async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise<void> {
+    this.#assertActive(options);
    value = this.#identityRedactor.redact(value);
    if (!isCompleteInput(value)) throw new Error("native_structured_result_required");
    // Capture compatibility diagnostics before canonical validation removes
@@ -341,6 +365,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
      const run = await tx.select().from(heartbeatRuns)
        .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1)
        .then((rows) => rows[0] ?? null);
+      this.#assertActive(options);
      if (!run || !this.#matchesPersistedBinding(run)) {
        throw new Error("native_result_binding_mismatch");
      }
@@ -352,6 +377,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
        .where(and(eq(nativeRunResults.runId, this.#binding.runId), or(...callerConditions)))
        .limit(1).then((rows) => rows[0] ?? null);
      if (existing) {
+        this.#assertActive(options);
        if (existing.canonicalSha256 !== canonicalSha256) throw new Error("structured_result_replay_conflict");
        return;
      }
@@ -404,6 +430,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
        },
        updatedAt: new Date(),
      }).where(eq(heartbeatRuns.id, this.#binding.runId));
+      this.#assertActive(options);
    });
  }
 }
@@ -1,3 +1,4 @@
+import { priceAnthropicReceipt } from "./anthropic-pricing.js";
 import { priceCodexReceipt } from "./codex-pricing.js";
 import { randomUUID } from "node:crypto";
 import { promises as fs } from "node:fs";
@@ -237,8 +238,8 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string;
    const attempt = parsed.attemptId ?? currentAttempt;
    if (!attempts.has(attempt)) currentAttempt = attempt;
    const previous = attempts.get(attempt);
-    const priced = priceCodexReceipt({ ...parsed, usage: parsed.usage ?? previous?.usage,
-      complete: parsed.complete && !(previous?.usage && !parsed.usage) });
+    const priced = priceAnthropicReceipt(priceCodexReceipt({ ...parsed, usage: parsed.usage ?? previous?.usage,
+      complete: parsed.complete && !(previous?.usage && !parsed.usage) }));
    priced.pricingProvenance ??= {
      source: priced.costUsd != null || priced.costUsdExact != null || priced.cacheAdjustedCostUsd != null ? "provider_reported" : "unknown",
      version: "accounting-receipt/v1",