fix(server): schedule automatic completion review retirement

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

server/src/services/native-runtime/automatic-completion-reviews.ts

@@ -1,4 +1,8 @@
-import { and, eq, inArray, isNull, sql } from "drizzle-orm";
+import type { AutomaticCompletionReviewScope } from "../automatic-completion-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 {
   completionContracts,
   issueThreadInteractions,
@@ -24,7 +28,7 @@ const automaticPrompt =
   "Review the persisted native-run evidence and confirm whether this issue may be completed.";
 
 /** Identify only proven system fallback cards; this lookup never changes state. */
-export async function findAutomaticCompletionReviews(db: Db, issueId?: string) {
+function findAutomaticReviewCandidates(db: Db, options: { issueId?: string; companyId?: string; id?: string; after?: string } = {}) {
   return db
     .select({ interaction: issueThreadInteractions, decision: statusDecisions })
     .from(issueThreadInteractions)
@@ -83,17 +87,20 @@ export async function findAutomaticCompletionReviews(db: Db, issueId?: string) {
         sql`${issueThreadInteractions.payload}->'target'->>'key' = 'native_completion_review'`,
         sql`${issueThreadInteractions.payload}->'target'->>'revisionId' = ${statusDecisions.id}::text`,
         sql`split_part(${issueThreadInteractions.payload}->>'prompt', E'\n', 1) = ${automaticPrompt}`,
-        ...(issueId ? [eq(issueThreadInteractions.issueId, issueId)] : []),
+        options.issueId ? eq(issueThreadInteractions.issueId, options.issueId) : undefined,
+        options.companyId ? eq(issueThreadInteractions.companyId, options.companyId) : undefined,
+        options.id ? eq(issueThreadInteractions.id, options.id) : undefined,
+        options.after ? gt(issueThreadInteractions.id, options.after) : undefined,
       ),
     )
-    .limit(100)
-    .catch((err) => {
-      logger.warn(
-        { err },
-        "Automatic completion review lookup failed; will retry",
-      );
-      return [];
-    });
+    .orderBy(asc(issueThreadInteractions.id)).limit(100);
+}
+
+export async function findAutomaticCompletionReviews(db: Db, issueId?: string) {
+  return findAutomaticReviewCandidates(db, { issueId }).catch((err) => {
+    logger.warn({ err }, "Automatic completion review lookup failed; will retry");
+    return [];
+  });
 }
 
 /** Narrow, replay-safe retirement. Explicit requests and answered cards are immutable here. */
@@ -101,7 +108,10 @@ export async function dismissAutomaticCompletionReviews(
   db: Db,
   issueId?: string,
 ) {
-  const candidates = await findAutomaticCompletionReviews(db, issueId);
+  await dismissCandidates(db, await findAutomaticCompletionReviews(db, issueId));
+}
+
+async function dismissCandidates(db: Db, candidates: Awaited<ReturnType<typeof findAutomaticReviewCandidates>>) {
   for (const { interaction, decision } of candidates) {
     const publications: ActivityPublication[] = [];
     try {
@@ -147,6 +157,10 @@ export async function dismissAutomaticCompletionReviews(
               eq(issueThreadInteractions.companyId, issue.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();
@@ -213,3 +227,33 @@ export async function decisionHasRetiredAutomaticReview(
     .limit(1);
   return rows.length > 0;
 }
+
+export function registerAutomaticCompletionReviewWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+  type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+  async function load(scope?: AutomaticCompletionReviewScope, signal?: AbortSignal): Promise<Entry[]> {
+    const entries = new Map<string, Entry>();
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await findAutomaticReviewCandidates(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("Automatic completion review discovery stopped");
+  }
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.automaticCompletionReviews, batchSize: 20,
+    decodeScope: (hint: string): AutomaticCompletionReviewScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+    matches: (entry: Entry, scope?: AutomaticCompletionReviewScope) => !scope || entry.companyId === scope.companyId && entry.issueId === scope.issueId,
+    load,
+    check: async (entry: Entry, signal?: AbortSignal) => {
+      const candidates = await findAutomaticReviewCandidates(db, entry);
+      if (!canRun() || signal?.aborted) throw new Error("Automatic completion review cleanup stopped");
+      await dismissCandidates(db, candidates);
+      const pending = await findAutomaticReviewCandidates(db, entry);
+      return pending.length ? { ...entry, dueAt: Date.now() + 30_000 } : null;
+    },
+  }, canRun);
+}

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

@@ -32,6 +32,14 @@ describe("native finalization reconciliation timing", () => {
     timing.phase.mockClear(); timing.finish.mockClear();
   });
 
+  it("omits automatic review discovery when its worker owns that duty", async () => {
+    timing.phase.mockClear();
+    const db = { select: vi.fn(() => { throw new Error("candidate scan reached"); }) };
+    await expect(reconcileNativeFinalizations(db as never, ["run"], { skipAutomaticCompletionReviews: true })).rejects.toThrow("candidate scan reached");
+    expect(timing.phase.mock.calls.map(([phase]) => phase)).toEqual(["legacy_workspace_exports", "obsolete_policy_reviews", "candidate_scan"]);
+    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

@@ -632,6 +632,7 @@ export async function claimNativeSessionResumptions(input: {
 
 type NativeReconciliationOptions = {
   skipObsoletePolicyReviews?: boolean;
+  skipAutomaticCompletionReviews?: boolean;
   pluginWorkerManager?: WorkspaceWorkers;
   environmentRuntime?: EnvironmentRuntimeService;
   onWorkspaceSettled?: (input: {
@@ -724,13 +725,15 @@ async function reconcileNativeFinalizationsPass(
       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)
-      .where(inArray(nativeRunFinalizations.runId, runIds));
-    for (const scope of scopes) await dismissAutomaticCompletionReviews(db, scope.issueId);
-  } else {
-    await dismissAutomaticCompletionReviews(db);
+  if (!options.skipAutomaticCompletionReviews) {
+    timing.phase("automatic_completion_reviews");
+    if (runIds?.length) {
+      const scopes = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations)
+        .where(inArray(nativeRunFinalizations.runId, runIds));
+      for (const scope of scopes) await dismissAutomaticCompletionReviews(db, scope.issueId);
+    } else {
+      await dismissAutomaticCompletionReviews(db);
+    }
   }
   timing.phase("candidate_scan");
   const now = new Date();

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

@@ -1,3 +1,4 @@
+import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js";
 import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
 import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
@@ -1940,6 +1941,7 @@ export async function commitNativeStatusDecision(input: {
     }
     await notifyChatRunMilestones(tx, input);
     await notifyObsoletePolicyReviews(tx, input);
+    await notifyAutomaticCompletionReviews(tx, input);
     if (!decisionRow) {
       const latestDecisionVersion = await tx
         .select({