fix(manager): record the manager's verdict when it reports done

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

@@ -153,6 +153,7 @@ function harness(db: ReturnType<typeof createDB>["db"]) {
     openHeadsFor: async () => async () => [{ ...HEAD, updatedAt: ago(60 * MINUTE).toISOString(), draft: false }],
     activity: { stale: () => true },
     observeTenant: observer.observe,
+    managerRuns: async () => NO_RUNS,
     headChecks: async () => [],
     verdicts: { latest: async () => [] },
     store: {

apps/hub/src/github/deployment.ts

@@ -72,7 +72,7 @@ type DeploymentQuery = { tenantId: string; definitionName?: string; kind: Deploy
  * 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.
  */
-const workflowName = sql<string>`coalesce(${schema.workflowRunLaunchSpec.frozenApprovalBundle} -> 'projection' ->> 'id', ${schema.workflowDefinition.name})`;
+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`. */
 async function deploymentsWhere(db: DB["db"], query: DeploymentQuery): Promise<TenantDeployment[]> {

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

@@ -345,6 +345,7 @@ test("PostgreSQL: approving a held Do needs its kind's grant and mails pr-triage
     async liveDeployments() { return [{ runId: "wfr_anchor", address: "wfr_anchor@read-access.test", createdAt: new Date(), cancelling: false }]; },
     async deliver(_tenantId, address, payload) { mailed.push({ address, payload }); },
     async toListener() { return false; },
+    async passRequest() { throw new Error("pr-triage's verdict is in its run's log"); },
     log() {},
   });
   async function approve(user: string, step: { effectId: string; kind: ActionKind }, runId = "wfr_verdict") {
@@ -372,3 +373,39 @@ test("PostgreSQL: approving a held Do needs its kind's grant and mails pr-triage
   expect(await approve("usr_approver", ship)).toMatchObject({ status: 409, body: { error: { code: "verdict_superseded" } } });
   expect(mailed).toHaveLength(2);
 });
+
+test("PostgreSQL: a held Do of a verdict a manager reported is approved from the recorded verdict and the mirror request its pass kept, and goes to the listener; once the run's newer verdict replaced it, it is superseded", async () => {
+  await handle.db.insert(schema.workflowDefinition).values({ id: "wfd_manager", tenantId: TENANT, name: "triage-manager-github", status: "stopped" });
+  await handle.db.insert(schema.workflowRun).values({ id: "wfr_manager", definitionId: "wfd_manager", anchorRunId: "wfr_manager", tenantId: TENANT, status: "running" });
+  const target = { labels: ["api"] };
+  const tag = { list: "pass" as const, index: 0, kind: "labels" as const, target, held: true, effectId: doEffectId({ repo: REPO, number: 8, headSha: HEAD, list: "pass", index: 0, kind: "labels", target }) };
+  const request = { repo: REPO, number: 8, labels: ["Triage: Pass"], owned: [], comment: "Ready for review." };
+  const recorded: Array<Pick<VerdictRecord, "runId" | "headSha" | "degraded" | "runStatus" | "detail">> = [{ runId: "wfr_manager", headSha: HEAD, degraded: null, runStatus: "completed", detail: { headSha: HEAD, actions: [tag] } }];
+  const kept: unknown[] = [];
+  const listened: unknown[] = [];
+  const heldDos = createGithubHeldDos({
+    ...deps,
+    runReader: { async readRunEvents() { return [{ seq: 0, type: "RunStarted", body: { type: "RunStarted", seq: 0 } }]; } },
+    verdicts: { async forPull() { return recorded as VerdictRecord[]; } },
+    async liveDeployments() { return []; },
+    async deliver() { throw new Error("a manager's approval goes to the listener"); },
+    async toListener(_tenantId, payload) {
+      listened.push(payload);
+      return true;
+    },
+    async passRequest(_tenantId, pr, runId, head) {
+      kept.push({ pr, runId, head });
+      return request;
+    },
+    log() {},
+  });
+  const res = await heldDos({ runId: "wfr_manager", repo: REPO, number: 8, effectId: tag.effectId, kind: "labels" }, member("usr_approver"));
+  expect({ status: res.status, body: await res.json() }).toEqual({ status: 202, body: { status: "sent" } });
+  expect(kept).toEqual([{ pr: { source: "github", repo: REPO, number: 8 }, runId: "wfr_manager", head: HEAD }]);
+  expect(listened).toEqual([{ kind: "approve-do", runId: "wfr_manager", principalId: "prn_approver", list: "pass", index: 0, request, step: { effectId: tag.effectId, headSha: HEAD, kind: "labels", target } }]);
+
+  recorded[0] = { ...recorded[0]!, headSha: "newer", detail: { headSha: "newer", actions: [] } };
+  const replaced = await heldDos({ runId: "wfr_manager", repo: REPO, number: 8, effectId: tag.effectId, kind: "labels" }, member("usr_approver"));
+  expect({ status: replaced.status, body: await replaced.json() }).toMatchObject({ status: 409, body: { error: { code: "verdict_superseded" } } });
+  expect(listened).toHaveLength(1);
+});

apps/hub/src/github/held-dos.ts

@@ -14,11 +14,12 @@ import { doEffectId } from "@corbits/triage-workflows/logic/actions";
 import { unrunnable } from "@corbits/triage-workflows/logic/execute-do";
 import { doRefused, failure, type PortalCredentialDeps } from "./portal-credential.js";
 import type { Member } from "../integrations/app.js";
+import type { PullRequestRef } from "../manager-state.js";
 import type { VerdictRecordStore } from "./verdict-record-store.js";
 import { DeploymentNotReadyError, newestLive, type LiveDeployment } from "./deployment.js";
 import { prWorkflowVariantOf } from "./pr-workflow.js";
 import { triageWorkflowFor } from "./source-workflow.js";
-import { repoRecords, sourceOf, triageNs } from "./tenant-config.js";
+import { repoRecords, sourceOf, triageNs, type RepoRecord } from "./tenant-config.js";
 import { verdictFor } from "./triage-runs.js";
 
 export const GITHUB_HELD_DOS_PATH = "/api/integrations/github-held-dos";
@@ -46,6 +47,8 @@ export type GithubHeldDosDeps = PortalCredentialDeps & {
   deliver: (tenantId: string, address: string, payload: unknown) => Promise<void>;
   /** Hands the approval to the workspace's listener, whose manager runs it; false when the workspace has none. */
   toListener: (tenantId: string, payload: unknown) => Promise<boolean>;
+  /** The mirror request a manager's pass kept with the verdict it posted on the head. */
+  passRequest: (tenantId: string, pr: PullRequestRef, runId: string, head: string) => Promise<Record<string, unknown> | undefined>;
   log: (entry: Record<string, unknown>) => void;
 };
 
@@ -117,6 +120,19 @@ export function createGithubHeldDos(deps: GithubHeldDosDeps) {
     return at >= 0 && verdicts.slice(at + 1).some((verdict) => verdict.degraded === null && (verdict.runStatus === null || verdict.runStatus === "completed"));
   }
 
+  /**
+   * A manager reports its verdict when its pass is done, so its Dos are the recorded verdict's, with the mirror request
+   * the pass kept. A run's newest verdict replaces its earlier one, so a Do the row no longer holds was on a verdict a
+   * newer one superseded.
+   */
+  async function managedVerdict(tenantId: string, record: RepoRecord, { number, runId, effectId }: Body): Promise<Record<string, unknown> | Response | undefined> {
+    const recorded = (await deps.verdicts.forPull(tenantId, record.name, number)).findLast((verdict) => verdict.runId === runId);
+    if (!recorded?.detail) return undefined;
+    if (!recorded.detail.actions?.some((action) => action.effectId === effectId)) return failure(409, "verdict_superseded", "A newer triage result exists.");
+    const request = await deps.passRequest(tenantId, { source: sourceOf(record), repo: record.name, number }, runId, recorded.headSha);
+    return request && { headSha: recorded.detail.headSha, actions: recorded.detail.actions, request };
+  }
+
   return async function approve(body: typeof HeldDoBody.infer, { tenantId, principalId }: Member): Promise<Response> {
     // A verdict recorded before agent Dos were removed may still hold one.
     if (body.kind === "agent") return failure(409, "do_not_runnable", AGENT_DO);
@@ -132,7 +148,8 @@ export function createGithubHeldDos(deps: GithubHeldDosDeps) {
     const run = await runEvents(tenantId, tenant.domain, runId);
     const events = run?.events;
     if (!run || !events?.length) return failure(404, "run_not_found", "No such run in this workspace.");
-    const verdict = verdictFor(events, repo, number);
+    const verdict = verdictFor(events, repo, number) ?? (await managedVerdict(tenantId, record, body));
+    if (verdict instanceof Response) return verdict;
     if (!verdict) return failure(404, "verdict_not_found", `The run has no verdict for ${repo}#${number}.`);
     const approval = heldDo(verdict, body, principalId);
     if (approval instanceof Response) return approval;

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

@@ -52,6 +52,7 @@ function route(store: TriageStateStore, deliver: Deliver) {
     readCheckPack: async () => ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }),
     liveDeployments: async () => [DEPLOYMENT],
     observeRuns: noRuns,
+    managerRuns: noRuns,
     store,
     send: async ({ payload }) => deliver(payload),
     policy: DEFAULT_RECONCILE_POLICY,
@@ -71,6 +72,7 @@ function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver,
     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 () => [] },
     store,

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

@@ -0,0 +1,154 @@
+// Requires PostgreSQL at TEST_DB.
+import { expect, test } from "bun:test";
+import { randomBytes } from "node:crypto";
+import { sql } from "drizzle-orm";
+import { createDB, runMigrations, schema } from "@intx/db";
+import type { PrTriageRow } from "@corbits/triage-contracts";
+import { done, recordEvent, take, type PullRequestRef } from "../manager-state.js";
+import { migrateTriageTables } from "../triage-db.js";
+import { createManagerRuns } from "./manager-runs.js";
+import { DEFAULT_RECONCILE_POLICY, planTenant, runKey, type OpenPr } from "./reconcile-plan.js";
+import { createVerdictRecordStore, replaceVerdict } from "./verdict-record-store.js";
+
+const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
+const TENANT = "tnt_manager_runs";
+const REPO = "acme/widgets";
+const MINUTE = 60_000;
+
+type Db = ReturnType<typeof createDB>["db"];
+
+async function withDatabase(run: (db: Db) => Promise<void>): Promise<void> {
+  const database = `corbits_manager_runs_${randomBytes(6).toString("hex")}`;
+  const admin = createDB(TEST_DB);
+  await admin.db.execute(sql.raw(`create database ${database}`));
+  const config = { ...TEST_DB, database };
+  let handle: ReturnType<typeof createDB> | undefined;
+  try {
+    await runMigrations(config, { schema: "public" });
+    handle = createDB(config);
+    await migrateTriageTables(handle.db, "public");
+    await handle.db.insert(schema.tenant).values({ id: TENANT, name: "Manager runs", slug: "manager-runs", domain: "manager-runs.test" });
+    await handle.db.insert(schema.workflowDefinition).values({ id: "wfd_manager", tenantId: TENANT, name: "triage-manager-github" });
+    await run(handle.db);
+  } finally {
+    await handle?.close();
+    await admin.db.execute(sql.raw(`drop database if exists ${database} with (force)`));
+    await admin.close();
+  }
+}
+
+async function managerUp(db: Db, runId: string): Promise<void> {
+  await db.insert(schema.workflowRun).values({ id: runId, anchorRunId: runId, definitionId: "wfd_manager", tenantId: TENANT, status: "running" });
+}
+
+async function mailed(db: Db, pr: PullRequestRef, head: string): Promise<void> {
+  await recordEvent(db, TENANT, pr, `opened-${pr.number}`, { type: "webhook", head, payload: { kind: "pr", event: "pull_request", action: "opened", headSha: head } });
+}
+
+const pr7: PullRequestRef = { source: "github", repo: REPO, number: 7 };
+const pr8: PullRequestRef = { source: "github", repo: REPO, number: 8 };
+
+/** The manager posts a settled verdict on h1 and reports done, recording it as the done route does, in done's transaction. */
+async function reportedDone(db: Db, pr: PullRequestRef, runId: string, now: Date): Promise<void> {
+  const result = await done(db, TENANT, pr, runId, [], { head: "h1", request: { repo: REPO, number: pr.number, labels: [], owned: [], comment: "" } }, async function record(tx, closing) {
+    if (!closing) throw new Error("a reporting pass has a closing");
+    await replaceVerdict(tx, TENANT, {
+      runId, repo: REPO, number: pr.number, headSha: closing.head, startedAt: new Date(closing.startedAt), verdictAt: now, settled: true, state: "ready-monitoring",
+      actor: "author", humanGated: false, degraded: null, checks: [], dos: [], commented: true, detail: { headSha: closing.head, checks: [] }, runStatus: "completed",
+    });
+  });
+  if (result.outcome === "refused") throw new Error("done refused the manager the claim names");
+}
+
+function open(number: number, now: Date): OpenPr {
+  return { number, headSha: "h1", updatedAt: new Date(now.getTime() - 30 * MINUTE).toISOString(), draft: false };
+}
+
+test("PostgreSQL: the reconciler does not mail a head again while a live manager holds it, and settles a head a manager triaged or held for CI", async () => {
+  await withDatabase(async function reconciled(db) {
+    const now = new Date();
+    const verdicts = createVerdictRecordStore(db);
+    const managerRuns = createManagerRuns(db, verdicts, () => now);
+    const triaged = pr7;
+    const held = pr8;
+    await managerUp(db, "run_triaged");
+    await managerUp(db, "run_held");
+    await mailed(db, triaged, "h1");
+    await take(db, TENANT, triaged, "run_triaged");
+
+    const queuedAt = new Date(now.getTime() - 20 * MINUTE).toISOString();
+    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)!;
+    }
+
+    const running = plan([queued], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
+    expect(running.rows.find((row) => row.number === 7)).toMatchObject({ status: "running", runId: "run_triaged" });
+    expect(running.enqueue.map((queuedPr) => queuedPr.number)).not.toContain(7);
+
+    await reportedDone(db, triaged, "run_triaged", now);
+    await mailed(db, held, "h1");
+    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 settled = plan([...running.rows.filter((row) => row.number === 7), fresh], await managerRuns(TENANT, [pr7, pr8], "1.0.0"));
+    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([]);
+  });
+}, 60_000);
+
+test("PostgreSQL: on the listener path a head is mailed once while its manager runs for hours, and not again once it reported done", async () => {
+  await withDatabase(async function oneMail(db) {
+    const start = new Date();
+    let now = start;
+    const verdicts = createVerdictRecordStore(db);
+    const managerRuns = createManagerRuns(db, verdicts, () => now);
+    const pr: PullRequestRef = { source: "github", repo: REPO, number: 7 };
+    let rows: PrTriageRow[] = [];
+    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)!;
+      rows = plan.rows;
+      mails += plan.enqueue.length;
+      return rows[0]!;
+    }
+
+    expect(await reconcile(0)).toMatchObject({ status: "queued" });
+    await managerUp(db, "run_manager");
+    await mailed(db, pr, "h1");
+    await take(db, TENANT, pr, "run_manager");
+    for (const afterMs of [MINUTE, 11 * MINUTE, 2 * 60 * MINUTE, 7 * 60 * MINUTE]) {
+      expect(await reconcile(afterMs)).toMatchObject({ status: "running", runId: "run_manager" });
+    }
+
+    await reportedDone(db, pr, "run_manager", now);
+    for (const afterMs of [7 * 60 * MINUTE + MINUTE, 14 * 60 * MINUTE]) {
+      expect(await reconcile(afterMs)).toMatchObject({ status: "triaged", runId: "run_manager" });
+    }
+    expect(mails).toBe(1);
+  });
+}, 60_000);
+
+test("PostgreSQL: a manager that went away mid-pass fails its pass's head, so the head is retried with backoff", async () => {
+  await withDatabase(async function gone(db) {
+    const now = new Date();
+    const managerRuns = createManagerRuns(db, createVerdictRecordStore(db), () => now);
+    await managerUp(db, "run_gone");
+    await mailed(db, pr7, "h1");
+    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);
+    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)!;
+    expect(plan.rows).toEqual([expect.objectContaining({ status: "failed", runId: "run_gone", error: "run failed" })]);
+    expect(plan.enqueue).toEqual([]);
+  });
+}, 60_000);