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

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native finalization repairs saved results and related task state. > - Its periodic pass also searches for obsolete policy-review cards. > - Those cards have a precise historical signature and existing durable rows. > - This PR gives that cleanup its own committed signals and pending-work retries. > - The existing owner still protects real reviews and later task changes. > - Settled review history no longer causes repeated cleanup queries. ## Linked Issues or Issue Description Refs #17677. This PR is stacked on cancelled workspace preservation. It owns only obsolete policy-review cleanup within the native finalization reconciler. **What needs to be improved?** Every periodic native finalization pass queries for old version-change review cards, even after all such cards are gone. The parent reaper combines this cleanup with unrelated finalization and recovery duties. **Why is this improvement important?** Empty queries prevent idle tenants from becoming quiet. Removing the query must preserve historical repair, real review requests, task status authority, and cleanup after a failed write. **Proposed solution** Reconstruct matching pending cards at startup. Signal the existing native decision transaction before its writes. Schedule bounded cleanup only while an exact card remains pending. Skip this duty in periodic reaping while keeping startup and manual entry points. ROADMAP.md and related PR search found no duplicate of this migration. ## What Changed - Register obsolete-review cleanup with the existing scoped deadline worker. Reconstruct in 100-row keyset pages and process fair batches of twenty. - Use company/issue hints from the existing status-decision commit boundary. Serialize only the two IDs; preserve outer commit and rollback. - Reuse the exact historical signature and existing cleanup owner. Keep status-version, decision-pointer, separate-review, activity and chat-publication guards. - Recheck selected payload and system-card identity in the conditional withdrawal so a concurrent edit cannot be cancelled from a stale snapshot. - Retain a 30-second retry only while a matching pending card survives. Preserve drain, standby, startup and joined shutdown. - Skip only this duty in the periodic finalization pass. Automatic completion reviews, unsafe exports, status effects, evidence reassessment and workspace finalization remain separate migrations. - Add the owner audit and validated external-test database support for the existing status-authority corpus. Add no schema, table, independent timer or provider branch. ## Verification - Current source `4cc0d03ae96bd92a4fdad44168c2a8647d42466c`: all 443 cases across seven complete files pass with no skips in 46.46 seconds. This covers the full status-authority corpus and cleanup worker, startup/timing, heartbeat recovery, native finalization, external chat and governed milestones. The audit gives per-file counts without double-counting fixtures inside the corpus matrix. - Five distinct fresh PostgreSQL fixtures migrated through parent 0409 before setup. Database build, server typecheck/build, module boundaries and target migration order pass. The whole database source tree matches the parent; no migration or table is added. - Normal integration preserves substantive owned edits and unrelated upstream blobs. The earlier circular notification-payload failure remains in the audit. Actual parent `51c3f8924cb0bd5fe7cbda15ad9b8f06a6a11055` is reconciled through a normal merge, with its full tree equal to the reviewed ancestor and the entire accepted child tree unchanged. Fresh current-head Greptile 5/5 and clean CI are still required. No deployed sleep qualification is claimed. - Review correction: the old head reproduces a fixed-prefix collision on a same-database corpus rerun. The initial deletion attempt passed assertions but failed teardown on non-cascading audit foreign keys; it was removed. The final two-value change uses each fixture company UUID in its prefix and changes no production behavior, assertions, timing limits or teardown. - The complete 56-case corpus passes a second time on the same database in 10.15 seconds. Retained PSC/OUT rows and an unrelated sentinel company remain unchanged after both runs. The repeat is not double-counted among the 443 distinct cases. The audit retains both failures and their causes. Fresh CI and Greptile are required for the updated head. ## Risks - Startup reconstruction still scans historical candidates and joins; its cost scales with history. Scoped commits use existing indexes. No recurring query remains after this worker settles. - The signal covers instrumented single-process writes. Independent SQL writers and concurrent older controllers need explicit integration. - A normal native decision can cause one scoped lookup, even when it creates no obsolete review. It does not retain a periodic scan. - Failed withdrawals retain retries while matching cards remain. Actual cleanup still uses the existing locking, authority and activity owner. - Other finalization duties still run. This PR does not establish whole-instance quiet, external alarm coverage, provider suspension or fleet savings. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and PostgreSQL tests. The runtime does not expose a more precise model ID or context-window size for this turn. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Current published head `9aa67199da64d700fd6230829d38f2ef522a07fe` has fresh Greptile 5/5, Ponytail 5/5 with an unconditional ship verdict, and all current CI checks passing or skipped. The reproduced rerun P2 is fixed and its thread is resolved. --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-30 13:20:54 +01:00 · 2026-10-28 22:53:19 -05:00
1 parent 51c3f8924c
commit dfd5b1ac62
@@ -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,10 +384,11 @@ 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);
-    await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: "PSC" });
-    await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: "OUT" });
+    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${companyId.replaceAll("-", "").toUpperCase()}` });
+    await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: `OUT${outsideCompanyId.replaceAll("-", "").toUpperCase()}` });
    await db.insert(agents).values([
      {
        id: agentId,
@@ -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.";
@@ -116,6 +116,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() })),
@@ -554,6 +555,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();
@@ -973,7 +981,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 {
@@ -1824,6 +1824,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,
@@ -1956,7 +1958,7 @@ async function startServerWithDatabaseTeardown(
          // Periodically reap orphaned runs (5-min staleness threshold) and make sure
          // persisted queued work is still being driven forward.
          runPeriodicRecovery(() => 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();
@@ -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",
@@ -1,3 +1,4 @@
+import { registerObsoletePolicyReviewWork } from "./native-runtime/obsolete-policy-reviews.js";
 import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js";
 import { readUpgradeReadiness } from "./heartbeat/upgrade-readiness.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -4085,6 +4086,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,
@@ -1947,7 +1947,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();
@@ -1965,6 +1965,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,
@@ -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");

@@ -637,6 +637,7 @@ export async function claimNativeSessionResumptions(input: {
 }

 type NativeReconciliationOptions = {
+  skipObsoletePolicyReviews?: boolean;
  pluginWorkerManager?: WorkspaceWorkers;
  environmentRuntime?: EnvironmentRuntimeService;
  onWorkspaceSettled?: (input: {
@@ -723,10 +724,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)
@@ -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);
+}
@@ -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";
@@ -1981,6 +1982,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({
@@ -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 }));
+}