fix(server): wake tool-review card repair from committed changes

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

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

@@ -69,6 +69,7 @@ const {
         settleInterruptedActionReviews: vi.fn(async () => ({ scanned: 0 })),
         nextActionReviewInterruptionAt: vi.fn(async () => null),
         recoverApprovedActionReviews: vi.fn(async () => ({ scanned: 0, retry: false })),
+        repairActionReviewProjections: vi.fn(async () => ({ scanned: 0, retry: false })),
         sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
         sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })),
         nextActionReviewExpiryAt: vi.fn(async () => null),

server/src/__tests__/tool-gateway-service.test.ts

@@ -1,3 +1,5 @@
+import { registerToolReviewProjectionWork } from "../services/tool-review-projection-work.js";
+import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
 import { registerToolReviewInterruptionWork } from "../services/tool-review-interruption-work.js";
 import { writeToolReviewChange } from "../services/tool-review-work-notifications.js";
 import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
@@ -277,6 +279,8 @@ describeEmbeddedPostgres("tool gateway service", () => {
       await coordinator?.stop(); coordinator = undefined;
       vi.useRealTimers(); vi.restoreAllMocks();
       await db.execute(sql`drop trigger if exists sleep_review_fail on tool_invocations`);
+      await db.execute(sql`drop trigger if exists sleep_projection_fail on issue_thread_interactions`);
+      await db.execute(sql`drop function if exists sleep_projection_fail()`);
       await db.execute(sql`drop function if exists sleep_review_fail()`);
       await db.execute(sql`drop sequence if exists sleep_review_failure_attempt`);
       expect(errors).toEqual([]);
@@ -342,6 +346,155 @@ describeEmbeddedPostgres("tool gateway service", () => {
       return { provider, release, started };
     }
 
+    describe("tool review projection worker", () => {
+      async function startProjection() {
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
+          canReconcile: () => true, onError: error => errors.push(error) });
+        await registerToolReviewProjectionWork(coordinator, { sweep: gateway.repairActionReviewProjections }, () => !draining).ready;
+      }
+      async function savedOutcome(pending: Awaited<ReturnType<typeof request>>, source = db,
+        status: "executed" | "failed" | "expired" | "cancelled" = "executed") {
+        await writeToolReviewChange(source, async tx => {
+          await tx.update(toolInvocations).set({ status: status === "executed" ? "succeeded" : "failed",
+            errorCode: status === "executed" ? null : "saved_failure", errorMessage: null,
+            resultSummary: { summary: "Durable result" }, completedAt: new Date() })
+            .where(eq(toolInvocations.id, pending.invocationId));
+          await tx.update(toolActionRequests).set({ status, updatedAt: new Date() })
+            .where(eq(toolActionRequests.id, pending.id));
+        });
+      }
+      async function acceptedCard(pending: Awaited<ReturnType<typeof request>>) {
+        expect(pending.interactionId).not.toBeNull();
+        await db.update(issueThreadInteractions).set({ status: "accepted", result: { version: 1, outcome: "accepted" } })
+          .where(eq(issueThreadInteractions.id, pending.interactionId!));
+      }
+      async function projection(pending: Awaited<ReturnType<typeof request>>) {
+        const [row] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        return (row.result as { toolAction?: { status: string; resultSummary: string; errorCode: string | null } } | null)?.toolAction;
+      }
+
+      it("becomes quiet with configuration and repairs a real foreground approval without replay", async () => {
+        await startProjection(); await quiet();
+        const pending = await request();
+        await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
+        await quiet(); expect(await projection(pending)).toMatchObject({ status: "executed" });
+        expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1);
+        const before = (await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_updated"))).length;
+        await gateway.repairActionReviewProjections();
+        expect((await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_updated"))).length).toBe(before);
+      });
+
+      it.each(["executed", "failed", "expired", "cancelled"] as const)("reconstructs a saved %s result after restart and leaves terminal history quiet", async status => {
+        const pending = await request(); await acceptedCard(pending); await savedOutcome(pending, db, status);
+        await startProjection(); await quiet();
+        expect(await projection(pending)).toMatchObject({ status: status === "cancelled" ? "expired" : status,
+          resultSummary: "Durable result", errorCode: status === "executed" ? null : "saved_failure" });
+        expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]);
+      });
+
+      it("retains a first terminal write during drain without passive scans", async () => {
+        const pending = await request(); await acceptedCard(pending);
+        await startProjection(); await quiet(); draining = true;
+        await savedOutcome(pending); await settle();
+        const reads = vi.spyOn(db, "select");
+        await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+        expect(await projection(pending)).toBeUndefined();
+        draining = false; await settle(); await quiet();
+        expect(await projection(pending)).toMatchObject({ status: "executed" });
+      });
+
+      it.each([false, true])("waits for the outer outcome transaction (rollback %s)", async rollback => {
+        const pending = await request(); await acceptedCard(pending); await startProjection(); await quiet();
+        let entered!: () => void, release!: () => void;
+        const ready = new Promise<void>(resolve => { entered = resolve; });
+        const gate = new Promise<void>(resolve => { release = resolve; });
+        const aborted = new Error("fixture rollback");
+        const running = db.transaction(async tx => {
+          await savedOutcome(pending, tx as unknown as typeof db);
+          entered(); await gate; if (rollback) throw aborted;
+        }).catch(error => { if (error !== aborted) throw error; });
+        await ready;
+        try { await vi.advanceTimersByTimeAsync(60_000); expect(await projection(pending)).toBeUndefined(); }
+        finally { release(); await running; }
+        await vi.waitFor(async () => {
+          await vi.advanceTimersByTimeAsync(30_000);
+          expect(idleWorkSnapshot().active).toBe(0);
+          if (rollback) expect(await projection(pending)).toBeUndefined();
+          else expect(await projection(pending)).toMatchObject({ status: "executed" });
+        }, { timeout: 5000 });
+        await quiet();
+      });
+
+      it("retries a failed projection through drain without executing the saved action", async () => {
+        const pending = await request(); await acceptedCard(pending); await savedOutcome(pending);
+        await db.execute(sql`create function sleep_projection_fail() returns trigger language plpgsql as $$
+          begin raise exception 'projection fixture failure'; end $$`);
+        await db.execute(sql`create trigger sleep_projection_fail before update on issue_thread_interactions
+          for each row when (new.result->'toolAction'->>'status' = 'executed') execute function sleep_projection_fail()`);
+        await startProjection(); expect(errors.length).toBeGreaterThan(0); errors.length = 0;
+        expect(await projection(pending)).toBeUndefined();
+        draining = true;
+        await db.execute(sql`drop trigger sleep_projection_fail on issue_thread_interactions`);
+        await settle(60 * 60_000); draining = false;
+        await vi.waitFor(async () => {
+          await vi.advanceTimersByTimeAsync(30_000);
+          expect(await projection(pending)).toMatchObject({ status: "executed" });
+        }, { timeout: 5000 });
+        await quiet();
+        expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]);
+      });
+
+      it("rechecks the projection under lock when foreground repair wins the race", async () => {
+        const pending = await request(); await acceptedCard(pending); await savedOutcome(pending);
+        const transaction = db.transaction.bind(db);
+        let entered!: () => void, release!: () => void;
+        const ready = new Promise<void>(resolve => { entered = resolve; });
+        const gate = new Promise<void>(resolve => { release = resolve; });
+        const intercepted = vi.spyOn(db, "transaction").mockImplementationOnce(async callback => {
+          entered(); await gate; return transaction(callback);
+        });
+        const repair = gateway.repairActionReviewProjections();
+        await ready;
+        try { await gateway.repairActionReviewProjections(); } finally { release(); }
+        await repair; intercepted.mockRestore();
+        const updates = await db.select().from(activityLog).where(and(eq(activityLog.companyId, fixture.company.id),
+          eq(activityLog.action, "issue.thread_interaction_updated")));
+        expect(updates).toHaveLength(1);
+        expect(await projection(pending)).toMatchObject({ status: "executed" });
+        await startProjection(); await quiet();
+      });
+
+      it("does not starve later pages behind a card that cannot yet reflect its outcome", async () => {
+        const pending = await request(); await savedOutcome(pending, db, "failed");
+        const [template] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        const copies = Array.from({ length: 101 }, () => ({ ...template, id: randomUUID(), idempotencyKey: randomUUID(),
+          status: "accepted" as const, result: { version: 1 as const, outcome: "accepted" as const } }));
+        await db.insert(issueThreadInteractions).values(copies);
+        await db.insert(toolActionRequests).values(copies.map((card, index) => ({ ...pending,
+          id: `ffffffff-ffff-4fff-afff-${String(index).padStart(12, "0")}`,
+          interactionId: card.id, status: "failed" as const })));
+        await startProjection();
+        expect(coordinator!.nextWakeAt()).not.toBeNull();
+        const cards = await db.select().from(issueThreadInteractions);
+        expect(cards.filter(card => (card.result as { toolAction?: { status: string } } | null)?.toolAction?.status === "failed")).toHaveLength(101);
+        await db.update(issueThreadInteractions).set({ addresseeAgentId: fixture.agent.id }).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        await issueThreadInteractionService(db).cancelPendingForDeletedAddressee(fixture.company.id, fixture.agent.id);
+        await settle(); await quiet();
+        expect(await projection(pending)).toMatchObject({ status: "failed" });
+      });
+
+      it("reactivates when an administrative card mutation overwrites a settled projection", async () => {
+        const pending = await request();
+        await db.update(issueThreadInteractions).set({ addresseeAgentId: fixture.agent.id,
+          result: { version: 1, outcome: "accepted" } }).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        await savedOutcome(pending, db, "failed"); await startProjection(); await quiet();
+        expect(await projection(pending)).toMatchObject({ status: "failed" });
+        await issueThreadInteractionService(db).cancelPendingForDeletedAddressee(fixture.company.id, fixture.agent.id);
+        await settle(); await quiet();
+        expect(await projection(pending)).toMatchObject({ status: "failed", errorCode: "saved_failure" });
+      });
+    });
+
     describe("interrupted review recovery worker", () => {
       async function startInterruption() {
         coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,

server/src/index.ts

@@ -1,3 +1,4 @@
+import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js";
 import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js";
 import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js";
 import { registerToolReviewExpiryWork } from "./services/tool-review-expiry-work.js";
@@ -1458,7 +1459,10 @@ async function startServerWithDatabaseTeardown(
     } else {
       await app.locals.toolGateway.settleInterruptedActionReviews();
     }
-    return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: false });
+    return config.heartbeatSchedulerEnabled
+      ? registerToolReviewProjectionWork(deliveryWork, { sweep: app.locals.toolGateway.repairActionReviewProjections },
+        () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready
+      : app.locals.toolGateway.repairActionReviewProjections();
   }).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed"));
   await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled
     ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway,
@@ -1883,8 +1887,6 @@ async function startServerWithDatabaseTeardown(
             logger.error({ err }, "periodic tool connection health sweep failed");
           }));
 
-        trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed")));
-
         if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
         if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
           // Periodically reap orphaned runs (5-min staleness threshold) and make sure

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

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   executionStatus: "execution-status",
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
+  toolReviewProjection: "tool-review-projection",
   toolReviewInterruption: "tool-review-interruption",
   toolReviewRecovery: "tool-review-recovery",
   toolReviewExpiry: "tool-review-expiry",

server/src/services/issue-thread-interactions.ts

@@ -2240,6 +2240,7 @@ export function issueThreadInteractionService(
             })
           : undefined;
 
+      if (lockedCurrent.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
       const [updated] = await tx
         .update(issueThreadInteractions)
         .set({
@@ -2517,6 +2518,7 @@ export function issueThreadInteractionService(
         now,
       });
 
+      if (lockedCurrent.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
       const [resolved] = await tx
         .update(issueThreadInteractions)
         .set({
@@ -3246,6 +3248,7 @@ export function issueThreadInteractionService(
       const cancelled: IssueThreadInteraction[] = [];
       for (const row of rows) {
         const updated = await db.transaction(async (tx) => {
+          if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
           const [resolved] = await tx
             .update(issueThreadInteractions)
             .set({
@@ -3330,6 +3333,7 @@ export function issueThreadInteractionService(
       const expired: IssueThreadInteraction[] = [];
       for (const { row, replacementInteractionId } of supersededRows) {
         const updated = await db.transaction(async (tx) => {
+          if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
           const [updatedRow] = await tx
             .update(issueThreadInteractions)
             .set({
@@ -3749,6 +3753,7 @@ export function issueThreadInteractionService(
             data.kind === "ask_user_questions"
               ? buildSupersededByNewerInteractionResult(row.id)
               : buildSupersededByNewerRequestResult(row.id);
+          if (data.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
           const supersededRows = await tx
             .update(issueThreadInteractions)
             .set({
@@ -4400,6 +4405,7 @@ export function issueThreadInteractionService(
               .delete(toolOauthStates)
               .where(eq(toolOauthStates.interactionId, row.id));
           }
+          if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
           const [resolved] = await tx
             .update(issueThreadInteractions)
             .set({
@@ -4568,6 +4574,7 @@ export function issueThreadInteractionService(
           );
           if (!sampleConfirmationRow) continue;
           const updatedInteractions = await db.transaction(async (tx) => {
+            await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection);
             const updatedRows = await tx
               .update(issueThreadInteractions)
               .set({

server/src/services/tool-gateway.ts

@@ -1,3 +1,4 @@
+import { toolReviewProjectionService } from "./tool-review-projection.js";
 import { toolReviewInterruptionService } from "./tool-review-interruption.js";
 import { claimToolTestApproval, persistToolTestOutcome, startToolTestInvocation, type ToolTestApprovalClaim } from "./tool-test-approval.js";
 import { recoverApprovedToolReviews } from "./tool-review-recovery.js";
@@ -2326,6 +2327,11 @@ export function createToolGatewayService(
                 ? "stale_target"
                 : null;
       if (!outcome) return;
+      const projection = asRecord(currentResult?.toolAction);
+      if (projection?.status === (input.status === "cancelled" ? "expired" : input.status)
+        && projection.errorCode === (input.errorCode ?? null)
+        && projection.errorMessage === (input.errorMessage ?? null)
+        && projection.resultSummary === (input.resultSummary ?? null)) return false;
 
       const now = new Date();
       await tx
@@ -7916,10 +7922,10 @@ export function createToolGatewayService(
       return { projectId: issue.projectId };
     }
     const expiredAt = new Date();
-    await db
+    await writeToolReviewChange(db, tx => tx
       .update(toolActionRequests)
       .set({ status: "expired", resolvedAt: expiredAt, updatedAt: expiredAt })
-      .where(eq(toolActionRequests.id, claimed.id));
+      .where(eq(toolActionRequests.id, claimed.id)));
     await reflectToolActionInteractionLifecycle({
       actionRequestId: claimed.id,
       status: "expired",
@@ -8626,6 +8632,7 @@ export function createToolGatewayService(
     };
   }
 
+  const reviewProjection = toolReviewProjectionService(db, reflectToolActionInteractionLifecycle);
   const reviewInterruption = toolReviewInterruptionService(db, async settled => {
     void emitConnectionInvoked(db, settled.invocationId);
     await reflectToolActionInteractionLifecycle(settled);
@@ -9649,6 +9656,7 @@ export function createToolGatewayService(
       return buildTestCallStatus(actionRequest, invocation);
     },
 
+    repairActionReviewProjections: reviewProjection.sweep,
     settleInterruptedActionReviews: reviewInterruption.sweep,
     nextActionReviewInterruptionAt: reviewInterruption.next,
     sweepExpiredActionReviews: reviewExpiry.sweep,
@@ -9679,40 +9687,7 @@ export function createToolGatewayService(
     },
 
     async sweepActionReviews(input: { includeExpiry?: boolean; includeApproved?: boolean; includeInterrupted?: boolean } = {}) {
-      // A process may stop between persisting a provider outcome and updating the
-      // feed projection. Reconcile from authoritative rows before delivering it.
-      const unreflected = await db
-        .select({ request: toolActionRequests, invocation: toolInvocations })
-        .from(toolActionRequests)
-        .innerJoin(
-          issueThreadInteractions,
-          eq(issueThreadInteractions.id, toolActionRequests.interactionId),
-        )
-        .innerJoin(
-          toolInvocations,
-          eq(toolInvocations.id, toolActionRequests.invocationId),
-        )
-        .where(
-          and(
-            inArray(toolActionRequests.status, [
-              "executed",
-              "failed",
-              "expired",
-              "cancelled",
-            ]),
-            sql`coalesce(${issueThreadInteractions.result}->'toolAction'->>'status', '') <> case when ${toolActionRequests.status} = 'cancelled' then 'expired' else ${toolActionRequests.status} end`,
-          ),
-        )
-        .limit(100);
-      for (const { request, invocation } of unreflected)
-        await reflectToolActionInteractionLifecycle({
-          actionRequestId: request.id,
-          status: request.status as
-            "executed" | "failed" | "expired" | "cancelled",
-          errorCode: invocation.errorCode,
-          errorMessage: invocation.errorMessage,
-          resultSummary: invocation.resultSummary?.summary,
-        });
+      await reviewProjection.sweep();
       let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep()).scanned;
       if (input.includeApproved !== false) scanned += (await this.recoverApprovedActionReviews()).scanned;
       if (input.includeInterrupted !== false) scanned += (await reviewInterruption.sweep()).scanned;

server/src/services/tool-review-projection-work.ts

@@ -0,0 +1,20 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import type { toolReviewProjectionService } from "./tool-review-projection.js";
+
+export function registerToolReviewProjectionWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof toolReviewProjectionService>,
+  canRun: () => boolean,
+) {
+  let next: number | null = null;
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.toolReviewProjection, {
+    retryMs: 30_000, canRun,
+    run: async () => {
+      const result = await service.sweep();
+      next = result.retry ? Date.now() + 30_000 : null;
+    },
+    next: async () => next,
+  });
+}

server/src/services/tool-review-projection.ts

@@ -0,0 +1,46 @@
+import { and, asc, eq, gt, inArray, sql } from "drizzle-orm";
+import { issueThreadInteractions, toolActionRequests, toolInvocations, type Db } from "@paperclipai/db";
+
+export function toolReviewProjectionService(db: Db, reflect: (input: {
+  actionRequestId: string;
+  status: "executed" | "failed" | "expired" | "cancelled";
+  errorCode: string | null;
+  errorMessage: string | null;
+  resultSummary: string | null;
+}) => Promise<void>) {
+  const mismatch = and(
+    inArray(toolActionRequests.status, ["executed", "failed", "expired", "cancelled"]),
+    sql`coalesce(${issueThreadInteractions.result}->'toolAction'->>'status', '') <> case when ${toolActionRequests.status} = 'cancelled' then 'expired' else ${toolActionRequests.status} end`,
+  );
+  const candidates = (cursor?: string) => db
+    .select({ request: toolActionRequests, invocation: toolInvocations })
+    .from(toolActionRequests)
+    .innerJoin(issueThreadInteractions, eq(issueThreadInteractions.id, toolActionRequests.interactionId))
+    .innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId))
+    .where(and(mismatch, cursor ? gt(toolActionRequests.id, cursor) : undefined))
+    .orderBy(asc(toolActionRequests.id));
+
+  return {
+    async sweep() {
+      let cursor: string | undefined;
+      let scanned = 0;
+      const failures: unknown[] = [];
+      for (;;) {
+        const rows = await candidates(cursor).limit(100);
+        for (const { request, invocation } of rows) {
+          try {
+            await reflect({ actionRequestId: request.id,
+              status: request.status as "executed" | "failed" | "expired" | "cancelled",
+              errorCode: invocation.errorCode, errorMessage: invocation.errorMessage,
+              resultSummary: invocation.resultSummary?.summary ?? null });
+          } catch (error) { failures.push(error); }
+        }
+        scanned += rows.length;
+        if (rows.length < 100) break;
+        cursor = rows[rows.length - 1].request.id;
+      }
+      if (failures.length) throw new AggregateError(failures, "Tool review projection repair failed");
+      return { scanned, retry: (await candidates().limit(1)).length > 0 };
+    },
+  };
+}

server/src/services/tool-review-work-notifications.ts

@@ -5,6 +5,7 @@ export async function notifyToolReviewChange(transaction: object) {
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewInterruption);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewProjection);
 }
 
 export function writeToolReviewChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {