fix(hub): run catch-up, Triage again and the Settings status from the listener when one is live

apps/hub/src/github/deployment-heal.postgres.test.ts

@@ -145,6 +145,7 @@ function harness(db: ReturnType<typeof createDB>["db"]) {
     rotationKey: "rotation",
     tenants: async () => [{ id: TENANT, domain: DOMAIN, config: CONFIG }],
     liveDeployments,
+    listener: async () => undefined,
     latestDeployments: (tenantId, limit) => latestDeployments(db, tenantId, WORKFLOW, limit),
     rotation: {
       afterRuns: 1_000,

apps/hub/src/github/interleave.postgres.test.ts

@@ -51,6 +51,7 @@ function route(store: TriageStateStore, deliver: Deliver) {
     pullHead: async () => "abc123",
     readCheckPack: async () => ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }),
     liveDeployments: async () => [DEPLOYMENT],
+    listener: async () => undefined,
     observeRuns: noRuns,
     managerRuns: noRuns,
     store,
@@ -67,6 +68,7 @@ function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver,
     rotationKey: "rotation",
     tenants: async () => [{ id: TENANT, domain: "tenant.example", config: CONFIG }],
     liveDeployments: async () => [DEPLOYMENT],
+    listener: async () => undefined,
     latestDeployments: async () => [],
     rotation: { afterRuns: 15, redeploy: async () => ({ status: "skipped", reason: "unused" }), sameSource: async () => false, release: async () => {}, claim: async () => false, clear: async () => {} },
     openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })),

apps/hub/src/github/pr-triage.test.ts

@@ -29,7 +29,7 @@ function stubDb(repo: Record<string, unknown>) {
   } as never;
 }
 
-function observed(runs: ObservedRun[]): GithubPrTriageDeps["observeRuns"] {
+function observed(runs: ObservedRun[]): GithubPrTriageDeps["observeRuns"] & GithubPrTriageDeps["managerRuns"] {
   return async function observeRuns() {
     return { byRepo: new Map([[REPO, new Map([["8@abc123", runs]])]]), runCount: runs.length, verdicts: [], approved: [] };
   };
@@ -73,6 +73,7 @@ function handler(repo: Record<string, unknown>, s: Stub, overrides: Partial<Gith
     pullHead: async (_tenantId, _record, number) => (number === 8 ? "abc123" : undefined),
     readCheckPack: async () => ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }),
     liveDeployments: async () => [ANCHOR],
+    listener: async () => undefined,
     observeRuns: observed([]),
     managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }),
     store: stored([], s.saved),
@@ -166,6 +167,17 @@ describe("createGithubPrTriage", () => {
     expect(s.delivered).toHaveLength(1);
   });
 
+  test("with the host's listener live, mails with no pr-triage deployment and refuses a pull request whose manager is up", async () => {
+    const s = stub();
+    const listening: Partial<GithubPrTriageDeps> = { listener: async () => ({ ...ANCHOR, runId: "listener-1", address: "listener-1@tenant.example" }), liveDeployments: async () => [] };
+    expect((await handler(ENABLED, s, listening)(request(), MEMBER)).status).toBe(202);
+    expect(s.delivered).toHaveLength(1);
+    const managed = stub();
+    const running: ObservedRun = { runId: "run-manager", source: "github", status: "running", startedAt: NOW.toISOString() };
+    await refused(await handler(ENABLED, managed, { ...listening, managerRuns: observed([running]) })(request(), MEMBER), 409, "already_running");
+    expect(managed.delivered).toEqual([]);
+  });
+
   test("refuses a head still running on a replaced deployment or stored as running, but not one stored running past the stuck time", async () => {
     const live: ObservedRun = { runId: "run-live", source: "github", status: "running", startedAt: new Date(NOW.getTime() - 60_000).toISOString() };
     const replaced = { ...ANCHOR, runId: "anchor-0", address: "anchor-0@tenant.example" };

apps/hub/src/github/pr-triage.ts

@@ -1,6 +1,7 @@
-// Queues one open pull request's head for the live pr-triage deployment at a
-// maintainer's request. The row is marked queued before the mail goes out so
-// repeated clicks and reconcile passes see it; a failed mail puts it back.
+// Queues one open pull request's head at a maintainer's request, for the host's
+// listener while one is live and the live pr-triage deployment otherwise. The
+// row is marked queued before the mail goes out so repeated clicks and reconcile
+// passes see it; a failed mail puts it back.
 // Attempts are left alone: a head the hub gave up on is only ever retried here.
 import { eq } from "drizzle-orm";
 import { type } from "arktype";
@@ -32,6 +33,8 @@ export type GithubPrTriageDeps = PortalCredentialDeps & {
   readCheckPack: PolicyReader;
   /** The named workflow's, newest first; mail goes to the first, and a head may still be running on a replaced one. */
   liveDeployments: (tenantId: string, workflow: string) => Promise<LiveDeployment[]>;
+  /** The workspace's live listener for the host, if it has one; its managers then run the pull requests. */
+  listener: (tenantId: string, source: RepoSource) => Promise<LiveDeployment | undefined>;
   observeRuns: ObserveRuns;
   managerRuns: ManagerRuns;
   store: TriageStateStore;
@@ -94,9 +97,9 @@ export function createGithubPrTriage(deps: GithubPrTriageDeps) {
     if (ready instanceof Response) return ready;
     const { tenant, record, policy, pack } = ready;
     const triage = triageWorkflowFor(sourceOf(record));
-    const deployments = await deps.liveDeployments(tenantId, triage.id);
-    const deployment = newestLive(deployments);
-    if (!deployment) return failure(503, "service_not_running", `The ${triage.id} service is not running.`);
+    const listening = (await deps.listener(tenantId, sourceOf(record))) !== undefined;
+    const deployments = listening ? [] : await deps.liveDeployments(tenantId, triage.id);
+    if (!listening && !newestLive(deployments)) return failure(503, "service_not_running", `The ${triage.id} service is not running.`);
 
     let openHead: string | undefined;
     try {

apps/hub/src/github/triage-reconciler.test.ts

@@ -5,7 +5,7 @@ import { formatRunAddress } from "@intx/types";
 import { workflowRunRepoIdForAddress, type RepoId, type WorkflowRunEvent, type WorkflowRunReader } from "@intx/hub-sessions";
 import { emptyPack, stockTriggerMail, type ActionKind, type CheckPack, type PrTriageRow } from "@corbits/triage-contracts";
 import type { CheckRun } from "@corbits/github-reads";
-import { DEFAULT_RECONCILE_POLICY, planTenant, type ObservedRun, type OpenPr } from "./reconcile-plan.js";
+import { DEFAULT_RECONCILE_POLICY, planTenant, runKey, type ObservedRun, type OpenPr } from "./reconcile-plan.js";
 import { triageReconcilePolicy } from "../env.js";
 import { TriageStateConflictError, type TriageStateStore } from "./triage-state-store.js";
 import { createRunTriggerDeliverer } from "@corbits/webhooks";
@@ -34,6 +34,8 @@ const LONG_AGO = "2026-10-07T00:00:00.000Z";
 const VERSION = "0.1.0-sha-11111111";
 const OFFERING = "off_decision";
 const LIVE = deployment("run_live", new Date(LONG_AGO));
+/** A listener names no workflow version or offering. */
+const LISTENER = { runId: "run_listener", address: formatRunAddress("run_listener", DOMAIN), createdAt: new Date(LONG_AGO), cancelling: false };
 const MINUTE = 60_000;
 const BASE_MOVE_WINDOW = DEFAULT_RECONCILE_POLICY.baseMoveWindowMs / MINUTE;
 
@@ -188,6 +190,9 @@ type HarnessOptions = {
   rotateAfterRuns?: number;
   now?: () => Date;
   deployments?: LiveDeployment[];
+  listener?: LiveDeployment;
+  /** The managers' runs by head, read on each pass. */
+  managed?: () => Record<string, ObservedRun[]>;
   whileRedeploying?: (deployments: LiveDeployment[]) => void;
   rotation?: RotationRecord;
   sameSource?: boolean;
@@ -259,6 +264,7 @@ function harness(logs: Logs, prs: OpenPr[] | Record<string, OpenPr[]>, options:
       return [tenant()];
     },
     liveDeployments,
+    listener: async () => options.listener,
     latestDeployments: async () => [],
     rotation: {
       afterRuns: options.rotateAfterRuns ?? 1_000,
@@ -295,7 +301,7 @@ function harness(logs: Logs, prs: OpenPr[] | Record<string, OpenPr[]>, options:
     },
     activity: options.activity ?? { stale: () => true },
     observeTenant: observer.observe,
-    managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }),
+    managerRuns: async () => ({ byRepo: new Map(options.managed ? [[REPO, new Map(Object.entries(options.managed()))]] : []), runCount: 0, verdicts: [], approved: [] }),
     headChecks: async () => options.headChecks ?? [],
     verdicts: { latest: async () => options.recorded ?? [] },
     store: store(),
@@ -347,6 +353,24 @@ function rotationLogs(logged: ReadonlyArray<Record<string, unknown>>): unknown[]
 }
 
 describe("triage reconciler", () => {
+  test("a host whose listener is live is caught up with no pr-triage deployment", async () => {
+    const run = harness({}, [head(1)], { deployments: [], listener: LISTENER });
+    await run.reconcile();
+    expect(mailedNumbers(run.delivered)).toEqual([1]);
+  });
+
+  test("a head mailed through the listener is mailed once while its manager runs and once it reported", async () => {
+    let minutes = 0;
+    let managed: Record<string, ObservedRun[]> = {};
+    const run = harness({}, [head(1)], { deployments: [], listener: LISTENER, now: () => at(minutes), managed: () => managed });
+    await run.reconcile();
+    managed = { [runKey(1, "sha1")]: [{ runId: "run_manager", source: "github", status: "running", startedAt: at(1).toISOString() }] };
+    for (minutes = 1; minutes <= 3; minutes++) await run.reconcile();
+    managed = { [runKey(1, "sha1")]: [{ runId: "run_manager", source: "github", status: "completed", startedAt: at(1).toISOString() }] };
+    for (minutes = 4; minutes <= 60; minutes += 14) await run.reconcile();
+    expect(mailedNumbers(run.delivered)).toEqual([1]);
+  });
+
   test("a host without build webhooks: a head whose verdict waits on CI is queued again once its builds stop pending", async () => {
     const waiting = { ...current(1), repo: REPO, number: 1, state: "awaiting-review", actor: "system", humanGated: false, checks: [{ check: "ci-green", kind: "machine", result: "unconfirmed" }] };
     const logs: Logs = { [repoKey(LIVE.runId)]: { run_1: [started(1), rendered(waiting), completed] } };

apps/hub/src/github/triage-reconciler.ts

@@ -1,7 +1,8 @@
 // Brings every open pull request on an enabled repository to a triaged head
 // without relying on any webhook, sidecar or portal visit having survived:
 // reads GitHub's open pull requests, the live deployments' run logs and the
-// stored state, then queues what the plan says, each head to a worker of its own.
+// stored state, then queues what the plan says, each head to a worker of its own,
+// or to the host's listener while one is live.
 // A head waiting for CI is released here once its CI finished or its wait ran
 // out, on hosts and pull requests whose CI results send no webhook.
 import { dependsOnCi, hasEnabledRules, marksWaiting, repoPolicy, type PrTriageRow, type RepoPolicy, type RepoSource, type VerdictDetail } from "@corbits/triage-contracts";
@@ -43,6 +44,12 @@ export type TriageReconcilerDeps = {
   rotationKey: string;
   tenants: () => Promise<ReconcileTenant[]>;
   liveDeployments: (tenantId: string) => Promise<LiveDeployment[]>;
+  /**
+   * The host's live listener, whose managers run its pull requests: the pass then needs no pr-triage deployment, and
+   * neither rotates nor heals one, so a rotation under way is abandoned. A listener names no workflow version, so no
+   * head starts over for one.
+   */
+  listener: (tenantId: string, source: RepoSource) => Promise<LiveDeployment | undefined>;
   latestDeployments: (tenantId: string, limit: number) => Promise<DeploymentRecord[]>;
   rotation: DeploymentRotation;
   /** Undefined when the tenant holds no credential for any pull request host. */
@@ -456,9 +463,10 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) {
   async function reconcileTenant(tenant: ReconcileTenant): Promise<boolean> {
     const repos = enabledRepos(tenant.config, deps.source);
     if (repos.length === 0) return false;
+    const listener = await deps.listener(tenant.id, deps.source);
     // Only live deployments' and recent workers' logs are read; heads triaged under a released one stay settled in the stored state until the live deployment runs another workflow version.
     const { deployments, observed, workers } = await deps.observeTenant(tenant.id, tenant.domain);
-    if (deployments.length === 0) {
+    if (!listener && deployments.length === 0) {
       deps.log({ level: "warn", msg: "triage_reconcile_skipped", tenantId: tenant.id, reason: "no_live_deployment" });
       return heal(tenant.id);
     }
@@ -475,11 +483,13 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) {
       return false;
     }
     const runs = mergeObservedRuns([...observed, ...workers]);
-    let deployment = newestLive(deployments);
-    try {
-      deployment = await route(tenant, deployments, observed);
-    } catch (err) {
-      deps.log({ level: "error", msg: "triage_rotation_failed", tenantId: tenant.id, error: String(err) });
+    let deployment = listener ?? newestLive(deployments);
+    if (!listener) {
+      try {
+        deployment = await route(tenant, deployments, observed);
+      } catch (err) {
+        deps.log({ level: "error", msg: "triage_rotation_failed", tenantId: tenant.id, error: String(err) });
+      }
     }
     if (!deployment) {
       deps.log({ level: "warn", msg: "triage_reconcile_skipped", tenantId: tenant.id, reason: "no_live_deployment" });

apps/hub/src/integrations/app.ts

@@ -397,7 +397,7 @@ export function createIntegrationsApp(guards: IntegrationGuards, handlers: Integ
         ...invalid,
         ...failures,
         502: json("`source_failed`: the pull request could not be read; `hub_unavailable`: it could not be queued", Failure),
-        503: json("`service_not_running`: the pr-triage service is not running; `github_rate_limited`: GitHub's rate limit is reached, `Retry-After` and `resetAt` say when it lifts", Failure),
+        503: json("`service_not_running`: neither the listener nor the pr-triage service is running; `github_rate_limited`: GitHub's rate limit is reached, `Retry-After` and `resetAt` say when it lifts", Failure),
       },
     }),
     member,

apps/hub/src/manager-state.postgres.test.ts

@@ -4,7 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto";
 import { sql } from "drizzle-orm";
 import { createDB, runMigrations, schema } from "@intx/db";
 import type { Tx } from "./github/admin-github-grant.js";
-import { done, recordEvent, take, type PullRequestRef } from "./manager-state.js";
+import { done, liveListener, recordEvent, take, type PullRequestRef } from "./manager-state.js";
 import { migrateTriageTables } from "./triage-db.js";
 
 const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -396,6 +396,20 @@ test("PostgreSQL: a manager deployed inside the startup window, or one with a si
   });
 }, 120_000);
 
+test("PostgreSQL: a listener counts as live as a manager does, and one deployed without a sidecar past the startup window is released", async () => {
+  await withDatabase(async function listener(db) {
+    await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" });
+    await db.insert(schema.workflowRun).values({ id: "run_listener", anchorRunId: "run_listener", definitionId: "wfd_listener", tenantId: TENANT, status: "deployed", createdAt: new Date() });
+    expect(await liveListener(db, TENANT, "github")).toMatchObject({ runId: "run_listener" });
+    expect(await liveListener(db, TENANT, "bitbucket-datacenter")).toBeUndefined();
+    await db.update(schema.workflowRun).set({ createdAt: new Date(Date.now() - 3 * MINUTE) }).where(sql`${schema.workflowRun.id} = 'run_listener'`);
+    expect(await liveListener(db, TENANT, "github")).toBeUndefined();
+    const [left] = await db.select({ cancellationRequestedAt: schema.workflowRun.cancellationRequestedAt, cancellationReason: schema.workflowRun.cancellationReason }).from(schema.workflowRun).where(sql`${schema.workflowRun.id} = 'run_listener'`);
+    expect(left?.cancellationRequestedAt).toBeInstanceOf(Date);
+    expect(left?.cancellationReason).toBe("listener deployed without a sidecar");
+  });
+}, 120_000);
+
 test("PostgreSQL: a malformed event payload or reviewer is refused", async () => {
   await withDatabase(async function badPayload(db) {
     await startManager(db, GITHUB, "run_a");