perf(queue): coalesce the four event-driven job types by their per-invocation id

src/queue/processors.ts

@@ -1810,6 +1810,15 @@ async function reReviewStoredPullRequest(
     resyncAdmissionKey,
   );
   primeLiveMergeState(liveFacts, repoFullName, prNumber, resyncToken, live?.mergeable_state);
+  // Terminal early-exit (#1942): the PR is CLOSED/merged on GitHub even though the stored row still reads open — a
+  // dropped `closed` webhook (relay down). Reconcile the stored row from the live payload and RETURN before the
+  // expensive resync (files) + readiness + re-review reads. A stale sweep must never spend GitHub budget — or post
+  // visible output — re-reviewing a PR that can no longer produce a valid outcome. Fail-open: only a live NON-open
+  // state early-exits (a fetch hiccup leaves `live` undefined → proceed with the stored open PR).
+  if (live && live.state !== "open") {
+    await upsertPullRequestFromGitHub(env, repoFullName, live).catch(() => undefined);
+    return;
+  }
   if (live?.head?.sha && live.head.sha !== pr.headSha) {
     await upsertPullRequestFromGitHub(env, repoFullName, live).catch(
       () => undefined,

src/selfhost/queue-common.ts

@@ -598,6 +598,10 @@ export function jobCoalesceKey(payload: string): string | null {
       dryRun?: unknown;
       variant?: unknown;
       paths?: unknown;
+      runId?: unknown;
+      deliveryId?: unknown;
+      draftId?: unknown;
+      event?: { dedupKey?: unknown } | null;
       payload?: GitHubWebhookPayload | null;
     };
     const type = typeof message.type === "string" ? message.type : "";
@@ -687,6 +691,25 @@ export function jobCoalesceKey(payload: string): string | null {
           normalizedRepo(message.repoFullName) ?? "all",
           normalizedPathScope(message.paths) ?? "full",
         );
+      // Event-driven jobs carry a stable per-invocation id, so coalescing only ever merges a DUPLICATE re-enqueue of
+      // the SAME job (e.g. a webhook redelivery / retry) — never two distinct invocations, which have distinct ids.
+      // No id (a malformed payload) → null (uncoalesced), never a shared key that could drop a distinct job. (#1942)
+      case "run-agent": {
+        const runId = normalizedId(message.runId);
+        return runId ? keyOf(type, runId) : null;
+      }
+      case "notify-deliver": {
+        const deliveryId = normalizedId(message.deliveryId);
+        return deliveryId ? keyOf(type, deliveryId) : null;
+      }
+      case "notify-evaluate": {
+        const dedupKey = normalizedId(message.event?.dedupKey);
+        return dedupKey ? keyOf(type, dedupKey) : null;
+      }
+      case "submit-draft": {
+        const draftId = normalizedId(message.draftId);
+        return draftId ? keyOf(type, draftId) : null;
+      }
     }
     if (type !== "github-webhook") return null;
     const eventName =
@@ -742,6 +765,11 @@ function normalizedCursor(value: unknown): string | null {
   return typeof value === "string" && value.trim() ? value.trim() : null;
 }
 
+// A stable, case-preserving opaque id (runId / deliveryId / draftId / dedupKey) for coalesce keys.
+function normalizedId(value: unknown): string | null {
+  return typeof value === "string" && value.trim() ? value.trim() : null;
+}
+
 function normalizedDate(value: unknown): string | null {
   return typeof value === "string" && /^\d{4}-\d{2}-\d{2}$/.test(value.trim())
     ? value.trim()

test/unit/queue.test.ts

@@ -825,6 +825,35 @@ describe("queue processors", () => {
     resyncUpsertSpy.mockRestore();
   });
 
+  it("#regate-terminal-exit: a swept PR CLOSED on GitHub reconciles the stored row then early-exits — no files/CI reads, no review (#1942)", async () => {
+    const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() });
+    await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001);
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" });
+    // STORED row still reads open — the `closed` webhook was dropped (relay down); GitHub's LIVE state is closed.
+    await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Closed PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" });
+    let filesFetched = false;
+    let ciFetched = false;
+    vi.stubGlobal("fetch", async (input: RequestInfo | URL) => {
+      const url = input.toString();
+      if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
+      if (url.endsWith("/pulls/7")) return Response.json({ number: 7, title: "Closed PR", state: "closed", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" });
+      if (url.includes("/pulls/7/files")) { filesFetched = true; return Response.json([]); }
+      if (url.includes("/commits/")) { ciFetched = true; return Response.json({ total_count: 0, check_runs: [] }); }
+      return Response.json({});
+    });
+    vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));
+
+    await processJob(env, { type: "agent-regate-pr", deliveryId: "resync-closed", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 });
+
+    // Reconciled: the stored row now reflects the live terminal state, so the NEXT sweep skips it outright.
+    const stored = await getPullRequest(env, "owner/agent-repo", 7);
+    expect(stored?.state).toBe("closed");
+    // Early-exit BEFORE the expensive resync + readiness reads: no files, no CI reads (and no review output).
+    expect(filesFetched).toBe(false);
+    expect(ciFetched).toBe(false);
+  });
+
   // REST-budget dedup (#audit-rate-headroom): one per-PR re-review threads request-local live GitHub facts through
   // readiness and auto-maintain, while post-gate planning refreshes facts that can change after the bot publishes
   // review/check state. Mergeability can advance to clean; CI can flip red and must still suppress merge.

test/unit/selfhost-queue-common.test.ts

@@ -542,6 +542,22 @@ describe("self-host queue common helpers", () => {
     ).toBeNull();
   });
 
+  it("coalesces the event-driven jobs by their stable per-invocation id — and only true duplicates (#1942)", () => {
+    // A DUPLICATE re-enqueue of the SAME job (same id — e.g. a webhook redelivery) coalesces.
+    expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-abc123" }))).toBe("run-agent:run-abc123");
+    expect(jobCoalesceKey(payload({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: "del-77" }))).toBe("notify-deliver:del-77");
+    expect(jobCoalesceKey(payload({ type: "submit-draft", requestedBy: "api", draftId: "draft-9" }))).toBe("submit-draft:draft-9");
+    expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "webhook", event: { dedupKey: "review_requested:o/r#3:bob" } }))).toBe("notify-evaluate:review_requested:o/r#3:bob");
+    // Two DISTINCT invocations have distinct ids → distinct keys, so they never merge.
+    expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-xyz789" }))).toBe("run-agent:run-xyz789");
+    // A payload missing its id → null (uncoalesced), never a shared key that could drop a distinct job.
+    expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "test" }))).toBeNull();
+    expect(jobCoalesceKey(payload({ type: "notify-deliver", requestedBy: "test" }))).toBeNull();
+    expect(jobCoalesceKey(payload({ type: "submit-draft", requestedBy: "test" }))).toBeNull();
+    expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test" }))).toBeNull();
+    expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test", event: {} }))).toBeNull();
+  });
+
   it("coalesces recurring maintenance jobs while preserving their semantic scope", () => {
     expect(
       jobCoalesceKey(