fix(native): fence continuation settlement during controller restart

Reject stale approval reports bound through unsatisfied criteria. Revoke
controller checkpoint and completion writes after restart detachment and
roll back in-flight port mutations after cancellation. Preserve bounded
same-turn accounting settlement and terminal subscription teardown.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/SPEC-implementation.md

@@ -1531,6 +1531,9 @@ 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

packages/paperclip-runner/src/native-session-runtime.test.ts

@@ -4637,7 +4637,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 +4653,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 +4680,7 @@ describe("executeNativeSession recovery", () => {
           lineage: [],
         };
       },
+      cancel: () => ({ cleanup: Promise.resolve() }),
       close,
     };
     const backend: NativeSessionBackend = {
@@ -4720,10 +4721,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 +7643,9 @@ describe("executeNativeSession recovery", () => {
     ]);
   });
 
-  it("parks immediately but retains shutdown usage before finalizing a governed wait", async () => {
+  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",
@@ -7711,6 +7715,7 @@ 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;
@@ -7780,7 +7785,7 @@ describe("executeNativeSession recovery", () => {
       completeRun,
     };
 
-    const completed = await executeNativeSession({
+    const execution = executeNativeSession({
       input,
       backend,
       controlPlane: port,
@@ -7790,6 +7795,8 @@ 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({
@@ -7807,6 +7814,7 @@ describe("executeNativeSession recovery", () => {
       "run.result.accepted",
       "run.terminal",
     ]);
+     } finally { vi.useRealTimers(); }
   });
 
   it("hands a committed structured input to the durable wait after its live window", async () => {

packages/paperclip-runner/src/native-session-runtime.ts

@@ -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;
 
@@ -876,6 +881,7 @@ async function consumeTurn(
   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>>();
@@ -947,7 +953,7 @@ async function consumeTurn(
       const drainAbort = new AbortController();
       const stopDrain = () => drainAbort.abort(appendAbort.signal.reason);
       appendAbort.signal.addEventListener("abort", stopDrain, { once: true });
-      const drainMs = Math.min(5_000, timeoutMs > 0 ? Math.max(1, Math.floor(timeoutMs / 4)) : 5_000);
+      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) {
@@ -961,7 +967,10 @@ async function consumeTurn(
             if (drainAbort.signal.aborted) return;
             eventCount += receipt.disposition === "committed" ? 1 : 0;
             highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq);
-            if (terminal) return;
+            if (terminal) {
+              governedProviderTerminalObserved = true;
+              return;
+            }
           }
         }
       })();
@@ -1346,15 +1355,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,
       );

server/src/services/native-runtime/native-completion-feedback.test.ts

@@ -55,11 +55,11 @@ describe("native final-response feedback", () => {
     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();
+    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: randomUUID(), 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" } },
+      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.
@@ -71,6 +71,16 @@ describe("native final-response feedback", () => {
       ...(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");
+    // 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}` }],

server/src/services/native-runtime/native-completion-feedback.ts

@@ -176,6 +176,9 @@ export async function nativeCompletionFeedback(
       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 === "not_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 : [],

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

@@ -5185,6 +5185,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;

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

@@ -7986,6 +7986,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,
     {
@@ -8001,6 +8006,7 @@ async function executePaperclipNativeSessionWithinScope(
     },
     {
       privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY,
+      assertControllerActive,
       onCommittedEvent: async (event) => {
         await observeAccountingEvent(event);
         await toolTrace.observe(event);
@@ -8530,8 +8536,10 @@ async function executePaperclipNativeSessionWithinScope(
             controlPlane,
             runnerInstanceId: effectiveRunnerInstanceId,
             controlPlaneInstanceId,
-            resolveGovernedWait: ({ event }) =>
-              governedWaitObservation.consume(event),
+            resolveGovernedWait: ({ event }) => {
+              assertControllerActive();
+              return governedWaitObservation.consume(event);
+            },
             resolveMissingResult: async ({ terminalEvent }) => {
               // Governed waits take precedence over an ordinary chat reply.
               // Execution tasks still require their normal semantic finish.
@@ -8762,6 +8770,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

server/src/services/native-runtime/paperclip-control-plane-port.test.ts

@@ -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();

server/src/services/native-runtime/paperclip-control-plane-port.ts

@@ -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);
     });
   }
 }