refactor(server): expose guarded governed-response recovery

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

doc/plans/2026-10-26-governed-review-response-sleep-audit.md

@@ -1,7 +1,7 @@
 # Governed-review response recovery: sleep migration audit
 
-Status: owner and producer audit only; no worker or periodic removal implemented.
-Inspected parent: `c4beddad1e` (slice 48, draft PR #17696).
+Status: raw owner contract and initial worker verified; producer wiring and periodic removal remain incomplete.
+Inspected parent: `fad35f2d7a` (slice 48, draft PR #17696).
 
 ## One reason for work
 
@@ -136,3 +136,27 @@ files. Establish a baseline before editing the owner. Cover:
 
 No deployment, database suspension, Cloud alarm coverage, or cost-savings claim
 follows from this migration. The full original stack remains in scope.
+
+## Owner-contract checkpoint
+
+The scheduled entry point holds the issue transaction, checks the coordinator
+claim and actual retry deadline under lock, restores the proof through its owner,
+and publishes through addComment in that same transaction. Operational errors
+propagate to the worker; existing foreground callers retain their logging contract.
+The scoped worker reconstructs existing coordinators and excludes materialized
+receipts, including deleted comments. No startup registration or periodic removal
+has happened yet.
+
+The unchanged owner baseline passed 156 cases. The first expanded run passed
+164 of 165; the new principal-revocation fixture incorrectly wrote a nonexistent
+principal status field, also caught by typechecking. Using the existing company
+membership suspension path repaired the fixture. The complete expanded suite now
+passes all 165 cases and server typecheck passes. Evidence:
+`/tmp/sleep49-baseline.log`, `/tmp/sleep49-contract-tests.log`,
+`/tmp/sleep49-contract-tests-v2.log`, `/tmp/sleep49-contract-typecheck-v3.log`,
+and `/tmp/sleep49-contract-v2.status`. The cases exercise concurrency, late claim/
+deadline races, real authorization-lock contention and rollback, revoked authority,
+startup reconstruction, deleted receipts, and actual quiet SQL after settlement.
+
+The producer audit, query-cost measurements, startup/periodic wiring, remaining
+lifecycle acceptance, full verification and remote gates are still outstanding.

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  nativeReviewResponse: "native-review-response",
   nativeResponsePresentation: "native-response-presentation",
   nativePendingEffects: "native-pending-effects",
   unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery",

server/src/services/native-review-response-notifications.ts

@@ -0,0 +1,8 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type NativeReviewResponseScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
+
+export function notifyNativeReviewResponse(tx: object, scope: NativeReviewResponseScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeReviewResponse,
+    JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
+}

server/src/services/native-runtime/external-chat-wait.integration.test.ts

@@ -47,7 +47,10 @@ import type {
 } from "../../vendor/paperclip-runner/index.js";
 
 import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
-import { finalizeNativeRun } from "./native-run-finalizer.js";
+import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js";
+import { nativeReviewResponseWork } from "./native-review-response-work.js";
+import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../task-admission.js";
 import { nativeCompletionFeedback } from "./native-completion-feedback.js";
 import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js";
 import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
@@ -2781,6 +2784,110 @@ describe("native external-chat response wait", () => {
     return run!.resultJson!;
   }
 
+  async function seedRecoverableReviewResponse() {
+    const fixture = await seedWaitTurn();
+    const gate = await seedPriorCompletionReview(fixture);
+    await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() })
+      .where(eq(heartbeatRuns.id, fixture.runId));
+    const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
+    expect(coordinator!.phase).toBe("committed");
+    const input = { companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId,
+      decisionId: coordinator!.decisionId!, resultId: coordinator!.resultId!, assessmentId: coordinator!.assessmentId!, onlyWhenDue: true };
+    return { ...fixture, gate, input };
+  }
+
+  it("recovers scheduled governed responses once under concurrent callers and preserves deleted receipts", async () => {
+    const fixture = await seedRecoverableReviewResponse();
+    const work = nativeReviewResponseWork(db);
+    const [entry] = await work.load({ companyId: fixture.companyId });
+    expect(entry?.id).toBe(fixture.runId);
+    expect(await Promise.all([recoverCommittedNativeReviewResponse(db, fixture.input),
+      recoverCommittedNativeReviewResponse(db, fixture.input)])).toEqual(expect.arrayContaining([true, false]));
+    const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId));
+    expect(comments).toHaveLength(1);
+    expect(await db.select().from(chatPublications).where(eq(chatPublications.commentId, comments[0].id))).toHaveLength(1);
+    expect(await work.load({ companyId: fixture.companyId })).toEqual([]);
+    await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comments[0].id));
+    expect(await work.load({ companyId: fixture.companyId })).toEqual([]);
+    expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false);
+    expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
+  });
+
+  it.each(["future", "claim"])("defers scheduled governed responses behind a new %s under the owner lock", async kind => {
+    const fixture = await seedRecoverableReviewResponse();
+    const work = nativeReviewResponseWork(db);
+    const [entry] = await work.load({ companyId: fixture.companyId });
+    const at = new Date(Date.now() + 60_000);
+    const transaction = db.transaction.bind(db);
+    const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => {
+      await transaction(async tx => {
+        await tx.update(nativeRunFinalizations).set(kind === "future" ? { nextAttemptAt: at }
+          : { leaseOwner: "another-finalizer", leaseExpiresAt: at }).where(eq(nativeRunFinalizations.runId, fixture.runId));
+      });
+      return transaction(callback, options);
+    }) as typeof db.transaction);
+    try {
+      expect(await work.check(entry)).toEqual(kind === "future" ? { ...entry, dueAt: at.getTime() } : null);
+    } finally { intercept.mockRestore(); }
+    expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null })
+      .where(eq(nativeRunFinalizations.runId, fixture.runId));
+    expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true);
+  });
+
+  it("propagates scheduled governed response contention and rolls back its proof restoration", async () => {
+    const fixture = await seedRecoverableReviewResponse();
+    await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` })
+      .where(eq(heartbeatRuns.id, fixture.runId));
+    let release!: () => void, observed!: () => void;
+    const hold = new Promise<void>(resolve => { release = resolve; });
+    const locked = new Promise<void>(resolve => { observed = resolve; });
+    const holder = db.transaction(async tx => {
+      await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update");
+      observed(); await hold;
+    });
+    await locked;
+    try {
+      await expect(recoverCommittedNativeReviewResponse(db, fixture.input)).rejects.toBeDefined();
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+      expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined();
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
+    } finally { release(); await holder; }
+    expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true);
+  });
+
+  it.each(["gate", "agent", "principal", "company"])("leaves scheduled governed response authority with its owner after %s changes", async kind => {
+    const fixture = await seedRecoverableReviewResponse();
+    if (kind === "gate") await db.update(issueThreadInteractions).set({ status: "accepted" }).where(eq(issueThreadInteractions.id, fixture.gate.id));
+    if (kind === "agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, fixture.agentId));
+    if (kind === "principal") await db.update(companyMemberships).set({ status: "suspended" })
+      .where(and(eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId)));
+    expect(await recoverCommittedNativeReviewResponse(db, { ...fixture.input,
+      ...(kind === "company" ? { companyId: randomUUID() } : {}) })).toBe(false);
+    expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0);
+  });
+
+  it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => {
+    const fixture = await seedRecoverableReviewResponse();
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const worker = nativeReviewResponseWork(db).register(coordinator, () => true);
+    try {
+      await worker.ready;
+      await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }),
+        { timeout: 10_000 }).toEqual({ active: 0, next: null });
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1);
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      const previousDebug = db.$client.options.debug;
+      const queries: string[] = [];
+      db.$client.options.debug = (_connection, query) => { queries.push(query); };
+      try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); }
+      finally { db.$client.options.debug = previousDebug; }
+      expect(errors).toEqual([]);
+    } finally { await coordinator.stop(); vi.useRealTimers(); }
+  });
+
   it.each(["telegram", "discord"] as const)(
     "presents the exact %s response without resolving a pre-existing native completion review",
     async (provider) => {

server/src/services/native-runtime/native-chat-review-presentation.ts

@@ -500,6 +500,7 @@ export async function restoreNativeChatReviewPresentationInTransaction(
     decisionId: string;
     resultId: string;
     assessmentId: string;
+    onlyWhenDue?: boolean;
   },
 ): Promise<boolean> {
   const [coordinator] = await db
@@ -519,6 +520,8 @@ export async function restoreNativeChatReviewPresentationInTransaction(
     .for("update")
     .limit(1);
   if (!coordinator) return false;
+  if (input.onlyWhenDue && (coordinator.leaseOwner !== null
+    || coordinator.nextAttemptAt !== null && coordinator.nextAttemptAt.getTime() > Date.now())) return false;
   const [decision] = await db
     .select()
     .from(statusDecisions)

server/src/services/native-runtime/native-review-response-work.ts

@@ -0,0 +1,83 @@
+import { and, asc, eq, gt, isNotNull, isNull, notExists, sql } from "drizzle-orm";
+import { heartbeatRuns, issueComments, nativeRunFinalizations, statusDecisions, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { NativeReviewResponseScope } from "../native-review-response-notifications.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import { recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js";
+
+type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number };
+
+export function nativeReviewResponseWork(db: Db) {
+  function candidates(scope?: NativeReviewResponseScope, after?: string, runId?: string) {
+    return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId,
+      issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId,
+      decisionId: nativeRunFinalizations.decisionId, resultId: nativeRunFinalizations.resultId,
+      assessmentId: nativeRunFinalizations.assessmentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
+    }).from(nativeRunFinalizations)
+      .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
+        eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
+      .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId),
+        eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId),
+        eq(statusDecisions.runId, nativeRunFinalizations.runId), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId)))
+      .where(and(eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner),
+        isNotNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.assessmentId),
+        eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"),
+        eq(statusDecisions.reasonCode, "governed_response_waiting"), eq(statusDecisions.applicationState, "applied"),
+        sql`${statusDecisions.decisionJson}->'externalChatReviewPresentation'->>'schema' = 'paperclip.chat_review_response_presentation.v1'`,
+        notExists(db.select({ id: issueComments.id }).from(issueComments).where(and(
+          eq(issueComments.companyId, nativeRunFinalizations.companyId), eq(issueComments.issueId, nativeRunFinalizations.issueId),
+          eq(issueComments.createdByRunId, nativeRunFinalizations.runId), eq(issueComments.authorAgentId, heartbeatRuns.agentId),
+          eq(issueComments.authorType, "agent"), sql`${issueComments.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`,
+          sql`${heartbeatRuns.resultJson}->'externalChatReviewPresentation'->>'decisionId' = ${nativeRunFinalizations.decisionId}::text`,
+        ))),
+        scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined,
+        scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
+        scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined,
+        scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined,
+        runId ? eq(nativeRunFinalizations.runId, runId) : undefined,
+        after ? gt(nativeRunFinalizations.runId, after) : undefined,
+      )).orderBy(asc(nativeRunFinalizations.runId)).limit(100);
+  }
+  async function load(scope?: NativeReviewResponseScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
+    const entries: Entry[] = [];
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await candidates(scope, after);
+      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId,
+        issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      if (rows.length < 100) return entries;
+      after = rows[rows.length - 1].id;
+    }
+    throw new Error("Governed review response discovery stopped");
+  }
+  async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
+    if (signal?.aborted) throw new Error("Governed review response recovery stopped");
+    const scope = { companyId: entry.companyId, issueId: entry.issueId };
+    const [current] = await candidates(scope, undefined, entry.id);
+    if (!current) return null;
+    if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() };
+    if (signal?.aborted) throw new Error("Governed review response recovery stopped");
+    const recoveryStartedAt = Date.now();
+    await recoverCommittedNativeReviewResponse(db, { ...scope, runId: entry.id,
+      decisionId: current.decisionId!, resultId: current.resultId!, assessmentId: current.assessmentId!, onlyWhenDue: true });
+    const [remaining] = await candidates(scope, undefined, entry.id);
+    return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt
+      ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null;
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.nativeReviewResponse,
+      decodeScope: (hint: string): NativeReviewResponseScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: NativeReviewResponseScope) => !scope || entry.companyId === scope.companyId
+        && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId)
+        && (!scope.agentId || entry.agentId === scope.agentId),
+      load: (scope, signal) => load(scope, signal, canRun),
+      check: (entry, signal) => {
+        if (!canRun()) throw new Error("Governed review response recovery paused");
+        return check(entry, signal);
+      },
+    }, canRun);
+  }
+  return { load, check, register };
+}

server/src/services/native-runtime/native-run-finalizer.ts

@@ -721,64 +721,61 @@ async function projectCommittedRun(input: {
   }
 }
 
-async function materializeCommittedReviewResponse(db: Db, runId: string) {
-  try {
-    const [run] = await db
-      .select()
-      .from(heartbeatRuns)
-      .where(eq(heartbeatRuns.id, runId))
-      .limit(1);
-    if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
-      return;
-    const decisionId = record(
-      run.resultJson.externalChatReviewPresentation,
-    ).decisionId;
-    if (
-      typeof decisionId !== "string" ||
-      (await hasMaterializedNativeReviewResponse(db, {
-        companyId: run.companyId,
-        issueId: run.nativeIssueId,
-        runId,
-        decisionId,
-      }))
-    )
-      return;
-    const summary = record(run.resultJson.nativeResult).summary;
-    if (
-      typeof summary !== "string" ||
-      !(await authorizeNativeChatReviewPresentation(db, {
-        companyId: run.companyId,
-        issueId: run.nativeIssueId,
-        runId,
-        resultJson: run.resultJson,
-      }))
-    )
-      return;
-    // The coordinator transaction is already committed. addComment owns a
-    // fresh issue transaction, rechecks the proof and carries selected files;
-    // its same-run/text and publication keys make recovery replay idempotent.
-    await issueService(db).addComment(
-      run.nativeIssueId,
-      summary,
-      { agentId: run.agentId, runId },
-      {
-        authorizationReason: "allow_chat_run_presentation",
-      },
-    );
-    publishChatPublicationCommitSignal({
+async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) {
+  const [run] = await dbOrTx
+    .select()
+    .from(heartbeatRuns)
+    .where(eq(heartbeatRuns.id, runId))
+    .limit(1);
+  if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
+    return null;
+  const decisionId = record(
+    run.resultJson.externalChatReviewPresentation,
+  ).decisionId;
+  if (
+    typeof decisionId !== "string" ||
+    (await hasMaterializedNativeReviewResponse(dbOrTx, {
       companyId: run.companyId,
       issueId: run.nativeIssueId,
-      runId: run.id,
-      agentId: run.agentId,
-      eventType: "run.presentation.resolved",
-    });
+      runId,
+      decisionId,
+    }))
+  )
+    return null;
+  const summary = record(run.resultJson.nativeResult).summary;
+  if (
+    typeof summary !== "string" ||
+    !(await authorizeNativeChatReviewPresentation(dbOrTx, {
+      companyId: run.companyId,
+      issueId: run.nativeIssueId,
+      runId,
+      resultJson: run.resultJson,
+    }))
+  )
+    return null;
+  await issueService(db).addComment(
+    run.nativeIssueId,
+    summary,
+    { agentId: run.agentId, runId },
+    {
+      authorizationReason: "allow_chat_run_presentation",
+    },
+    dbOrTx,
+  );
+  return run;
+}
+
+function publishNativeReviewResponse(run: typeof heartbeatRuns.$inferSelect) {
+  publishChatPublicationCommitSignal({ companyId: run.companyId, issueId: run.nativeIssueId!,
+    runId: run.id, agentId: run.agentId, eventType: "run.presentation.resolved" });
+}
+
+async function materializeCommittedReviewResponse(db: Db, runId: string) {
+  try {
+    const run = await materializeNativeReviewResponse(db, runId);
+    if (run) publishNativeReviewResponse(run);
   } catch (error) {
-    // The durable committed proof remains retryable by the next reconciler
-    // sweep. Presentation failure must not reclassify a successful native run.
-    logger.warn(
-      { runId, error },
-      "Committed chat review response is awaiting presentation retry",
-    );
+    logger.warn({ runId, error }, "Committed chat review response is awaiting presentation retry");
   }
 }
 
@@ -1153,6 +1150,22 @@ export async function repairCommittedNativeReviewResponse(
   );
   if (restored) await materializeCommittedReviewResponse(db, input.runId);
 }
+export async function recoverCommittedNativeReviewResponse(db: Db, input: {
+  companyId: string; issueId: string; runId: string; decisionId: string;
+  resultId: string; assessmentId: string; onlyWhenDue?: boolean;
+}): Promise<boolean> {
+  const materialized = await db.transaction(async transaction => {
+    const tx = transaction as unknown as Db;
+    const [issue] = await tx.select({ id: issues.id }).from(issues)
+      .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
+    if (!issue || await hasMaterializedNativeReviewResponse(tx, input)) return null;
+    if (!await restoreNativeChatReviewPresentationInTransaction(tx, input)) return null;
+    return materializeNativeReviewResponse(db, input.runId, tx);
+  });
+  if (materialized) publishNativeReviewResponse(materialized);
+  return materialized !== null;
+}
+
 export async function finalizeNativeRun(input: {
   db: Db;
   runId: string;