diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index cd56f4460ba..fc6434e7cb0 100644 --- a/doc/SPEC-implementation.md +++ b/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 diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 1d6aa5a0c0f..629b273fc56 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/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 = []; 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(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 () => { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 93e0d81545f..16491bc99de 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/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>(); let governedCancellationCommitted = false; let semanticResultObserved = false; + let governedProviderTerminalObserved = false; let deferredGovernedCleanupSettlement: Promise | null = null; let deferredSessionCancellationSettlement: Promise | null = null; const inputTimers = new Map>(); @@ -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 | 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, ); diff --git a/server/src/services/native-runtime/native-completion-feedback.test.ts b/server/src/services/native-runtime/native-completion-feedback.test.ts index 7d043c93208..a8490f5b954 100644 --- a/server/src/services/native-runtime/native-completion-feedback.test.ts +++ b/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}` }], diff --git a/server/src/services/native-runtime/native-completion-feedback.ts b/server/src/services/native-runtime/native-completion-feedback.ts index de374de1166..2261abf6d2e 100644 --- a/server/src/services/native-runtime/native-completion-feedback.ts +++ b/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 : [], diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 8b838e3fd30..723fa4570ca 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/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; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index d4dec4ee160..293f293b589 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/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 diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index f6651aedc88..129ffdcf23a 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/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 | 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(); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index e41fd522c8e..bfd1cdc56b2 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/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; readonly #onDuplicateEvent?: (event: PrpEvent) => Promise; @@ -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; onDuplicateEvent?: (event: PrpEvent) => Promise; } = {}, ) { 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 { + 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 { + async checkpointSession(snapshot: PersistedNativeSession, options?: { signal: AbortSignal }): Promise { + 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 }, - nativeSource: { - sourceInstanceId: event.sourceInstanceId, - sourceEventId: event.sourceEventId, - sourceSeq: event.sourceSeq, - protocolSchemaVersion: event.schemaVersion, - canonicalPayload: event as unknown as Record, - }, + 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 }, + nativeSource: { + sourceInstanceId: event.sourceInstanceId, + sourceEventId: event.sourceEventId, + sourceSeq: event.sourceSeq, + protocolSchemaVersion: event.schemaVersion, + canonicalPayload: event as unknown as Record, + }, + }); + // 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 { + async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise { + 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); }); } }