fix(server): wake obsolete policy review cleanup from commits

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

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -2166,3 +2166,9 @@ is still required. The full remaining stack and older unresolved gates remain
 in scope.
 
 Slice 43 final local acceptance: 120 owner/startup cases plus 209 affected consumer cases pass, without skips. Workspace typecheck/build, final server types, migration application/replay and index-plan checks passed. The negative fixture FK rejection and its correction remain in the owner audit. Remote CI and review are still required.
+
+## Slice 43 accepted; slice 44 obsolete policy reviews
+
+Cancelled workspace preservation is draft [#17677](https://github.com/paperclipai/paperclip/pull/17677) at `33567f91551d60053722c013f0d9b402c1db29e6`. All 53 checks passed, Greptile reviewed that head at 5/5, and there are no unresolved review threads. It remains unmerged.
+
+The native finalization audit found separate historical unsafe-export recovery, obsolete policy-review cleanup, automatic completion-review cleanup, effect delivery, committed response repair, evidence reassessment and workspace/result finalization. Slice 44 owns obsolete policy-review cleanup only. See [its audit](2026-10-25-obsolete-policy-review-sleep-audit.md). The owner/startup selection passes 116 tests after repairing a reproduced notification serialization bug. Consumer and final workspace verification remain pending. Older stack gates and final master reconciliation remain outstanding.

doc/plans/2026-10-25-obsolete-policy-review-sleep-audit.md

@@ -0,0 +1,113 @@
+# Obsolete native policy-review cleanup
+
+Date: 2026-10-25. Parent: cancelled workspace preservation #17677 at
+`33567f91551d60053722c013f0d9b402c1db29e6`.
+
+## Owner and scope
+
+The native finalization reconciler combines independent duties. Its initial
+calls recover historical unsafe exports, dismiss obsolete policy reviews, and
+retire automatic completion reviews. Its main loop also dispatches status
+effects, repairs committed response presentation, reassesses changed evidence,
+and resumes workspace/result finalization. Each duty needs its own migration;
+removing the combined scan would lose work.
+
+This slice owns obsolete version-change policy reviews only. The existing
+`obsolete-policy-reviews.ts` service identifies their full historical signature:
+pending system confirmation, exact prompt, applied native review decision,
+exact two-effect payload, and matching company/issue/run/effect/assessment.
+Real completion reviews, authored requests and answered cards are excluded.
+The current status arbiter does not generate this old prompt. The typed decision
+committer still supports the old effect shape, so a startup-only scan is not
+used as the complete writer contract.
+
+## Producers and durable truth
+
+The existing interaction, status decision, effect, and assessment rows are truth.
+The only production inserter of status decisions and decision effects is
+`commitNativeStatusDecision`. Its existing transaction registers an issue-scoped
+notification before creating the decision/effects. The signal settles at outer
+commit, including a caller-owned transaction. It adds no new transaction around
+provider work and no table. A normal current decision can cause one scoped
+lookup without leaving a recurring deadline.
+
+Other interaction writers cannot independently create the matching applied
+native decision/effect lineage. Existing decision mutation paths update lineage
+or delivery state, not the historical signature. Current-policy producers do
+not create new obsolete reviews, but their common commit boundary remains covered.
+Startup reconstructs old records. Independent SQL writers and concurrent older
+controllers are not covered by process-local signals; they require an explicit
+integration or restart reconstruction.
+
+## Scheduling and recovery
+
+The service registers with the existing scoped deadline helper. Startup and
+committed issue hints reconstruct exact candidates in 100-row keyset pages.
+The worker stores interaction/company/issue IDs and due times, not the decision
+payload. Batches of twenty use the shared fairness mechanism. Each check rereads
+the exact candidate and delegates to the existing transaction owner.
+
+The owner retains coordinator/issue/interaction locking, current status-version
+and decision-pointer guards, other-review checks, chat terminal publication,
+and activity persistence. The conditional withdrawal additionally matches the
+selected payload, kind, source run, idempotency key, and system author fields.
+A card changed after discovery is not withdrawn from a stale snapshot.
+
+After an attempt the worker rereads the exact pending predicate. Pending cards
+retain a 30-second retry; successful, answered, changed, or deleted cards disarm.
+An owner failure must not starve another candidate. Discovery failures use the
+shared backoff and retain the dirty scope. No expiry or retry budget is invented
+for historical records. No unresolved card is silently discarded for quiet.
+
+This preserves manual/scoped reconciliation and startup repair. Only periodic
+orphan reaping passes the new skip flag, so that path stops repeating this
+cleanup. The worker is registered when heartbeat maintenance is enabled and
+shares the existing admission/drain/standby gates. Shutdown joins admitted
+cleanup and removes the registration. Paused dispatch retains dirty requests.
+
+## Cost and boundaries
+
+The change is a worker migration, not proof of provider suspension. The combined
+reaper and unrelated finalization duties still run until migrated separately.
+Cloud alarms and sleep eligibility remain unchanged.
+
+Startup cost scales with historical pending candidates and their joins. Scoping
+ordinary commit hints to company/issue uses the existing interaction indexes.
+There is no new schema or unconditional hasPending query. Query-plan evidence
+and final test results are recorded below when available.
+
+## Verification record
+
+Before production changes, all 44 existing status-authority corpus tests passed
+on a fresh disposable PostgreSQL database. The host's embedded harness remains
+unavailable; the corpus now accepts the repository's validated external-test DB
+URL convention and closes its pool. No production connection behavior changed.
+
+The initial server typecheck and the typecheck including new worker tests both
+passed. Full owner/startup, affected consumers, final workspace checks and exact
+PR head CI/review remain required. No local or remote completion is claimed here.
+
+The first combined run had 67 passing tests and 49 failures. The producer helper
+serialized the structurally assignable commit input rather than projecting its
+company/issue fields. That included the database object and threw a circular
+JSON error on real commits. The helper now serializes only its two domain IDs.
+The failed run is retained at `/tmp/sleep44-owners-v1.log`. A fresh full selection
+is required after this production repair.
+
+After the payload repair, all 116 tests in the owner/startup/timing selection pass
+without skips (56 corpus tests, 58 startup tests and two timing tests). The corpus
+includes the existing 53 native/legacy fixtures inside its matrix test.
+
+A rolled-back query fixture added 10,000 completed interaction rows on the same
+issue and restored one matching pending historical review. The company/issue
+lookup used existing decision, interaction-idempotency, effect-ordinal and
+assessment indexes. It returned one candidate in 0.058 ms locally. This supports
+the scoped-query access path, not a production latency claim. Startup remains
+a global reconstruction whose cost scales with history; it does not recur when
+settled. SQL and plan are in `/tmp/sleep44-query-plan.sql` and `.log`.
+
+Final affected consumer acceptance passed all 235 cases across heartbeat recovery,
+native finalizer telemetry, external-chat response waits and governed chat
+milestones. Combined with the 116 owner/startup cases, 351 tests passed without
+skips. Workspace typecheck and build also passed. Remote CI and review remain
+required for the published head. No workflow, lockfile or schema change is made.

server/src/__tests__/native-status-arbiter-corpus.test.ts

@@ -1,10 +1,12 @@
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
 import { dismissAutomaticCompletionReviews } from "../services/native-runtime/automatic-completion-reviews.js";
 import * as childCompletionDelivery from "../services/native-runtime/native-child-completion-delivery.js";
 import { nativeCompletionFeedback } from "../services/native-runtime/native-completion-feedback.js";
 import { randomUUID } from "node:crypto";
 import { readFileSync } from "node:fs";
 import { fileURLToPath } from "node:url";
-import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { and, eq } from "drizzle-orm";
 import {
   agents,
@@ -67,10 +69,11 @@ import {
   resolveNativeReconciliationStatus,
 } from "../services/native-runtime/native-finalization-reconciler.js";
 import * as activityLog from "../services/activity-log.js";
-import { dismissObsoleteNativePolicyReviews } from "../services/native-runtime/obsolete-policy-reviews.js";
+import { dismissObsoleteNativePolicyReviews, registerObsoletePolicyReviewWork } from "../services/native-runtime/obsolete-policy-reviews.js";
 import { issueService } from "../services/issues.js";
 import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 
 type Fixture = {
   id: string;
@@ -381,8 +384,9 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
   const outsideAgentId = randomUUID();
 
   beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-");
-    db = createDb(temporary.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_CORPUS_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-");
+    db = createDb(external ?? temporary!.connectionString);
     await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: "PSC" });
     await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: "OUT" });
     await db.insert(agents).values([
@@ -411,7 +415,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
     ]);
   }, 30_000);
 
-  afterAll(async () => temporary?.cleanup());
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
 
   async function seedFixture(fixture: Fixture) {
     const issueId = randomUUID();
@@ -2007,7 +2011,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
       .rejects.toThrow("native_pending_effect_target_missing:enqueue_continuation");
   }, 30_000);
 
-  async function seedPolicyReview(options: { genuine?: boolean; priorStatus?: "in_progress" | "blocked" | "in_review" } = {}) {
+  async function seedPolicyReview(options: { genuine?: boolean; priorStatus?: "in_progress" | "blocked" | "in_review"; commit?: typeof commitNativeStatusDecision } = {}) {
     const template = corpus.fixtures.find((candidate) => candidate.mode === "native")!;
     const priorStatus = options.priorStatus ?? "in_progress";
     const seeded = await seedFixture({
@@ -2022,7 +2026,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
     });
     await db.update(workAssessments).set({ supersedesAssessmentId: previousAssessmentId })
       .where(eq(workAssessments.id, seeded.assessmentId));
-    const committed = await commitNativeStatusDecision({
+    const committed = await (options.commit ?? commitNativeStatusDecision)({
       db, companyId, issueId: seeded.issueId, runId: seeded.runId,
       assessmentId: seeded.assessmentId, priorStatus, priorStatusVersion: 0, priorDecisionId: null,
       decision: {
@@ -2045,6 +2049,165 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
     return { ...seeded, decision: decision!, interaction: interaction! };
   }
 
+
+  describe("obsolete policy review scheduling", () => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let admitted = true;
+    const errors: unknown[] = [];
+    beforeEach(() => {
+      admitted = true; errors.length = 0;
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    });
+    afterEach(async () => {
+      await coordinator?.stop(); coordinator = undefined;
+      vi.useRealTimers(); vi.restoreAllMocks();
+      await dismissObsoleteNativePolicyReviews(db);
+      expect(errors).toEqual([]);
+    });
+    async function start() {
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
+        onError: error => errors.push(error) });
+      await registerObsoletePolicyReviewWork(db, coordinator, () => admitted).ready;
+    }
+    async function flush() {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(idleWorkSnapshot().active).toBe(0);
+        expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 15_000 });
+    }
+    async function quiet() {
+      const read = vi.spyOn(db, "select"), write = vi.spyOn(db, "transaction");
+      try { await vi.advanceTimersByTimeAsync(120_000); expect(read).not.toHaveBeenCalled(); expect(write).not.toHaveBeenCalled(); }
+      finally { read.mockRestore(); write.mockRestore(); }
+    }
+    const card = async (id: string) => (await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, id)))[0];
+    function gate() {
+      let release!: () => void;
+      const promise = new Promise<void>(resolve => { release = resolve; });
+      return { promise, release };
+    }
+
+    it("leaves genuine pending reviews and completed history quiet", async () => {
+      const genuine = await seedPolicyReview({ genuine: true });
+      const settled = await seedPolicyReview();
+      await dismissObsoleteNativePolicyReviews(db, [settled.runId]);
+      await start(); await flush();
+      expect((await card(genuine.interaction.id)).status).toBe("pending");
+      expect((await card(settled.interaction.id)).status).toBe("cancelled");
+      await quiet();
+    });
+
+    it("wakes from the real decision commit after becoming idle and preserves decision history", async () => {
+      await start(); await flush(); await quiet();
+      const created = await seedPolicyReview();
+      await flush();
+      expect((await card(created.interaction.id)).status).toBe("cancelled");
+      expect((await issueService(db).getById(created.issueId))!.status).toBe("in_progress");
+      expect((await db.select().from(statusDecisions).where(eq(statusDecisions.id, created.decision.id)))[0]).toEqual(created.decision);
+      await quiet();
+    });
+
+    it.each(["commit", "rollback"] as const)("waits for the outer transaction to %s", async outcome => {
+      await start(); await flush();
+      const entered = gate(), release = gate();
+      let issueId = "";
+      const creating = seedPolicyReview({ commit: async input => db.transaction(async tx => {
+        issueId = input.issueId;
+        const result = await commitNativeStatusDecision({ ...input, db: tx as unknown as typeof db });
+        entered.release(); await release.promise;
+        if (outcome === "rollback") throw new Error("rollback obsolete review");
+        return result;
+      }) });
+      const settled = creating.then(value => ({ value, error: null }), error => ({ value: null, error }));
+      await entered.promise;
+      await vi.advanceTimersByTimeAsync(2000);
+      expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, issueId))).toEqual([]);
+      release.release(); const result = await settled;
+      await flush();
+      if (outcome === "rollback") {
+        expect(result.error).toMatchObject({ message: "rollback obsolete review" });
+        expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, issueId))).toEqual([]);
+      } else {
+        expect(result.error).toBeNull();
+        expect((await card(result.value!.interaction.id)).status).toBe("cancelled");
+      }
+      await quiet();
+    });
+
+    it.each(["restart", "drain", "standby"] as const)("recovers committed cards after %s", async mode => {
+      if (mode !== "restart") { await start(); await flush(); admitted = false; }
+      const created = await seedPolicyReview();
+      if (mode === "restart") await start();
+      else {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect((await card(created.interaction.id)).status).toBe("pending");
+        await quiet(); admitted = true;
+        await vi.advanceTimersByTimeAsync(30_000);
+      }
+      await flush(); expect((await card(created.interaction.id)).status).toBe("cancelled"); await quiet();
+    });
+
+    it("retains startup discovery while admission is closed", async () => {
+      const created = await seedPolicyReview(); admitted = false;
+      await start(); await quiet(); expect((await card(created.interaction.id)).status).toBe("pending");
+      admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush();
+      expect((await card(created.interaction.id)).status).toBe("cancelled"); await quiet();
+    });
+
+    it("retries only a failed cleanup while another candidate settles", async () => {
+      const first = await seedPolicyReview(), second = await seedPolicyReview();
+      const write = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("injected obsolete cleanup failure"));
+      await start();
+      await vi.waitFor(async () => {
+        const states = [(await card(first.interaction.id)).status, (await card(second.interaction.id)).status];
+        expect(states.sort()).toEqual(["cancelled", "pending"]);
+      });
+      write.mockRestore(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+      await vi.advanceTimersByTimeAsync(5000);
+      expect([(await card(first.interaction.id)).status, (await card(second.interaction.id)).status].sort()).toEqual(["cancelled", "pending"]);
+      admitted = false; await quiet(); admitted = true;
+      await vi.advanceTimersByTimeAsync(30_000); await flush();
+      expect((await card(first.interaction.id)).status).toBe("cancelled");
+      expect((await card(second.interaction.id)).status).toBe("cancelled"); await quiet();
+    });
+
+    it("does not withdraw a review edited after discovery", async () => {
+      const created = await seedPolicyReview();
+      const original = db.transaction.bind(db);
+      const write = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, config) => {
+        await db.update(issueThreadInteractions).set({ payload: { ...created.interaction.payload, prompt: "Approve this deployment." } })
+          .where(eq(issueThreadInteractions.id, created.interaction.id));
+        return original(callback, config);
+      });
+      await start(); await flush(); write.mockRestore();
+      expect((await card(created.interaction.id)).status).toBe("pending");
+      expect((await issueService(db).getById(created.issueId))!.status).toBe("in_review"); await quiet();
+    });
+
+    it("joins admitted cleanup on shutdown and reconstructs quiet state on restart", async () => {
+      const created = await seedPolicyReview(), entered = gate(), release = gate();
+      const original = db.transaction.bind(db);
+      const write = vi.spyOn(db, "transaction").mockImplementationOnce((callback, config) => original(async tx => {
+        entered.release(); await release.promise; return callback(tx);
+      }, config));
+      const starting = start(); await entered.promise;
+      let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; });
+      await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false);
+      release.release(); await starting; await stopping; write.mockRestore();
+      expect((await card(created.interaction.id)).status).toBe("cancelled");
+      await start(); await flush(); await quiet();
+    });
+
+    it("drains more than one reconstruction page without replaying settled reviews", async () => {
+      const created = [];
+      for (let index = 0; index < 103; index++) created.push(await seedPolicyReview());
+      await start(); await flush();
+      for (const value of created) expect((await card(value.interaction.id)).status).toBe("cancelled");
+      await quiet();
+    }, 60_000);
+  });
+
   async function seedAutomaticReview() {
     const seeded = await seedPolicyReview();
     const prompt = "Review the persisted native-run evidence and confirm whether this issue may be completed.";

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -108,6 +108,7 @@ const {
     reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
     registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
+    registerObsoletePolicyReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerCancelledWorkspacePreservationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerInstructionDirectoryCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
@@ -543,6 +544,13 @@ describe("startServer feedback export wiring", () => {
     process.env.BETTER_AUTH_SECRET = "test-secret";
   });
 
+  it.each([false, true])("registers obsolete policy review cleanup only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerObsoletePolicyReviewWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    if (enabled) expect(heartbeatServiceMock.registerObsoletePolicyReviewWork.mock.calls[0][1]()).toBe(true);
+  });
+
   it.each([false, true])("registers cancelled workspace preservation only when maintenance is enabled (%s)", async enabled => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
     await startServer(); await settleStartupHousekeeping();
@@ -928,7 +936,7 @@ describe("startServer feedback export wiring", () => {
       intervalCallback?.();
       await new Promise<void>((resolve) => setImmediate(resolve));
       expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
-      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true });
+      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true });
       expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
       expect(retiredDetector).not.toHaveBeenCalled();
     } finally {

server/src/index.ts

@@ -1777,6 +1777,8 @@ async function startServerWithDatabaseTeardown(
       () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork,
       () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
+    if (config.heartbeatSchedulerEnabled) await heartbeat.registerObsoletePolicyReviewWork(deliveryWork,
+      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerCancelledWorkspacePreservationWork(deliveryWork,
       () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork,
@@ -1909,7 +1911,7 @@ async function startServerWithDatabaseTeardown(
           // Periodically reap orphaned runs (5-min staleness threshold) and make sure
           // persisted queued work is still being driven forward.
           trackHeartbeatSchedulerWork(heartbeat
-            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true })
+            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true })
             .then(() => heartbeat.promoteDueScheduledRetries())
             .then(async (promotion) => {
               await heartbeat.resumeQueuedRuns();

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 = {
+  obsoletePolicyReviews: "obsolete-policy-reviews",
   cancelledWorkspacePreservation: "cancelled-workspace-preservation",
   instructionCollection: "instruction-collection",
   directoryCleanup: "directory-cleanup",

server/src/services/heartbeat.ts

@@ -1,3 +1,4 @@
+import { registerObsoletePolicyReviewWork } from "./native-runtime/obsolete-policy-reviews.js";
 import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
@@ -4063,6 +4064,8 @@ export function heartbeatService(
     triggerIssueMonitor,
 
     reportRunActivity: clearDetachedRunWarning,
+    registerObsoletePolicyReviewWork: (coordinator: Parameters<typeof registerObsoletePolicyReviewWork>[1], canRun: () => boolean) =>
+      registerObsoletePolicyReviewWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
     registerCancelledWorkspacePreservationWork: (
       coordinator: Parameters<typeof registerCancelledWorkspacePreservationWork>[1], canRun: () => boolean,
     ) => registerCancelledWorkspacePreservationWork(db, coordinator,

server/src/services/heartbeat/recovery.ts

@@ -1915,7 +1915,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
 
   let settledProcessesDiscovered = false;
 
-  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean }) {
+  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean }) {
     if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
     const staleThresholdMs = opts?.staleThresholdMs ?? 0;
     const now = new Date();
@@ -1933,6 +1933,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     // reconciler reads the durable workspace barrier and persisted runtime
     // mode, never the current feature flag.
     await reconcileNativeFinalizations(db, undefined, {
+      skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews,
       environmentRuntime,
       pluginWorkerManager: dependencies.pluginWorkerManager,
       onWorkspaceSettled: settleRecoveredNativeWorkspace,

server/src/services/native-runtime/native-finalization-reconciler-timing.test.ts

@@ -25,6 +25,13 @@ vi.mock("./automatic-completion-reviews.js", async (importOriginal) => ({
 const { reconcileNativeFinalizations } = await import("./native-finalization-reconciler.js");
 
 describe("native finalization reconciliation timing", () => {
+  it("omits migrated policy cleanup from periodic reconciliation", async () => {
+    timing.phase.mockClear();
+    await expect(reconcileNativeFinalizations({} as never, undefined, { skipObsoletePolicyReviews: true })).rejects.toThrow("review lookup failed");
+    expect(timing.phase.mock.calls.map(([phase]) => phase)).toEqual(["legacy_workspace_exports", "automatic_completion_reviews"]);
+    timing.phase.mockClear(); timing.finish.mockClear();
+  });
+
   it("records the phases and a failed outcome when a pass fails, and rethrows", async () => {
     await expect(reconcileNativeFinalizations({} as never)).rejects.toThrow("review lookup failed");
 

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -631,6 +631,7 @@ export async function claimNativeSessionResumptions(input: {
 }
 
 type NativeReconciliationOptions = {
+  skipObsoletePolicyReviews?: boolean;
   pluginWorkerManager?: WorkspaceWorkers;
   environmentRuntime?: EnvironmentRuntimeService;
   onWorkspaceSettled?: (input: {
@@ -717,10 +718,12 @@ async function reconcileNativeFinalizationsPass(
   await recoverLegacyUnsafeWorkspaceExports(db, runIds).catch((err) => {
     logger.warn({ err }, "Historical unsafe export recovery remains pending");
   });
-  timing.phase("obsolete_policy_reviews");
-  await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => {
-    logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation");
-  });
+  if (!options.skipObsoletePolicyReviews) {
+    timing.phase("obsolete_policy_reviews");
+    await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => {
+      logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation");
+    });
+  }
   timing.phase("automatic_completion_reviews");
   if (runIds?.length) {
     const scopes = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations)

server/src/services/native-runtime/obsolete-policy-reviews.ts

@@ -1,4 +1,8 @@
-import { and, eq, inArray, isNull, sql } from "drizzle-orm";
+import type { ObsoletePolicyReviewScope } from "../obsolete-policy-review-notifications.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import { and, asc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
 import {
   approvals, issueApprovals, issueThreadInteractions, issues,
   nativeRunFinalizations, statusDecisionEffects, statusDecisions, workAssessments,
@@ -14,8 +18,8 @@ import { persistActivity, publishActivity, type ActivityPublication } from "../a
 const obsoletePrompt = "Review the superseding native policy assessment.";
 
 /** Retire only the old version-change gate, never a real completion review. */
-export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string[]) {
-  const candidates = await db.select({
+function findObsoletePolicyReviews(db: Db, options: { runIds?: string[]; scope?: ObsoletePolicyReviewScope; id?: string; after?: string } = {}) {
+  return db.select({
     interaction: issueThreadInteractions,
     decision: statusDecisions,
     priorDecisionId: workAssessments.priorDecisionId,
@@ -53,9 +57,18 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string
         { kind: "bind_reviewer", prompt: obsoletePrompt, ownerUserId: null },
         { kind: "append_superseding_assessment" },
       ])}::jsonb`,
-      ...(runIds?.length ? [inArray(statusDecisions.runId, runIds)] : []),
-    )).limit(100);
+      ...(options.runIds?.length ? [inArray(statusDecisions.runId, options.runIds)] : []),
+      options.scope ? and(eq(issueThreadInteractions.companyId, options.scope.companyId), eq(issueThreadInteractions.issueId, options.scope.issueId)) : undefined,
+      options.id ? eq(issueThreadInteractions.id, options.id) : undefined,
+      options.after ? gt(issueThreadInteractions.id, options.after) : undefined,
+    )).orderBy(asc(issueThreadInteractions.id)).limit(100);
+}
 
+export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string[]) {
+  await dismissCandidates(db, await findObsoletePolicyReviews(db, { runIds }));
+}
+
+async function dismissCandidates(db: Db, candidates: Awaited<ReturnType<typeof findObsoletePolicyReviews>>) {
   for (const { interaction, decision, priorDecisionId } of candidates) {
     const publications: ActivityPublication[] = [];
     try {
@@ -80,6 +93,11 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string
           eq(issueThreadInteractions.id, interaction.id),
           eq(issueThreadInteractions.companyId, decision.companyId),
           eq(issueThreadInteractions.status, "pending"),
+          eq(issueThreadInteractions.payload, interaction.payload),
+          eq(issueThreadInteractions.kind, interaction.kind),
+          eq(issueThreadInteractions.sourceRunId, decision.runId),
+          eq(issueThreadInteractions.idempotencyKey, interaction.idempotencyKey!),
+          isNull(issueThreadInteractions.createdByAgentId), isNull(issueThreadInteractions.createdByUserId),
         )).returning({ id: issueThreadInteractions.id });
         if (!cancelled) return;
         const terminalInteraction = await issueThreadInteractionService(tx as unknown as Db).getById(cancelled.id);
@@ -150,3 +168,33 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string
     }
   }
 }
+
+export function registerObsoletePolicyReviewWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+  type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+  async function load(scope?: ObsoletePolicyReviewScope, signal?: AbortSignal): Promise<Entry[]> {
+    const entries = new Map<string, Entry>();
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await findObsoletePolicyReviews(db, { scope, after });
+      for (const { interaction } of rows) entries.set(interaction.id, {
+        id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: 0,
+      });
+      if (rows.length < 100) return [...entries.values()];
+      after = rows[rows.length - 1].interaction.id;
+    }
+    throw new Error("Obsolete policy review discovery stopped");
+  }
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.obsoletePolicyReviews, batchSize: 20,
+    decodeScope: (hint: string): ObsoletePolicyReviewScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+    matches: (entry: Entry, scope?: ObsoletePolicyReviewScope) => !scope || entry.companyId === scope.companyId && entry.issueId === scope.issueId,
+    load,
+    check: async (entry: Entry, signal?: AbortSignal) => {
+      const candidates = await findObsoletePolicyReviews(db, { scope: entry, id: entry.id });
+      if (!canRun() || signal?.aborted) throw new Error("Obsolete policy review cleanup stopped");
+      await dismissCandidates(db, candidates);
+      const pending = await findObsoletePolicyReviews(db, { scope: entry, id: entry.id });
+      return pending.length ? { ...entry, dueAt: Date.now() + 30_000 } : null;
+    },
+  }, canRun);
+}

server/src/services/native-runtime/status-decision-committer.ts

@@ -1,3 +1,4 @@
+import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
 import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
@@ -1938,6 +1939,7 @@ export async function commitNativeStatusDecision(input: {
       return { decision: decisionRow, issue, replayed: true };
     }
     await notifyChatRunMilestones(tx, input);
+    await notifyObsoletePolicyReviews(tx, input);
     if (!decisionRow) {
       const latestDecisionVersion = await tx
         .select({

server/src/services/obsolete-policy-review-notifications.ts

@@ -0,0 +1,7 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type ObsoletePolicyReviewScope = { companyId: string; issueId: string };
+
+export async function notifyObsoletePolicyReviews(transaction: object, scope: ObsoletePolicyReviewScope) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.obsoletePolicyReviews, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
+}