refactor(hub): read verdicts and catch-up from the managers

apps/hub/src/github/coalescer.test.ts

@@ -375,7 +375,7 @@ test("a paused-label head whose pull request cannot be read is left failed, and
   expect(mails).toEqual([]);
   expect(state.rows()).toEqual([expect.objectContaining({ status: "failed", error: "delivery failed: Error: 502 Bad Gateway" })]);
   const pr = { number: 8, headSha: "sha1", updatedAt: new Date(START).toISOString(), draft: false };
-  const plan = planTenant({ repos: [{ name: REPO, prs: [pr], rows: state.rows() }], runs: new Map(), now: new Date(START + DEFAULT_RECONCILE_POLICY.backoffMaxMs), policy: DEFAULT_RECONCILE_POLICY, workflowVersion: undefined });
+  const plan = planTenant({ repos: [{ name: REPO, prs: [pr], rows: state.rows() }], runs: new Map(), now: new Date(START + DEFAULT_RECONCILE_POLICY.backoffMaxMs), policy: DEFAULT_RECONCILE_POLICY });
   expect(plan.get(REPO)?.enqueue.map((entry) => entry.number)).toEqual([8]);
 });
 

apps/hub/src/github/deployment.ts

@@ -1,4 +1,4 @@
-import { and, desc, eq, gt, inArray, isNotNull, isNull, ne, or, sql } from "drizzle-orm";
+import { and, desc, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm";
 import { schema, type DB, type ResolvedSenderPrincipal, type WorkflowRunLaunchSpecStore } from "@intx/db";
 import { MAIL_ADMISSION_DENIED_CODE } from "@intx/hub-sessions";
 import type { RunTriggerMaterialize } from "@corbits/webhooks";
@@ -50,12 +50,8 @@ export function systemSenderMaterializer(materialize: RunTriggerMaterialize, sen
   };
 }
 
-/**
- * `cancelling` once cancellation was requested: the run stays live until the lifecycle sweep stops it. `workflowVersion`
- * once read from its launch spec, for one pinned from the package registry; `offeringId`, the decision-model offering
- * the deployment resolved, once read from its launch spec.
- */
-export type LiveDeployment = { runId: string; address: string; createdAt: Date; cancelling: boolean; endedAt?: Date; workflowVersion?: string; offeringId?: string };
+/** `cancelling` once cancellation was requested: the run stays live until the lifecycle sweep stops it. */
+export type LiveDeployment = { runId: string; address: string; createdAt: Date; cancelling: boolean };
 
 export type TenantDeployment = LiveDeployment & { workflow: string };
 
@@ -72,26 +68,24 @@ function markedAs(kind: DeploymentKind) {
   return kind === "worker" ? eq(marker, "true") : isNull(marker);
 }
 
-type DeploymentQuery = { tenantId: string; definitionName?: string; kind: DeploymentKind; endedSince?: Date };
+type DeploymentQuery = { tenantId: string; definitionName?: string; kind: DeploymentKind };
 
 /**
  * A workflow the portal publishes per person deploys from an asset named for its publisher and release, which names
  * its definition too; its frozen projection keeps the workflow's own id, so that is the name a deployment is found by.
  */
 export const workflowName = sql<string>`coalesce(${schema.workflowRunLaunchSpec.frozenApprovalBundle} -> 'projection' ->> 'id', ${schema.workflowDefinition.name})`;
 
-/** Matches how the hub lists deployments: a deployment is its anchor run. Newest first; live, or ended since `endedSince`. */
+/** Matches how the hub lists deployments: a deployment is its anchor run. Newest first, live only. */
 async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<TenantDeployment[]> {
   const { workflowRun, workflowDefinition, workflowRunLaunchSpec, tenant, liveWorkflowRunStatuses } = schema;
-  const live = inArray(workflowRun.status, [...liveWorkflowRunStatuses]);
   const rows = await db
     .select({
       runId: workflowRun.id,
       workflow: workflowName,
       domain: tenant.domain,
       createdAt: workflowRun.createdAt,
       cancellationRequestedAt: workflowRun.cancellationRequestedAt,
-      endedAt: workflowRun.endedAt,
     })
     .from(workflowRun)
     .innerJoin(workflowDefinition, eq(workflowRun.definitionId, workflowDefinition.id))
@@ -102,7 +96,7 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
       query.definitionName === undefined ? undefined : eq(workflowName, query.definitionName),
       isNotNull(workflowRun.anchorRunId),
       eq(workflowRun.id, workflowRun.anchorRunId),
-      query.endedSince === undefined ? live : or(live, gt(workflowRun.endedAt, query.endedSince)),
+      inArray(workflowRun.status, [...liveWorkflowRunStatuses]),
       markedAs(query.kind),
     ))
     .orderBy(desc(workflowRun.createdAt), desc(workflowRun.id));
@@ -111,7 +105,6 @@ async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<T
     address: `${row.runId}@${row.domain}`,
     createdAt: row.createdAt,
     cancelling: row.cancellationRequestedAt !== null,
-    ...(row.endedAt !== null && { endedAt: row.endedAt }),
     workflow: row.workflow,
   }));
 }
@@ -121,34 +114,6 @@ export function resolveLiveDeployments(db: DB["db"], tenantId: string, definitio
   return deploymentsWhere(db, { tenantId, definitionName, kind });
 }
 
-/** Workers live or released since `endedSince`: a worker's one body run landed in its log before its anchor was cancelled, so the log outlives the deployment. */
-export function recentWorkers(db: DB["db"], tenantId: string, definitionName: string, endedSince: Date): Promise<TenantDeployment[]> {
-  return deploymentsWhere(db, { tenantId, definitionName, kind: "worker", endedSince });
-}
-
-export type ActiveTenant = { id: string; domain: string };
-
-/**
- * Tenants with a child run of the definition in flight, or ended or created since `endedSince`: a run inserted already
- * completed, as a Bitbucket run is, keeps a null `endedAt` (stock leaves a terminal run routable that way), so its
- * creation time is what says it just ended, and its verdict is recorded within a pass.
- */
-export async function activeTenants(db: DB["db"], definitionName: string, what: "runs" | "deployments", endedSince?: Date): Promise<ActiveTenant[]> {
-  const { workflowRun, workflowDefinition, tenant, liveWorkflowRunStatuses } = schema;
-  const running = eq(workflowRun.status, "running");
-  const childRuns = endedSince === undefined ? running : or(running, gt(workflowRun.endedAt, endedSince), gt(workflowRun.createdAt, endedSince));
-  return db
-    .selectDistinct({ id: tenant.id, domain: tenant.domain })
-    .from(workflowRun)
-    .innerJoin(workflowDefinition, eq(workflowRun.definitionId, workflowDefinition.id))
-    .innerJoin(tenant, eq(workflowRun.tenantId, tenant.id))
-    .where(and(
-      eq(workflowDefinition.name, definitionName),
-      what === "runs" ? ne(workflowRun.id, workflowRun.anchorRunId) : eq(workflowRun.id, workflowRun.anchorRunId),
-      what === "runs" ? childRuns : inArray(workflowRun.status, [...liveWorkflowRunStatuses]),
-    ));
-}
-
 export function newestLive(deployments: readonly LiveDeployment[]): LiveDeployment | undefined {
   return deployments.find((live) => !live.cancelling);
 }
@@ -201,20 +166,6 @@ function pinFor(bundle: FrozenApprovalBundle): string | undefined {
   return pinOf(bundle.closure);
 }
 
-export async function withWorkflowVersions<T extends LiveDeployment>(
-  launchSpecs: Pick<WorkflowRunLaunchSpecStore, "get">,
-  deployments: readonly T[],
-): Promise<T[]> {
-  const versioned: T[] = [];
-  for (const deployment of deployments) {
-    const spec = await launchSpecs.get(deployment.runId);
-    const workflowVersion = spec ? closureVersions(spec.frozenApprovalBundle)?.workflow.version : undefined;
-    const offeringId = spec?.defaultSourceOfferingId ?? undefined;
-    versioned.push({ ...deployment, ...(workflowVersion !== undefined && { workflowVersion }), ...(offeringId !== undefined && { offeringId }) });
-  }
-  return versioned;
-}
-
 /** The reason the hub gave when it replaced a busy deployment with a fresh copy, which earlier audit entries carry. */
 export const ROTATION_REASON = "Replaced by a fresh deployment of the same workflow";
 

apps/hub/src/github/github-read-access.postgres.test.ts

@@ -308,7 +308,7 @@ test("PostgreSQL: a code review is sent once per head as one COMMENT review, inl
 test("PostgreSQL: every verdict route needs workflow-run read, which holding the GitHub credential does not give", async () => {
   const unused = {} as never;
   const app = guardedApp(handle.db, sessionOf, grants, {
-    triageVerdicts: createTriageVerdicts({ ...deps, list, verdicts: unused, observer: unused, ready: unused, versions: unused, openPulls: unused, triageState: unused, now: unused }),
+    triageVerdicts: createTriageVerdicts({ ...deps, list, verdicts: unused, managerRuns: unused, openPulls: unused, triageState: unused, now: unused }),
     triageTimeline: createTriageTimeline({ verdicts: unused }),
     triageStats: createTriageStats({ verdicts: unused, outcomes: unused, now: unused, log: unused }),
     audit: createAuditLog({ db: unused, artifacts: unused, readCheckPack: unused, runReader: unused, sameSource: unused, authorize: unused, workflows: ["pr-triage"] }),

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

@@ -12,7 +12,7 @@ import { createGithubPrTriage } from "./pr-triage.js";
 import { DEFAULT_RECONCILE_POLICY } from "./reconcile-plan.js";
 import { createTriageReconciler } from "./triage-reconciler.js";
 import { createTriageStateStore, type TriageStateStore } from "./triage-state-store.js";
-import type { ObservedRuns } from "./triage-runs.js";
+import type { ObservedRuns } from "./manager-runs.js";
 
 /** No deployment holds the credential in these tests. */
 async function noKeyPush(): Promise<void> {}
@@ -39,7 +39,7 @@ function stubDb() {
 }
 
 async function noRuns(): Promise<ObservedRuns> {
-  return { byRepo: new Map(), runCount: 0, verdicts: [], approved: [] };
+  return { byRepo: new Map() };
 }
 
 function route(store: TriageStateStore, deliver: Deliver) {
@@ -67,7 +67,6 @@ function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver,
     listener: async () => DEPLOYMENT,
     openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })),
     activity: { stale: () => true },
-    observeTenant: async () => ({ deployments: [DEPLOYMENT], observed: [await noRuns()], workers: [] }),
     managerRuns: noRuns,
     headChecks: async () => checks,
     verdicts: { latest: async () => [] },

apps/hub/src/github/manager-runs.postgres.test.ts

@@ -80,10 +80,10 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
     const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
     function plan(rows: PrTriageRow[], runs: Awaited<ReturnType<typeof managerRuns>>) {
       const ci = new Map([[runKey(8, "h1"), "pending" as const]]);
-      return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows, ci }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: "1.0.0" }).get(REPO)!;
+      return planTenant({ repos: [{ name: REPO, prs: [open(7, now), open(8, now)], rows, ci }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
     }
 
-    const running = plan([queued], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
+    const running = plan([queued], await managerRuns(TENANT, [pr7, pr8]));
     expect(running.rows.find((row) => row.number === 7)).toMatchObject({ status: "running", runId: "run_triaged" });
     expect(running.enqueue.map((queuedPr) => queuedPr.number)).not.toContain(7);
 
@@ -92,9 +92,9 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager
     await take(db, TENANT, held, "run_held");
     await done(db, TENANT, held, "run_held", [], { head: "h1", waiting: true });
     const firstSeenAt = new Date(now.getTime() - MINUTE).toISOString();
-    const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, workflowVersion: "1.0.0", firstSeenAt, updatedAt: firstSeenAt };
+    const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, firstSeenAt, updatedAt: firstSeenAt };
 
-    const settled = plan([...running.rows.filter((row) => row.number === 7), fresh], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
+    const settled = plan([...running.rows.filter((row) => row.number === 7), fresh], await managerRuns(TENANT, [pr7, pr8]));
     expect(settled.rows).toEqual([expect.objectContaining({ number: 7, status: "triaged", runId: "run_triaged" }), expect.objectContaining({ number: 8, status: "waiting", runId: "run_held" })]);
     expect(settled.enqueue).toEqual([]);
   });
@@ -111,8 +111,8 @@ test("PostgreSQL: on the listener path a head is mailed once while its manager r
     let mails = 0;
     async function reconcile(afterMs: number): Promise<PrTriageRow> {
       now = new Date(start.getTime() + afterMs);
-      const runs = await managerRuns(TENANT, [pr], undefined);
-      const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: undefined }).get(REPO)!;
+      const runs = await managerRuns(TENANT, [pr]);
+      const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, start)], rows }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
       rows = plan.rows;
       mails += plan.enqueue.length;
       return rows[0]!;
@@ -143,11 +143,11 @@ test("PostgreSQL: a manager that went away mid-pass fails its pass's head, so th
     await take(db, TENANT, pr7, "run_gone");
     await db.update(schema.workflowRun).set({ status: "failed" }).where(sql`${schema.workflowRun.id} = 'run_gone'`);
 
-    const runs = await managerRuns(TENANT, [pr7], undefined);
+    const runs = await managerRuns(TENANT, [pr7]);
     expect(runs.byRepo.get(REPO)?.get(runKey(7, "h1"))).toEqual([expect.objectContaining({ runId: "run_gone", status: "failed" })]);
     const queuedAt = new Date(now.getTime() - MINUTE).toISOString();
     const queued: PrTriageRow = { number: 7, headSha: "h1", status: "queued", attempts: 1, firstSeenAt: queuedAt, queuedAt, updatedAt: queuedAt };
-    const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY, workflowVersion: undefined }).get(REPO)!;
+    const plan = planTenant({ repos: [{ name: REPO, prs: [open(7, now)], rows: [queued] }], runs: runs.byRepo, now, policy: DEFAULT_RECONCILE_POLICY }).get(REPO)!;
     expect(plan.rows).toEqual([expect.objectContaining({ status: "failed", runId: "run_gone", error: "run failed" })]);
     expect(plan.enqueue).toEqual([]);
   });

apps/hub/src/github/manager-runs.ts

@@ -6,37 +6,38 @@ import type { DB } from "@intx/db";
 import { unsettledBy } from "@corbits/triage-workflows/logic/settled";
 import { managedPulls, type PullRequestRef } from "../manager-state.js";
 import { runKey, type ObservedRun } from "./reconcile-plan.js";
-import type { ObservedRuns } from "./triage-runs.js";
 import type { VerdictRecord, VerdictRecordStore } from "./verdict-record-store.js";
 
-/** The runs of `pulls`' managers; `workflowVersion` is the version new mail goes to, which a manager's runs count as. */
-export type ManagerRuns = (tenantId: string, pulls: readonly PullRequestRef[], workflowVersion: string | undefined) => Promise<ObservedRuns>;
+/** Runs by repository, then by `${number}@${headSha}`. */
+export type ObservedRuns = { byRepo: Map<string, Map<string, ObservedRun[]>> };
+
+/** The runs of `pulls`' managers. */
+export type ManagerRuns = (tenantId: string, pulls: readonly PullRequestRef[]) => Promise<ObservedRuns>;
 
 function settlesBy(verdict: VerdictRecord): Pick<ObservedRun, "unsettled" | "unconfirmed"> {
   const { detail } = verdict;
   return unsettledBy({ degraded: verdict.degraded, reason: detail?.reason, headSha: detail?.headSha, state: verdict.state, checks: detail?.checks }, verdict.headSha);
 }
 
 export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStore, "ofRuns">, now: () => Date): ManagerRuns {
-  return async function managerRuns(tenantId, asked, workflowVersion) {
+  return async function managerRuns(tenantId, asked) {
     const pulls = await managedPulls(db, tenantId, asked);
     const recorded = await verdicts.ofRuns(tenantId, [...new Set(pulls.flatMap((pull) => pull.history.map((pass) => pass.runId)))], asked);
-    const stamp = workflowVersion === undefined ? {} : { workflowVersion };
     const byRepo: ObservedRuns["byRepo"] = new Map();
 
     function observed(repo: string, number: number, head: string, run: ObservedRun): void {
       const byHead = byRepo.get(repo) ?? new Map<string, ObservedRun[]>();
       byRepo.set(repo, byHead);
       const key = runKey(number, head);
       const listed = byHead.get(key);
-      if (listed) listed.push({ ...run, ...stamp });
-      else byHead.set(key, [{ ...run, ...stamp }]);
+      if (listed) listed.push(run);
+      else byHead.set(key, [run]);
     }
 
     for (const { pr, history, live, gone } of pulls) {
       const { source } = pr;
       // The hub checked the run is up, so it is running now and never stuck.
-      if (live?.head !== undefined) observed(pr.repo, pr.number, live.head, { runId: live.runId, source, status: "running", startedAt: now().toISOString() });
+      if (live?.head !== undefined) observed(pr.repo, pr.number, live.head, { runId: live.runId, source, status: "running", startedAt: now().toISOString(), ...(live.pass && { passAt: live.pass.takenAt }) });
       if (gone) observed(pr.repo, pr.number, gone.pass.head, { runId: gone.runId, source, status: "failed", startedAt: gone.pass.takenAt });
       // A head's last pass decides whether it still waits for CI; a waiting run is listed first so it outranks an older verdict.
       const lastOfHead = new Map(history.map((pass) => [pass.head, pass]));
@@ -47,6 +48,6 @@ export function createManagerRuns(db: DB["db"], verdicts: Pick<VerdictRecordStor
         observed(pr.repo, pr.number, verdict.headSha, { runId: verdict.runId, source, status: "completed", startedAt: verdict.startedAt.toISOString(), ...settlesBy(verdict), ...(verdict.trigger !== null && { trigger: verdict.trigger }) });
       }
     }
-    return { byRepo, runCount: 0, verdicts: [], approved: [] };
+    return { byRepo };
   };
 }