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

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Tool requests and invocations store the execution outcome. Review cards show that outcome. > - The maintenance tick repeatedly joins these tables even when every card matches. > - Committed changes can wake repair. Existing rows can restore work after restart. > - A large company backlog also needs a bounded pass so other companies and shutdown can proceed. > - This PR uses the shared coordinator for scoped repair, continuation and failure deadlines. Settled history becomes quiet. ## Linked Issues or Issue Description **What happened?** The aggregate maintenance tick scans tool-review projections on every tick. Matching terminal history still causes database work. A company with many stale cards can also hold one worker in an unlimited drain. **What did you expect?** Relevant writes and startup should wake repair. Failed or unresolved cards should retain recovery. Matching history should cause no recurring SQL. Each normal worker pass should honor drain and shutdown. **Steps to reproduce** Start an instance with completed tool requests and matching cards. Leave it idle. The old maintenance tick keeps querying for mismatches. A large mismatch backlog also drains without a row limit or cancellation check. This is slice 07 of the sleep migration. It now targets master after #17458 landed. Related owners are pending expiry #17453 and approved recovery #17454. Their execution duties remain separate from card repair. The prior open-PR search found no duplicate projection migration. ## What Changed - Register company-scoped projection repair with the existing delivery coordinator. Remove its recurring aggregate-tick scan. - Signal through the existing review-write transaction owner and linked interaction mutations. Ordinary unlinked confirmations do not wake repair. - Keep interpretation, locks, current-state checks, activity and continuation callbacks in the gateway. A repair never repeats a provider tool call. - Limit each normal worker pass to 500 rows. Check abort and admission between rows. Save company progress through truncated and failed passes so blocked early pages cannot hide later cards. - Schedule truncated work after one second. Back off unresolved cards from thirty seconds to five minutes; fresh committed hints reset the delay. Clear obsolete cursors after empty reconstruction. - Skip repeated company discovery for scoped hints and the extra candidate query on an empty sweep with no saved cursor. Attempted or resumed passes still verify remaining mismatches because reflection returns no settlement result. - Add generated migration `0386_steady_epoch.sql` for the existing `(company_id, interaction_id)` lookup. Use the existing concurrent-index executor. Add no table. ## Verification Current candidate: `d4706fbf2a04a00ade30181eb1bd9322e8455818`. All 222 cases in the full gateway/projection/interruption, startup and scoped-deadline suites pass without skips after the query/retry efficiency repair; server types and build pass. A single committed company hint with matching retained history and a separate empty direct sweep each perform one candidate query. The unresolved-card regression failed before repair and now verifies backoff through the five-minute cap, prompt producer activation and quiet after settlement. An initial assertion incorrectly assumed an entire approval emitted one wake; its separate real writes require isolating the single-hint measurement. The earlier master integration `fe6e10629b` additionally established the following evidence before this efficiency follow-up: - **320 cases pass** across full gateway/projection/interruption, startup, scoped-deadline, narrowed identity-read and authenticated gateway HTTP suites. The HTTP selection includes all 81 cases. - **Six real PostgreSQL regressions failed before their fixes:** pass bounds, abort, drain, progress past unprojectable pages, progress past failed pages, and first repair after empty reconstruction. The complete owning selection passes all 220 cases. - Server production typechecking and build pass. Generated index migration, snapshot drift and numbering/safety checks pass. The five newest snapshots remain; master journal entries are unchanged. - A real missing-link query on 30,000 history rows changed from a sequential scan that filtered all 30,000 rows and touched 429 buffers to the new index with two buffers. Both treatments returned no rows. - Normal master integration preserves 35 independent performance files exactly. The gateway also retains master's narrowed identity read. Commands include `pnpm --filter @paperclipai/server typecheck`, `pnpm --filter @paperclipai/server build`, and server Vitest selections for `tool-gateway-service`, `server-startup-feedback-export`, `scoped-deadline-work`, `run-identity-projections`, and `connection-management.integration`, with separate PostgreSQL test fixtures. [The behavior and failure audit](doc/idle-tool-review-projection.md) records the producer contract, earlier fixture failures, index measurement, red/green regressions and integration evidence. Fresh current-head CI and Greptile/Ponytail remain required before merge. ## Risks - Signals use the existing single-active-writer database owner. Another replica or independent SQL writer needs a wake contract. - An already issued database query or reflection finishes before cancellation takes effect. Checks between rows bound further dispatch. - Legacy manual/global and disabled-scheduler one-shot repairs keep their complete-sweep behavior. The normal coordinator worker always supplies its row budget and admission controls. - Unprojectable cards and thrown failures retain recovery with a five-minute backoff cap. The company mismatch query still grows with terminal history. Empty history does not retain a recurring scan. - The index migration uses concurrent creation and the existing invalid-index recovery path. Prior SQL and journal entries remain unchanged. - Cloud sleep eligibility and external-wake coverage do not change. Local worker quiet is not proof of host or database suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-29 04:25:52 +01:00 · 2026-10-27 17:09:31 -05:00
1 parent d40f9ac377
commit 8161f2a813
@@ -72,6 +72,8 @@ 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 })),
+        actionReviewProjectionCompanies: vi.fn(async () => []),
        approvedActionReviewCompanies: vi.fn(async () => []),
        sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
        sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })),
@@ -1,3 +1,6 @@
+import { toolReviewProjectionService } from "../services/tool-review-projection.js";
+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 { deleteCompany } from "../services/company-deletion.js";
@@ -281,6 +284,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([]);
@@ -352,6 +357,357 @@ 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, companies: gateway.actionReviewProjectionCompanies }, () => !draining).ready;
+      }
+      async function savedOutcome(pending: Awaited<ReturnType<typeof request>>, source = db,
+        status: "executed" | "failed" | "expired" | "cancelled" = "executed") {
+        await writeToolReviewChange(source, pending.companyId, 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("keeps projection scans scoped to the company whose review changed", async () => {
+        const other = await createRunFixture(db);
+        await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company));
+        const foreign = await request(gateway, db, other);
+        await gateway.approveActionRequest({ companyId: other.company.id, actionRequestId: foreign.id, actor: { userId: "board" } });
+        await startProjection(); await quiet();
+        const before = await projection(foreign);
+        const reads: Array<{ query: string; params: unknown[] }> = [];
+        const debug = db.$client.options.debug;
+        db.$client.options.debug = (_id, query, params) => {
+          if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce("))
+            reads.push({ query, params: [...params] });
+        };
+        try {
+          for (let index = 0; index < 3; index++) {
+            const pending = await request();
+            await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
+            await settle(1500); expect(await projection(pending)).toMatchObject({ status: "executed" });
+          }
+        } finally { db.$client.options.debug = debug; }
+        expect(reads.length).toBeGreaterThan(0);
+        for (const read of reads) {
+          expect(read.query).toContain('"tool_action_requests"."company_id" =');
+          expect(read.params).toContain(fixture.company.id);
+          expect(read.params).not.toContain(other.company.id);
+        }
+        expect(await projection(foreign)).toEqual(before); await quiet();
+      });
+
+      it("uses one candidate scan for a committed company hint with matching retained history", async () => {
+        const pending = await request();
+        await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
+        await startProjection(); await quiet();
+        const reads: string[] = [];
+        const debug = db.$client.options.debug;
+        db.$client.options.debug = (_id, query) => {
+          if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce(")) reads.push(query);
+        };
+        try {
+          await writeToolReviewChange(db, pending.companyId, async tx => {
+            await tx.update(toolActionRequests).set({ updatedAt: new Date() }).where(eq(toolActionRequests.id, pending.id));
+          });
+          await settle();
+          expect(reads).toHaveLength(1);
+          await quiet();
+          expect(reads).toHaveLength(1);
+          reads.length = 0;
+          expect(await gateway.repairActionReviewProjections(fixture.company.id)).toMatchObject({ scanned: 0, retry: false });
+          expect(reads).toHaveLength(1);
+        } finally { db.$client.options.debug = debug; }
+      });
+
+      it("backs off unresolved cards to five minutes and clears the retry after settlement", async () => {
+        const pending = await request();
+        await savedOutcome(pending, db, "failed");
+        await startProjection();
+        for (const delay of [30_000, 60_000, 120_000, 240_000, 300_000, 300_000]) {
+          const next = coordinator!.nextWakeAt();
+          expect(next).not.toBeNull();
+          expect(next! - Date.now()).toBeGreaterThan(delay - 1500);
+          expect(next! - Date.now()).toBeLessThanOrEqual(delay);
+          await settle(next! - Date.now() + 100);
+        }
+        expect(await projection(pending)).toBeUndefined();
+        await acceptedCard(pending);
+        await writeToolReviewChange(db, pending.companyId, async tx => {
+          await tx.update(toolActionRequests).set({ updatedAt: new Date() }).where(eq(toolActionRequests.id, pending.id));
+        });
+        await settle();
+        expect(await projection(pending)).toMatchObject({ status: "failed" });
+        await quiet();
+      });
+
+      it("does not wake projection repair for ordinary confirmation creation, replacement, or cancellation", async () => {
+        await startProjection(); await quiet();
+        const interactions = issueThreadInteractionService(db);
+        const scope = { id: fixture.issue.id, companyId: fixture.company.id };
+        const [recipient] = await db.insert(agents).values({ companyId: fixture.company.id,
+          name: "Plan recipient", role: "engineer", adapterType: "process" }).returning();
+        const input = { kind: "request_confirmation" as const, addresseeAgentId: recipient.id,
+          payload: { version: 1 as const, prompt: "Approve this plan?" } };
+        const signal = vi.fn();
+        const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.toolReviewProjection, signal);
+        const reads: string[] = [];
+        const debug = db.$client.options.debug;
+        db.$client.options.debug = (_id, query) => {
+          if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce(")) reads.push(query);
+        };
+        try {
+          const first = await interactions.create(scope, input, { agentId: fixture.agent.id });
+          const second = await interactions.create(scope, input, { agentId: fixture.agent.id });
+          const cancelled = await interactions.cancelPendingForDeletedAddressee(fixture.company.id, recipient.id);
+          expect(cancelled.map(row => row.id)).toEqual([second.id]);
+          expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, first.id)))[0].status).toBe("expired");
+          await settle();
+          expect(signal).not.toHaveBeenCalled();
+          expect(reads).toEqual([]);
+          await quiet();
+        } finally { unsubscribe(); db.$client.options.debug = debug; }
+      });
+
+      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("backs off a failing company while another company's projection completes", async () => {
+        const other = await createRunFixture(db);
+        await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company));
+        const foreign = await request(gateway, db, other); await acceptedCard(foreign); await savedOutcome(foreign, db, "failed");
+        await db.execute(sql`create function sleep_projection_fail() returns trigger language plpgsql as $$
+          begin raise exception 'company 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' = 'failed') execute function sleep_projection_fail()`);
+        await startProjection(); expect(errors).toHaveLength(1); errors.length = 0;
+        const retryAt = coordinator!.nextWakeAt(); expect(retryAt).not.toBeNull();
+        const pending = await request(); await acceptedCard(pending); await savedOutcome(pending);
+        await request(gateway, db, other);
+        await settle(5000);
+        expect(await projection(pending)).toMatchObject({ status: "executed" });
+        expect(await projection(foreign)).toBeUndefined();
+        expect(errors).toEqual([]); expect(coordinator!.nextWakeAt()).toBe(retryAt);
+        await db.execute(sql`drop trigger sleep_projection_fail on issue_thread_interactions`);
+        await settle(); await quiet();
+        expect(await projection(foreign)).toMatchObject({ status: "failed" });
+        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();
+      });
+
+      async function projectionBacklog(count: number) {
+        const pending = await request(); await acceptedCard(pending); await savedOutcome(pending, db, "failed");
+        const [template] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        const cards = Array.from({ length: count - 1 }, () => ({ ...template, id: randomUUID(), idempotencyKey: randomUUID() }));
+        if (cards.length) await db.insert(issueThreadInteractions).values(cards);
+        const requests = cards.map((card, index) => ({ ...pending,
+          id: `ffffffff-ffff-4fff-afff-${String(index).padStart(12, "0")}`,
+          interactionId: card.id, status: "failed" as const }));
+        if (requests.length) await db.insert(toolActionRequests).values(requests);
+        const cardByRequest = new Map([pending, ...requests].map(row => [row.id, row.interactionId!]));
+        const reflect = async (id: string) => {
+          await db.update(issueThreadInteractions).set({ result: sql`${issueThreadInteractions.result}
+            || ${JSON.stringify({ toolAction: { status: "failed" } })}::jsonb` })
+            .where(eq(issueThreadInteractions.id, cardByRequest.get(id)!));
+        };
+        return { cardByRequest, reflect };
+      }
+
+      it("bounds a projection worker pass and continues its saved backlog", async () => {
+        await projectionBacklog(501);
+        await startProjection();
+        const projected = async () => (await db.select().from(issueThreadInteractions))
+          .filter(card => (card.result as { toolAction?: { status: string } } | null)?.toolAction?.status === "failed").length;
+        expect(await projected()).toBe(500);
+        expect(coordinator!.nextWakeAt()).not.toBeNull();
+        await settle(2000); await quiet();
+        expect(await projected()).toBe(501);
+        expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]);
+      });
+
+      it("clears an abandoned projection cursor after empty reconstruction before the next write", async () => {
+        const backlog = await projectionBacklog(3);
+        const service = toolReviewProjectionService(db, input => backlog.reflect(input.actionRequestId));
+        expect(await service.sweep(fixture.company.id, { maxRows: 1 })).toMatchObject({ scanned: 1, truncated: true });
+        for (const id of backlog.cardByRequest.keys()) await backlog.reflect(id);
+        expect(await service.companies(fixture.company.id)).toEqual([]);
+        const firstId = [...backlog.cardByRequest.keys()].sort()[0]!;
+        await db.update(issueThreadInteractions).set({ result: sql`${issueThreadInteractions.result} - 'toolAction'` })
+          .where(eq(issueThreadInteractions.id, backlog.cardByRequest.get(firstId)!));
+        expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 1, retry: false });
+      });
+
+      it.each(["abort", "drain"] as const)("stops projection between rows on %s and resumes without losing cards", async mode => {
+        const backlog = await projectionBacklog(5), controller = new AbortController();
+        const seen: string[] = []; let admitted = true;
+        const service = toolReviewProjectionService(db, async input => {
+          seen.push(input.actionRequestId); await backlog.reflect(input.actionRequestId);
+          if (seen.length === 1) { if (mode === "abort") controller.abort(); else admitted = false; }
+        });
+        expect(await service.sweep(fixture.company.id, { maxRows: 500, signal: controller.signal, shouldContinue: () => admitted }))
+          .toMatchObject({ scanned: 1, retry: true, truncated: true });
+        expect(seen).toHaveLength(1);
+        admitted = true;
+        expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 4, retry: false });
+        expect(new Set(seen).size).toBe(5);
+      });
+
+      it.each(["unprojectable", "failed"] as const)("continues beyond a full %s page before revisiting it", async mode => {
+        const backlog = await projectionBacklog(501);
+        const blocked = new Set([...backlog.cardByRequest.keys()].sort().slice(0, 500)), seen: string[] = [];
+        const service = toolReviewProjectionService(db, async input => {
+          seen.push(input.actionRequestId);
+          if (blocked.has(input.actionRequestId)) { if (mode === "failed") throw new Error("saved card unavailable"); return; }
+          await backlog.reflect(input.actionRequestId);
+        });
+        const first = service.sweep(fixture.company.id, { maxRows: 500 });
+        if (mode === "failed") await expect(first).rejects.toBeInstanceOf(AggregateError);
+        else expect(await first).toMatchObject({ scanned: 500, retry: true, truncated: true });
+        expect(seen).toHaveLength(500); seen.length = 0;
+        expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 1, retry: true, truncated: false });
+        expect(seen).toHaveLength(1); expect(blocked.has(seen[0]!)).toBe(false);
+      });
+
+      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.each([false, true])("reactivates when an administrative card mutation overwrites a settled projection (legacy %s)", async legacy => {
+        const pending = await request();
+        if (legacy) {
+          const [card] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!));
+          const payload = { ...card.payload } as Record<string, unknown>;
+          delete payload.toolAction; delete payload.target;
+          await db.update(issueThreadInteractions).set({ payload }).where(eq(issueThreadInteractions.id, pending.interactionId!));
+        }
+        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,
@@ -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";
@@ -1486,7 +1487,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, companies: app.locals.toolGateway.actionReviewProjectionCompanies },
+        () => !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,
@@ -1917,8 +1921,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
@@ -9,6 +9,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",
@@ -14,7 +14,7 @@ type SelectRow = Record<string, unknown>;
 function createSelectChain(rows: SelectRow[] | ((table: unknown) => SelectRow[])) {
  return {
    from(table: unknown) {
-      const selectedRows = typeof rows === "function" ? rows(table) : rows;
+      let selectedRows = typeof rows === "function" ? rows(table) : rows;
      const query = {
        innerJoin() {
          return query;
@@ -25,6 +25,10 @@ function createSelectChain(rows: SelectRow[] | ((table: unknown) => SelectRow[])
        for() {
          return query;
        },
+        limit(count: number) {
+          selectedRows = selectedRows.slice(0, count);
+          return query;
+        },
        then(callback: (rows: SelectRow[]) => unknown) {
          return Promise.resolve(callback(selectedRows));
        },
@@ -292,7 +296,7 @@ describe("issueThreadInteractionService", () => {
    ]);
  });

-  it("withdraws a pending interaction with attribution and rejects repeats", async () => {
+  it("withdraws an unlinked pending interaction with attribution and rejects repeats", async () => {
    const { issueThreadInteractionService } = await import("./issue-thread-interactions.js");
    const interactionRow = {
      id: "interaction-withdraw", companyId: "company-1", issueId: "11111111-1111-4111-8111-111111111111",
@@ -308,8 +312,7 @@ describe("issueThreadInteractionService", () => {
    expect(withdrawn.status).toBe("cancelled");
    expect(withdrawn.result).toEqual({ version: 1, outcome: "withdrawn", reason: "Replanning" });
    expect(withdrawn.resolvedByAgentId).toBe("agent-1");
-    expect(state.toolActionRequestUpdates).toHaveLength(1);
-    expect(state.toolActionRequestUpdates[0]).toMatchObject({ status: "cancelled", resolvedByAgentId: "agent-1" });
+    expect(state.toolActionRequestUpdates).toHaveLength(0);
    const resolvedState = createFakeDb({ interactionRow: { ...interactionRow, status: "accepted" } });
    const resolvedSvc = issueThreadInteractionService(resolvedState.db as never);
    await expect(resolvedSvc.withdrawInteraction(
@@ -388,7 +391,9 @@ describe("issueThreadInteractionService", () => {
      },
      result: null, resolvedAt: null, createdAt: new Date("2026-07-25T10:00:00.000Z"), updatedAt: new Date("2026-07-25T10:00:00.000Z"),
    };
-    const state = createFakeDb({ interactionRow });
+    const state = createFakeDb({ interactionRow, parentRows: [{
+      id: "33333333-3333-4333-8333-333333333333", companyId: "company-1", interactionId: interactionRow.id,
+    }] });
    const svc = issueThreadInteractionService(state.db as never);
    const expired = await svc.expirePendingInteractionsForTerminalIssue(
      { id: interactionRow.issueId, companyId: "company-1", status: "cancelled" },
@@ -1073,8 +1073,22 @@ class InteractionResolvedConcurrentlyError extends Error {
 // tool_action_requests row. Administrative resolutions (withdraw, terminal-issue
 // expiry) must settle that row too, or the parked call stays approvable under
 // its own one-hour lifecycle after its card is gone.
+async function hasLinkedToolActionRequests(db: Pick<Db, "select">, companyId: string, interactionIds: string[]) {
+  if (interactionIds.length === 0) return false;
+  const [request] = await db.select({ id: toolActionRequests.id }).from(toolActionRequests)
+    .where(and(eq(toolActionRequests.companyId, companyId), inArray(toolActionRequests.interactionId, interactionIds)))
+    .limit(1);
+  return request !== undefined;
+}
+
+async function notifyToolProjectionForInteractions(db: Pick<Db, "select">, companyId: string, interactionIds: string[]) {
+  // The persisted link also covers legacy cards without current payload flags.
+  if (await hasLinkedToolActionRequests(db, companyId, interactionIds))
+    await notifyDeliveryWork(db, DELIVERY_QUEUES.toolReviewProjection, companyId);
+}
+
 async function resolveLinkedToolActionRequests(
-  db: Pick<Db, "update">,
+  db: Pick<Db, "select" | "update">,
  interaction: Pick<IssueThreadInteractionRow, "id" | "companyId" | "kind">,
  outcome: {
    status: "expired" | "cancelled";
@@ -1083,7 +1097,8 @@ async function resolveLinkedToolActionRequests(
    now: Date;
  },
 ) {
-  if (interaction.kind !== "request_confirmation") return;
+  if (interaction.kind !== "request_confirmation" ||
+    !(await hasLinkedToolActionRequests(db, interaction.companyId, [interaction.id]))) return;
  await notifyToolReviewChange(db, interaction.companyId);
  await db
    .update(toolActionRequests)
@@ -2240,6 +2255,7 @@ export function issueThreadInteractionService(
            })
          : undefined;

+      if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]);
      const [updated] = await tx
        .update(issueThreadInteractions)
        .set({
@@ -2517,6 +2533,7 @@ export function issueThreadInteractionService(
        now,
      });

+      if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]);
      const [resolved] = await tx
        .update(issueThreadInteractions)
        .set({
@@ -3273,6 +3290,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 notifyToolProjectionForInteractions(tx, row.companyId, [row.id]);
          const [resolved] = await tx
            .update(issueThreadInteractions)
            .set({
@@ -3357,6 +3375,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 notifyToolProjectionForInteractions(tx, row.companyId, [row.id]);
          const [updatedRow] = await tx
            .update(issueThreadInteractions)
            .set({
@@ -3776,6 +3795,19 @@ export function issueThreadInteractionService(
            data.kind === "ask_user_questions"
              ? buildSupersededByNewerInteractionResult(row.id)
              : buildSupersededByNewerRequestResult(row.id);
+          const siblingCondition = and(
+            eq(issueThreadInteractions.companyId, issue.companyId),
+            eq(issueThreadInteractions.issueId, issue.id),
+            eq(issueThreadInteractions.kind, data.kind),
+            eq(issueThreadInteractions.createdByAgentId, actor.agentId),
+            eq(issueThreadInteractions.status, "pending"),
+            ne(issueThreadInteractions.id, row.id),
+          );
+          if (data.kind === "request_confirmation") {
+            // The issue lock above excludes concurrent interaction creation.
+            const siblings = await tx.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(siblingCondition);
+            await notifyToolProjectionForInteractions(tx, issue.companyId, siblings.map(sibling => sibling.id));
+          }
          const supersededRows = await tx
            .update(issueThreadInteractions)
            .set({
@@ -3786,17 +3818,7 @@ export function issueThreadInteractionService(
              resolvedAt: now,
              updatedAt: now,
            })
-            .where(
-              and(
-                eq(issueThreadInteractions.companyId, issue.companyId),
-                eq(issueThreadInteractions.issueId, issue.id),
-                eq(issueThreadInteractions.kind, data.kind),
-                eq(issueThreadInteractions.createdByAgentId, actor.agentId),
-                eq(issueThreadInteractions.status, "pending"),
-                ne(issueThreadInteractions.id, row.id),
-
-              ),
-            )
+            .where(siblingCondition)
            .returning();
          for (const supersededRow of supersededRows) {
            await resolveLinkedToolActionRequests(tx, supersededRow, {
@@ -4442,6 +4464,7 @@ export function issueThreadInteractionService(
              .delete(toolOauthStates)
              .where(eq(toolOauthStates.interactionId, row.id));
          }
+          if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]);
          const [resolved] = await tx
            .update(issueThreadInteractions)
            .set({
@@ -4610,6 +4633,7 @@ export function issueThreadInteractionService(
          );
          if (!sampleConfirmationRow) continue;
          const updatedInteractions = await db.transaction(async (tx) => {
+            await notifyToolProjectionForInteractions(tx, issue.companyId, confirmationRowIds);
            const updatedRows = await tx
              .update(issueThreadInteractions)
              .set({
@@ -1,3 +1,4 @@
+import { toolReviewProjectionService } from "./tool-review-projection.js";
 import { toolReviewInterruptionService } from "./tool-review-interruption.js";
 import { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, persistToolTestOutcome, startToolInvocation, type ToolApprovalClaim } from "./tool-approval-execution.js";
 import { approvedToolReviewCompanies, prepareApprovedToolReview, recoverApprovedToolReviews } from "./tool-review-recovery.js";
@@ -2398,6 +2399,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
@@ -8125,10 +8131,10 @@ export function createToolGatewayService(
      return { projectId: issue.projectId };
    }
    const expiredAt = new Date();
-    await db
+    await writeToolReviewChange(db, claimed.companyId, 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",
@@ -8820,6 +8826,7 @@ export function createToolGatewayService(
    };
  }

+  const reviewProjection = toolReviewProjectionService(db, reflectToolActionInteractionLifecycle);
  const reviewInterruption = toolReviewInterruptionService(db, async settled => {
    void emitConnectionInvoked(db, settled.invocationId);
    await reflectToolActionInteractionLifecycle(settled);
@@ -9917,6 +9924,8 @@ export function createToolGatewayService(
      return buildTestCallStatus(actionRequest, invocation);
    },

+    repairActionReviewProjections: reviewProjection.sweep,
+    actionReviewProjectionCompanies: reviewProjection.companies,
    settleInterruptedActionReviews: reviewInterruption.sweep,
    nextActionReviewInterruptionAt: reviewInterruption.next,
    sweepExpiredActionReviews: reviewExpiry.sweep,
@@ -9951,40 +9960,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;
@@ -0,0 +1,33 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
+import type { toolReviewProjectionService } from "./tool-review-projection.js";
+
+export function registerToolReviewProjectionWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof toolReviewProjectionService>,
+  canRun: () => boolean,
+) {
+  const unresolvedPasses = new Map<string, number>();
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.toolReviewProjection,
+    batchSize: 1,
+    decodeScope: hint => hint === "*" ? undefined : hint,
+    matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId,
+    load: async (companyId?: string) => {
+      const rows = companyId ? [{ companyId }] : await service.companies();
+      for (const scope of unresolvedPasses.keys()) {
+        if (!companyId || scope === companyId) unresolvedPasses.delete(scope);
+      }
+      return rows.map(row => ({ ...row, id: row.companyId, dueAt: Date.now() }));
+    },
+    check: async (entry, signal) => {
+      const result = await service.sweep(entry.companyId, { maxRows: 500, signal, shouldContinue: canRun });
+      if (!result.retry) { unresolvedPasses.delete(entry.id); return null; }
+      if (result.truncated) return { ...entry, dueAt: Date.now() + 1000 };
+      const count = Math.min((unresolvedPasses.get(entry.id) ?? 0) + 1, 5);
+      unresolvedPasses.set(entry.id, count);
+      return { ...entry, dueAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) };
+    },
+  }, canRun);
+}
@@ -0,0 +1,77 @@
+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 = (companyId?: string, 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, companyId ? eq(toolActionRequests.companyId, companyId) : undefined,
+      cursor ? gt(toolActionRequests.id, cursor) : undefined))
+    .orderBy(asc(toolActionRequests.id));
+
+  const continuations = new Map<string, string>();
+  return {
+    async companies(companyId?: string) {
+      const rows = await db.select({ companyId: toolActionRequests.companyId }).from(toolActionRequests)
+        .innerJoin(issueThreadInteractions, eq(issueThreadInteractions.id, toolActionRequests.interactionId))
+        .innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId))
+        .where(and(mismatch, companyId ? eq(toolActionRequests.companyId, companyId) : undefined))
+        .groupBy(toolActionRequests.companyId);
+      const pending = new Set(rows.map(row => row.companyId));
+      for (const scope of continuations.keys()) {
+        if (companyId && scope !== companyId) continue;
+        if (scope === "*" ? rows.length === 0 : !pending.has(scope)) continuations.delete(scope);
+      }
+      return rows;
+    },
+    async sweep(companyId?: string, options: {
+      maxRows?: number;
+      signal?: AbortSignal;
+      shouldContinue?: () => boolean;
+    } = {}) {
+      const maxRows = options.maxRows ?? Infinity;
+      const scope = companyId ?? "*";
+      let cursor = maxRows === Infinity ? undefined : continuations.get(scope);
+      const resumed = cursor !== undefined;
+      let scanned = 0, completed = false;
+      const canContinue = () => !options.signal?.aborted && (options.shouldContinue?.() ?? true);
+      const failures: unknown[] = [];
+      while (scanned < maxRows && canContinue()) {
+        const pageSize = Math.min(100, maxRows - scanned);
+        const rows = await candidates(companyId, cursor).limit(pageSize);
+        for (const { request, invocation } of rows) {
+          if (!canContinue()) break;
+          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++;
+          cursor = request.id;
+        }
+        if (!canContinue()) break;
+        if (rows.length < pageSize) { completed = true; break; }
+      }
+      if (maxRows !== Infinity) {
+        if (completed) continuations.delete(scope);
+        else if (cursor) continuations.set(scope, cursor);
+      }
+      if (failures.length) throw new AggregateError(failures, "Tool review projection repair failed");
+      return { scanned, truncated: !completed,
+        retry: !completed || ((scanned > 0 || resumed) && (await candidates(companyId).limit(1)).length > 0) };
+    },
+  };
+}
@@ -5,6 +5,7 @@ export async function notifyToolReviewChange(transaction: object, companyId: str
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId);
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery, companyId);
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewInterruption, companyId);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewProjection, companyId);
 }

 export function writeToolReviewChange<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>): Promise<T> {