diff --git a/apps/hub/src/github/deployment-heal.postgres.test.ts b/apps/hub/src/github/deployment-heal.postgres.test.ts index 54289d7cb..103268bb3 100644 --- a/apps/hub/src/github/deployment-heal.postgres.test.ts +++ b/apps/hub/src/github/deployment-heal.postgres.test.ts @@ -153,6 +153,7 @@ function harness(db: ReturnType["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: { diff --git a/apps/hub/src/github/deployment.ts b/apps/hub/src/github/deployment.ts index c913fce96..5be30c22b 100644 --- a/apps/hub/src/github/deployment.ts +++ b/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`coalesce(${schema.workflowRunLaunchSpec.frozenApprovalBundle} -> 'projection' ->> 'id', ${schema.workflowDefinition.name})`; +export const workflowName = sql`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 { diff --git a/apps/hub/src/github/github-read-access.postgres.test.ts b/apps/hub/src/github/github-read-access.postgres.test.ts index a1e32c93f..18cda0716 100644 --- a/apps/hub/src/github/github-read-access.postgres.test.ts +++ b/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> = [{ 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); +}); diff --git a/apps/hub/src/github/held-dos.ts b/apps/hub/src/github/held-dos.ts index 944d35306..e4e7b8f1b 100644 --- a/apps/hub/src/github/held-dos.ts +++ b/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; /** Hands the approval to the workspace's listener, whose manager runs it; false when the workspace has none. */ toListener: (tenantId: string, payload: unknown) => Promise; + /** 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 | undefined>; log: (entry: Record) => 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 | 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 { // 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; diff --git a/apps/hub/src/github/interleave.postgres.test.ts b/apps/hub/src/github/interleave.postgres.test.ts index 60f01a1a3..30694a450 100644 --- a/apps/hub/src/github/interleave.postgres.test.ts +++ b/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, diff --git a/apps/hub/src/github/manager-runs.postgres.test.ts b/apps/hub/src/github/manager-runs.postgres.test.ts new file mode 100644 index 000000000..f6a52a4ea --- /dev/null +++ b/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["db"]; + +async function withDatabase(run: (db: Db) => Promise): Promise { + 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 | 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 { + 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 { + 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 { + 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>) { + 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 { + 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); diff --git a/apps/hub/src/github/manager-runs.ts b/apps/hub/src/github/manager-runs.ts new file mode 100644 index 000000000..a35014f88 --- /dev/null +++ b/apps/hub/src/github/manager-runs.ts @@ -0,0 +1,52 @@ +// Pull requests' managers as the reconciler reads a deployment's runs, from their claims and the verdicts they +// reported: a manager that is up runs the newest head the hub recorded a mail for, one that went away mid-pass failed +// on its pass's head, a head its last pass held for CI waits for it, and each verdict a manager recorded settles its +// head as a pr-triage run's does. +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; + +function settlesBy(verdict: VerdictRecord): Pick { + 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, now: () => Date): ManagerRuns { + return async function managerRuns(tenantId, asked, workflowVersion) { + 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(); + 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 }]); + } + + 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 (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])); + for (const pass of lastOfHead.values()) { + if (pass.waiting) observed(pr.repo, pr.number, pass.head, { runId: pass.runId, source, status: "completed", startedAt: pass.takenAt, heldForCi: true }); + } + for (const verdict of recorded.filter((row) => row.repo === pr.repo && row.number === pr.number)) { + 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: [] }; + }; +} diff --git a/apps/hub/src/github/pr-triage.test.ts b/apps/hub/src/github/pr-triage.test.ts index a9f8c375b..055a25395 100644 --- a/apps/hub/src/github/pr-triage.test.ts +++ b/apps/hub/src/github/pr-triage.test.ts @@ -74,6 +74,7 @@ function handler(repo: Record, s: Stub, overrides: Partial ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }), liveDeployments: async () => [ANCHOR], observeRuns: observed([]), + managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }), store: stored([], s.saved), send: async ({ payload }) => { s.delivered.push({ payload }); @@ -188,6 +189,19 @@ describe("createGithubPrTriage", () => { expect(s.saved).toEqual([[{ ...old, queuedAt: NOW.toISOString(), updatedAt: NOW.toISOString() }]]); }); + test("a head a live manager holds is running, and one whose manager reported done or went away since it was queued is queued again at once", async () => { + const queued = queuedRow(60_000); + const manager = (run: ObservedRun): Partial => ({ store: stored([queued], s.saved), managerRuns: async () => observed([run])("", "") }); + const s = stub(); + await refused(await handler(ENABLED, s, manager({ runId: "run_manager", source: "github", status: "running", startedAt: NOW.toISOString() }))(request(), MEMBER), 409, "already_running"); + await refused(await handler(ENABLED, s, manager({ runId: "run_manager", source: "github", status: "completed", startedAt: new Date(NOW.getTime() - 120_000 - DEFAULT_RECONCILE_POLICY.clockSkewMs).toISOString() }))(request(), MEMBER), 409, "already_queued"); + expect(s.delivered).toEqual([]); + expect((await handler(ENABLED, s, manager({ runId: "run_manager", source: "github", status: "completed", startedAt: new Date(NOW.getTime() - 30_000).toISOString() }))(request(), MEMBER)).status).toBe(202); + expect(s.delivered).toHaveLength(1); + expect((await handler(ENABLED, s, manager({ runId: "run_manager", source: "github", status: "failed", startedAt: new Date(NOW.getTime() - 30_000).toISOString() }))(request(), MEMBER)).status).toBe(202); + expect(s.delivered).toHaveLength(2); + }); + test("a state write refused as stale is retried once on a fresh load, gate included; refused twice it answers conflict, never mailing", async () => { const s = stub(); expect((await handler(ENABLED, s, { store: stored([], s.saved, 1) })(request(), MEMBER)).status).toBe(202); diff --git a/apps/hub/src/github/pr-triage.ts b/apps/hub/src/github/pr-triage.ts index 9c26cec06..f8f3e43cd 100644 --- a/apps/hub/src/github/pr-triage.ts +++ b/apps/hub/src/github/pr-triage.ts @@ -5,7 +5,7 @@ import { eq } from "drizzle-orm"; import { type } from "arktype"; import { schema } from "@intx/db"; -import { hasEnabledRules, repoPolicy, type RepoSource } from "@corbits/triage-contracts"; +import { hasEnabledRules, repoPolicy, type PrTriageRow, type RepoSource } from "@corbits/triage-contracts"; import { errorText } from "@corbits/common"; import { mailPayload } from "./bridge.js"; import type { TriageJob } from "./coordinator.js"; @@ -13,11 +13,12 @@ import type { PolicyReader } from "./check-pack-store.js"; import { githubLimited } from "./github-refusal.js"; import { failure, sourceCredentialGrant, type PortalCredentialDeps } from "./portal-credential.js"; import type { Member } from "../integrations/app.js"; -import { isStuck, runKey, type ReconcilePolicy } from "./reconcile-plan.js"; +import type { ManagerRuns } from "./manager-runs.js"; +import { isStuck, runKey, type ObservedRun, type ReconcilePolicy } from "./reconcile-plan.js"; import { triageWorkflowFor } from "./source-workflow.js"; import { repoRecords, sourceOf, triageNs, triageStateKey, type RepoRecord } from "./tenant-config.js"; import { newestLive, type LiveDeployment } from "./deployment.js"; -import { mergeObservedRuns, observeDeployments, type ObserveRuns } from "./triage-runs.js"; +import { mergeObservedRuns, observeDeployments, type ObservedRuns, type ObserveRuns } from "./triage-runs.js"; import { isQueuedSince, isRunningSince, queueHead, restoreHead } from "./triage-queue.js"; import type { TriageStateStore } from "./triage-state-store.js"; @@ -32,6 +33,7 @@ export type GithubPrTriageDeps = PortalCredentialDeps & { /** 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; observeRuns: ObserveRuns; + managerRuns: ManagerRuns; store: TriageStateStore; send: (job: TriageJob) => Promise; policy: ReconcilePolicy; @@ -72,6 +74,19 @@ async function readyToTriage(deps: GithubPrTriageDeps, member: Member, repo: str return { ...granted, ...inputs }; } +/** The runs that named the head; none when no run did. */ +function headRuns(runs: ObservedRuns, repo: string, key: string): readonly ObservedRun[] { + const listed = runs.byRepo.get(repo)?.get(key); + return listed === undefined ? [] : listed; +} + +/** A manager that reported done on the head, or went away on it, since it was queued settles it before the reconciler's next pass marks it so. */ +function doneSince(row: PrTriageRow | undefined, runs: readonly ObservedRun[], policy: ReconcilePolicy): boolean { + if (row?.queuedAt === undefined) return false; + const since = new Date(row.queuedAt).getTime() - policy.clockSkewMs; + return runs.some((run) => run.status !== "running" && new Date(run.startedAt).getTime() >= since); +} + export function createGithubPrTriage(deps: GithubPrTriageDeps) { return async function handle({ repo, number }: typeof TriageBody.infer, member: Member): Promise { const { tenantId, principalId } = member; @@ -94,13 +109,15 @@ export function createGithubPrTriage(deps: GithubPrTriageDeps) { const headSha = openHead; const now = deps.now(); - const runs = mergeObservedRuns(await observeDeployments(deps.observeRuns, deployments, tenant.domain)).byRepo.get(repo)?.get(runKey(number, headSha)) ?? []; + const key = runKey(number, headSha); + const managed = headRuns(await deps.managerRuns(tenantId, [{ source: sourceOf(record), repo, number }], undefined), repo, key); + const runs = [...headRuns(mergeObservedRuns(await observeDeployments(deps.observeRuns, deployments, tenant.domain)), repo, key), ...managed]; if (runs.some((run) => run.status === "running" && !isStuck(run, now, deps.policy))) { return failure(409, "already_running", `${repo}#${number} is already being triaged.`); } const head = { tenantId, repo: triageStateKey(record), number, headSha }; const queued = await queueHead(deps.store, head, now, function busy(prior) { - return isQueuedSince(prior, now, deps.policy) || isRunningSince(prior, now, deps.policy); + return (isQueuedSince(prior, now, deps.policy) || isRunningSince(prior, now, deps.policy)) && !doneSince(prior, managed, deps.policy); }, false); if (queued.status === "kept") { if (isQueuedSince(queued.prior, now, deps.policy)) return failure(409, "already_queued", `${repo}#${number} is already queued for triage.`); diff --git a/apps/hub/src/github/source-workflow.ts b/apps/hub/src/github/source-workflow.ts index 3ef2698e5..e5c47ad5b 100644 --- a/apps/hub/src/github/source-workflow.ts +++ b/apps/hub/src/github/source-workflow.ts @@ -3,7 +3,7 @@ // its host's workflow. import { and, eq, inArray } from "drizzle-orm"; import { schema, type DB } from "@intx/db"; -import { CODE_REVIEW_PROVIDERS, codeReviewProviderOf, type RepoSource } from "@corbits/triage-contracts"; +import { CODE_REVIEW_PROVIDERS, codeReviewProviderOf, TRIAGE_MANAGERS, type RepoSource } from "@corbits/triage-contracts"; import { PR_TRIAGE_ADDRESS } from "@corbits/triage-workflows/pr-triage"; import { PR_TRIAGE_BITBUCKET_DATACENTER_ADDRESS, workflow as prTriageBitbucket } from "@corbits/triage-workflows/pr-triage-bitbucket-datacenter"; import { PR_TRIAGE_GITHUB_ADDRESS, workflow as prTriageGithub } from "@corbits/triage-workflows/pr-triage-github"; @@ -61,6 +61,11 @@ export function jobWorkflowsFor(source: RepoSource): string[] { return [triageWorkflowFor(source).id, codeReviewWorkflowFor(source).id, ...variants]; } +/** Every workflow whose runs record a host's verdicts: its jobs' and its pull request manager. */ +export function verdictWorkflowsFor(source: RepoSource): string[] { + return [...jobWorkflowsFor(source), TRIAGE_MANAGERS[source]]; +} + /** The host whose credential a package entry binds; undefined for an entry that binds none. */ export function entrySource(entry: string): RepoSource | undefined { if (entry === HISTORICAL_ENTRY) return "github"; diff --git a/apps/hub/src/github/triage-reconciler.test.ts b/apps/hub/src/github/triage-reconciler.test.ts index d3873f45c..e71bbb53f 100644 --- a/apps/hub/src/github/triage-reconciler.test.ts +++ b/apps/hub/src/github/triage-reconciler.test.ts @@ -295,6 +295,7 @@ function harness(logs: Logs, prs: OpenPr[] | Record, options: }, activity: options.activity ?? { stale: () => true }, observeTenant: observer.observe, + managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }), headChecks: async () => options.headChecks ?? [], verdicts: { latest: async () => options.recorded ?? [] }, store: store(), diff --git a/apps/hub/src/github/triage-reconciler.ts b/apps/hub/src/github/triage-reconciler.ts index a02c70128..8f7f02e9f 100644 --- a/apps/hub/src/github/triage-reconciler.ts +++ b/apps/hub/src/github/triage-reconciler.ts @@ -19,6 +19,7 @@ import { repoRecords, rotationRecord, sourceOf, triageNs, triageStateKey, type R import type { HeadChecksReader, OpenHeadsReader } from "./tenant-open-heads.js"; import type { RepoActivity } from "./repo-activity.js"; import { newestLive, type DeploymentRecord, type LiveDeployment, type Redeployed } from "./deployment.js"; +import type { ManagerRuns } from "./manager-runs.js"; import { isRunning, mergeObservedRuns, type ObservedRuns } from "./triage-runs.js"; import type { TenantObserver } from "./tenant-observer.js"; import type { VerdictRecord, VerdictRecordStore } from "./verdict-record-store.js"; @@ -58,6 +59,8 @@ export type TriageReconcilerDeps = { verdicts: Pick; /** Reads and records the live deployments' runs, shared with the recording loop. */ observeTenant: TenantObserver["observe"]; + /** The host's pull request managers, read from their claims and the verdicts they reported. */ + managerRuns: ManagerRuns; store: TriageStateStore; readCheckPack: PolicyReader; send: (job: TriageJob) => Promise; @@ -493,9 +496,11 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { if (reconcileFailed({ tenantId: tenant.id, repo: repo.record.name }, err)) failed = true; } } + const pulls = loaded.flatMap((repo) => repo.prs.map((pr) => ({ source: deps.source, repo: repo.record.name, number: pr.number }))); + const managed = await deps.managerRuns(tenant.id, pulls, pass.deployment.workflowVersion); const plans = planTenant({ repos: loaded.map((repo) => ({ name: repo.record.name, prs: repo.prs, rows: repo.stored.rows, ci: repo.ci })), - runs: pass.runs.byRepo, + runs: mergeObservedRuns([pass.runs, managed]).byRepo, now: deps.now(), policy: deps.policy, workflowVersion: pass.deployment.workflowVersion, diff --git a/apps/hub/src/github/triage-runs.ts b/apps/hub/src/github/triage-runs.ts index a173c6d7c..ba7fab1e9 100644 --- a/apps/hub/src/github/triage-runs.ts +++ b/apps/hub/src/github/triage-runs.ts @@ -7,21 +7,18 @@ import { asRecord } from "@corbits/common"; import { schema, type DB } from "@intx/db"; import { formatRunAddress } from "@intx/types"; import { classifyTerminalEvent, WORKFLOW_RUN_REF, workflowRunRepoIdForAddress, type WorkflowRunEvent, type WorkflowRunReader } from "@intx/hub-sessions"; -import { NOT_ASKED_REASON } from "@corbits/triage-workflows/logic/render"; import { heldForCi, unsettledBy } from "@corbits/triage-workflows/logic/settled"; import type { Judged } from "@corbits/triage-workflows/logic/item"; import type { LiveDeployment } from "./deployment.js"; import { runKey, type ObservedRun } from "./reconcile-plan.js"; -import { VerdictDetail } from "@corbits/triage-contracts"; -import { type ApprovedDo, type RecordedDo, type RecordedVerdict } from "./verdict-record-store.js"; +import type { ApprovedDo, RecordedVerdict } from "./verdict-record-store.js"; +import { DoResult, doResultsOf, eventOf, mirrorOutcomesOf, ruledOf, summaryOf, withResults, type Ruled, type Summary, type Trigger } from "./verdict-row.js"; const INLINE_PREFIX = "inline:"; const NO_VERDICT = "run completed without a verdict"; -type Trigger = { repo: string; source: RepoSource; number: number; headSha: string; startedAt: string; event?: string }; type Unsettled = Pick; type Terminal = Unsettled & { status: Exclude }; -type Summary = Omit; type Head = { trigger: Trigger; terminal?: Terminal; summary?: Summary; judged?: Judged }; /** No heads when the trigger names no pull request head, such as a portal run typed without a sha; a catch-up mail names several. */ /** An approval run names no head; `approved` is set once it ended, empty when it wrote nothing. `principalId` is who approved. */ @@ -75,12 +72,6 @@ function headOf(item: unknown, repo: string, source: RepoSource, startedAt: stri return [{ repo, source, number, headSha, startedAt, ...(event !== undefined && { event }) }]; } -/** A webhook mail's `event:action`; catch-up and portal mails name none. */ -function eventOf(payload: Record): string | undefined { - const { event, action } = payload; - return typeof event === "string" && typeof action === "string" ? `${event}:${action}` : undefined; -} - function isBacklog(started: Record): boolean { return triggerRequestOf(asRecord(started["trigger"])?.["payload"])?.["kind"] === "backlog"; } @@ -147,27 +138,8 @@ export function inlineOutput(event: WorkflowRunEvent | undefined): Record { - const dos = inlineOutput(lastCompleted(events, steps))?.["dos"]; - return new Map((Array.isArray(dos) ? dos : []).flatMap(function parsed(result): Array<[string, typeof DoResult.infer]> { - const read = DoResult(result); - return read instanceof type.errors ? [] : [[read.effectId, read]]; - })); -} - -function withResults(verdict: Record | undefined, results: ReadonlyMap): Record | undefined { - const actions = verdict?.["actions"]; - if (!verdict || !Array.isArray(actions) || results.size === 0) return verdict; - function withResult(action: unknown): unknown { - const result = results.get(String(asRecord(action)?.["effectId"])); - if (!result) return action; - // The PR workflow holds an approve or merge its code review did not clear; it waits as one evaluate held. - if (result.status === "held") return { ...asRecord(action), held: true, ...(result.error !== undefined && { reason: result.error }) }; - return { ...asRecord(action), status: result.status, ...(result.error !== undefined && { error: result.error }) }; - } - return { ...verdict, actions: actions.map(withResult) }; +function doResults(events: readonly WorkflowRunEvent[], steps = EXECUTE_STEPS): Map { + return doResultsOf(inlineOutput(lastCompleted(events, steps))); } function listOf(value: unknown): unknown[] { @@ -206,93 +178,12 @@ function verdictsOf(events: readonly WorkflowRunEvent[]): Array withResults(item, results)); } -type Ruled = { facts: Record; evidence: string[] }; - -/** The reasons the rules step's findings give, which the portal shows when the verdict names none. */ -function evidenceOf(det: unknown): string[] { - const findings = asRecord(det)?.["findings"]; - const reasons = [...(Array.isArray(findings) ? findings.map((finding) => asRecord(finding)?.["reason"]) : []), asRecord(det)?.["reason"]]; - return [...new Set(reasons.filter((reason): reason is string => typeof reason === "string" && reason !== ""))]; -} - -/** The facts the rules step read for each pull request, and its findings' reasons, by `repo#number`. */ -function ruledOf(events: readonly WorkflowRunEvent[]): Map { - const items = inlineOutput(lastCompleted(events, RULES_STEPS))?.["items"]; - return new Map((Array.isArray(items) ? items : []).flatMap(function pull(item): Array<[string, Ruled]> { - const facts = asRecord(asRecord(item)?.["facts"]); - return facts ? [[`${String(facts["repo"])}#${String(facts["number"])}`, { facts, evidence: evidenceOf(asRecord(item)?.["det"]) }]] : []; - })); +function ruledIn(events: readonly WorkflowRunEvent[]): Map { + return ruledOf(inlineOutput(lastCompleted(events, RULES_STEPS))); } -/** Gates decide actions, not the verdict; a model check the verdict never asked about says nothing about the pull request. */ -function recorded(c: { kind?: string; reason?: string; gate?: boolean }): boolean { - return c.gate !== true && !(c.kind === "model" && c.reason === NOT_ASKED_REASON); -} - -/** Mirror outcomes by call: a batch run outputs `{ results }`, a single run the one outcome. */ function mirrorOutcomes(events: readonly WorkflowRunEvent[]): Map> { - const output = inlineOutput(lastCompleted(events, MIRROR_STEPS)); - const outcomes = Array.isArray(output?.["results"]) ? output["results"] : [output]; - return new Map(outcomes.map(asRecord).flatMap((outcome) => outcome && typeof outcome["call"] === "string" ? [[outcome["call"], outcome] as const] : [])); -} - -/** Why the mirror posted nothing for the verdict, when it said. */ -function notPostedOf(outcome: Record | undefined): string | undefined { - return outcome?.["skipped"] === true && typeof outcome["reason"] === "string" ? outcome["reason"] : undefined; -} - -function mirrorCallId(repo: string, number: number): string { - return `mirror:${repo.replace("/", "--")}:${number}`; -} - -const SummaryFields = type({ - state: "string", - actor: "string", - humanGated: "boolean", - "degraded?": "string | null", - "checks?": type({ check: "string", result: "string", "kind?": "string", "reason?": "string", "gate?": "boolean" }).array(), - "actions?": "unknown[]", - "request?": { comment: "string" }, -}); - -function suggestedDos(actions: readonly unknown[]): RecordedDo[] { - return actions.map(asRecord).flatMap(function suggested(action) { - const list = action?.["list"]; - const effectId = action?.["effectId"]; - return typeof list === "string" && typeof effectId === "string" && action?.["skipped"] !== true ? [{ list, effectId }] : []; - }); -} - -function detailOf(verdict: Record, ruled: Ruled | undefined, notPosted: string | undefined): VerdictDetail | undefined { - const reason = verdict["reason"]; - const evidence = typeof reason === "string" && reason !== "" ? [reason] : (ruled?.evidence ?? []); - const detail = VerdictDetail({ ...verdict, evidence, ...(ruled && { facts: ruled.facts }), ...(notPosted !== undefined && { notPosted }) }); - return detail instanceof type.errors ? undefined : detail; -} - -/** - * What the stats and the portal read of a verdict; none for one written before it carried these fields, made for - * another head, still waiting on the host's checks, or made on a closed pull request. - */ -function summaryOf(verdict: Record, at: unknown, trigger: Trigger, mirrored: ReadonlyMap>, ruled: Ruled | undefined): Summary | undefined { - if (verdict["waiting"] === true || verdict["closed"] === true) return undefined; - const fields = SummaryFields(verdict); - if (fields instanceof type.errors || typeof at !== "string" || verdict["repo"] !== trigger.repo || verdict["number"] !== trigger.number) return undefined; - const outcome = mirrored.get(mirrorCallId(trigger.repo, trigger.number)); - const detail = detailOf(verdict, ruled, notPostedOf(outcome)); - return { - verdictAt: new Date(at), - settled: unsettledBy(verdict, trigger.headSha).unsettled === undefined, - state: fields.state, - actor: fields.actor, - humanGated: fields.humanGated, - degraded: fields.degraded ?? null, - checks: (fields.checks ?? []).filter(recorded).map(({ check, result }) => ({ check, result })), - dos: suggestedDos(fields.actions ?? []), - commented: (fields.request?.comment ?? "") !== "" && outcome?.["ok"] === true, - ...(detail && { detail }), - ...(trigger.event !== undefined && { trigger: trigger.event }), - }; + return mirrorOutcomesOf(inlineOutput(lastCompleted(events, MIRROR_STEPS))); } export function verdictFor(events: readonly WorkflowRunEvent[], repo: string, number: number): Record | undefined { @@ -341,7 +232,7 @@ function headsOf(events: readonly WorkflowRunEvent[], started: Record`case when ${inArray(schema.workflowDefinition.name, jobWorkflowsFor("bitbucket-datacenter"))} then 'bitbucket-datacenter' else 'github' end`; +const verdictSource = sql`case when ${inArray(workflowName, verdictWorkflowsFor("bitbucket-datacenter"))} then 'bitbucket-datacenter' else 'github' end`; /** The verdicts `where` picks, each with its host, to group or pick from by host. */ function hostedVerdicts(db: DB["db"], where: SQL | undefined) { - const { workflowRun, workflowDefinition } = schema; + const { workflowRun, workflowDefinition, workflowRunLaunchSpec } = schema; return db .select({ ...getTableColumns(verdictRecord), source: verdictSource.as("source") }) .from(verdictRecord) .leftJoin(workflowRun, and(eq(workflowRun.id, verdictRecord.runId), eq(workflowRun.tenantId, verdictRecord.tenantId))) .leftJoin(workflowDefinition, eq(workflowDefinition.id, workflowRun.definitionId)) + .leftJoin(workflowRunLaunchSpec, eq(workflowRunLaunchSpec.anchorRunId, workflowRun.anchorRunId)) .where(where) .as("hosted"); } @@ -103,6 +107,46 @@ function latestOf(db: DB["db"], where: SQL | undefined): Promise { + for (const done of approved) { + const pull = and(eq(verdictRecord.tenantId, tenantId), eq(verdictRecord.repo, done.repo), eq(verdictRecord.number, done.number)); + const rows = await db.select({ runId: verdictRecord.runId, detail: verdictRecord.detail }).from(verdictRecord).where(pull); + for (const { runId, detail } of rows) { + if (!detail?.actions?.some((action) => action.effectId === done.effectId)) continue; + await db.update(verdictRecord).set({ detail: withApproved(detail, [done]) }).where(and(pull, eq(verdictRecord.runId, runId))); + } + } +} + +/** The Dos already written on the verdict's head, as the verdicts recorded there show them. */ +async function writtenOn(db: DB["db"] | Tx, tenantId: string, { repo, number, headSha }: Pick): Promise { + const rows = await db.select({ detail: verdictRecord.detail }).from(verdictRecord) + .where(and(eq(verdictRecord.tenantId, tenantId), eq(verdictRecord.repo, repo), eq(verdictRecord.number, number), eq(verdictRecord.headSha, headSha))); + const written: ApprovedDo[] = []; + for (const { detail } of rows) { + if (!detail?.actions) continue; + for (const { effectId, status, error } of detail.actions) { + if (effectId !== undefined && status !== undefined && WRITTEN.has(status)) written.push({ repo, number, effectId, status, ...(error !== undefined && { error }) }); + } + } + return written; +} + +/** + * A manager's verdict: its run makes one per pass, and the newest replaces the row its run recorded before, keeping the + * Dos already written on its head, as a re-triage of a head does. + */ +export async function replaceVerdict(db: DB["db"] | Tx, tenantId: string, verdict: RecordedVerdict): Promise { + const written = await writtenOn(db, tenantId, verdict); + const kept = verdict.detail && written.length > 0 ? { ...verdict, detail: withApproved(verdict.detail, written) } : verdict; + const { runId: _runId, repo: _repo, number: _number, ...fields } = kept; + await db.insert(verdictRecord).values({ ...kept, tenantId }).onConflictDoUpdate({ + target: [verdictRecord.tenantId, verdictRecord.runId, verdictRecord.repo, verdictRecord.number], + set: { detail: null, trigger: null, runStatus: null, ...fields }, + }); +} + export function createVerdictRecordStore(db: DB["db"]) { return { /** @@ -122,16 +166,8 @@ export function createVerdictRecordStore(db: DB["db"]) { }); }, - /** Sets each approved Do's outcome on every recorded verdict of its pull request that suggests it; a verdict not recorded yet gains it when it is. */ - async recordApproved(tenantId: string, approved: readonly ApprovedDo[]): Promise { - for (const done of approved) { - const pull = and(eq(verdictRecord.tenantId, tenantId), eq(verdictRecord.repo, done.repo), eq(verdictRecord.number, done.number)); - const rows = await db.select({ runId: verdictRecord.runId, detail: verdictRecord.detail }).from(verdictRecord).where(pull); - for (const { runId, detail } of rows) { - if (!detail?.actions?.some((action) => action.effectId === done.effectId)) continue; - await db.update(verdictRecord).set({ detail: withApproved(detail, [done]) }).where(and(pull, eq(verdictRecord.runId, runId))); - } - } + recordApproved(tenantId: string, approved: readonly ApprovedDo[]): Promise { + return recordApproved(db, tenantId, approved); }, /** The newest verdict of each pull request on each host. */ @@ -175,6 +211,12 @@ export function createVerdictRecordStore(db: DB["db"]) { return row !== undefined; }, + /** The verdicts `runIds` recorded on `pulls`. */ + ofRuns(tenantId: string, runIds: readonly string[], pulls: readonly PullRef[]): Promise { + if (runIds.length === 0 || pulls.length === 0) return Promise.resolve([]); + return db.select().from(verdictRecord).where(and(eq(verdictRecord.tenantId, tenantId), inArray(verdictRecord.runId, [...runIds]), ofPulls(pulls))); + }, + /** Oldest first. */ since(tenantId: string, repo: string, from: Date): Promise { const where = and(eq(verdictRecord.tenantId, tenantId), eq(verdictRecord.repo, repo), gte(verdictRecord.verdictAt, from)); diff --git a/apps/hub/src/github/verdict-row.ts b/apps/hub/src/github/verdict-row.ts new file mode 100644 index 000000000..624fdd2d1 --- /dev/null +++ b/apps/hub/src/github/verdict-row.ts @@ -0,0 +1,134 @@ +// A verdict row as the verdict table records it, built from what a triage pass's evaluate, rules, mirror and execute +// steps output, however those outputs reached the hub. +import { type } from "arktype"; +import { asRecord } from "@corbits/common"; +import { VerdictDetail, type RepoSource } from "@corbits/triage-contracts"; +import { NOT_ASKED_REASON } from "@corbits/triage-workflows/logic/render"; +import { unsettledBy } from "@corbits/triage-workflows/logic/settled"; +import type { RecordedDo, RecordedVerdict } from "./verdict-record-store.js"; + +export type Trigger = { repo: string; source: RepoSource; number: number; headSha: string; startedAt: string; event?: string }; +export type Summary = Omit; +export type Ruled = { facts: Record; evidence: string[] }; + +/** A webhook mail's `event:action`; catch-up and portal mails name none. */ +export function eventOf(payload: Record): string | undefined { + const { event, action } = payload; + return typeof event === "string" && typeof action === "string" ? `${event}:${action}` : undefined; +} + +export const DoResult = type({ effectId: "string", status: "string", "error?": "string" }); +export type DoResult = typeof DoResult.infer; + +/** An execute step's Do outcomes by effect id. */ +export function doResultsOf(output: Record | undefined): Map { + const dos = output?.["dos"]; + return new Map((Array.isArray(dos) ? dos : []).flatMap(function parsed(result): Array<[string, DoResult]> { + const read = DoResult(result); + return read instanceof type.errors ? [] : [[read.effectId, read]]; + })); +} + +export function withResults(verdict: Record | undefined, results: ReadonlyMap): Record | undefined { + const actions = verdict?.["actions"]; + if (!verdict || !Array.isArray(actions) || results.size === 0) return verdict; + function withResult(action: unknown): unknown { + const result = results.get(String(asRecord(action)?.["effectId"])); + if (!result) return action; + // The PR workflow holds an approve or merge its code review did not clear; it waits as one evaluate held. + if (result.status === "held") return { ...asRecord(action), held: true, ...(result.error !== undefined && { reason: result.error }) }; + return { ...asRecord(action), status: result.status, ...(result.error !== undefined && { error: result.error }) }; + } + return { ...verdict, actions: actions.map(withResult) }; +} + +/** The reasons the rules step's findings give, which the portal shows when the verdict names none. */ +function evidenceOf(det: unknown): string[] { + const findings = asRecord(det)?.["findings"]; + const reasons = [...(Array.isArray(findings) ? findings.map((finding) => asRecord(finding)?.["reason"]) : []), asRecord(det)?.["reason"]]; + return [...new Set(reasons.filter((reason): reason is string => typeof reason === "string" && reason !== ""))]; +} + +/** The facts the rules step read for each pull request, and its findings' reasons, by `repo#number`. */ +export function ruledOf(output: Record | undefined): Map { + const items = output?.["items"]; + return new Map((Array.isArray(items) ? items : []).flatMap(function pull(item): Array<[string, Ruled]> { + const facts = asRecord(asRecord(item)?.["facts"]); + return facts ? [[`${String(facts["repo"])}#${String(facts["number"])}`, { facts, evidence: evidenceOf(asRecord(item)?.["det"]) }]] : []; + })); +} + +/** Mirror outcomes by call: a batch run outputs `{ results }`, a single run the one outcome. */ +export function mirrorOutcomesOf(output: Record | undefined): Map> { + const outcomes = Array.isArray(output?.["results"]) ? output["results"] : [output]; + return new Map(outcomes.map(asRecord).flatMap((outcome) => outcome && typeof outcome["call"] === "string" ? [[outcome["call"], outcome] as const] : [])); +} + +/** Gates decide actions, not the verdict; a model check the verdict never asked about says nothing about the pull request. */ +function recorded(c: { kind?: string; reason?: string; gate?: boolean }): boolean { + return c.gate !== true && !(c.kind === "model" && c.reason === NOT_ASKED_REASON); +} + +/** Why the mirror posted nothing for the verdict, when it said. */ +function notPostedOf(outcome: Record | undefined): string | undefined { + return outcome?.["skipped"] === true && typeof outcome["reason"] === "string" ? outcome["reason"] : undefined; +} + +function mirrorCallId(repo: string, number: number): string { + return `mirror:${repo.replace("/", "--")}:${number}`; +} + +const SummaryFields = type({ + state: "string", + actor: "string", + humanGated: "boolean", + "degraded?": "string | null", + "checks?": type({ check: "string", result: "string", "kind?": "string", "reason?": "string", "gate?": "boolean" }).array(), + "actions?": "unknown[]", + "request?": { comment: "string" }, +}); + +function suggestedDos(actions: readonly unknown[]): RecordedDo[] { + return actions.map(asRecord).flatMap(function suggested(action) { + const list = action?.["list"]; + const effectId = action?.["effectId"]; + return typeof list === "string" && typeof effectId === "string" && action?.["skipped"] !== true ? [{ list, effectId }] : []; + }); +} + +/** The verdict's own reason, else its rules' findings' reasons; a verdict made without rules gives none. */ +function evidenceFor(reason: unknown, ruled: Ruled | undefined): string[] { + if (typeof reason === "string" && reason !== "") return [reason]; + return ruled ? ruled.evidence : []; +} + +function detailOf(verdict: Record, ruled: Ruled | undefined, notPosted: string | undefined): VerdictDetail | undefined { + const evidence = evidenceFor(verdict["reason"], ruled); + const detail = VerdictDetail({ ...verdict, evidence, ...(ruled && { facts: ruled.facts }), ...(notPosted !== undefined && { notPosted }) }); + return detail instanceof type.errors ? undefined : detail; +} + +/** + * What the stats and the portal read of a verdict; none for one written before it carried these fields, made for + * another head, still waiting on the host's checks, or made on a closed pull request. + */ +export function summaryOf(verdict: Record, at: unknown, trigger: Trigger, mirrored: ReadonlyMap>, ruled: Ruled | undefined): Summary | undefined { + if (verdict["waiting"] === true || verdict["closed"] === true) return undefined; + const fields = SummaryFields(verdict); + if (fields instanceof type.errors || typeof at !== "string" || verdict["repo"] !== trigger.repo || verdict["number"] !== trigger.number) return undefined; + const outcome = mirrored.get(mirrorCallId(trigger.repo, trigger.number)); + const detail = detailOf(verdict, ruled, notPostedOf(outcome)); + return { + verdictAt: new Date(at), + settled: unsettledBy(verdict, trigger.headSha).unsettled === undefined, + state: fields.state, + actor: fields.actor, + humanGated: fields.humanGated, + degraded: fields.degraded ?? null, + checks: fields.checks ? fields.checks.filter(recorded).map(({ check, result }) => ({ check, result })) : [], + dos: fields.actions ? suggestedDos(fields.actions) : [], + commented: fields.request !== undefined && fields.request.comment !== "" && outcome?.["ok"] === true, + ...(detail && { detail }), + ...(trigger.event !== undefined && { trigger: trigger.event }), + }; +} diff --git a/apps/hub/src/manager-routes.postgres.test.ts b/apps/hub/src/manager-routes.postgres.test.ts index 8db31694c..d0f45b8cd 100644 --- a/apps/hub/src/manager-routes.postgres.test.ts +++ b/apps/hub/src/manager-routes.postgres.test.ts @@ -12,7 +12,8 @@ import { matchesRunTokenRoute } from "../../../vendor/interchange/packages/hub-a import { MANAGER_RESULT_ACTION, MANAGER_RESULT_RESOURCE, MANAGER_STATE_ACTION, MANAGER_STATE_RESOURCE } from "@corbits/triage-contracts"; import { createManagerRoutes, MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; import { MAX_BODY_BYTES } from "./read-json.js"; -import { recordEvent, type PullRequestRef } from "./manager-state.js"; +import { passRequest, recordEvent, type PullRequestRef } from "./manager-state.js"; +import { createVerdictRecordStore } from "./github/verdict-record-store.js"; import { mountTenantRoutes } from "./tenant-routes.js"; import { migrateTriageTables } from "./triage-db.js"; @@ -164,6 +165,81 @@ test("PostgreSQL: the manager's run token takes, finishes and a reviewer's recor }); }, 60_000); +/** The run as the stock deploy route launched it, from a published workflow whose frozen projection keeps its id. */ +async function launchedAs(db: Db, runId: string, workflow: string): Promise { + await db.insert(schema.workflowRunLaunchSpec).values({ + anchorRunId: runId, sessionId: `ses_${runId}`, deploymentDomain: "manager.test", sourceAuthorityPrincipalId: "prn_member", + frozenApprovalBundle: { projection: { id: workflow } }, sourceOfferingIds: [], defaultSourceOfferingId: "off_default", deployContent: {}, + }); +} + +async function mailed(db: Db, pr: PullRequestRef, head: string, action: string): Promise { + await recordEvent(db, TENANT, pr, `${action}-${pr.source}`, { type: "webhook", head, payload: { kind: "pr", event: "pull_request", action, headSha: head } }); +} + +/** What a pass that posted its verdict on `head` reports, as its evaluate, mirror, execute and rules steps output it. */ +function postedOn(pr: PullRequestRef, head: string) { + const target = { labels: ["api"] }; + const request = { repo: pr.repo, number: pr.number, labels: ["Triage: Pass"], owned: [], comment: "Ready for review." }; + const actions = [{ list: "pass", index: 0, effectId: `held-${head}`, kind: "labels", target, held: true }, { list: "pass", index: 1, effectId: `auto-${head}`, kind: "labels", target }]; + return { + request, + report: { + head, + verdict: { repo: pr.repo, number: pr.number, headSha: head, verdict: "pass", state: "ready-monitoring", actor: "author", humanGated: false, degraded: null, checks: [{ check: "title", kind: "machine", result: "pass" }], actions, request }, + mirror: { call: `mirror:${pr.repo.replace("/", "--")}:${pr.number}`, ok: true }, + executed: { dos: [{ effectId: `auto-${head}`, status: "done" }] }, + rules: { items: [{ facts: { repo: pr.repo, number: pr.number, title: "Fix the widget", author: "octocat" }, det: { findings: [{ reason: "The title says what changed." }] } }] }, + }, + }; +} + +test("PostgreSQL: done records the verdict a pass posted as its run's on its host, the run's newest replaces it keeping the Dos written on its head, approvals set their outcomes and a held head records none", async () => { + await withDatabase(async function recorded(db) { + const app = buildHub(db); + const verdicts = createVerdictRecordStore(db); + for (const pr of [GITHUB, BITBUCKET]) { + const runId = `run_${pr.source}`; + const manager = bearer(await runWithToken(db, runId, [STATE])); + await launchedAs(db, runId, `triage-manager-${pr.source}`); + + await mailed(db, pr, "h1", "opened"); + await post(app, "take", pr, manager); + const first = postedOn(pr, "h1"); + expect(await (await post(app, "done", { ...pr, reviewers: [], report: first.report }, manager)).json()).toEqual({ outcome: "ended" }); + expect(await verdicts.latestFor(TENANT, pr)).toMatchObject({ + runId, source: pr.source, headSha: "h1", runStatus: "completed", state: "ready-monitoring", settled: true, commented: true, trigger: "pull_request:opened", + checks: [{ check: "title", result: "pass" }], dos: [{ list: "pass", effectId: "held-h1" }, { list: "pass", effectId: "auto-h1" }], + detail: { headSha: "h1", facts: { title: "Fix the widget", author: "octocat" }, evidence: ["The title says what changed."], actions: [{ effectId: "held-h1", held: true }, { effectId: "auto-h1", status: "done" }] }, + }); + expect(await passRequest(db, TENANT, pr, runId, "h1")).toEqual(first.request); + + await mailed(db, pr, "h2", "synchronize"); + await post(app, "take", pr, manager); + const approvals = { results: [{ dos: [{ effectId: "held-h2", status: "done" }] }] }; + await post(app, "done", { ...pr, reviewers: [], report: postedOn(pr, "h2").report }, manager); + await post(app, "take", pr, manager); + await post(app, "done", { ...pr, reviewers: [], approvals }, manager); + const rows = await verdicts.forPull(TENANT, pr.repo, pr.number); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ runId, headSha: "h2", trigger: "pull_request:synchronize", detail: { actions: [{ effectId: "held-h2", status: "done" }, { effectId: "auto-h2", status: "done" }] } }); + + // Mailed again on the same head, its next verdict keeps the Do already written there, the run's start on the head and the head's newest mail. + await mailed(db, pr, "h2", "edited"); + await post(app, "take", pr, manager); + await post(app, "done", { ...pr, reviewers: [], report: postedOn(pr, "h2").report }, manager); + expect(await verdicts.forPull(TENANT, pr.repo, pr.number)).toEqual([expect.objectContaining({ + startedAt: rows[0]!.startedAt, trigger: "pull_request:edited", detail: expect.objectContaining({ actions: [expect.objectContaining({ effectId: "held-h2", status: "done" }), expect.objectContaining({ effectId: "auto-h2", status: "done" })] }), + })]); + + await mailed(db, pr, "h3", "reopened"); + await post(app, "take", pr, manager); + expect(await (await post(app, "done", { ...pr, reviewers: [], report: { head: "h3", waiting: true } }, manager)).json()).toEqual({ outcome: "ended" }); + expect((await verdicts.forPull(TENANT, pr.repo, pr.number)).map((row) => row.headSha)).toEqual(["h2"]); + } + }); +}, 60_000); + test("PostgreSQL: a run the claim does not name, a session and a run without the grant are refused", async () => { await withDatabase(async function refusals(db) { const app = buildHub(db); diff --git a/apps/hub/src/manager-routes.ts b/apps/hub/src/manager-routes.ts index f67e25ba2..752dd1329 100644 --- a/apps/hub/src/manager-routes.ts +++ b/apps/hub/src/manager-routes.ts @@ -9,11 +9,16 @@ import { HTTPException } from "hono/http-exception"; import { describeRoute } from "hono-openapi"; import { schema, type DB } from "@intx/db"; import type { RequireGrant, RunTokenRoute, TenantEnv } from "@intx/hub-api"; +import { asRecord } from "@corbits/common"; import { MANAGER_RESULT_ACTION, MANAGER_RESULT_RESOURCE, MANAGER_STATE_ACTION, MANAGER_STATE_RESOURCE, REPO_SOURCES, StageResult } from "@corbits/triage-contracts"; +import { heldForCi } from "@corbits/triage-workflows/logic/settled"; import { failure } from "./github/portal-credential.js"; import { MAX_BODY_BYTES } from "./read-json.js"; import { body, Failure, json } from "./route-schema.js"; -import { done, EventState, PassState, recordEvent, ReviewerState, take } from "./manager-state.js"; +import type { Tx } from "./github/admin-github-grant.js"; +import { recordApproved, replaceVerdict } from "./github/verdict-record-store.js"; +import { doResultsOf, eventOf, mirrorOutcomesOf, ruledOf, summaryOf, withResults } from "./github/verdict-row.js"; +import { done, EventState, PassState, recordEvent, ReviewerState, take, type Closing, type PassReport } from "./manager-state.js"; const TAKE_PATH = "/manager/take"; const DONE_PATH = "/manager/done"; @@ -26,7 +31,20 @@ export const MANAGER_RUN_TOKEN_ROUTES: readonly RunTokenRoute[] = [ ]; const PullRequest = type({ source: type.enumerated(...REPO_SOURCES), repo: /^[\w.-]+\/[\w.-]+$/, number: "number.integer > 0" }); -const DoneBody = PullRequest.and({ reviewers: ReviewerState.array() }); +/** + * What the pass did on its head: held it for CI, or posted a verdict, given as its evaluate, mirror, execute and rules + * steps output it. A pass that posted nothing reports nothing. + */ +const Report = type({ + head: "string > 0", + "waiting?": "true", + "verdict?": "Record", + "mirror?": "Record", + "executed?": "Record", + "rules?": "Record", +}); +/** `approvals` is the pass's approved Dos as its approvals step ran them. */ +const DoneBody = PullRequest.and({ reviewers: ReviewerState.array(), "report?": Report, "approvals?": { results: "Record[]" } }); const ReviewerResultBody = PullRequest.and(StageResult); const Latest = { "request?": "Record", reviewers: ReviewerState.array() } as const; @@ -53,6 +71,21 @@ type RunEnv = TenantEnv & { Variables: TenantEnv["Variables"] & { runId: string export type ManagerRoutesDeps = { db: DB["db"]; requireGrant: RequireGrant; log: (entry: Record) => void }; +type Reported = typeof Report.infer; + +/** The verdict the pass posted, with the outcomes of the Dos it ran. */ +function postedOf(report: Reported | undefined): Record | undefined { + return report?.verdict && withResults(report.verdict, doResultsOf(report.executed)); +} + +/** What done keeps on the closed pass: whether it held its head for CI, or the mirror request of its verdict. */ +function passReport(report: Reported | undefined, posted: Record | undefined): PassReport | undefined { + if (!report) return undefined; + if (report.waiting || (posted && heldForCi(posted))) return { head: report.head, waiting: true }; + const request = asRecord(posted?.["request"]); + return { head: report.head, ...(request && { request }) }; +} + export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps) { const app = new Hono(); // Hono throws on a body that is not JSON; it keeps the hub's error envelope. @@ -96,8 +129,8 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps app.post( DONE_PATH, describeRoute({ - summary: "Close the manager's pass and record its reviewers", - description: `Run token only. \`ended\`: no event is pending and no reviewer of the current head is due, so the claim ended and the run should end. \`more\`: the manager takes again at once, or with \`wait\` after waiting for its reviewer. ${NOT_MANAGER}`, + summary: "Close the manager's pass and record its reviewers and its verdict", + description: `Run token only. In the same transaction, a \`report\` of a verdict the pass posted is recorded as the run's verdict on the pull request, replacing one the run recorded before, and \`approvals\` sets each approved Do's outcome on the verdicts that suggest it. \`ended\`: no event is pending and no reviewer of the current head is due, so the claim ended and the run should end. \`more\`: the manager takes again at once, or with \`wait\` after waiting for its reviewer. ${NOT_MANAGER}`, tags: TAGS, responses: { 200: json("Whether the manager is done", Done), ...refusals, 409: json(NOT_MANAGER, Failure) }, }), @@ -106,8 +139,22 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps granted, body(DoneBody), async function finishPass(c) { - const valid = c.req.valid("json"); - const result = await done(db, c.get("tenant").id, valid, c.get("runId"), valid.reviewers); + const { reviewers, report, approvals, ...pr } = c.req.valid("json"); + const tenantId = c.get("tenant").id; + const runId = c.get("runId"); + const posted = postedOf(report); + const kept = passReport(report, posted); + async function recordPass(tx: Tx, closing: Closing | undefined): Promise { + if (posted && closing && !kept?.waiting) { + const event = closing.mail && eventOf(closing.mail); + const trigger = { ...pr, headSha: closing.head, startedAt: closing.startedAt, ...(event !== undefined && { event }) }; + const summary = summaryOf(posted, new Date().toISOString(), trigger, mirrorOutcomesOf(report?.mirror), ruledOf(report?.rules).get(`${pr.repo}#${pr.number}`)); + if (summary) await replaceVerdict(tx, tenantId, { runId, repo: pr.repo, number: pr.number, headSha: closing.head, startedAt: new Date(closing.startedAt), ...summary, runStatus: "completed" }); + } + const approved = approvals === undefined ? [] : approvals.results.flatMap((ran) => [...doResultsOf(ran).values()].map((outcome) => ({ repo: pr.repo, number: pr.number, ...outcome }))); + await recordApproved(tx, tenantId, approved); + } + const result = await done(db, tenantId, pr, runId, reviewers, kept, recordPass); if (result.outcome === "refused") return refused(); return c.json(result); }, diff --git a/apps/hub/src/manager-state.ts b/apps/hub/src/manager-state.ts index fa1765294..63ad6b52b 100644 --- a/apps/hub/src/manager-state.ts +++ b/apps/hub/src/manager-state.ts @@ -26,7 +26,7 @@ const ClaimState = type({ "runId?": "string > 0", "pass?": PassState, reviewers: ReviewerState.array(), - history: PassState.and({ runId: "string > 0", doneAt: "string.date.iso" }).array(), + history: PassState.and({ runId: "string > 0", doneAt: "string.date.iso", "waiting?": "true", "request?": "Record" }).array(), starts: "number.integer >= 0", }); const EventPayload = type("Record").narrow(function plainObject(value) { @@ -51,6 +51,13 @@ type TakenEvent = { key: string; event: ManagerEvent }; type Latest = { request?: Record; reviewers: Reviewer[] }; type TakeResult = { outcome: "refused" } | ({ outcome: "idle" } & Latest) | ({ outcome: "pass"; pass: Pass; events: TakenEvent[]; resumed: boolean } & Latest); +type ClosedPass = Claim["history"][number]; +/** What a pass reports as done closes it: the head it worked on, held for CI, or the mirror request of the verdict it posted. */ +export type PassReport = { head: string; waiting?: true; request?: Record }; +/** A reporting pass's head, when this run's first pass on it was taken, and the newest pull request mail for it. */ +export type Closing = { head: string; startedAt: string; mail?: Record }; +/** Records what the closed pass reported in done's transaction, so the claim never moves on without it. */ +export type RecordDone = (tx: Tx, closing: Closing | undefined) => Promise; type DoneResult = { outcome: "refused" } | { outcome: "ended" } | { outcome: "more"; wait: boolean }; function pullKey({ source, repo, number }: PullRequestRef): string { @@ -222,11 +229,12 @@ async function latest(tx: Tx, tenantId: string, pr: PullRequestRef, claim: Claim return { ...(request && { request }), reviewers: claim.reviewers }; } -/** The newest pull request mail recorded for the pull request, which triage runs on. */ -async function newestRequest(tx: Tx, tenantId: string, pr: PullRequestRef): Promise | undefined> { +/** The newest pull request mail recorded for the pull request, or for its `head`, which triage runs on. */ +async function newestRequest(tx: Tx, tenantId: string, pr: PullRequestRef, head?: string): Promise | undefined> { const prefix = eventPrefix(pr); + const onHead = head === undefined ? undefined : sql`${managerState.state}->>'head' = ${head}`; const [row] = await tx.select({ key: managerState.key, state: managerState.state }).from(managerState) - .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, like(managerState.key, `${prefix}%`), sql`${managerState.state}->'payload'->>'kind' = 'pr'`)) + .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, like(managerState.key, `${prefix}%`), sql`${managerState.state}->'payload'->>'kind' = 'pr'`, onHead)) .orderBy(desc(sql`${managerState.state}->>'receivedAt'`), desc(managerState.key)) .limit(1); return row && eventOf(row.key, row.state).payload; @@ -266,29 +274,106 @@ export async function take(db: DB["db"], tenantId: string, pr: PullRequestRef, r }); } +/** The open pass closed by `runId`, with what it reported; a pass that reported without taking events, as one past a reviewer's deadline, is closed as its own. */ +function closing(claim: Claim, runId: string, now: Date, report: PassReport | undefined): ClosedPass | undefined { + const at = now.toISOString(); + const pass = claim.pass ?? (report && { id: randomUUID(), head: report.head, events: [], takenAt: at, takes: 1 }); + return pass && { ...pass, runId, doneAt: at, ...(report?.waiting && { waiting: true }), ...(report?.request && { request: report.request }) }; +} + +/** The history with `closed` added; only the newest pass of a run on a head keeps a mirror request, the verdict that run's row now holds. */ +function closedInto(history: Claim["history"], closed: ClosedPass): Claim["history"] { + if (closed.request === undefined) return [...history, closed]; + const older = history.map(function withoutRequest(pass) { + if (pass.runId !== closed.runId || pass.head !== closed.head) return pass; + const { request: _request, ...rest } = pass; + return rest; + }); + return [...older, closed]; +} + +/** The reporting pass's closing: when this run first took `head`, and the newest pull request mail for it. */ +async function closingOf(tx: Tx, tenantId: string, pr: PullRequestRef, runId: string, head: string, history: Claim["history"]): Promise { + const first = history.find((pass) => pass.runId === runId && pass.head === head); + if (!first) throw new Error(`${claimKey(pr)} has no pass of ${runId} on ${head} to report`); + const mail = await newestRequest(tx, tenantId, pr, head); + return { head, startedAt: first.takenAt, ...(mail && { mail }) }; +} + /** - * Closes the open pass and records `reviewers`. The claim ends when no event is pending and no reviewer of the current - * head is still due; otherwise the manager has more to do: at once when an event is pending, else after it waits for - * its reviewer. + * Closes the open pass with what it reported and records `reviewers`. The claim ends when no event is pending and no + * reviewer of the current head is still due; otherwise the manager has more to do: at once when an event is pending, + * else after it waits for its reviewer. */ -export async function done(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, reviewers: readonly Reviewer[]): Promise { +export async function done(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, reviewers: readonly Reviewer[], report?: PassReport, record?: RecordDone): Promise { return db.transaction(async function finishPass(tx) { const claim = await lockClaim(tx, tenantId, pr); if (!fencedTo(claim, runId)) return { outcome: "refused" }; const recorded = ReviewerState.array()(asStored(reviewers)); if (recorded instanceof type.errors) throw new Error(`reviewers for ${claimKey(pr)} are malformed: ${recorded.summary}`); const now = new Date(); - const history = claim.pass ? [...claim.history, { ...claim.pass, runId, doneAt: now.toISOString() }] : claim.history; + const closed = closing(claim, runId, now, report); + const history = closed ? closedInto(claim.history, closed) : claim.history; const head = history.at(-1)?.head; const due = recorded.some((reviewer) => reviewer.head === head && reviewer.result === undefined && Date.parse(reviewer.deadline) > now.getTime()); const rows = await tx.select({ key: managerState.key, state: managerState.state }).from(managerState).where(pending(tenantId, pr)); const waiting = passHeadOf(eventsOf(rows), history) !== undefined; - const closed = { claimId: claim.claimId, claimedAt: claim.claimedAt, reviewers: recorded, history, starts: claim.starts }; + const kept = { claimId: claim.claimId, claimedAt: claim.claimedAt, reviewers: recorded, history, starts: claim.starts }; + if (record) await record(tx, report && await closingOf(tx, tenantId, pr, runId, report.head, history)); if (due || waiting) { - await writeClaim(tx, tenantId, pr, { ...closed, status: "live", runId }); + await writeClaim(tx, tenantId, pr, { ...kept, status: "live", runId }); return { outcome: "more", wait: !waiting }; } - await writeClaim(tx, tenantId, pr, { ...closed, status: "ended" }); + await writeClaim(tx, tenantId, pr, { ...kept, status: "ended" }); return { outcome: "ended" }; }); } + +/** The mirror request of the verdict `runId` posted on `head`, which approving one of its held Dos sends again. */ +export async function passRequest(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, head: string): Promise | undefined> { + const key = claimKey(pr); + const [row] = await db.select({ state: managerState.state }).from(managerState).where(and(eq(managerState.tenantId, tenantId), eq(managerState.key, key), eq(managerState.kind, "claim"))); + return row && claimOf(key, row.state).history.findLast((pass) => pass.runId === runId && pass.head === head && pass.request !== undefined)?.request; +} + +/** + * A pull request a manager claimed: its closed passes; while its manager is up, that run and the head of the newest pull + * request mail recorded; and the pass a manager that went away left open. + */ +export type ManagedPull = { pr: PullRequestRef; history: ClosedPass[]; live?: { runId: string; head?: string }; gone?: { runId: string; pass: Pass } }; + +/** The newest head a pull request mail named, which a live manager takes next or has taken. */ +async function newestHead(tx: Tx, tenantId: string, pr: PullRequestRef): Promise { + const [row] = await tx.select({ key: managerState.key, state: managerState.state }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, like(managerState.key, `${eventPrefix(pr)}%`), sql`${managerState.state}->>'type' = 'webhook'`, sql`${managerState.state}->'payload'->>'kind' IS DISTINCT FROM ${APPROVE_DO_KIND}`)) + .orderBy(desc(sql`${managerState.state}->>'receivedAt'`), desc(managerState.key)) + .limit(1); + return row && eventOf(row.key, row.state).head; +} + +/** The claims of `pulls` a manager made. */ +export async function managedPulls(db: DB["db"], tenantId: string, pulls: readonly PullRequestRef[]): Promise { + if (pulls.length === 0) return []; + return db.transaction(async function readClaims(tx) { + const now = new Date(); + const byKey = new Map(pulls.map((pr) => [claimKey(pr), pr])); + const rows = await tx.select({ key: managerState.key, state: managerState.state }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), eq(managerState.kind, "claim"), inArray(managerState.key, [...byKey.keys()]))); + const managed: ManagedPull[] = []; + for (const { key, state } of rows) { + const claim = claimOf(key, state); + const pr = byKey.get(key); + if (!pr) throw new Error(`manager_state ${key} is not a claim asked for`); + const runId = claim.status === "live" ? claim.runId : undefined; + if (runId === undefined) { + managed.push({ pr, history: claim.history }); + } else if (await runGone(tx, runId, now)) { + managed.push({ pr, history: claim.history, ...(claim.pass && { gone: { runId, pass: claim.pass } }) }); + } else { + const head = await newestHead(tx, tenantId, pr); + managed.push({ pr, history: claim.history, live: { runId, ...(head !== undefined && { head }) } }); + } + } + return managed; + }); +} diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index fb0841420..59fba8c67 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -34,6 +34,7 @@ import { TRIAGE_LISTENERS, TRIAGE_MAIL_SENDER, type RepoSource } from "@corbits/ import { createInterchangeHub } from "./interchange-hub.js"; import { createManagerIntake } from "./manager-intake.js"; import { MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; +import { passRequest } from "./manager-state.js"; import { mountTenantRoutes } from "./tenant-routes.js"; import { connectDockerSidecars, dockerPlaces, sweepDockerSidecarsOf } from "./sidecar/docker.js"; import { connectE2bSidecars, tendE2bSidecarsOf, type E2bSidecarProvisioner } from "./sidecar/e2b.js"; @@ -80,6 +81,7 @@ import { createRepoReads } from "./github/repo-reads.js"; import { createGithubPrActions } from "./github/pr-actions.js"; import { createGithubHeldDos } from "./github/held-dos.js"; import { createVerdictRecordStore } from "./github/verdict-record-store.js"; +import { createManagerRuns } from "./github/manager-runs.js"; import { createSessions } from "./sessions.js"; import { createVerifiedGate } from "./verified-gate.js"; import { createHubOwner } from "./hub-owner.js"; @@ -571,6 +573,7 @@ function reconcileTenants() { return composition.db.select({ id: schema.tenant.id, domain: schema.tenant.domain, config: schema.tenant.config }).from(schema.tenant); } const launchSpecs = createWorkflowRunLaunchSpecStore(composition.db); +const managerRuns = createManagerRuns(composition.db, verdictRecordStore, now); async function liveTriageDeployments(tenantId: string, workflow: string) { return withWorkflowVersions(launchSpecs, await resolveLiveDeployments(composition.db, tenantId, workflow)); } @@ -660,6 +663,7 @@ function triageStack(triage: TriageWorkflow) { openHeadsFor: tenantOpenHeads, activity: repoActivity, observeTenant: observer.observe, + managerRuns, store: triageStateStore, readCheckPack, send: sendTriageMail, @@ -715,6 +719,7 @@ const githubPrTriage = createGithubPrTriage({ readCheckPack, liveDeployments: liveTriageDeployments, observeRuns: observeTriageRuns, + managerRuns, store: triageStateStore, send: sendTriageMail, policy: reconcilePolicy, @@ -729,6 +734,9 @@ const githubHeldDos = createGithubHeldDos({ liveDeployments: liveTriageDeployments, deliver: deliverToDeployment, toListener, + passRequest: function managerPassRequest(tenantId, pr, runId, head) { + return passRequest(composition.db, tenantId, pr, runId, head); + }, log: logJson, }); const reconcileSourceDeployments = createSourceDeployments({ diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7455ff05a..2a02d290a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -261,10 +261,15 @@ any other tenant or an unknown sender. review report, or an overdue reviewer, pr-triage's own steps run on the newest mail for the head, handed over as `{ taken }`, which they trust as the hub's: a head held for CI is marked waiting when that mail is the newest - for it. Anything else is triaged. A policy of only deterministic rules posts - on this pass. One whose rules need the judge, or that asks for code review, - posts once, after each of those has reported or given up: the machine - verdict, the judge's verdict and the code review together. Done then answers + for it. Anything else is triaged. Every pass runs its approvals step, which + approves nothing when there is nothing to run. A policy of only deterministic + rules posts on this pass. One whose rules need the judge, or that asks for + code review, posts once, after each of those has reported or given up: the machine + verdict, the judge's verdict and the code review together. Done reports what + the pass ran: the approved Dos' outcomes, a head it held for CI, or the + verdict it posted as its evaluate, mirror, execute and rules steps output + it, with the facts the verdict keeps rather than the diff. Each branch that + reports one of these has its own done step. Done then answers `ended`, or `more`: the next pass runs at once when an event is pending, or every 30 seconds while it waits for its judge or its code review. Once the passes end, fail or run out, or take finds another run holds the @@ -310,6 +315,20 @@ any other tenant or an unknown sender. one an earlier manager deployed runs out its own deadline. Stock grants the deployer `read` on a run it deployed, and the vendored Interchange carries `manage` instead (`vendor/interchange/VENDORED.md`). + The hub records the verdict done reports as the manager run's, in done's + transaction (`apps/hub/src/manager-routes.ts`), with the row-building code + pr-triage's run logs use (`apps/hub/src/github/verdict-row.ts`). The + reconciler and Triage again read the managers of the pull requests they look + at from their claims and those rows (`apps/hub/src/github/manager-runs.ts`): + a manager that is up runs the head of the newest pull request mail recorded + for it, so its head is not mailed again; one that went away with a pass open + failed on that pass's head, which is retried with backoff; a head whose last + pass was held for CI waits for its CI; a recorded verdict settles its head as + a pr-triage run's does. Triage again can queue a head again as soon as its + manager has reported done, or went away, since it was queued. + Approving a held Do of a manager's verdict reads the Do from the recorded + verdict and sends it with the mirror request its pass kept; a Do the run's + newer verdict no longer holds answers `verdict_superseded`. - `pr-code-review-reviewer-github` and `pr-code-review-reviewer-bitbucket-datacenter` (`codeReviewReviewer` in `@corbits/pr-code-review`): the code review a manager deploys, alone on its @@ -810,8 +829,13 @@ read it, and passes the rest of the request to stock. recorded file is older than its own last one; the `@corbits/*` packages' migrations keep no record of their own. - Triage's tables, in the `corbits_triage` schema: `verdict` (verdicts per - head, with the outcomes of the actions their run executed; a verdict's host - is its run's workflow's, and GitHub when that run is gone), `device_code` + head, with the outcomes of the actions their run executed, copied from a + pr-triage run's log or reported by a manager's done; a manager run's newest + verdict on a pull request replaces the one it recorded before, keeping the + outcomes of Dos already written on its head. The verdict's `startedAt` is + when the run first took its head, and its trigger the newest mail for it. A verdict's + host is its run's workflow's, by the workflow id its deployment's frozen + projection keeps, and GitHub when that run is gone), `device_code` (better-auth's device authorization, for agents a signed-in person approves) and `manager_state` (below), migrated from `apps/hub/migrations` by `db:migrate` in one transaction under an advisory lock. Nothing else is @@ -819,7 +843,9 @@ read it, and passes the rest of the request to stock. invited principal. - `manager_state` (`apps/hub/src/manager-state.ts`) holds a pull request's manager claim (`mgr::#`: status, the run it names, the - open pass, reviewers and closed passes) and its pending events + open pass, reviewers and closed passes, each closed pass with whether it + held its head for CI, and the newest pass of a run on a head with the mirror + request of the verdict it posted) and its pending events (`evt::#:`), each change one short transaction. Recording an event is a no-op on a redelivery; the database stamps `receivedAt` to the microsecond as it inserts the row, so events @@ -849,7 +875,12 @@ read it, and passes the rest of the request to stock. `triage-manager:state`; reviewer-result needs `create` on `triage-manager:result` and nothing more. The calling run is the token's run, never a body field. Take and done - answer 409 `not_manager` to a run the claim does not name. + answer 409 `not_manager` to a run the claim does not name. Done's body + carries the pass's `report` and `approvals`: the hub keeps the report on the + closed pass (a pass that reported without taking events, as one past a + reviewer's deadline, is closed as its own), and in the same transaction + records a posted verdict and sets each approved Do's outcome on the verdicts + that suggest it. Reviewer-result is not fenced: any run with the grant records under its own run id, and the manager matches results to its reviewers by run id. Its body is a `StageResult` (`@corbits/triage-contracts`): the reviewer's diff --git a/packages/triage-contracts/src/hub-responses.ts b/packages/triage-contracts/src/hub-responses.ts index 28496a566..06c8d2fc9 100644 --- a/packages/triage-contracts/src/hub-responses.ts +++ b/packages/triage-contracts/src/hub-responses.ts @@ -35,6 +35,9 @@ const DetailAction = type({ "error?": "string", }); +/** The pull request's facts a verdict keeps. */ +export const VerdictFacts = type({ "+": "delete", "title?": "string", "author?": "string", "draft?": "boolean", "mergeable?": "boolean | null", "checks?": "string", "reviewAnswered?": "boolean" }); + /** What the portal shows of a verdict beyond the columns: the verdict's own fields, and the facts it was made from. */ export const VerdictDetail = type({ "+": "delete", @@ -55,7 +58,7 @@ export const VerdictDetail = type({ "automatic?": "string[]", /** Why the run's mirror posted nothing on the host: `stale`, `waiting`, `no_request` or `no_grants`. */ "notPosted?": "string", - "facts?": { "+": "delete", "title?": "string", "author?": "string", "draft?": "boolean", "mergeable?": "boolean | null", "checks?": "string", "reviewAnswered?": "boolean" }, + "facts?": VerdictFacts, }); export type VerdictDetail = typeof VerdictDetail.infer; diff --git a/packages/triage-contracts/src/listener.ts b/packages/triage-contracts/src/listener.ts index e327aee34..dbac70dd1 100644 --- a/packages/triage-contracts/src/listener.ts +++ b/packages/triage-contracts/src/listener.ts @@ -4,6 +4,9 @@ import { MANAGER_RESULT_ACTION, MANAGER_RESULT_RESOURCE, MANAGER_STATE_ACTION, M /** Each host's listener by its definition id, also its mail address: a workspace with one live routes pull requests to managers. */ export const TRIAGE_LISTENERS: Record = { github: "triage-listener-github", "bitbucket-datacenter": "triage-listener-bitbucket-datacenter" }; +/** Each host's pull request manager by its definition id, also its mail address. */ +export const TRIAGE_MANAGERS: Record = { github: "triage-manager-github", "bitbucket-datacenter": "triage-manager-bitbucket-datacenter" }; + /** The system role a person holds to publish the listener's chain and deploy it, which every deployment of it then acts under. */ export const DEPLOYER_ROLE = "Deployer"; diff --git a/packages/triage-manager/src/actions/manager.ts b/packages/triage-manager/src/actions/manager.ts index ba336ede1..ec44ac7c0 100644 --- a/packages/triage-manager/src/actions/manager.ts +++ b/packages/triage-manager/src/actions/manager.ts @@ -1,6 +1,6 @@ import type { EffectContext } from "@intx/workflow"; import { type, type Type } from "arktype"; -import { APPROVE_DO_KIND, CodeReviewOutput, JudgeVerdict, StageResult, WAITING_KIND } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, CodeReviewOutput, JudgeVerdict, StageResult, VerdictFacts, WAITING_KIND } from "@corbits/triage-contracts"; import { decided } from "@corbits/triage-workflows/logic/settled"; import { answered, callHub, endRun, managerMailOf, tenantPath, type ManagerMail } from "./hub.js"; @@ -104,13 +104,39 @@ export async function take(input: unknown, ctx: EffectContext, _signal: AbortSig return { run: answer.outcome === "pass" || overdue, approvals, triage, ...(taken && { taken }), head, ...report, ...(verdict && { verdict }), overdue, reviewers }; } +const Output = { output: "Record" } as const; + const DoneInput = type({ payload: "unknown", take: { output: TakeOutput }, + "approvals?": { output: { results: "Record[]" } }, + "waiting?": { output: { waiting: "boolean" } }, + "rules?": { output: { items: type({ facts: "Record", "det?": "unknown" }).array() } }, + "evaluateRules?": Output, + "mirrorRules?": Output, + "executeRules?": Output, "reviewMailed?": { output: { "reviewer?": Reviewer } }, "judgeMailed?": { output: { "reviewer?": Reviewer } }, }); +type DoneInput = typeof DoneInput.infer; + +/** The rules' reasons, which a verdict that gives none of its own keeps as its evidence. */ +const RuledReasons = type({ "+": "delete", "reason?": "string", "findings?": type({ "+": "delete", "reason?": "string" }).array() }); + +/** + * What the pass did on its head, for the hub to record: held it for CI, or posted the verdict its mirror ran on, with + * only the facts and reasons the verdict keeps, never the diff. + */ +function doneReport({ take, waiting, rules, evaluateRules, mirrorRules, executeRules }: DoneInput): Record | undefined { + const { head } = take.output; + if (head === undefined) return undefined; + if (waiting?.output.waiting) return { head, waiting: true }; + if (!rules || !evaluateRules || !mirrorRules) return undefined; + const items = rules.output.items.map(({ facts, det }) => ({ facts: { ...VerdictFacts.assert(facts), repo: facts["repo"], number: facts["number"] }, ...(det !== undefined && { det: RuledReasons.assert(det) }) })); + return { head, verdict: evaluateRules.output, mirror: mirrorRules.output, ...(executeRules && { executed: executeRules.output }), rules: { items } }; +} + /** This pass's report, or a passed deadline, recorded on the reviewer it belongs to. */ function remembered(take: TakeOutput): Reviewer[] { const { head, reviewers } = take; @@ -172,8 +198,8 @@ function abandoned({ head, reviewers }: TakeOutput, runId: string): Reviewer[] { } /** - * Closes the pass and stops the code reviews it no longer waits for. `more` runs the next pass at once, or with `wait` - * after waiting for a code review. + * Closes the pass with what it ran and stops the code reviews it no longer waits for. `more` runs the next pass at once, + * or with `wait` after waiting for a code review. */ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ outcome: "ended" | "refused" } | { outcome: "more"; wait: boolean }> { const parsed = DoneInput(input); @@ -181,7 +207,9 @@ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSig if (parsed.take.output.refused) return { outcome: "refused" }; const mail = managerMailOf(parsed.payload); const mailed = parsed.judgeMailed?.output.reviewer ?? parsed.reviewMailed?.output.reviewer; - const answer = await asManager(ctx, mail, "done", { ...mail.pullRequest, reviewers: reviewersAfter(parsed.take.output, mailed) }, Done); + const report = doneReport(parsed); + const body = { ...mail.pullRequest, reviewers: reviewersAfter(parsed.take.output, mailed), ...(report && { report }), ...(parsed.approvals && { approvals: parsed.approvals.output }) }; + const answer = await asManager(ctx, mail, "done", body, Done); if (answer === undefined) return { outcome: "refused" }; for (const reviewer of abandoned(parsed.take.output, mail.runId)) await endRun(ctx, mail.tenantId, reviewer.runId); return answer; diff --git a/packages/triage-manager/src/loops.ts b/packages/triage-manager/src/loops.ts index e4350270c..657a656bf 100644 --- a/packages/triage-manager/src/loops.ts +++ b/packages/triage-manager/src/loops.ts @@ -1,10 +1,12 @@ // The manager's pass loop. Each fn is pure: it reads a finished pass's step outputs, keyed by step id. import { asRecord } from "@corbits/common"; -/** The pass's done, on whichever branch it ran. A skipped step is missing, so the first of these that ran is the pass's. */ +const DONES = ["done", "donePosted", "doneWaiting", "doneReviewing", "doneJudging"]; + +/** The pass's done, on whichever branch it ran. The loop reads a pass's outputs from its log, where each done it skipped holds the gate's skip marker. */ function doneOf(output: unknown): Record | undefined { const steps = asRecord(output); - return asRecord(steps?.["done"]) ?? asRecord(steps?.["doneReviewing"]) ?? asRecord(steps?.["doneJudging"]); + return DONES.map((id) => asRecord(steps?.[id])).find((done) => done !== undefined && done["skipped"] !== true); } /** Another pass while done answers `more`. */ diff --git a/packages/triage-manager/src/triage-manager-bitbucket-datacenter.ts b/packages/triage-manager/src/triage-manager-bitbucket-datacenter.ts index 6614eb657..3be74a1a4 100644 --- a/packages/triage-manager/src/triage-manager-bitbucket-datacenter.ts +++ b/packages/triage-manager/src/triage-manager-bitbucket-datacenter.ts @@ -1,6 +1,5 @@ +import { TRIAGE_MANAGERS } from "@corbits/triage-contracts"; import { publishedFor } from "./release.js"; import { triageManager } from "./triage-manager.js"; -export const TRIAGE_MANAGER_BITBUCKET_DATACENTER_ADDRESS = "triage-manager-bitbucket-datacenter"; - -export const workflow = triageManager({ id: TRIAGE_MANAGER_BITBUCKET_DATACENTER_ADDRESS, sources: ["bitbucket-datacenter"], codeReview: publishedFor("bitbucket-datacenter", "codeReview"), judge: publishedFor("bitbucket-datacenter", "judge") }); +export const workflow = triageManager({ id: TRIAGE_MANAGERS["bitbucket-datacenter"], sources: ["bitbucket-datacenter"], codeReview: publishedFor("bitbucket-datacenter", "codeReview"), judge: publishedFor("bitbucket-datacenter", "judge") }); diff --git a/packages/triage-manager/src/triage-manager-github.ts b/packages/triage-manager/src/triage-manager-github.ts index 2b4b64313..4da4dbbfa 100644 --- a/packages/triage-manager/src/triage-manager-github.ts +++ b/packages/triage-manager/src/triage-manager-github.ts @@ -1,6 +1,5 @@ +import { TRIAGE_MANAGERS } from "@corbits/triage-contracts"; import { publishedFor } from "./release.js"; import { triageManager } from "./triage-manager.js"; -export const TRIAGE_MANAGER_GITHUB_ADDRESS = "triage-manager-github"; - -export const workflow = triageManager({ id: TRIAGE_MANAGER_GITHUB_ADDRESS, sources: ["github"], codeReview: publishedFor("github", "codeReview"), judge: publishedFor("github", "judge") }); +export const workflow = triageManager({ id: TRIAGE_MANAGERS["github"], sources: ["github"], codeReview: publishedFor("github", "codeReview"), judge: publishedFor("github", "judge") }); diff --git a/packages/triage-manager/src/triage-manager.test.ts b/packages/triage-manager/src/triage-manager.test.ts index 1e66d7f5f..48c1e8f34 100644 --- a/packages/triage-manager/src/triage-manager.test.ts +++ b/packages/triage-manager/src/triage-manager.test.ts @@ -163,8 +163,13 @@ function githubHost(head = HEAD, { pr = PR, pull = {}, checkRuns }: { pr?: Snaps } const TAKE = { method: "POST", path: `${TENANT_PATH}/manager/take`, body: PULL }; -const DONE = { method: "POST", path: `${TENANT_PATH}/manager/done`, body: { ...PULL, reviewers: [] } }; -const STAGE_1 = ["approvals", "done", "evaluateRules", "executeRules", "facts", "held", "mirrorRules", "paused", "postReview", "posting", "ran", "reviewScope", "reviewing", "route", "rules", "take", "triaged", "unready", "waiting"]; +const DONE = { method: "POST", path: `${TENANT_PATH}/manager/done`, body: expect.objectContaining({ ...PULL, reviewers: [] }) }; + +/** What the pass reported to the hub as done. */ +function doneBody(calls: readonly HubCall[]): Record { + return calls.find((call) => call.path === DONE.path)?.body as Record; +} +const STAGE_1 = ["approvals", "donePosted", "evaluateRules", "executeRules", "facts", "held", "mirrorRules", "paused", "postReview", "posting", "ran", "reviewScope", "reviewing", "route", "rules", "take", "triaged", "unready", "waiting"]; /** Publishes `release` as the release.json the portal writes beside the workflows; the repository's own holds none. */ function publish(release: Record): void { @@ -197,10 +202,20 @@ describe("triage-manager", () => { expect(hub.calls).toEqual([TAKE, DONE, END]); }); + test("done reports the verdict the pass posted, as its evaluate, mirror and execute steps output it, with the facts the verdict keeps and not the diff", async () => { + const { host } = githubHost(); + const hub = fakeHub({ takes: [PASS], dones: [ENDED] }); + const { outputs } = await runMail(github, wake(), host, hub.hub); + const { report, approvals } = doneBody(hub.calls) as { report: { rules: { items: { facts: Record }[] } }; approvals: unknown }; + expect(report).toEqual({ head: HEAD, verdict: outputs["evaluateRules"], mirror: outputs["mirrorRules"], executed: outputs["executeRules"], rules: { items: [{ facts: expect.objectContaining({ repo: PR.repo, number: PR.number, title: PR.title }), det: { reason: expect.any(String), findings: expect.any(Array) } }] } }); + expect(Object.keys(report.rules.items[0]!.facts)).not.toContain("files"); + expect(approvals).toEqual({ results: [] }); + }); + test("with an event pending after done, the manager runs its next pass at once, and a refused newcomer is harmless", async () => { const { writes, host } = githubHost(); const newcomer = fakeHub({ takes: [REFUSED] }); - expect(await runMail(github, wake(), host, newcomer.hub)).toMatchObject({ terminalStatus: "completed", stop: { stopped: true }, passes: 1, ran: ["done", "paused", "ran", "take"] }); + expect(await runMail(github, wake(), host, newcomer.hub)).toMatchObject({ terminalStatus: "completed", stop: { stopped: true }, passes: 1, ran: ["approvals", "done", "paused", "ran", "take"] }); expect(newcomer.calls).toEqual([TAKE, END]); expect(writes).toEqual([]); @@ -233,6 +248,7 @@ describe("triage-manager", () => { const { ran } = await runMail(github, wake(), host, hub.hub); expect(ran).toEqual(["approvals", "done", "facts", "held", "judgeScope", "judgeWait", "judging", "paused", "ran", "route", "rules", "take", "triaged", "unready", "waiting"]); expect(writes).toEqual([]); + expect(doneBody(hub.calls)).not.toHaveProperty("report"); }); test("a head the hub holds for CI gets the Waiting label and is not triaged", async () => { @@ -241,13 +257,14 @@ describe("triage-manager", () => { const { ran } = await runMail(github, wake(), host, hub.hub); expect(ran).toContain("markWaiting"); expect(ran).not.toContain("facts"); + expect(doneBody(hub.calls)["report"]).toEqual({ head: HEAD, waiting: true }); expect(writes).toContainEqual(expect.objectContaining({ method: "POST", path: `/repos/${PR.repo}/issues/${PR.number}/labels`, body: { labels: [WAITING_LABEL] } })); }); test("an old review's reply finds nothing for its head, runs no stage 1 and stops the superseded review", async () => { const { writes, host } = githubHost(); const hub = fakeHub({ takes: [{ ...IDLE, request: { ...REQUEST, headSha: "newer" }, reviewers: [{ ...reviewer(), head: "older" }] }], dones: [ENDED] }); - expect((await runMail(github, wake(), host, hub.hub)).ran).toEqual(["done", "paused", "ran", "take"]); + expect((await runMail(github, wake(), host, hub.hub)).ran).toEqual(["approvals", "done", "paused", "ran", "take"]); expect(writes).toEqual([]); expect(hub.calls).toEqual([TAKE, DONE, STOP_REVIEW, END]); }); @@ -267,6 +284,7 @@ describe("triage-manager", () => { const { ran, outputs } = await runMail(github, wake(), host, hub.hub); expect(ran).toEqual(["approvals", "done", "paused", "ran", "take", "triaged"]); expect(outputs["approvals"]).toEqual({ results: [{ dos: [expect.objectContaining({ effectId, status: "done" })] }] }); + expect(doneBody(hub.calls)["approvals"]).toEqual(outputs["approvals"]); expect(writes).toEqual([expect.objectContaining({ method: "POST", path: `/repos/${PR.repo}/issues/${PR.number}/comments`, body: { body: expect.stringMatching(/\nhi$/) } })]); expect(hub.calls).toEqual([TAKE, DONE, END]); }); @@ -311,7 +329,7 @@ describe("triage-manager", () => { const hub = fakeHub({ takes: [{ ...PASS, pass: { ...PASS.pass, head: "abc" }, events: [mailEvent("abc", request)], request }], dones: [ENDED] }); expect((await runMail(bitbucket, wake(bbPull), host, hub.hub)).terminalStatus).toBe("completed"); expect(builds).toEqual([expect.objectContaining({ state: "FAILED" })]); - expect(hub.calls.map(({ body }) => body)).toEqual([bbPull, { ...bbPull, reviewers: [] }, undefined]); + expect(hub.calls.map(({ body }) => body)).toEqual([bbPull, expect.objectContaining({ ...bbPull, reviewers: [] }), undefined]); }); }); @@ -338,7 +356,7 @@ describe("triage-manager with code review", () => { expect(ran).not.toContain("deployReview"); expect(writes).toContainEqual(READY); expect(writes).toContainEqual(POSTED_REVIEW); - expect(second.calls).toEqual([TAKE, { ...DONE, body: { ...PULL, reviewers: [{ ...reviewer(), result: REVIEW }] } }, END]); + expect(second.calls).toEqual([TAKE, { ...DONE, body: expect.objectContaining({ ...PULL, reviewers: [{ ...reviewer(), result: REVIEW }] }) }, END]); expect(mailed).toHaveLength(1); }); @@ -352,7 +370,7 @@ describe("triage-manager with code review", () => { expect(writes).toContainEqual(READY); expect(writes).toContainEqual(POSTED_REVIEW); expect(mailed).toEqual([]); - expect(hub.calls).toEqual([TAKE, { ...DONE, body: { ...PULL, reviewers: [{ ...reviewer(), result: REVIEW }] } }, END]); + expect(hub.calls).toEqual([TAKE, { ...DONE, body: expect.objectContaining({ ...PULL, reviewers: [{ ...reviewer(), result: REVIEW }] }) }, END]); }); test("a stored review is not posted once this pass's policy no longer asks for code review", async () => { @@ -447,10 +465,18 @@ describe("triage-manager with a judge", () => { expect(ran).not.toContain("deployJudge"); expect(outputs["evaluateRules"]).toEqual(verdict); expect(writes).toContainEqual(JUDGE_POSTED); - expect(second.calls[1]).toEqual({ ...DONE, body: { ...PULL, reviewers: [{ ...JUDGE, result: verdict }] } }); + expect(second.calls[1]).toEqual({ ...DONE, body: expect.objectContaining({ ...PULL, reviewers: [{ ...JUDGE, result: verdict }] }) }); + expect(doneBody(second.calls)["report"]).toMatchObject({ head: HEAD, verdict }); expect(mailed).toHaveLength(1); }); + test("a pass that mailed its judge and answers more runs the next pass, which ends", async () => { + const { host } = githubHost(); + const hub = fakeHub({ takes: [{ ...PASS, request: hubRequest(JUDGED) }, IDLE], dones: [MORE, ENDED] }); + expect(await runMail(github, wake(), host, hub.hub)).toMatchObject({ terminalStatus: "completed", passes: 2 }); + expect(hub.calls.filter((call) => call.path === DONE.path)).toHaveLength(2); + }); + test("a later event on the same head deploys no second judge and posts the judge's verdict, whatever changed on the pull request since", async () => { const verdict = await judgeVerdict(JSON.parse((await firstJudgeMail()).content).rules); const { writes, host } = githubHost(HEAD, { pr: { ...PR, title: "Renamed since", body: "", bodyHtml: "" } }); @@ -460,7 +486,8 @@ describe("triage-manager with a judge", () => { expect(ran).not.toContain("deployJudge"); expect(outputs["evaluateRules"]).toEqual(verdict); expect(writes).toContainEqual(JUDGE_POSTED); - expect(hub.calls).toEqual([TAKE, { ...DONE, body: { ...PULL, reviewers: [{ ...JUDGE, result: verdict }] } }, END]); + expect(hub.calls).toEqual([TAKE, { ...DONE, body: expect.objectContaining({ ...PULL, reviewers: [{ ...JUDGE, result: verdict }] }) }, END]); + expect(doneBody(hub.calls)["report"]).toMatchObject({ head: HEAD, verdict }); }); test("a stored verdict gives way when this pass's rules no longer need the judge, as on a draft", async () => { @@ -563,12 +590,12 @@ describe("triage-manager with a judge and code review", () => { const { writes, host } = githubHost(); const mailed: unknown[] = []; const request = hubRequest(JUDGED, { ...CODE_REVIEW, ...REVIEW_REQUEST }); - const first = fakeHub({ takes: [{ ...PASS, request }], dones: [{ outcome: "more", wait: true }] }); + const first = fakeHub({ takes: [{ ...PASS, request }], dones: [ENDED] }); expect((await runMail(github, wake(), host, first.hub, mailed)).ran).toContain("doneJudging"); expect(writes).toEqual([]); const verdict = await judgeVerdict(JSON.parse((mailed[0] as { content: string }).content).rules); - const second = fakeHub({ takes: [judgeReported(verdict, request)], dones: [{ outcome: "more", wait: true }] }); + const second = fakeHub({ takes: [judgeReported(verdict, request)], dones: [ENDED] }); const again = await runMail(github, wake(), host, second.hub, mailed); expect(again.ran).toContain("deployReview"); expect(again.ran).not.toContain("mirrorRules"); @@ -581,7 +608,7 @@ describe("triage-manager with a judge and code review", () => { expect(ran).not.toContain("deployJudge"); expect(writes).toContainEqual(JUDGE_POSTED); expect(writes).toContainEqual(POSTED_REVIEW); - expect(third.calls[1]).toEqual({ ...DONE, body: { ...PULL, reviewers: [{ ...JUDGE, result: verdict }, { ...reviewer(), result: REVIEW }] } }); + expect(third.calls[1]).toEqual({ ...DONE, body: expect.objectContaining({ ...PULL, reviewers: [{ ...JUDGE, result: verdict }, { ...reviewer(), result: REVIEW }] }) }); }); }); diff --git a/packages/triage-manager/src/triage-manager.ts b/packages/triage-manager/src/triage-manager.ts index 13ac130ae..aaca9a3a2 100644 --- a/packages/triage-manager/src/triage-manager.ts +++ b/packages/triage-manager/src/triage-manager.ts @@ -48,7 +48,7 @@ function reviewSteps(sources: Sources): Record { reviewMailed: action({ handler: "reviewMailed", input: stepsOf("deployReview", "mailReview"), after: ["mailReview"] }), reviewSent: gate({ when: { from: "steps.reviewMailed.output.post" }, then: "mirrorRules", else: "doneReviewing", after: ["reviewMailed"] }), // A pass that mailed its review records it as a reviewer; a skipped step is missing from every selector, so this branch has its own done. - doneReviewing: hubAction("done", { merge: [MAIL, stepsOf("take", "reviewMailed")] }, ["reviewSent"]), + doneReviewing: hubAction("done", { merge: [MAIL, stepsOf("take", "approvals", "reviewMailed")] }, ["reviewSent"]), posting: gate({ when: { from: "steps.reviewScope.output.post" }, then: "mirrorRules", else: "done", after: ["reviewing"] }), postReview: action({ handler: "postReview", input: { merge: [MAIL, stepsOf("take", "reviewScope")] }, after: ["mirrorRules"], credentials: sources, effect: { requires: [WRITE_PULL_REQUEST] } }), }; @@ -70,7 +70,7 @@ function judgeSteps(): Record { judgeMailed: action({ handler: "judgeMailed", input: stepsOf("deployJudge", "mailJudge"), after: ["mailJudge"] }), judgeSent: gate({ when: { from: "steps.judgeMailed.output.post" }, then: "evaluateRules", else: "doneJudging", after: ["judgeMailed"] }), // A pass that mailed its judge records it as a reviewer; a skipped step is missing from every selector, so this branch has its own done. - doneJudging: hubAction("done", { merge: [MAIL, stepsOf("take", "judgeMailed")] }, ["judgeSent"]), + doneJudging: hubAction("done", { merge: [MAIL, stepsOf("take", "approvals", "judgeMailed")] }, ["judgeSent"]), judgeWait: gate({ when: { from: "steps.judgeScope.output.post" }, then: "evaluateRules", else: "done", after: ["judging"] }), }; } @@ -91,9 +91,10 @@ function passOf(id: string, sources: Sources) { // Nothing signals it: the timeout is the wait, which a loop body cannot sleep. poll: awaitSignal({ name: "manager-poll", timeout: POLL_MS, onTimeout: "take", after: ["paused"] }), take: hubAction("take", { from: "trigger.payload.payload" }, ["paused", "poll"]), - ran: gate({ when: { from: "steps.take.output.run" }, then: "approvals", else: "done", after: ["take"] }), - approvals: { ...hostAction("execute", sources, stepsOf("take"), ["ran"]), handler: "executeApprovals" }, - triaged: gate({ when: { from: "steps.take.output.triage" }, then: "waiting", else: "done", after: ["approvals"] }), + // Runs on every pass, approving nothing when there is nothing to run, so every branch's done reports its outcomes. + approvals: { ...hostAction("execute", sources, stepsOf("take"), ["take"]), handler: "executeApprovals" }, + ran: gate({ when: { from: "steps.take.output.run" }, then: "triaged", else: "done", after: ["approvals"] }), + triaged: gate({ when: { from: "steps.take.output.triage" }, then: "waiting", else: "done", after: ["ran"] }), waiting: { ...waiting, input: TAKEN, after: ["triaged"] }, held, markWaiting, @@ -107,7 +108,10 @@ function passOf(id: string, sources: Sources) { mirrorRules: { ...mirrorRules, after: ["posting", "reviewSent"] }, // The verdict's automatic Dos run as pr-triage runs them, after the verdict and any code review are up. executeRules: { ...executeRules, after: ["postReview"] }, - done: hubAction("done", { merge: [MAIL, stepsOf("take")] }, ["ran", "triaged", "markWaiting", "judgeWait", "posting", "executeRules"]), + // A skipped step is missing from every selector, so the branches that report a held head or a posted verdict have their own done. + doneWaiting: hubAction("done", { merge: [MAIL, stepsOf("take", "approvals", "waiting")] }, ["markWaiting"]), + donePosted: hubAction("done", { merge: [MAIL, stepsOf("take", "approvals", "rules", "evaluateRules", "mirrorRules", "executeRules")] }, ["executeRules"]), + done: hubAction("done", { merge: [MAIL, stepsOf("take", "approvals")] }, ["ran", "triaged", "judgeWait", "posting"]), }, }); }