feat(server): stage scoped GitHub review check scheduling

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

doc/plans/2026-10-25-github-review-check-sleep-audit.md

@@ -0,0 +1,99 @@
+# GitHub review-check scheduling audit
+
+Baseline: `3d7c79c262f92563555b278d101de8240e6f011b`, parent
+[#17661](https://github.com/paperclipai/paperclip/pull/17661).
+Parent acceptance: 1,571 tests across complete chat, publication, and startup
+suites passed in 517.41 seconds. All 53 remote checks passed on that commit;
+Greptile reviewed the same head at 5/5 with no findings. The parent is unmerged.
+
+## Scope and owner
+
+This slice owns publication of the GitHub Review check derived from ordinary
+chat admission, review binding, run state, and assessment ownership. Existing
+`chat_actions.kind = github_review_check` rows remain durable truth. No new
+queue table is needed. GitHub state interpretation belongs under
+`chat-channels/github/`; the shared coordinator receives identifiers/deadlines.
+
+Late stale-head projection repair is a separate duty. The manual combined
+`processPending` API remains compatible. The legacy maintenance callback now
+calls only `repairSupersededReviews`; removing that scan is the next slice.
+Assessment publication remains owned by its existing governed outbox, including
+its authorization, score, delivery receipts, and recovery guarantees.
+
+## Actual external obligation
+
+The old dispatcher checks the current remote PR head before publishing. A saved
+nonterminal check receives a five-second `retryAt` even when its local state has
+not changed. The current webhook parser accepts opened, synchronize, reopened,
+and ready-for-review events, but receipt is not a proof that all future pushes
+will arrive. A manual review can also exist without a new automatic admission.
+This slice must not turn a missed webhook into a permanently active obsolete
+check or authorize publication against an old head.
+
+Retain the existing five-second remote-head reconciliation deadline only while
+that particular check is received/nonterminal. Local state changes can make the
+entry due immediately; a successful unchanged inspection does not rewrite the
+provider check. Failure uses its saved exponential backoff and existing eight-
+attempt exhaustion policy. Processed, cancelled, and exhausted history have no
+worker deadline. This is not a claim that an active review needs no SQL/network
+traffic: unresolved checks retain external reconciliation until they settle.
+The startup global scan and committed scope rebuilds enumerate unfinished rows;
+deadline dispatch rechecks only known entry IDs. No configuration-only timer.
+
+## Producer and lifecycle boundary
+
+- `githubReviewCheckService.enqueue`: automatic webhook and authorized manual
+  creation use one awaited notification before insertion, inside a database
+  transaction. Caller-owned/nested transaction settlement stays with `createDb`.
+- `retainGitHubReviewReceipt`: task admission signals its repository/PR before
+  inserting the queued review in the existing caller-owned transaction.
+- `reviewForHead`: manual review insertion and binding an existing review to a
+  run signal inside their short database transactions. Provider calls remain
+  outside those transactions.
+- Existing run/lineage and endpoint-configuration notification ownership fans
+  out to the new topic by company. These hints conservatively rebuild that
+  company's outstanding checks; they do not inspect completed history. Narrow
+  repository/PR hints are used by check enqueue, review writes, and lease release.
+- Assessment and publication receipts are serialized by the existing GitHub
+  publication lease. Its transactional release wakes the affected repository/PR.
+  A competing holder's saved expiry is a deadline if it dies without release.
+- Delivery admission failures retain discovery through the outstanding check's
+  remote reconciliation deadline. This does not claim every shared delivery
+  mutation now emits a check-specific immediate signal.
+- Restart reconstructs existing received/retryable-failed rows. Shared scheduling
+  retains dirty scopes across drain/standby. HTTP admission is checked before each
+  provider request; receipt fencing remains with the publication lease.
+- Supported writers are the existing process-local database ownership boundary.
+  Independent SQL writers and other replicas do not receive a new signal protocol.
+
+Current provider-state reads preserve the prior latest-review/run selection,
+legacy delivery provenance, current configuration checks, and completed-check
+replacement behavior. Introducing a scheduler does not authorize changing those
+product decisions or passing GitHub payloads into generic scheduling machinery.
+
+## Evidence ledger
+
+- Provider-state extraction: server typecheck initially found widened returned
+  state literals. Explicit domain-state typing repaired the extraction.
+- Three existing check/late-projection/manual-review acceptance cases passed after
+  extraction. This was a targeted selection, not all GitHub coverage.
+- First worker selection: 12 passed, one retry test failed. The fixture installed
+  fake timers after a real 60-second timer had been armed, so fake advancement did
+  not deliver that timer. Move fake timer installation before registration; keep
+  the production deadline unchanged. Corrected selection pending.
+
+Remaining acceptance: immediate producer transitions, scope/history pagination,
+provider failure and drain boundaries, startup wiring, migration/index evidence,
+full relevant acceptance, workspace checks, and remote PR review/CI. No deployed
+host sleep, Cloud alarm coverage expansion, or fleet savings are claimed.
+
+The corrected first selection passed all 13 cases. Expanded coverage added real
+run-identity commit/rollback, a 101-row revoked prefix, and shutdown during HTTP.
+It exposed a real stop-admission error: the HTTP boundary wrapped the stop as a
+provider error and spent a retry attempt. Check admission in the catch boundary
+as well; leave the durable action for resume. That regression then passed. The
+next run passed 15 cases and failed the history fixture: fixed UUIDs collided
+with intentionally retained rows from earlier tests. Use a per-fixture UUID
+prefix while retaining deterministic before-candidate ordering. No database was
+dropped and no production timeout was enlarged. Workspace typecheck passed
+before the final shutdown guard change; final acceptance remains pending.

packages/db/src/schema/chat_channels.ts

@@ -679,6 +679,8 @@ export const chatActions = pgTable(
       table.endpointId,
       table.providerActionId,
     ),
+    index("chat_actions_github_check_work_idx").on(table.id).where(sql`${table.kind} = 'github_review_check'
+      and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`),
     index("chat_actions_github_terminal_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment'
       and ${table.status} = 'processed' and ${table.deliveryId} is not null
       and coalesce(${table.result}->>'final', 'false') <> 'true'`),

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -99,6 +99,8 @@ describe("createChatReconciliationCoordinator", () => {
     expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()");
     expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()");
     expect(source).toContain("chatChannels.registerGitHubTerminalResponseWork(deliveryWork,");
+    expect(source).toContain("chatChannels.registerGitHubReviewCheckWork(deliveryWork,");
+    expect(github).not.toContain("githubReviewCheckService(db, fetchImpl).processPending()");
     expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,");
     expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
   });

server/src/__tests__/chat-channels.integration.test.ts

@@ -6612,6 +6612,232 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         runId: run.id,
       }, "read_pull_request", { section: "metadata" })).rejects.toThrow("older pull request head");
     });
+    async function reviewCheckWorkerFixture(status: "running" | "failed" = "failed") {
+      const f = await toolReplyFixture("github:paperclipai/paperclip:418");
+      const config = await f.management.configuration(f.endpoint.id, "owner-user");
+      const event = githubAutomaticReviewEvent({ action: "opened", repository: { id: 97531, full_name: "paperclipai/paperclip" }, sender: { id: 42, login: "octocat" },
+        pull_request: { number: 418, title: "Review", body: "", draft: false, base: { sha: "a".repeat(40), ref: "master" }, head: { sha: "b".repeat(40) }, user: { id: 42, login: "octocat" }, labels: [] } }, randomUUID())!;
+      const [review] = await db.insert(chatGitHubReviews).values({ companyId: f.companyId, endpointId: f.endpoint.id,
+        issueId: f.conversation.issueId!, runId: f.run.id, repositoryId: "97531", repository: event.repository, pullNumber: 418, headSha: event.headSha,
+        deliveryId: event.deliveryId, configurationRevision: config.revision, policySnapshot: config.configuration.defaults, event }).returning();
+      await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.run.id));
+      const writes: Record<string, unknown>[] = [], requests: string[] = [];
+      let remoteHead = event.headSha, unavailable = false;
+      f.setSupplementalProviderFetch(async (input, init) => {
+        const url = String(input);
+        if (url.endsWith("/pulls/418")) { requests.push(url); return Response.json({ head: { sha: remoteHead } }); }
+        if (url.includes("/check-runs?")) { requests.push(url); return Response.json({ check_runs: [] }); }
+        if (url.endsWith("/check-runs") && init?.method === "POST") {
+          requests.push(url);
+          if (unavailable) return Response.json({ message: "injected check outage" }, { status: 503 });
+          writes.push(JSON.parse(String(init.body)));
+          return Response.json({ id: 418, html_url: "https://github.com/checks/418" });
+        }
+        return undefined;
+      });
+      const checks = githubReviewCheckService(db, f.providerFetch);
+      const [endpoint] = await readChatEndpointsById(f.endpoint.id);
+      const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "github_review_check"))))[0];
+      return { ...f, event, review, endpoint, checks, writes, requests, action,
+        enqueue: () => checks.enqueue(endpoint, event, true, "authorized"),
+        setRemoteHead: (value: string) => { remoteHead = value; }, setUnavailable: (value: boolean) => { unavailable = value; } };
+    }
+
+    it("keeps GitHub review check history quiet and wakes from the real enqueue producer", async () => {
+      const f = await reviewCheckWorkerFixture();
+      const errors: unknown[] = [];
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        expect(coordinator.nextWakeAt()).toBeNull();
+        await f.enqueue();
+        await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed");
+        await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+        expect(f.writes).toEqual([expect.objectContaining({ status: "completed" })]);
+        vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+        const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction");
+        const count = f.requests.length;
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled();
+        expect(f.requests).toHaveLength(count);
+        reads.mockRestore(); transactions.mockRestore();
+        expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); vi.useRealTimers(); }
+    });
+
+    it("defers GitHub review check enqueue until outer commit and discards rollback", async () => {
+      const f = await reviewCheckWorkerFixture();
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
+      const enqueue = (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) =>
+        githubReviewCheckService(tx as unknown as Db, f.providerFetch).enqueue(f.endpoint, f.event, true, "authorized");
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        await expect(db.transaction(async tx => { await enqueue(tx); throw new Error("rollback check"); })).rejects.toThrow("rollback check");
+        expect(await f.action()).toBeUndefined(); expect(f.writes).toEqual([]);
+        let entered!: () => void, release!: () => void;
+        const ready = new Promise<void>(resolve => { entered = resolve; }), hold = new Promise<void>(resolve => { release = resolve; });
+        const pending = db.transaction(async tx => { await enqueue(tx); entered(); await hold; });
+        try { await ready; expect(f.writes).toEqual([]); } finally { release(); await pending; }
+        await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed");
+        expect(f.writes).toHaveLength(1);
+      } finally { await coordinator.stop(); }
+    });
+
+    it("recovers a missed GitHub head webhook only while a review check remains unresolved", async () => {
+      const f = await reviewCheckWorkerFixture("running");
+      await f.enqueue();
+      const errors: unknown[] = [];
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        await expect.poll(() => f.writes.length, { timeout: 5000 }).toBe(1);
+        expect(f.writes[0].status).toBe("in_progress");
+        const saved = await f.action();
+        expect(Date.parse(String(saved.result?.retryAt))).toBeGreaterThan(Date.now());
+        f.setRemoteHead("c".repeat(40));
+        await vi.advanceTimersByTimeAsync(6000);
+        await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("cancelled");
+        await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+        expect((await db.select().from(chatGitHubReviews).where(eq(chatGitHubReviews.id, f.review.id)))[0].state).toBe("superseded");
+        const reads = vi.spyOn(db, "select"), count = f.requests.length;
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(f.requests).toHaveLength(count); reads.mockRestore();
+        expect(f.writes).toHaveLength(1); expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); vi.useRealTimers(); }
+    });
+
+    it.each(["restart", "drain", "standby"] as const)("retains GitHub review check work across %s", async mode => {
+      const f = await reviewCheckWorkerFixture();
+      let canRun = true;
+      const errors: unknown[] = [];
+      let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+        canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) });
+      let worker = f.checks.registerCheckWork(coordinator, () => canRun);
+      try {
+        await worker.ready;
+        if (mode === "restart") await coordinator.stop(); else canRun = false;
+        await f.enqueue(); expect(f.writes).toEqual([]);
+        if (mode === "restart") {
+          coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+          worker = f.checks.registerCheckWork(coordinator, () => true); await worker.ready;
+        } else { canRun = true; worker.wake(); }
+        await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed");
+        await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+        expect(f.writes).toHaveLength(1); expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); }
+    });
+
+    it("retains the saved GitHub review check retry through restart without an early provider call", async () => {
+      const f = await reviewCheckWorkerFixture();
+      f.setUnavailable(true); await f.enqueue(); await f.checks.processPending();
+      const failed = await f.action();
+      expect(failed).toMatchObject({ status: "failed", result: { attempts: 1, retryable: true } });
+      const future = new Date(Date.now() + 60_000).toISOString();
+      await db.update(chatActions).set({ result: { ...failed.result, retryAt: future } }).where(eq(chatActions.id, failed.id));
+      f.setUnavailable(false);
+      const errors: unknown[] = [];
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        const count = f.requests.length;
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        expect(coordinator.nextWakeAt()).toBe(Date.parse(future)); expect(f.requests).toHaveLength(count);
+        await vi.advanceTimersByTimeAsync(61_000);
+        await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed");
+        expect(f.writes).toHaveLength(1); expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); vi.useRealTimers(); }
+    });
+
+    it("waits for and wakes from the actual GitHub review check publication lease", async () => {
+      const f = await reviewCheckWorkerFixture();
+      const errors: unknown[] = [];
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      let entered!: () => void, release!: () => void;
+      const ready = new Promise<void>(resolve => { entered = resolve; }), hold = new Promise<void>(resolve => { release = resolve; });
+      let lease: Promise<unknown> | undefined;
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, repositoryId: "97531", number: 418 }, f.providerFetch,
+          async () => { entered(); await hold; });
+        await ready; await f.enqueue();
+        await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).not.toBeNull();
+        expect(f.writes).toEqual([]); release(); await lease;
+        await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed");
+        expect(f.writes).toHaveLength(1); expect(errors).toEqual([]);
+      } finally { release(); await lease; await coordinator.stop(); }
+    });
+
+    it("wakes a GitHub review check from committed run identity before its future head deadline", async () => {
+      const f = await reviewCheckWorkerFixture("running");
+      await f.enqueue(); await f.checks.processPending();
+      const action = await f.action(), future = new Date(Date.now() + 60_000).toISOString();
+      await db.update(chatActions).set({ result: { ...action.result, retryAt: future } }).where(eq(chatActions.id, action.id));
+      const errors: unknown[] = [];
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        expect(coordinator.nextWakeAt()).toBe(Date.parse(future));
+        const change = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
+          await tx.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
+          await initializeRunIdentity(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!,
+            responsibleUserId: "owner-user", messageIds: [String(f.run.contextSnapshot?.wakeCommentId)], cause: "instruction" });
+        };
+        await expect(db.transaction(async tx => { await change(tx); throw new Error("rollback run check"); })).rejects.toThrow("rollback run check");
+        expect(f.writes).toHaveLength(1);
+        await db.transaction(change);
+        await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed");
+        expect(f.writes).toHaveLength(2); expect(f.writes[1].status).toBe("completed");
+        expect(Date.now()).toBeLessThan(Date.parse(future)); expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); }
+    });
+
+    it("finds a GitHub review check after a full page of revoked work and ignores settled history", async () => {
+      const f = await reviewCheckWorkerFixture();
+      await f.enqueue(); const original = await f.action();
+      const pageKey = randomUUID().replaceAll("-", "").slice(0, 9);
+      const prefix = Array.from({ length: 101 }, (_, index) => ({ ...original,
+        id: `00000000-0000-4000-8000-${pageKey}${String(index + 1).padStart(3, "0")}`,
+        providerActionId: `revoked-check-${index}`, payload: { ...original.payload, runtimeGeneration: -1 } }));
+      await db.insert(chatActions).values(prefix);
+      await db.insert(chatActions).values(Array.from({ length: 101 }, (_, index) => ({ ...original,
+        id: randomUUID(), providerActionId: `settled-check-${index}`, status: "processed" })));
+      const errors: unknown[] = [];
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        await expect.poll(async () => (await db.select().from(chatActions).where(and(eq(chatActions.id, original.id))))[0].status,
+          { timeout: 15_000 }).toBe("processed");
+        await expect.poll(() => coordinator.nextWakeAt(), { timeout: 15_000 }).toBeNull();
+        expect(f.writes).toHaveLength(1);
+        const cancelled = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.status, "cancelled")));
+        expect(cancelled).toHaveLength(101); expect(errors).toEqual([]);
+      } finally { await coordinator.stop(); }
+    });
+
+    it("stops GitHub review check HTTP after shutdown admission and recovers its durable action", async () => {
+      const f = await reviewCheckWorkerFixture();
+      await f.enqueue();
+      let entered!: () => void, release!: () => void;
+      const ready = new Promise<void>(resolve => { entered = resolve; }), hold = new Promise<void>(resolve => { release = resolve; });
+      const guardedProvider: typeof fetch = async (input, init) => {
+        if (String(input).endsWith("/pulls/418")) { entered(); await hold; }
+        return f.providerFetch(input, init);
+      };
+      const errors: unknown[] = [];
+      let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      try {
+        const starting = githubReviewCheckService(db, guardedProvider).registerCheckWork(coordinator, () => true).ready;
+        await ready;
+        const stopped = coordinator.stop(); release(); await stopped; await starting;
+        expect(f.writes).toEqual([]); expect((await f.action()).status).toBe("received");
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+        await f.checks.registerCheckWork(coordinator, () => true).ready;
+        await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed");
+        expect(f.writes).toHaveLength(1); expect(errors).toEqual([]);
+      } finally { release(); await coordinator.stop(); }
+    });
+
     it("supersedes a late review projection after its check already rejected the old head", async () => {
       const f = await reviewBotFixture();
       const deliveryId = randomUUID();

server/src/app.ts

@@ -1289,6 +1289,7 @@ export async function createApp(
   chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+  chatChannels.registerGitHubReviewCheckWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());

server/src/services/chat-channels.ts

@@ -1,3 +1,4 @@
+import { githubReviewCheckService } from "./chat-github-checks.js";
 import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js";
 import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js";
 import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js";
@@ -3507,6 +3508,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
   const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
 return {
     registerInteractionContinuationWork,
+    registerGitHubReviewCheckWork: githubReviewCheckService(db, fetchImpl).registerCheckWork,
     registerGitHubTerminalResponseWork: githubResponseCommentService(db, fetchImpl).registerTerminalResponseWork,
     registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork,
     registerTerminalConfirmationWork,

server/src/services/chat-channels/github/maintenance.ts

@@ -12,7 +12,7 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet
     options.scheduleMessageProcessing(async () => {
       try {
         // Provider I/O must not block Slack, Teams, or ordinary message receipts.
-        await githubReviewCheckService(db, fetchImpl).processPending();
+        await githubReviewCheckService(db, fetchImpl).repairSupersededReviews();
       } finally {
         githubMaintenancePending = false;
       }

server/src/services/chat-channels/github/review-check-state.ts

@@ -0,0 +1,133 @@
+import { and, desc, eq, sql } from "drizzle-orm";
+import { chatActions, chatDeliveries, chatEndpoints, chatEndpointResources,
+  chatGitHubConfigurations, chatGitHubReviews, chatMessageLinks, companies,
+  heartbeatRuns, type Db } from "@paperclipai/db";
+import type { GitHubReviewEventContext } from "@paperclipai/shared";
+
+export async function githubReviewCheckState(db: Db, action: typeof chatActions.$inferSelect) {
+  const event = action.payload.event as GitHubReviewEventContext;
+  const [source] = await db
+    .select({
+      endpoint: chatEndpoints,
+      resource: chatEndpointResources,
+      configuration: chatGitHubConfigurations.configuration,
+      companyPrefix: companies.issuePrefix,
+    })
+    .from(chatEndpoints)
+    .innerJoin(companies, eq(companies.id, chatEndpoints.companyId))
+    .innerJoin(
+      chatGitHubConfigurations,
+      eq(chatGitHubConfigurations.endpointId, chatEndpoints.id),
+    )
+    .innerJoin(
+      chatEndpointResources,
+      eq(chatEndpointResources.endpointId, chatEndpoints.id),
+    )
+    .where(
+      and(
+        eq(chatEndpoints.id, action.endpointId),
+        eq(chatEndpoints.companyId, action.companyId),
+        eq(chatEndpointResources.providerResourceId, event.repository),
+      ),
+    );
+  if (
+    !source ||
+    !source.configuration.toolsEnabled ||
+    !["active", "verifying"].includes(source.endpoint.status) ||
+    !source.resource.enabled ||
+    source.resource.availability !== "available" ||
+    String(source.resource.metadata?.providerRepositoryId) !==
+      event.repositoryId ||
+    ((source.endpoint.setup as { runtimeGeneration?: number })
+      .runtimeGeneration ?? 0) !== action.payload.runtimeGeneration
+  ) {
+    return { disposition: "connection_changed" as const };
+  }
+  const [review] = await db
+    .select()
+    .from(chatGitHubReviews)
+    .where(
+      and(
+        eq(chatGitHubReviews.endpointId, action.endpointId),
+        eq(chatGitHubReviews.repositoryId, event.repositoryId),
+        eq(chatGitHubReviews.pullNumber, event.pullNumber),
+        eq(chatGitHubReviews.headSha, event.headSha),
+      ),
+    )
+    .orderBy(desc(chatGitHubReviews.createdAt))
+    .limit(1);
+  if (review?.assessment) return { disposition: "assessment_owns_check" as const, reviewId: review.id };
+  let run = review?.runId
+    ? (
+        await db
+          .select({ id: heartbeatRuns.id, status: heartbeatRuns.status })
+          .from(heartbeatRuns)
+          .where(eq(heartbeatRuns.id, review.runId))
+      )[0]
+    : undefined;
+  if (review && !run) {
+    const [sourceRun] = await db
+      .select({ run: { id: heartbeatRuns.id, status: heartbeatRuns.status } })
+      .from(chatDeliveries)
+      .innerJoin(
+        chatMessageLinks,
+        eq(chatMessageLinks.deliveryId, chatDeliveries.id),
+      )
+      .innerJoin(
+        heartbeatRuns,
+        and(
+          eq(heartbeatRuns.companyId, action.companyId),
+          eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId),
+          sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${chatMessageLinks.commentId}::text`,
+        ),
+      )
+      .where(
+        and(
+          eq(chatDeliveries.endpointId, action.endpointId),
+          sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`,
+        ),
+      )
+      .orderBy(desc(heartbeatRuns.createdAt))
+      .limit(1);
+    run = sourceRun?.run;
+  }
+  const [delivery] = await db
+    .select({ state: chatDeliveries.state })
+    .from(chatDeliveries)
+    .where(
+      and(
+        eq(chatDeliveries.endpointId, action.endpointId),
+        sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`,
+      ),
+    )
+    .limit(1);
+  const admissionEnded =
+    !review &&
+    (delivery?.state === "filtered" || delivery?.state === "failed");
+  const terminal =
+    !!run &&
+    [
+      "succeeded",
+      "failed",
+      "cancelled",
+      "timed_out",
+      "interrupted",
+    ].includes(run.status);
+  const state: (typeof chatGitHubReviews.$inferSelect)["state"] =
+    (!action.payload.admitted && !review) || admissionEnded
+      ? "manual_required"
+      : terminal
+        ? run!.status === "succeeded"
+          ? "incomplete"
+          : "error"
+        : run?.status === "running"
+          ? "running"
+          : "queued";
+  const status =
+    state === "running"
+      ? "in_progress"
+      : state === "queued"
+        ? "queued"
+        : "completed";
+  return { disposition: "publish" as const, source, review, run, terminal, state, status };
+}

server/src/services/chat-channels/github/review-check-work.ts

@@ -0,0 +1,79 @@
+import { and, asc, eq, gt, or, sql } from "drizzle-orm";
+import { chatActions, chatEndpointLeases, type Db } from "@paperclipai/db";
+import type { GitHubReviewEventContext } from "@paperclipai/shared";
+import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js";
+import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js";
+import { githubReviewCheckState } from "./review-check-state.js";
+
+type Scope = { companyId: string; endpointId?: string; repositoryId?: string; number?: number };
+type Entry = Scope & { id: string; dueAt: number };
+
+export async function notifyGitHubReviewChecks(tx: object, scope: Scope) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify(scope));
+}
+
+export function registerGitHubReviewCheckWork(
+  db: Db,
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  dispatch: (id: string, canContinue: () => boolean) => Promise<unknown>,
+  canRun: () => boolean,
+) {
+  async function page(scope?: Scope, cursor?: string, actionId?: string) {
+    return db.select({ action: chatActions,
+      leaseExpiresAt: sql<string | null>`(select max(l.expires_at) from ${chatEndpointLeases} l
+        where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId}
+          and l.lease_key = 'github-publication:' || (${chatActions.payload}->'event'->>'repositoryId')
+            || ':' || (${chatActions.payload}->'event'->>'pullNumber'))`,
+    }).from(chatActions).where(and(eq(chatActions.kind, "github_review_check"),
+      or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)),
+      scope ? eq(chatActions.companyId, scope.companyId) : undefined,
+      scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined,
+      scope?.repositoryId ? sql`${chatActions.payload}->'event'->>'repositoryId' = ${scope.repositoryId}` : undefined,
+      scope?.number ? sql`${chatActions.payload}->'event'->>'pullNumber' = ${String(scope.number)}` : undefined,
+      cursor ? gt(chatActions.id, cursor) : undefined, actionId ? eq(chatActions.id, actionId) : undefined,
+    )).orderBy(asc(chatActions.id)).limit(100);
+  }
+  async function entry(row: Awaited<ReturnType<typeof page>>[number]): Promise<Entry> {
+    const { action } = row;
+    const event = action.payload.event as GitHubReviewEventContext;
+    let dueAt = action.result?.retryAt ? Date.parse(String(action.result.retryAt)) : 0;
+    if (!Number.isFinite(dueAt)) throw new Error("GitHub review check has an invalid retry deadline");
+    if (action.status === "received") {
+      const current = await githubReviewCheckState(db, action);
+      if (current.disposition !== "publish" || current.state !== action.result?.state) dueAt = 0;
+    }
+    if (row.leaseExpiresAt) dueAt = Math.max(dueAt, Date.parse(row.leaseExpiresAt));
+    return { id: action.id, companyId: action.companyId, endpointId: action.endpointId,
+      repositoryId: event.repositoryId, number: event.pullNumber, dueAt };
+  }
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.chatGitHubReviewChecks,
+    decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+    matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId &&
+      (!scope.endpointId || value.endpointId === scope.endpointId) &&
+      (!scope.repositoryId || value.repositoryId === scope.repositoryId) && (!scope.number || value.number === scope.number)),
+    load: async (scope?: Scope, signal?: AbortSignal) => {
+      const entries: Entry[] = [];
+      let cursor: string | undefined;
+      while (canRun() && !signal?.aborted) {
+        const rows = await page(scope, cursor);
+        for (const row of rows) {
+          if (!canRun() || signal?.aborted) throw signal?.reason ?? new Error("GitHub review check recovery stopped");
+          entries.push(await entry(row));
+        }
+        if (rows.length < 100) return entries;
+        cursor = rows[rows.length - 1].action.id;
+      }
+      throw signal?.reason ?? new Error("GitHub review check recovery stopped");
+    },
+    check: async (value: Entry, signal?: AbortSignal) => {
+      const [row] = await page(undefined, undefined, value.id);
+      const current = row ? await entry(row) : null;
+      if (!current || current.dueAt > Date.now() || !canRun() || signal?.aborted) return current;
+      await dispatch(current.id, () => canRun() && !signal?.aborted);
+      const [remaining] = await page(undefined, undefined, value.id);
+      return remaining ? entry(remaining) : null;
+    },
+  }, canRun);
+}

server/src/services/chat-channels/github/task-preparation.ts

@@ -1,3 +1,4 @@
+import { notifyGitHubReviewChecks } from "./review-check-work.js";
 import { agents, chatDeliveries, chatGitHubReviews } from "@paperclipai/db";
 import { and, eq, sql } from "drizzle-orm";
 import { LOW_TRUST_REVIEW_PRESET, type GitHubReviewEventContext, type GitHubIssueEventContext, type GitHubReviewPolicy } from "@paperclipai/shared";
@@ -41,6 +42,8 @@ export async function retainGitHubDeliveryAuthority(
 }
 
 export async function retainGitHubReviewReceipt(taskTx: DbOrTransaction, endpoint: EndpointRow, issueId: string, deliveryId: string, githubAutomatic: AutomaticRequest): Promise<void> {
+          await notifyGitHubReviewChecks(taskTx, { companyId: endpoint.companyId, endpointId: endpoint.id,
+            repositoryId: githubAutomatic.context.repositoryId, number: githubAutomatic.context.pullNumber });
           await taskTx.insert(chatGitHubReviews).values({
             companyId: endpoint.companyId, endpointId: endpoint.id, issueId: issueId,
             repositoryId: githubAutomatic.context.repositoryId, repository: githubAutomatic.context.repository,

server/src/services/chat-github-checks.ts

@@ -1,18 +1,17 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { notifyGitHubReviewChecks, registerGitHubReviewCheckWork } from "./chat-channels/github/review-check-work.js";
+import { githubReviewCheckState } from "./chat-channels/github/review-check-state.js";
 import {
   GitHubPublicationLeaseLost,
   withGitHubPublicationLease,
 } from "./chat-github-publication-lease.js";
-import { and, desc, eq, inArray, isNull, or, sql } from "drizzle-orm";
+import { and, eq, inArray, isNull, or, sql } from "drizzle-orm";
 import {
   chatActions,
   chatDeliveries,
   chatEndpoints,
-  chatEndpointResources,
   chatGitHubConfigurations,
   chatGitHubReviews,
-  chatMessageLinks,
-  companies,
-  heartbeatRuns,
   type Db,
 } from "@paperclipai/db";
 import type { GitHubReviewEventContext } from "@paperclipai/shared";
@@ -41,27 +40,32 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
       ...config.configuration.repositories[event.repositoryId],
     };
     if (policy.ratingThreshold === null && !admitted) return;
-    await db
-      .insert(chatActions)
-      .values({
-        companyId: endpoint.companyId,
-        endpointId: endpoint.id,
-        kind: "github_review_check",
-        providerActionId: `github-check:${event.deliveryId}`,
-        payload: {
-          version: 1,
-          event,
-          admitted,
-          reason,
-          runtimeGeneration:
-            (endpoint.setup as { runtimeGeneration?: number })
-              .runtimeGeneration ?? 0,
-        },
-        status: "received",
-      })
-      .onConflictDoNothing();
+    await db.transaction(async tx => {
+      await notifyGitHubReviewChecks(tx, { companyId: endpoint.companyId, endpointId: endpoint.id,
+        repositoryId: event.repositoryId, number: event.pullNumber });
+      await tx
+        .insert(chatActions)
+        .values({
+          companyId: endpoint.companyId,
+          endpointId: endpoint.id,
+          kind: "github_review_check",
+          providerActionId: `github-check:${event.deliveryId}`,
+          payload: {
+            version: 1,
+            event,
+            admitted,
+            reason,
+            runtimeGeneration:
+              (endpoint.setup as { runtimeGeneration?: number })
+                .runtimeGeneration ?? 0,
+          },
+          status: "received",
+        })
+        .onConflictDoNothing();
+    });
   }
-  async function dispatch(id: string) {
+  async function dispatch(id: string, canContinue = () => true) {
+    if (!canContinue()) return;
     const [action] = await db
       .select()
       .from(chatActions)
@@ -74,6 +78,11 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
       .limit(1);
     if (!action || ["processed", "cancelled"].includes(action.status)) return;
     const event = action.payload.event as GitHubReviewEventContext;
+    const guardedFetch: typeof fetch = (input, init) => {
+      if (!canContinue()) throw new GitHubPublicationLeaseLost();
+      return fetchImpl(input, init);
+    };
+    if (!canContinue()) return;
     await withGitHubPublicationLease(
       db,
       {
@@ -82,7 +91,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
         repositoryId: event.repositoryId,
         number: event.pullNumber,
       },
-      fetchImpl,
+      guardedFetch,
       async (lease) => {
         const [fresh] = await db
           .select()
@@ -91,155 +100,19 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
         if (!fresh || ["processed", "cancelled"].includes(fresh.status)) return;
         let attempts = Number(fresh.result?.attempts ?? 0);
         try {
-          const [source] = await db
-            .select({
-              endpoint: chatEndpoints,
-              resource: chatEndpointResources,
-              configuration: chatGitHubConfigurations.configuration,
-              companyPrefix: companies.issuePrefix,
-            })
-            .from(chatEndpoints)
-            .innerJoin(companies, eq(companies.id, chatEndpoints.companyId))
-            .innerJoin(
-              chatGitHubConfigurations,
-              eq(chatGitHubConfigurations.endpointId, chatEndpoints.id),
-            )
-            .innerJoin(
-              chatEndpointResources,
-              eq(chatEndpointResources.endpointId, chatEndpoints.id),
-            )
-            .where(
-              and(
-                eq(chatEndpoints.id, action.endpointId),
-                eq(chatEndpoints.companyId, action.companyId),
-                eq(chatEndpointResources.providerResourceId, event.repository),
-              ),
-            );
-          if (
-            !source ||
-            !source.configuration.toolsEnabled ||
-            !["active", "verifying"].includes(source.endpoint.status) ||
-            !source.resource.enabled ||
-            source.resource.availability !== "available" ||
-            String(source.resource.metadata?.providerRepositoryId) !==
-              event.repositoryId ||
-            ((source.endpoint.setup as { runtimeGeneration?: number })
-              .runtimeGeneration ?? 0) !== action.payload.runtimeGeneration
-          ) {
-            await lease.commit(async (tx) => {
-              await tx
-                .update(chatActions)
-                .set({
-                  status: "cancelled",
-                  result: { code: "connection_changed" },
-                  updatedAt: new Date(),
-                })
-                .where(eq(chatActions.id, id));
+          const currentState = await githubReviewCheckState(db, fresh);
+          if (currentState.disposition !== "publish") {
+            await lease.commit(async tx => {
+              await tx.update(chatActions).set({
+                status: currentState.disposition === "connection_changed" ? "cancelled" : "processed",
+                result: { code: currentState.disposition,
+                  ...(currentState.disposition === "assessment_owns_check" ? { reviewId: currentState.reviewId } : {}) },
+                updatedAt: new Date(),
+              }).where(eq(chatActions.id, id));
             });
             return;
           }
-          const [review] = await db
-            .select()
-            .from(chatGitHubReviews)
-            .where(
-              and(
-                eq(chatGitHubReviews.endpointId, action.endpointId),
-                eq(chatGitHubReviews.repositoryId, event.repositoryId),
-                eq(chatGitHubReviews.pullNumber, event.pullNumber),
-                eq(chatGitHubReviews.headSha, event.headSha),
-              ),
-            )
-            .orderBy(desc(chatGitHubReviews.createdAt))
-            .limit(1);
-          if (review?.assessment) {
-            // The assessment outbox owns the terminal score and its retries.
-            await lease.commit(async (tx) => {
-              await tx
-                .update(chatActions)
-                .set({
-                  status: "processed",
-                  result: {
-                    code: "assessment_owns_check",
-                    reviewId: review.id,
-                  },
-                  updatedAt: new Date(),
-                })
-                .where(eq(chatActions.id, id));
-            });
-            return;
-          }
-          let run = review?.runId
-            ? (
-                await db
-                  .select({ id: heartbeatRuns.id, status: heartbeatRuns.status })
-                  .from(heartbeatRuns)
-                  .where(eq(heartbeatRuns.id, review.runId))
-              )[0]
-            : undefined;
-          if (review && !run) {
-            const [sourceRun] = await db
-              .select({ run: { id: heartbeatRuns.id, status: heartbeatRuns.status } })
-              .from(chatDeliveries)
-              .innerJoin(
-                chatMessageLinks,
-                eq(chatMessageLinks.deliveryId, chatDeliveries.id),
-              )
-              .innerJoin(
-                heartbeatRuns,
-                and(
-                  eq(heartbeatRuns.companyId, action.companyId),
-                  eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId),
-                  sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${chatMessageLinks.commentId}::text`,
-                ),
-              )
-              .where(
-                and(
-                  eq(chatDeliveries.endpointId, action.endpointId),
-                  sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`,
-                ),
-              )
-              .orderBy(desc(heartbeatRuns.createdAt))
-              .limit(1);
-            run = sourceRun?.run;
-          }
-          const [delivery] = await db
-            .select({ state: chatDeliveries.state })
-            .from(chatDeliveries)
-            .where(
-              and(
-                eq(chatDeliveries.endpointId, action.endpointId),
-                sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`,
-              ),
-            )
-            .limit(1);
-          const admissionEnded =
-            !review &&
-            (delivery?.state === "filtered" || delivery?.state === "failed");
-          const terminal =
-            !!run &&
-            [
-              "succeeded",
-              "failed",
-              "cancelled",
-              "timed_out",
-              "interrupted",
-            ].includes(run.status);
-          const state =
-            (!action.payload.admitted && !review) || admissionEnded
-              ? "manual_required"
-              : terminal
-                ? run!.status === "succeeded"
-                  ? "incomplete"
-                  : "error"
-                : run?.status === "running"
-                  ? "running"
-                  : "queued";
-          const status =
-            state === "running"
-              ? "in_progress"
-              : state === "queued"
-                ? "queued"
-                : "completed";
+          const { source, review, run, terminal, state, status } = currentState;
           const policy = review?.policySnapshot ?? {
             ...source.configuration.defaults,
             ...source.configuration.repositories[event.repositoryId],
@@ -389,7 +262,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
               .where(eq(chatActions.id, id));
           });
         } catch (error) {
-          if (error instanceof GitHubPublicationLeaseLost) return;
+          if (error instanceof GitHubPublicationLeaseLost || !canContinue()) return;
           attempts++;
           await lease.commit(async (tx) => {
             await tx
@@ -412,7 +285,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
       },
     );
   }
-  async function processPending(limit = 10) {
+  async function repairSupersededReviews() {
     // Check delivery and task admission run independently. A rapid push can
     // cancel an obsolete check before admission inserts its review projection.
     // Reconcile that late row from the durable cancellation receipt, without
@@ -442,6 +315,9 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
       )`,
         ),
       );
+  }
+  async function processPending(limit = 10) {
+    await repairSupersededReviews();
     const rows = await db
       .select({ id: chatActions.id })
       .from(chatActions)
@@ -465,5 +341,8 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) {
       .limit(limit);
     for (const row of rows) await dispatch(row.id);
   }
-  return { enqueue, processPending };
+  return { enqueue, processPending, repairSupersededReviews,
+    registerCheckWork: (coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) =>
+      registerGitHubReviewCheckWork(db, coordinator, dispatch, canRun),
+  };
 }

server/src/services/chat-github-publication-lease.ts

@@ -115,6 +115,7 @@ export async function withGitHubPublicationLease<T>(
     clearInterval(timer);
     await renewal;
     await db.transaction(async tx => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify(scope));
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify(scope));
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope));

server/src/services/chat-github-reviews.ts

@@ -1,3 +1,4 @@
+import { notifyGitHubReviewChecks } from "./chat-channels/github/review-check-work.js";
 import { notifyChatResponseAvailability } from "./chat-response-availability.js";
 import { githubRunReplyState } from "./chat-run-publications.js";
 import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js";
@@ -350,16 +351,20 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
       .limit(1);
     if (prior && prior.runId === source.run.id) return prior;
     if (prior && !prior.runId && prior.deliveryId === source.delivery.id) {
-      const [claimed] = await db
-        .update(chatGitHubReviews)
-        .set({ runId: source.run.id, state: "running", updatedAt: new Date() })
-        .where(
-          and(
-            eq(chatGitHubReviews.id, prior.id),
-            isNull(chatGitHubReviews.runId),
-          ),
-        )
-        .returning();
+      const [claimed] = await db.transaction(async tx => {
+        await notifyGitHubReviewChecks(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
+          repositoryId: source.repositoryId, number: source.number });
+        return tx
+          .update(chatGitHubReviews)
+          .set({ runId: source.run.id, state: "running", updatedAt: new Date() })
+          .where(
+            and(
+              eq(chatGitHubReviews.id, prior.id),
+              isNull(chatGitHubReviews.runId),
+            ),
+          )
+          .returning();
+      });
       if (claimed) return claimed;
       return reviewForHead(source, pull);
     }
@@ -407,26 +412,30 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
       labels: pull.labels.map((l) => l.name),
       ...(automatic ? automatic.context : {}),
     };
-    const [review] = await db
-      .insert(chatGitHubReviews)
-      .values({
-        companyId: source.endpoint.companyId,
-        endpointId: source.endpoint.id,
-        issueId: source.issue.id,
-        runId: source.run.id,
-        repositoryId: source.repositoryId,
-        repository: source.repository,
-        pullNumber: source.number,
-        headSha: pull.head.sha,
-        deliveryId: `run:${source.run.id}:${pull.head.sha}`,
-        configurationRevision:
-          manual?.revision ?? automatic?.revision ?? source.config.revision,
-        policySnapshot: manual?.policy ?? automatic?.policy ?? source.policy,
-        event: context,
-        state: "running",
-      })
-      .onConflictDoNothing()
-      .returning();
+    const [review] = await db.transaction(async tx => {
+      await notifyGitHubReviewChecks(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
+        repositoryId: source.repositoryId, number: source.number });
+      return tx
+        .insert(chatGitHubReviews)
+        .values({
+          companyId: source.endpoint.companyId,
+          endpointId: source.endpoint.id,
+          issueId: source.issue.id,
+          runId: source.run.id,
+          repositoryId: source.repositoryId,
+          repository: source.repository,
+          pullNumber: source.number,
+          headSha: pull.head.sha,
+          deliveryId: `run:${source.run.id}:${pull.head.sha}`,
+          configurationRevision:
+            manual?.revision ?? automatic?.revision ?? source.config.revision,
+          policySnapshot: manual?.policy ?? automatic?.policy ?? source.policy,
+          event: context,
+          state: "running",
+        })
+        .onConflictDoNothing()
+        .returning();
+    });
     if (review) {
       await githubReviewCheckService(db, fetchImpl).enqueue(
         source.endpoint,

server/src/services/chat-run-milestone-notifications.ts

@@ -8,6 +8,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type ChatRunMilestoneScope = { companyId: string; issueId?: string };
 
 export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId }));
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
   // Lifecycle and coarse native progress share lineage and authority. Extend

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

@@ -38,6 +38,7 @@ export const DELIVERY_QUEUES = {
   chatProviderEffects: "chat-provider-effects",
   chatGitHubIngress: "chat-github-ingress",
   chatGitHubPublications: "chat-github-publications",
+  chatGitHubReviewChecks: "chat-github-review-checks",
   chatGitHubTerminalResponses: "chat-github-terminal-responses",
   chatGitHubAcknowledgements: "chat-github-acknowledgements",
   chatFailedRunRetries: "chat-failed-run-retries",