fix(hub,web): unlatch stalled banner on stale events and surface waiting PR (CL-12012)

Author: TheGreatAxios

apps/hub/src/integrations/app.ts

@@ -744,7 +744,7 @@ export function createIntegrationsApp(guards: IntegrationGuards, handlers: Integ
     `${WORKFLOW_VERSIONS_PATH}/:tenantId`,
     describeRoute({
       summary: "Package versions of the workspace's live workflow deployments and the hub's last roll of each host",
-      description: "Each live deployment, newest first, with the workflow package it pinned and the packages its frozen closure carries. `package` is null for a deployment without a frozen closure; `source` is the asset and commit a source-format deployment runs, else null. `rollouts` holds each host the hub rolled onto its build's release since it started: when that roll ended and, when it failed, the error; a failed roll leaves the host's previous listener running. `stalled` is present while the workspace's pull requests have waited past the hub's alarm age (`TRIAGE_STALLED_AFTER_MS`) for a manager to take them: `since` is when the oldest still waiting was queued, `lastTakeAt` when a manager last took one.",
+      description: "Each live deployment, newest first, with the workflow package it pinned and the packages its frozen closure carries. `package` is null for a deployment without a frozen closure; `source` is the asset and commit a source-format deployment runs, else null. `rollouts` holds each host the hub rolled onto its build's release since it started: when that roll ended and, when it failed, the error; a failed roll leaves the host's previous listener running. `stalled` is present while the workspace's pull requests have waited past the hub's alarm age (`TRIAGE_STALLED_AFTER_MS`) for a manager to take them: `since` is when the oldest still waiting was queued, `pull` which pull request it is, `count` how many wait, `reason` why if known, and `lastTakeAt` when a manager last took one.",
       tags: TAGS,
       responses: { 200: json("Live deployments and their versions", DeployedVersions), 401: failures[401], 403: failures[403] },
     }),

apps/hub/src/manager-state.ts

@@ -978,7 +978,7 @@ function pullOf(key: string): PullRequestRef {
 }
 
 /** How long an overdue wake no manager has taken the claim since holds off the next, while its manager may still be reaching a sidecar. */
-const OVERDUE_UNSTARTED_GRACE_MS = 30 * 60_000;
+export const OVERDUE_UNSTARTED_GRACE_MS = 30 * 60_000;
 
 /** Whether the claim's last overdue wake is still within its grace: a manager took the claim since, or none has yet. */
 function recentlyWoken(claim: Claim, now: Date): boolean {
@@ -1012,13 +1012,23 @@ export async function lastTakes(db: DB["db"]): Promise<{ tenantId: string; sourc
   });
 }
 
-/** When the oldest pull request mail of the tenant's host that no manager has taken was recorded, among those recorded after `after`. */
-export async function oldestPendingMail(db: DB["db"], tenantId: string, source: RepoSource, after: string | undefined): Promise<string | undefined> {
+const EVENT_KEY = /^evt:[^:]+:([^#]+)#(\d+):/;
+
+/** The oldest pull request mail of each pull request on the tenant's host that no manager has taken, among those recorded after `after`. */
+export async function pendingMails(db: DB["db"], tenantId: string, source: RepoSource, after: string | undefined): Promise<{ repo: string; number: number; at: string }[]> {
   const since = after === undefined ? undefined : sql`(${managerState.state}->>'receivedAt')::timestamptz > ${after}::timestamptz`;
   const ofHost = like(managerState.key, `evt:${source}:%`);
-  const [row] = await db.select({ at: sql<string | null>`min(${managerState.state}->>'receivedAt')` }).from(managerState)
+  const rows = await db.select({ key: managerState.key, at: sql<string>`${managerState.state}->>'receivedAt'` }).from(managerState)
     .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, sql`${managerState.state}->>'consumedBy' IS NULL`, sql`${managerState.state}->>'type' = 'webhook'`, ofHost, since));
-  return row?.at ?? undefined;
+  const oldest = new Map<string, { repo: string; number: number; at: string }>();
+  for (const row of rows) {
+    const [, repo, num] = EVENT_KEY.exec(row.key) ?? [];
+    if (repo === undefined || num === undefined) continue;
+    const pr = `${repo}#${num}`;
+    const prior = oldest.get(pr);
+    if (prior === undefined || Date.parse(row.at) < Date.parse(prior.at)) oldest.set(pr, { repo, number: Number(num), at: row.at });
+  }
+  return [...oldest.values()];
 }
 
 /** The tenant's pull requests whose claim holds a reviewer with no result. */

apps/hub/src/triage-liveness.postgres.test.ts

@@ -112,7 +112,7 @@ test("a mail no manager takes raises the alarm once past the alarm age, and a ta
     await hub.check();
     await hub.check();
     expect(hub.alarms()).toEqual(["triage_stalled github"]);
-    expect(hub.stalled()).toMatchObject([{ source: "github" }]);
+    expect(hub.stalled()).toMatchObject([{ source: "github", pull: "acme/widgets#7", count: 1 }]);
 
     await hub.took(PR);
     await hub.check();
@@ -159,6 +159,7 @@ test("a head whose mail finds no listener stays stalled while the reconciler ret
     hub.advance(STALLED_AFTER_MS + 60_000);
     await hub.check();
     expect(hub.alarms()).toEqual(["triage_stalled github"]);
+    expect(hub.stalled()).toMatchObject([{ source: "github", pull: "acme/widgets#7", count: 1, reason: triageStopped("github") }]);
 
     for (let retry = 0; retry < 3; retry++) {
       hub.advance(60 * 60_000);
@@ -265,3 +266,22 @@ test("a hub that sends no email says so once and still raises and clears the ala
     expect(hub.logged.filter((entry) => entry["msg"] === "triage_stall_not_emailed").map((entry) => entry["reason"])).toEqual(["SMTP_URL is not set"]);
   });
 }, 60_000);
+
+test("a raised alarm clears without a newer take once its pull request is no longer waiting in the store, and ignores disabled repositories", async () => {
+  await withDatabase(async (db) => {
+    const queuedAt = new Date().toISOString();
+    const store = memoryStore([{ number: 7, headSha: "h1", status: "queued", attempts: 2, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt }]);
+    const hub = stack(db, store);
+    await mailOf(PR, "d1", db);
+    await mailOf({ source: "github", repo: "acme/disabled", number: 1 }, "d2", db);
+
+    hub.advance(STALLED_AFTER_MS + 60_000);
+    await hub.check();
+    expect(hub.stalled()).toMatchObject([{ source: "github", pull: "acme/widgets#7", count: 1, reason: "run never started" }]);
+
+    await store.save(TENANT, RECORD.name, [{ number: 7, headSha: "h1", status: "failed", error: "run never started", attempts: 5, firstSeenAt: queuedAt, queuedAt, updatedAt: hub.now().toISOString() }]);
+    await hub.check();
+    expect(hub.alarms()).toEqual(["triage_stalled github", "triage_stall_cleared github"]);
+    expect(hub.stalled()).toEqual([]);
+  });
+}, 60_000);

apps/hub/src/triage-liveness.ts

@@ -11,10 +11,12 @@ import { REPO_SOURCES, repoPolicy, type RepoSource } from "@corbits/triage-contr
 import { repoRecords, sourceOf, triageNs, triageStateKey } from "./github/tenant-config.js";
 import { undelivered } from "./github/triage-queue.js";
 import type { TriageStateStore } from "./github/triage-state-store.js";
-import { lastTakes, oldestPendingMail } from "./manager-state.js";
+import { lastTakes, OVERDUE_UNSTARTED_GRACE_MS, pendingMails } from "./manager-state.js";
 
-/** A stalled host: since when its oldest waiting pull request waits, and when one of its managers last took one. */
-export type Stalled = { source: RepoSource; since: string; lastTakeAt?: string };
+/** A stalled host: since when its oldest waiting pull request waits, which pull request it is, how many wait, why if known, and when one of its managers last took one. */
+export type Stalled = { source: RepoSource; since: string; lastTakeAt?: string; pull?: string; count?: number; reason?: string };
+
+type WaitingPull = { pull: string; at: number; reason?: string };
 
 /** What the workspace's owners and admins are emailed when a host's alarm is raised, or cleared. */
 export type StallNotice = { to: string[]; workspace: string; source: RepoSource; since: string; cleared: boolean };
@@ -104,35 +106,67 @@ export function createTriageLiveness(deps: TriageLivenessDeps) {
    * When each waiting pull request of each host began to wait: a mail or queued head since the host's last take, and
    * any head whose mail did not go out, which no take can have picked up, until a run of it is seen.
    */
-  async function waiting(tenant: LivenessTenant): Promise<Map<RepoSource, number[]>> {
-    const byHost = new Map<RepoSource, number[]>();
+  async function waiting(tenant: LivenessTenant): Promise<Map<RepoSource, WaitingPull[]>> {
+    const byHost = new Map<RepoSource, WaitingPull[]>(REPO_SOURCES.map((source) => [source, []]));
+    const mails = new Map<RepoSource, { repo: string; number: number; at: string }[]>();
     for (const source of REPO_SOURCES) {
-      const mail = await oldestPendingMail(deps.db, tenant.id, source, deps.takes.last(tenant.id, source));
-      byHost.set(source, mail === undefined ? [] : [Date.parse(mail)]);
+      mails.set(source, await pendingMails(deps.db, tenant.id, source, deps.takes.last(tenant.id, source)));
     }
+    const now = deps.now().getTime();
     for (const record of repoRecords(triageNs(tenant.config))) {
       if (!record.connected || !repoPolicy(record).enabled) continue;
       const source = sourceOf(record);
       const taken = deps.takes.last(tenant.id, source);
-      const { rows } = await deps.store.load(tenant.id, triageStateKey(record));
+      const { rows, version } = await deps.store.load(tenant.id, triageStateKey(record));
+      const queued = new Set<number>();
       for (const row of rows) {
         const since = row.undeliveredSince ?? row.queuedAt;
         if (since === undefined || (row.status !== "queued" && !undelivered(row, source))) continue;
+        queued.add(row.number);
         const unsent = row.undeliveredSince !== undefined || undelivered(row, source);
-        if (unsent || taken === undefined || Date.parse(since) > Date.parse(taken)) byHost.get(source)?.push(Date.parse(since));
+        if (unsent || taken === undefined || Date.parse(since) > Date.parse(taken)) {
+          const reason = row.error ?? (row.attempts > 1 && row.runId === undefined ? "run never started" : undefined);
+          byHost.get(source)?.push({ pull: `${record.name}#${row.number}`, at: Date.parse(since), ...(reason !== undefined && { reason }) });
+        }
+      }
+      for (const mail of mails.get(source) ?? []) {
+        if (mail.repo !== record.name || queued.has(mail.number)) continue;
+        const at = Date.parse(mail.at);
+        if (now - at > OVERDUE_UNSTARTED_GRACE_MS) continue;
+        const ofPull = rows.filter((each) => each.number === mail.number);
+        const unstored = ofPull.length === 0 && version === null && rows.length === 0;
+        const newer = ofPull.length > 0 && ofPull.every((each) => at > Date.parse(each.updatedAt));
+        if (unstored || newer) byHost.get(source)?.push({ pull: `${record.name}#${mail.number}`, at });
       }
     }
     return byHost;
   }
 
-  /** Raises the host's alarm once it stalls, and clears it only once one of its managers opened a pass since and none waits. */
-  async function settle(tenantId: string, source: RepoSource, times: readonly number[]): Promise<void> {
+  function stalledOf(source: RepoSource, pulls: readonly WaitingPull[], taken: string | undefined): Stalled | undefined {
+    const now = deps.now().getTime();
+    const byPull = new Map<string, WaitingPull>();
+    for (const item of pulls) {
+      if (now - item.at < deps.stalledAfterMs) continue;
+      const prior = byPull.get(item.pull);
+      if (prior === undefined || item.at < prior.at) byPull.set(item.pull, item);
+    }
+    const overdue = [...byPull.values()].sort((a, b) => a.at - b.at || a.pull.localeCompare(b.pull));
+    const oldest = overdue[0];
+    if (oldest === undefined) return undefined;
+    return {
+      source,
+      since: new Date(oldest.at).toISOString(),
+      ...(taken !== undefined && { lastTakeAt: taken }),
+      pull: oldest.pull,
+      count: overdue.length,
+      ...(oldest.reason !== undefined && { reason: oldest.reason }),
+    };
+  }
+
+  /** Raises the host's alarm once it stalls, and clears it once nothing on the host waits past the alarm age. */
+  async function settle(tenantId: string, source: RepoSource, pulls: readonly WaitingPull[]): Promise<void> {
     const key = hostKey(tenantId, source);
-    const taken = deps.takes.last(tenantId, source);
-    const oldest = times.length === 0 ? undefined : Math.min(...times);
-    const stalled: Stalled | undefined = oldest === undefined || deps.now().getTime() - oldest < deps.stalledAfterMs
-      ? undefined
-      : { source, since: new Date(oldest).toISOString(), ...(taken !== undefined && { lastTakeAt: taken }) };
+    const stalled = stalledOf(source, pulls, deps.takes.last(tenantId, source));
     const raised = alarms.get(key);
     if (raised === undefined) {
       if (stalled === undefined) return;
@@ -141,7 +175,6 @@ export function createTriageLiveness(deps: TriageLivenessDeps) {
       await email(tenantId, stalled, false);
       return;
     }
-    if (taken === undefined || Date.parse(taken) <= Date.parse(raised.since)) return;
     if (stalled !== undefined) {
       alarms.set(key, stalled);
       return;

apps/web/src/components/TriageStalledNotice.dom.test.tsx

@@ -20,8 +20,12 @@ test("the workspace is told which host has pull requests waiting for Triage only
   expect(status?.textContent).toBe("");
 
   const since = new Date("2026-10-27T10:42:00Z");
+  const at = since.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" });
   act(() => root?.render(<MemoryRouter><TriageStalledNotice stalled={[{ source: "bitbucket-datacenter", since: since.toISOString() }]} /></MemoryRouter>));
   expect(document.querySelector("[role=status]")).toBe(status);
-  expect(status?.textContent).toContain(`Pull requests on Bitbucket Data Center have waited since ${since.toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" })} for Triage to start.`);
+  expect(status?.textContent).toContain(`Pull requests on Bitbucket Data Center have waited since ${at} for Triage to start.`);
   expect(document.querySelector("a")?.getAttribute("href")).toBe("/settings/triage");
+
+  act(() => root?.render(<MemoryRouter><TriageStalledNotice stalled={[{ source: "github", since: since.toISOString(), pull: "acme/widgets#7", count: 3, reason: "run never started" }]} /></MemoryRouter>));
+  expect(status?.textContent).toContain(`acme/widgets#7 and 2 other pull requests on GitHub have waited since ${at} for Triage to start: run never started.`);
 });

apps/web/src/components/TriageStalledNotice.tsx

@@ -1,12 +1,26 @@
 import { Link } from "react-router-dom";
+import { counted } from "@corbits/common";
 import type { DeployedVersions } from "@corbits/triage-contracts";
 import { hostFullName } from "../lib/pull-link.ts";
 import { useTriageStalled } from "../lib/tenant-entities.ts";
 
 type Stalled = NonNullable<DeployedVersions["stalled"]>[number];
 
-export function stalledText({ source, since }: Stalled): string {
-  return `Pull requests on ${hostFullName(source)} have waited since ${new Date(since).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" })} for Triage to start.`;
+function waitingSubject({ source, pull, count }: Stalled): string {
+  const host = hostFullName(source);
+  if (pull === undefined) return `Pull requests on ${host} have`;
+  if (count !== undefined && count > 1) return `${pull} and ${counted(count - 1, "other pull request", "other pull requests")} on ${host} have`;
+  return `${pull} on ${host} has`;
+}
+
+function reasonTail(reason: string | undefined): string {
+  if (reason === undefined) return ".";
+  return reason.endsWith(".") ? `: ${reason}` : `: ${reason}.`;
+}
+
+export function stalledText(stalled: Stalled): string {
+  const at = new Date(stalled.since).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" });
+  return `${waitingSubject(stalled)} waited since ${at} for Triage to start${reasonTail(stalled.reason)}`;
 }
 
 /** Tells everyone in the workspace while its pull requests on a host wait for Triage to start. */

packages/triage-contracts/src/hub-responses.ts

@@ -226,8 +226,8 @@ export const DeployedVersions = type({
    * because it was full (`overflow`).
    */
   sidecar: type({ active: "number", waiting: "number", cap: "number", overflow: "number" }),
-  /** Each host whose pull requests have waited for a manager past the hub's alarm age: since when the oldest waits, and when a manager of that host last took one. */
-  "stalled?": type({ source: type.enumerated(...REPO_SOURCES), since: "string", "lastTakeAt?": "string" }).array(),
+  /** Each host whose pull requests have waited for a manager past the hub's alarm age: since when the oldest waits, which pull request it is, how many wait, why if known, and when a manager of that host last took one. */
+  "stalled?": type({ source: type.enumerated(...REPO_SOURCES), since: "string", "lastTakeAt?": "string", "pull?": "string", "count?": "number", "reason?": "string" }).array(),
 });
 export type DeployedVersions = typeof DeployedVersions.infer;