diff --git a/apps/hub/src/code-review-runs.postgres.test.ts b/apps/hub/src/code-review-runs.postgres.test.ts index df8557ec..6d501c84 100644 --- a/apps/hub/src/code-review-runs.postgres.test.ts +++ b/apps/hub/src/code-review-runs.postgres.test.ts @@ -59,9 +59,9 @@ test("PostgreSQL: a pull request's latest code review is what its manager record await db.insert(schema.workflowRun).values({ id: runId, definitionId: "wfd_review", anchorRunId: runId, tenantId: TENANT, status: "running", createdAt }); return createdAt; } - async function claim(reviewers: unknown[], taken: string[]) { + async function claim(reviewers: unknown[], taken: string[], status: "live" | "ended" = "live") { const since = new Date(0).toISOString(); - const state = { status: "live", claimId: "claim-1", claimedAt: since, runId: "run_manager", reviewers, history: [{ id: "pass-1", head: "sha1", events: taken, takenAt: since, takes: 1, runId: "run_manager", doneAt: since }], starts: 1 }; + const state = { status, claimId: "claim-1", claimedAt: since, ...(status === "live" && { runId: "run_manager" }), reviewers, history: [{ id: "pass-1", head: "sha1", events: taken, takenAt: since, takes: 1, runId: "run_manager", doneAt: since }], starts: 1 }; await db.execute(sql`INSERT INTO corbits_triage.manager_state (tenant_id, key, kind, state) VALUES (${TENANT}, 'mgr:github:acme/widgets#7', 'claim', ${JSON.stringify(state)}::jsonb) ON CONFLICT (tenant_id, key) DO UPDATE SET state = excluded.state`); } let head: PullRequestToReview = { title: "Fix", description: "", headSha: "sha5", draft: false, open: true, diff: DIFF }; @@ -97,6 +97,10 @@ test("PostgreSQL: a pull request's latest code review is what its manager record const live = await deployed("run_live"); await claim([reviewer("run_live", "sha4")], [eventKey("run_old"), eventKey("run_new")]); expect(await reviewsOf(GITHUB)).toEqual({ review: { status: "running", headSha: "sha4", startedAt: live.toISOString() }, completed: finished, carriedTo: null }); + // A manager ends once it mailed its review, which still runs until it reports or its deadline passes, and no head takes over the finished one meanwhile. + await claim([reviewer("run_live", "sha4")], [eventKey("run_old"), eventKey("run_new")], "ended"); + expect(await reviewsOf(GITHUB, "sha5")).toEqual({ review: { status: "running", headSha: "sha4", startedAt: live.toISOString() }, completed: finished, carriedTo: null }); + expect(hostReads).toBe(0); // A finished review names its diff's hash, so a later head with the same diff takes it over. await recordEvent(db, TENANT, GITHUB, "run_live", { type: "reviewer-result", head: "sha4", payload: { kind: "code-review", result: OUTPUT, diffHash: diffHash(DIFF) } }); diff --git a/apps/hub/src/github/manager-runs.postgres.test.ts b/apps/hub/src/github/manager-runs.postgres.test.ts index ce4783f9..5dbdf174 100644 --- a/apps/hub/src/github/manager-runs.postgres.test.ts +++ b/apps/hub/src/github/manager-runs.postgres.test.ts @@ -50,13 +50,14 @@ 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) { + const report = { head: "h1", request: { repo: REPO, number: pr.number, labels: [], owned: [], comment: "" } }; + const result = await done(db, TENANT, pr, runId, [], { report, record: 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"); } @@ -90,7 +91,7 @@ test("PostgreSQL: the reconciler does not mail a head again while a live manager 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 }); + await done(db, TENANT, held, "run_held", [], { report: { head: "h1", waiting: true } }); const firstSeenAt = new Date(now.getTime() - MINUTE).toISOString(); const fresh: PrTriageRow = { number: 8, headSha: "h1", status: "new", attempts: 0, firstSeenAt, updatedAt: firstSeenAt }; diff --git a/apps/hub/src/manager-intake.postgres.test.ts b/apps/hub/src/manager-intake.postgres.test.ts index c75b8917..9f11e108 100644 --- a/apps/hub/src/manager-intake.postgres.test.ts +++ b/apps/hub/src/manager-intake.postgres.test.ts @@ -7,7 +7,8 @@ import { APPROVE_DO_KIND, emptyPack, repoPolicy } from "@corbits/triage-contract import type { PullRequestToReview } from "./pull-request-reader.js"; import { mailPayload, waitingPayload } from "./github/bridge.js"; import { createPauseGate } from "./github/pr-workflow.js"; -import { createManagerIntake } from "./manager-intake.js"; +import { createManagerIntake, createManagerWake, createOverdueWake, createSupersede } from "./manager-intake.js"; +import { recordEvent } from "./manager-state.js"; import { migrateTriageTables } from "./triage-db.js"; const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" }; @@ -50,6 +51,16 @@ const OPEN = [{ number: 7, headSha: "h1", draft: false }, { number: 9, headSha: function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = passed, open = OPEN) { const mailed: { address: string; payload: unknown }[] = []; const logged: Record[] = []; + const cancelled: { runId: string; reason: string }[] = []; + const refusing = new Set(); + const { supersede, supersedeAll } = createSupersede({ + db, + async cancel(_tenantId, runId, reason) { + if (refusing.has(runId)) throw new Error("cancel failed"); + cancelled.push({ runId, reason }); + }, + log: (entry) => logged.push(entry), + }); const toListener = createManagerIntake({ db, listener: async () => listener, @@ -59,10 +70,11 @@ function stack(db: Db, listener: string | undefined, pull = PULL, gate: Gate = p async deliver(_tenantId, address, payload) { mailed.push({ address, payload }); }, + supersede, gate, log: (entry) => logged.push(entry), }); - return { toListener, mailed, logged }; + return { toListener, mailed, logged, cancelled, refusing, supersedeAll }; } async function events(db: Db): Promise<{ key: string; state: { head: string; payload: Record } }[]> { @@ -114,10 +126,10 @@ test("PostgreSQL: each pull request a mail names is recorded as its own event, a }); }, 60_000); -async function claimManager(db: Db, runId: string, number: number, status: "running" | "deployed", createdAt?: Date): Promise { +async function claimManager(db: Db, runId: string, number: number, status: "running" | "deployed" | "completed", createdAt?: Date, reviewers: object[] = []): Promise { await db.insert(schema.workflowDefinition).values({ id: "wfd_intake", tenantId: TENANT, name: "triage-manager-github" }).onConflictDoNothing(); await db.insert(schema.workflowRun).values({ id: runId, anchorRunId: runId, definitionId: "wfd_intake", tenantId: TENANT, status, ...(createdAt && { createdAt }) }); - const claim = { status: "live", claimId: `claim-${runId}`, claimedAt: new Date().toISOString(), runId, reviewers: [], history: [], starts: 1 }; + const claim = { status: "live", claimId: `claim-${runId}`, claimedAt: new Date().toISOString(), runId, reviewers, history: [], starts: 1 }; await db.execute(sql`INSERT INTO corbits_triage.manager_state (tenant_id, key, kind, state) VALUES (${TENANT}, ${`mgr:github:${RECORD.name}#${number}`}, 'claim', ${JSON.stringify(claim)}::jsonb)`); } @@ -135,6 +147,137 @@ test("PostgreSQL: a pull request whose manager is up is mailed that manager, and }); }, 60_000); +test("PostgreSQL: a reviewer's result wakes the pull request's live manager, or has the listener deploy one once its claim ended, and none without a listener", async () => { + await withDatabase(async function wake(db) { + await claimManager(db, "run_mgr", 7, "running"); + const mailed: { address: string; payload: unknown }[] = []; + function waker(listener: string | undefined) { + return createManagerWake({ + db, + listener: async () => listener, + async deliver(_tenantId, address, payload) { + mailed.push({ address, payload }); + }, + }); + } + const pr = { source: "github" as const, repo: RECORD.name, number: 7 }; + expect(await waker(LISTENER)(TENANT, pr, "h1")).toBe(true); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = (state - 'runId') || '{"status":"ended"}' WHERE kind = 'claim'`); + expect(await waker(LISTENER)(TENANT, pr, "h1")).toBe(true); + expect(await waker(undefined)(TENANT, pr, "h1")).toBe(false); + expect(mailed).toEqual([ + { address: "run_mgr@intake.test", payload: { kind: "manage", tenantId: TENANT, repo: RECORD.name, repoSource: "github", prNumber: 7 } }, + { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } }, + ]); + }); +}, 60_000); + +test("PostgreSQL: a push stops the judge and the code review still out on the older head, once each, and one whose stop failed stays on the claim until a reconcile pass stops it", async () => { + await withDatabase(async function superseded(db) { + const deadline = new Date(Date.now() + 60 * 60_000).toISOString(); + const judge = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_old", deadline }; + const review = { ...judge, kind: "code-review", runId: "run_review" }; + const current = { ...judge, head: "h2", runId: "run_current" }; + await claimManager(db, "run_old", 7, "completed", undefined, [judge, review, current]); + const { toListener, cancelled, logged, refusing, supersedeAll } = stack(db, LISTENER); + async function held(): Promise { + const [claim] = await db.execute<{ state: { reviewers: unknown[] } }>(sql`SELECT state FROM corbits_triage.manager_state WHERE kind = 'claim'`); + return claim!.state.reviewers; + } + const where = { tenantId: TENANT, source: "github", repo: RECORD.name, number: 7 }; + refusing.add("run_review"); + expect(await toListener(TENANT, mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h2", deliveryId: "push-1" }))).toBe(true); + expect(cancelled).toEqual([{ runId: "run_judge", reason: "superseded by a newer head" }]); + expect(await held()).toEqual([review, current]); + expect(logged.filter((entry) => entry["msg"] === "reviewer_supersede_failed")).toEqual([{ level: "error", msg: "reviewer_supersede_failed", ...where, kind: "code-review", head: "h1", runId: "run_review", error: "Error: cancel failed" }]); + + refusing.clear(); + await supersedeAll(TENANT); + expect(await toListener(TENANT, mailPayload(RECORD, POLICY, STORED, { prNumber: 7, headSha: "h2", deliveryId: "push-2" }))).toBe(true); + expect(cancelled).toEqual([{ runId: "run_judge", reason: "superseded by a newer head" }, { runId: "run_review", reason: "superseded by a newer head" }]); + expect(logged.filter((entry) => entry["msg"] === "reviewer_superseded")).toEqual([ + { level: "info", msg: "reviewer_superseded", ...where, kind: "judge", head: "h1", runId: "run_judge" }, + { level: "info", msg: "reviewer_superseded", ...where, kind: "code-review", head: "h1", runId: "run_review" }, + ]); + expect(await held()).toEqual([current]); + }); +}, 60_000); + +test("PostgreSQL: the reconcile pass wakes a manager for a pull request whose reviewer of its newest head is past its deadline with no result and no manager up", async () => { + await withDatabase(async function overdue(db) { + const past = new Date(Date.now() - 60_000).toISOString(); + const late = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_gone", deadline: past }; + await claimManager(db, "run_gone", 7, "completed", undefined, [late]); + await claimManager(db, "run_up", 8, "running", undefined, [{ ...late, runId: "run_judge_8" }]); + await claimManager(db, "run_reported", 9, "completed", undefined, [{ ...late, runId: "run_judge_9" }]); + for (const number of [7, 8, 9]) await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number }, `d${number}`, { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } }); + await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number: 9 }, "run_judge_9", { type: "reviewer-result", head: "h1", payload: { kind: "judge", failed: "the judge did not finish" } }); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || '{"consumedBy":"pass_9"}' WHERE kind = 'event' AND key LIKE '%:run_judge_9'`); + const mailed: { address: string; payload: unknown }[] = []; + const logged: Record[] = []; + const wake = createManagerWake({ + db, + listener: async () => LISTENER, + async deliver(_tenantId, address, payload) { + mailed.push({ address, payload }); + }, + }); + const wakeOverdue = createOverdueWake({ db, wake, log: (entry) => logged.push(entry) }); + await wakeOverdue(TENANT); + await wakeOverdue(TENANT); + const woken = { address: LISTENER, payload: { kind: "pr", repo: RECORD.name, tenantId: TENANT, items: [{ prNumber: 7, headSha: "h1" }] } }; + expect(mailed).toEqual([woken]); + expect(logged).toEqual([{ level: "info", msg: "manager_woken_overdue", tenantId: TENANT, source: "github", repo: RECORD.name, number: 7, head: "h1" }]); + + // A manager the wake started took the claim and went away before done; past the start grace the next pass wakes again. + const earlier = new Date(Date.now() - 3 * 60_000); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = state || ${JSON.stringify({ overdueWokenAt: earlier.toISOString(), claimedAt: new Date(earlier.getTime() + 1_000).toISOString() })}::jsonb WHERE key = ${`mgr:github:${RECORD.name}#7`}`); + await wakeOverdue(TENANT); + expect(mailed).toEqual([woken, woken]); + }); +}, 60_000); + +async function overdueWokenAt(db: Db, number: number): Promise { + const [claim] = await db.execute<{ state: Record }>(sql`SELECT state FROM corbits_triage.manager_state WHERE key = ${`mgr:github:${RECORD.name}#${number}`}`); + return claim!.state["overdueWokenAt"]; +} + +test("PostgreSQL: an overdue wake is recorded only once it succeeded, and one pull request whose wake fails is logged while the next is still woken", async () => { + await withDatabase(async function partly(db) { + const late = { kind: "judge", head: "h1", runId: "run_judge_7", deployedBy: "run_gone_7", deadline: new Date(Date.now() - 60_000).toISOString() }; + await claimManager(db, "run_gone_7", 7, "completed", undefined, [late]); + await claimManager(db, "run_gone_8", 8, "completed", undefined, [{ ...late, runId: "run_judge_8" }]); + for (const number of [7, 8]) await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number }, `d${number}`, { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } }); + const woken: number[] = []; + const logged: Record[] = []; + let failing = true; + async function wake(_tenantId: string, pr: { number: number }): Promise { + if (pr.number === 7 && failing) throw new Error("delivery failed"); + woken.push(pr.number); + return true; + } + const wakeOverdue = createOverdueWake({ db, wake, log: (entry) => logged.push(entry) }); + await wakeOverdue(TENANT); + expect(woken).toEqual([8]); + expect(logged).toEqual([ + { level: "error", msg: "manager_overdue_wake_failed", tenantId: TENANT, source: "github", repo: RECORD.name, number: 7, head: "h1", error: "Error: delivery failed" }, + { level: "info", msg: "manager_woken_overdue", tenantId: TENANT, source: "github", repo: RECORD.name, number: 8, head: "h1" }, + ]); + expect(await overdueWokenAt(db, 7)).toBeUndefined(); + expect(await overdueWokenAt(db, 8)).toBeString(); + + failing = false; + await wakeOverdue(TENANT); + expect(woken).toEqual([8, 7]); + + const unrouted = createOverdueWake({ db, wake: async () => false, log: (entry) => logged.push(entry) }); + await claimManager(db, "run_gone_9", 9, "completed", undefined, [{ ...late, runId: "run_judge_9" }]); + await recordEvent(db, TENANT, { source: "github", repo: RECORD.name, number: 9 }, "d9", { type: "webhook", head: "h1", payload: { kind: "pr", headSha: "h1" } }); + await unrouted(TENANT); + expect(await overdueWokenAt(db, 9)).toBeUndefined(); + }); +}, 60_000); + test("PostgreSQL: a manager deployed without a sidecar past the startup window is not mailed, and the listener is", async () => { await withDatabase(async function stuck(db) { await claimManager(db, "run_stuck", 7, "deployed", new Date(Date.now() - 3 * 60_000)); diff --git a/apps/hub/src/manager-intake.ts b/apps/hub/src/manager-intake.ts index 5f3acfb6..b05aabed 100644 --- a/apps/hub/src/manager-intake.ts +++ b/apps/hub/src/manager-intake.ts @@ -1,7 +1,7 @@ // The hub's side of a pull request's manager. In a workspace with a host's listener live, each pull request, waiting, -// held-Do approval and backlog mail is recorded as one event per pull request it names. A pull request whose manager -// is up is mailed that manager. One with none is sent to the listener, which deploys a manager and wakes it. A manager -// reads the newest recorded mail back through take. +// held-Do approval and backlog mail is recorded as one event per pull request it names, as a reviewer's result is by +// its route. A pull request whose manager is up is mailed that manager. One with none is sent to the listener, which +// deploys a manager and wakes it. A manager reads the newest recorded mail back through take. import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { schema, type DB } from "@intx/db"; @@ -9,7 +9,7 @@ import { asRecord } from "@corbits/common"; import { APPROVE_DO_KIND, repoSourceOf, WAITING_KIND, type RepoSource } from "@corbits/triage-contracts"; import { reviewRequestOf, type PullRequestToReview } from "./pull-request-reader.js"; import { repoRecords, sourceOf, triageNs, type RepoRecord } from "./github/tenant-config.js"; -import { liveManager, recordEvent } from "./manager-state.js"; +import { dropReviewer, liveManager, overduePulls, recordEvent, recordOverdueWake, reviewedPulls, superseded, type PullRequestRef } from "./manager-state.js"; export type ManagerIntakeDeps = { db: DB["db"]; @@ -22,6 +22,8 @@ export type ManagerIntakeDeps = { /** The mail as a GitHub workspace's runs read it, naming the App's bot account. */ withAppLogin: (tenantId: string, payload: Record) => Promise; deliver: (tenantId: string, address: string, payload: unknown) => Promise; + /** Stops the pull request's reviewers on a head older than its newest, as `createSupersede` does. */ + supersede: (tenantId: string, pr: PullRequestRef) => Promise; /** The pull request's mail as its manager takes it, or undefined when none is to be taken, as for a paused pull request. */ gate: (tenantId: string, record: RepoRecord, number: number, request: Record) => Promise | undefined>; log: (entry: Record) => void; @@ -51,6 +53,110 @@ function itemsOf(request: Record, items: readonly unknown[]): N }); } +/** The tenant domain on a listener address, which a manager's address shares. */ +function domainOf(address: string): string { + const at = address.lastIndexOf("@"); + if (at <= 0 || at === address.length - 1) throw new Error(`listener address ${address} has no domain`); + return address.slice(at + 1); +} + +type Delivery = Pick; + +/** Mails the pull request's manager while one is up; false when none is, for the listener to deploy one. */ +async function mailedLive(deps: Delivery, tenantId: string, domain: string, pr: PullRequestRef): Promise { + const runId = await liveManager(deps.db, tenantId, pr); + if (!runId) return false; + await deps.deliver(tenantId, `${runId}@${domain}`, { kind: "manage", tenantId, repo: pr.repo, repoSource: pr.source, prNumber: pr.number }); + return true; +} + +/** Mails the listener the heads of `repo` it deploys a manager for. */ +async function deployFor(deps: Delivery, tenantId: string, address: string, source: RepoSource, repo: string, heads: readonly { number: number; head: string }[]): Promise { + await deps.deliver(tenantId, address, { kind: "pr", repo, ...(source !== "github" && { repoSource: source }), tenantId, items: heads.map((pull) => ({ prNumber: pull.number, headSha: pull.head })) }); +} + +/** + * Wakes the manager of a pull request whose reviewer result was recorded, as a host event does: its live manager is + * mailed, or the listener deploys one. False when the workspace has no listener for the host. + */ +export function createManagerWake(deps: Delivery & Pick) { + return async function wake(tenantId: string, pr: PullRequestRef, head: string): Promise { + const address = await deps.listener(tenantId, pr.source); + if (!address) return false; + if (!(await mailedLive(deps, tenantId, domainOf(address), pr))) await deployFor(deps, tenantId, address, pr.source, pr.repo, [{ number: pr.number, head }]); + return true; + }; +} + +type Wake = ReturnType; + +/** + * Wakes a manager for each of the tenant's pull requests whose manager has gone while a reviewer result waits for it, + * or a reviewer of its newest head is past its deadline with no result, so that manager posts the verdict. A wake is + * recorded once it succeeded; a pull request whose wake fails is logged, the rest are still woken, and the next + * reconcile pass tries it again. + */ +export function createOverdueWake(deps: Pick & { wake: Wake }) { + async function wakeOne(tenantId: string, pr: PullRequestRef, head: string): Promise { + const entry = { tenantId, source: pr.source, repo: pr.repo, number: pr.number, head }; + try { + if (!(await deps.wake(tenantId, pr, head))) return; + await recordOverdueWake(deps.db, tenantId, pr); + deps.log({ level: "info", msg: "manager_woken_overdue", ...entry }); + } catch (err) { + deps.log({ level: "error", msg: "manager_overdue_wake_failed", ...entry, error: String(err) }); + } + } + + return async function wakeOverdue(tenantId: string): Promise { + try { + for (const { pr, head } of await overduePulls(deps.db, tenantId)) await wakeOne(tenantId, pr, head); + } catch (err) { + deps.log({ level: "error", msg: "manager_overdue_wake_failed", tenantId, error: String(err) }); + } + }; +} + +type SupersedeDeps = Pick & { + /** Stops a run through the stock run lifecycle with the hub's own authority; stopping one already ending is a no-op. */ + cancel: (tenantId: string, runId: string, reason: string) => Promise; +}; + +/** + * Stops each reviewer a pull request's claim still waits on for a head older than its newest, then takes it off the + * claim. A reviewer whose stop fails is logged and stays on the claim, so the next reconcile pass, which runs this over + * every pull request with a reviewer out (`supersedeAll`), stops it again. + */ +export function createSupersede(deps: SupersedeDeps) { + async function supersede(tenantId: string, pr: PullRequestRef): Promise { + const base = { tenantId, source: pr.source, repo: pr.repo, number: pr.number }; + try { + for (const reviewer of await superseded(deps.db, tenantId, pr)) { + const entry = { ...base, kind: reviewer.kind, head: reviewer.head, runId: reviewer.runId }; + try { + await deps.cancel(tenantId, reviewer.runId, "superseded by a newer head"); + await dropReviewer(deps.db, tenantId, pr, reviewer); + deps.log({ level: "info", msg: "reviewer_superseded", ...entry }); + } catch (err) { + deps.log({ level: "error", msg: "reviewer_supersede_failed", ...entry, error: String(err) }); + } + } + } catch (err) { + deps.log({ level: "error", msg: "reviewer_supersede_failed", ...base, error: String(err) }); + } + } + + async function supersedeAll(tenantId: string): Promise { + try { + for (const pr of await reviewedPulls(deps.db, tenantId)) await supersede(tenantId, pr); + } catch (err) { + deps.log({ level: "error", msg: "reviewer_supersede_failed", tenantId, error: String(err) }); + } + } + + return { supersede, supersedeAll }; +} + export function createManagerIntake(deps: ManagerIntakeDeps) { async function recordOf(tenantId: string, repo: string, source: RepoSource): Promise { const tenant = await deps.db.query.tenant.findFirst({ where: eq(schema.tenant.id, tenantId), columns: { config: true } }); @@ -85,13 +191,6 @@ export function createManagerIntake(deps: ManagerIntakeDeps) { return request; } - /** The tenant domain on a listener address, which a manager's address shares. */ - function domainOf(address: string): string { - const at = address.lastIndexOf("@"); - if (at <= 0 || at === address.length - 1) throw new Error(`listener address ${address} has no domain`); - return address.slice(at + 1); - } - /** * Records each pull request a mail names. One whose manager is up is mailed that run; the rest go to the listener, * which deploys them, including those recorded before a later one failed. True with nothing recorded when the mail @@ -115,18 +214,13 @@ export function createManagerIntake(deps: ManagerIntakeDeps) { if (!request) continue; const pull = { ...asked, request }; const recorded = await deps.withAppLogin(tenantId, await withReviewRequest(tenantId, record, pull)); - await recordEvent(deps.db, tenantId, { source, repo, number: pull.number }, pull.id, { type: "webhook", head: pull.head, payload: recorded as Record }); - const runId = await liveManager(deps.db, tenantId, { source, repo, number: pull.number }); - if (runId) { - await deps.deliver(tenantId, `${runId}@${domain}`, { kind: "manage", tenantId, repo, repoSource: source, prNumber: pull.number }); - } else { - start.push(pull); - } + const pr = { source, repo, number: pull.number }; + await recordEvent(deps.db, tenantId, pr, pull.id, { type: "webhook", head: pull.head, payload: recorded as Record }); + await deps.supersede(tenantId, pr); + if (!(await mailedLive(deps, tenantId, domain, pr))) start.push(pull); } } finally { - if (start.length > 0) { - await deps.deliver(tenantId, address, { kind: "pr", repo, ...(source !== "github" && { repoSource: source }), tenantId, items: start.map((pull) => ({ prNumber: pull.number, headSha: pull.head })) }); - } + if (start.length > 0) await deployFor(deps, tenantId, address, source, repo, start); } return true; }; diff --git a/apps/hub/src/manager-routes.postgres.test.ts b/apps/hub/src/manager-routes.postgres.test.ts index d0f45b8c..21f988ba 100644 --- a/apps/hub/src/manager-routes.postgres.test.ts +++ b/apps/hub/src/manager-routes.postgres.test.ts @@ -10,6 +10,7 @@ import { createApp, createRequireGrant, HUB_CONDITION_REGISTRY, type GetSession import { createEventCollectorRegistry, createSidecarRouter, type SessionService } from "@intx/hub-sessions"; import { matchesRunTokenRoute } from "../../../vendor/interchange/packages/hub-api/src/middleware/run-token-auth.js"; import { MANAGER_RESULT_ACTION, MANAGER_RESULT_RESOURCE, MANAGER_STATE_ACTION, MANAGER_STATE_RESOURCE } from "@corbits/triage-contracts"; +import { createOverdueWake, createSupersede } from "./manager-intake.js"; import { createManagerRoutes, MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; import { MAX_BODY_BYTES } from "./read-json.js"; import { passRequest, recordEvent, type PullRequestRef } from "./manager-state.js"; @@ -65,13 +66,27 @@ function buildStock(db: Db) { function ignore(): void {} -function routeDeps(db: Db) { - return { db, requireGrant: createRequireGrant({ grantStore: createGrantStore(db), conditionRegistry: HUB_CONDITION_REGISTRY }), log: ignore }; +type Woken = { tenantId: string; pr: PullRequestRef; head: string }; + +/** The hub's wake, recording each pull request it wakes; `listening` false is a workspace with no listener for the host. */ +function waker(woken: Woken[] = [], listening = true) { + return async function wake(tenantId: string, pr: PullRequestRef, head: string): Promise { + if (listening) woken.push({ tenantId, pr, head }); + return listening; + }; } -function buildHub(db: Db) { +async function unsuperseded(): Promise {} + +type RouteOptions = { log?: (entry: Record) => void; wake?: ReturnType; supersede?: (tenantId: string, pr: PullRequestRef) => Promise }; + +function routeDeps(db: Db, { log = ignore, wake = waker(), supersede = unsuperseded }: RouteOptions = {}) { + return { db, requireGrant: createRequireGrant({ grantStore: createGrantStore(db), conditionRegistry: HUB_CONDITION_REGISTRY }), log, wake, supersede }; +} + +function buildHub(db: Db, deps: RouteOptions = {}) { const app = buildStock(db); - mountTenantRoutes(app, { ...routeDeps(db), now: () => new Date() }); + mountTenantRoutes(app, { ...routeDeps(db, deps), now: () => new Date() }); return app; } @@ -165,6 +180,99 @@ test("PostgreSQL: the manager's run token takes, finishes and a reviewer's recor }); }, 60_000); +test("PostgreSQL: a reviewer's result wakes the pull request's manager as a host event does, and one for a head the pull request moved past is dropped and logged", async () => { + await withDatabase(async function woken(db) { + const woken: Woken[] = []; + const logged: Record[] = []; + function log(entry: Record): void { + logged.push(entry); + } + const app = buildHub(db, { log, wake: waker(woken) }); + const reviewer = bearer(await runWithToken(db, "run_reviewer", [RESULT])); + await webhook(db, GITHUB); + const result = { ...GITHUB, kind: "judge", head: "h1", failed: "the judge did not finish" }; + expect(await (await post(app, "reviewer-result", result, reviewer)).json()).toEqual({ recorded: true }); + expect(woken).toEqual([{ tenantId: TENANT, pr: GITHUB, head: "h1" }]); + + await recordEvent(db, TENANT, GITHUB, "pushed", { type: "webhook", head: "h2", payload: { kind: "pr", headSha: "h2" } }); + const late = bearer(await runWithToken(db, "run_late", [RESULT])); + expect(await (await post(app, "reviewer-result", result, late)).json()).toEqual({ recorded: false }); + expect(woken).toHaveLength(1); + expect(logged).toEqual([{ level: "info", msg: "reviewer_result_stale", tenantId: TENANT, ...GITHUB, kind: "judge", head: "h1", current: "h2", runId: "run_late" }]); + const [stored] = await db.execute<{ events: number }>(sql`SELECT count(*)::int AS events FROM corbits_triage.manager_state WHERE key LIKE '%:run_late'`); + expect(stored!.events).toBe(0); + + const unrouted = buildHub(db, { log, wake: waker([], false) }); + const quiet = bearer(await runWithToken(db, "run_quiet", [RESULT])); + expect(await (await post(unrouted, "reviewer-result", { ...result, head: "h2" }, quiet)).json()).toEqual({ recorded: true }); + expect(logged.at(-1)).toMatchObject({ level: "warn", msg: "reviewer_result_unrouted", runId: "run_quiet" }); + }); +}, 60_000); + +test("PostgreSQL: a result whose wake fails stays recorded with nothing else, the route answers 502, and a reconcile pass wakes a manager for it once the start grace has passed since it was recorded", async () => { + await withDatabase(async function unwoken(db) { + const logged: Record[] = []; + function log(entry: Record): void { + logged.push(entry); + } + async function failingWake(): Promise { + throw new Error("delivery failed"); + } + const app = buildHub(db, { log, wake: failingWake }); + const manager = bearer(await runWithToken(db, "run_manager", [STATE])); + await webhook(db, GITHUB); + await post(app, "take", GITHUB, manager); + expect(await (await post(app, "done", { ...GITHUB, reviewers: [] }, manager)).json()).toEqual({ outcome: "ended" }); + await db.execute(sql`UPDATE workflow_run SET status = 'completed' WHERE id = 'run_manager'`); + const before = await db.execute(sql`SELECT key, state FROM corbits_triage.manager_state WHERE kind = 'claim'`); + + const reviewer = bearer(await runWithToken(db, "run_reviewer", [RESULT])); + const failed = await post(app, "reviewer-result", { ...GITHUB, kind: "judge", head: "h1", failed: "the judge did not finish" }, reviewer); + expect(failed.status).toBe(502); + expect(await errorCode(failed)).toBe("wake_failed"); + expect(logged).toEqual([{ level: "error", msg: "reviewer_result_wake_failed", tenantId: TENANT, ...GITHUB, kind: "judge", head: "h1", runId: "run_reviewer", error: "Error: delivery failed" }]); + const [stored] = await db.execute<{ events: number }>(sql`SELECT count(*)::int AS events FROM corbits_triage.manager_state WHERE key LIKE '%:run_reviewer' AND state->>'consumedBy' IS NULL`); + expect(stored!.events).toBe(1); + expect(await db.execute(sql`SELECT key, state FROM corbits_triage.manager_state WHERE kind = 'claim'`)).toEqual(before); + + const woken: Woken[] = []; + const wakeOverdue = createOverdueWake({ db, wake: waker(woken), log }); + await wakeOverdue(TENANT); + expect(woken).toEqual([]); + const earlier = new Date(Date.now() - 2 * 60_000 - 1_000).toISOString(); + await db.execute(sql`UPDATE corbits_triage.manager_state SET state = jsonb_set(state, '{receivedAt}', to_jsonb(${earlier}::text)) WHERE key LIKE '%:run_reviewer'`); + await wakeOverdue(TENANT); + expect(woken).toEqual([{ tenantId: TENANT, pr: GITHUB, head: "h1" }]); + // The manager that wake deployed has not taken the claim yet, so the next pass does not deploy another. + await wakeOverdue(TENANT); + expect(woken).toHaveLength(1); + }); +}, 60_000); + +test("PostgreSQL: done stops a reviewer the pass deployed on a head a push moved past, and takes it off the claim", async () => { + await withDatabase(async function supersededMidPass(db) { + const cancelled: string[] = []; + const { supersede } = createSupersede({ + db, + async cancel(_tenantId, runId) { + cancelled.push(runId); + }, + log: ignore, + }); + const app = buildHub(db, { supersede }); + const manager = bearer(await runWithToken(db, "run_manager", [STATE])); + await webhook(db, GITHUB); + await post(app, "take", GITHUB, manager); + await recordEvent(db, TENANT, GITHUB, "pushed", { type: "webhook", head: "h2", payload: { kind: "pr", headSha: "h2" } }); + const deadline = new Date(Date.now() + 60 * 60_000).toISOString(); + const deployed = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_manager", deadline }; + expect(await (await post(app, "done", { ...GITHUB, reviewers: [deployed], wait: true }, manager)).json()).toEqual({ outcome: "more", wait: false }); + expect(cancelled).toEqual(["run_judge"]); + const [claim] = await db.execute<{ state: { reviewers: unknown[] } }>(sql`SELECT state FROM corbits_triage.manager_state WHERE kind = 'claim'`); + expect(claim!.state.reviewers).toEqual([]); + }); +}, 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({ diff --git a/apps/hub/src/manager-routes.ts b/apps/hub/src/manager-routes.ts index 752dd132..e6b7ea17 100644 --- a/apps/hub/src/manager-routes.ts +++ b/apps/hub/src/manager-routes.ts @@ -18,7 +18,7 @@ import { body, Failure, json } from "./route-schema.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"; +import { done, EventState, PassState, recordResult, ReviewerState, take, type Closing, type PassReport, type PullRequestRef } from "./manager-state.js"; const TAKE_PATH = "/manager/take"; const DONE_PATH = "/manager/done"; @@ -43,8 +43,8 @@ const Report = type({ "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[]" } }); +/** `approvals` is the pass's approved Dos as its approvals step ran them; `wait` that the pass posted to the host or mailed a reviewer. */ +const DoneBody = PullRequest.and({ reviewers: ReviewerState.array(), "wait?": "boolean", "report?": Report, "approvals?": { results: "Record[]" } }); const ReviewerResultBody = PullRequest.and(StageResult); const Latest = { "request?": "Record", reviewers: ReviewerState.array() } as const; @@ -69,7 +69,15 @@ function refused(): Response { type RunEnv = TenantEnv & { Variables: TenantEnv["Variables"] & { runId: string } }; -export type ManagerRoutesDeps = { db: DB["db"]; requireGrant: RequireGrant; log: (entry: Record) => void }; +export type ManagerRoutesDeps = { + db: DB["db"]; + requireGrant: RequireGrant; + log: (entry: Record) => void; + /** Wakes the pull request's manager for a reviewer result, as a host event does; false when the workspace has no listener for the host. */ + wake: (tenantId: string, pr: PullRequestRef, head: string) => Promise; + /** Stops the pull request's reviewers on a head older than its newest and takes them off the claim; never throws. */ + supersede: (tenantId: string, pr: PullRequestRef) => Promise; +}; type Reported = typeof Report.infer; @@ -86,7 +94,7 @@ function passReport(report: Reported | undefined, posted: Record(); // Hono throws on a body that is not JSON; it keeps the hub's error envelope. app.onError(function onError(err) { @@ -130,7 +138,7 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps DONE_PATH, describeRoute({ 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}`, + 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. A reviewer the claim recorded that ran past its deadline with no result and that \`reviewers\` no longer waits for is logged. Each reviewer the claim then holds on a head older than the newest a pull request mail named, as one the pass deployed while a push arrived, is stopped and taken off the claim. \`more\`: an event is pending, so the manager takes again at once, or the pass asked to \`wait\` because it posted to the host or mailed a reviewer, so it takes again after waiting. \`ended\`: otherwise; the claim ended and the run should end, and a later event or reviewer result starts the next manager. ${NOT_MANAGER}`, tags: TAGS, responses: { 200: json("Whether the manager is done", Done), ...refusals, 409: json(NOT_MANAGER, Failure) }, }), @@ -139,7 +147,7 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps granted, body(DoneBody), async function finishPass(c) { - const { reviewers, report, approvals, ...pr } = c.req.valid("json"); + const { reviewers, wait, report, approvals, ...pr } = c.req.valid("json"); const tenantId = c.get("tenant").id; const runId = c.get("runId"); const posted = postedOf(report); @@ -154,8 +162,9 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps 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); + const result = await done(db, tenantId, pr, runId, reviewers, { ...(wait !== undefined && { wait }), ...(kept && { report: kept }), record: recordPass, log }); if (result.outcome === "refused") return refused(); + await supersede(tenantId, pr); return c.json(result); }, ); @@ -164,9 +173,14 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps REVIEWER_RESULT_PATH, describeRoute({ summary: "Record a reviewer's result for the pull request's manager", - description: "Run token only, for a reviewer granted `create` on `triage-manager:result`, which reaches no other manager route. The body is a `StageResult`. The event is recorded under the calling run's id, so `recorded` is false when that run reported before. The reviewer then mails the manager that deployed it, which collects the result on its next take.", + description: "Run token only, for a reviewer granted `create` on `triage-manager:result`, which reaches no other manager route. The body is a `StageResult`. A result for a head the pull request has moved past is dropped and logged, and `recorded` is false. Otherwise the event is recorded under the calling run's id, so `recorded` is false when that run reported before, and the pull request's manager is woken as a host event wakes it: its live manager is mailed, or the host's listener deploys one, whose take collects the result. When that wake fails the result stays recorded, the route answers `wake_failed`, and the hub's next reconcile pass wakes a manager for it.", tags: TAGS, - responses: { 200: json("Whether the result was recorded now", Recorded), ...refusals, 403: json("Not a workflow run, or the run lacks `create` on `triage-manager:result`", Failure) }, + responses: { + 200: json("Whether the result was recorded now", Recorded), + ...refusals, + 403: json("Not a workflow run, or the run lacks `create` on `triage-manager:result`", Failure), + 502: json("`wake_failed`: the result is recorded, but the pull request's manager could not be woken", Failure), + }, }), limited, asRun, @@ -174,7 +188,21 @@ export function createManagerRoutes({ db, requireGrant, log }: ManagerRoutesDeps body(ReviewerResultBody), async function reviewerResult(c) { const { source, repo, number, head, ...reported } = c.req.valid("json"); - const recorded = await recordEvent(db, c.get("tenant").id, { source, repo, number }, c.get("runId"), { type: "reviewer-result", head, payload: reported }); + const tenantId = c.get("tenant").id; + const runId = c.get("runId"); + const pr = { source, repo, number }; + const { recorded, current } = await recordResult(db, tenantId, pr, runId, head, reported); + const entry = { tenantId, source, repo, number, kind: reported.kind, head, runId }; + if (!recorded && current !== undefined && current !== head) { + log({ level: "info", msg: "reviewer_result_stale", ...entry, current }); + return c.json({ recorded: false }); + } + try { + if (!(await wake(tenantId, pr, head))) log({ level: "warn", msg: "reviewer_result_unrouted", ...entry, reason: "the workspace has no listener for the host" }); + } catch (err) { + log({ level: "error", msg: "reviewer_result_wake_failed", ...entry, error: String(err) }); + return failure(502, "wake_failed", "The result is recorded, but the pull request's manager could not be woken."); + } return c.json({ recorded }); }, ); diff --git a/apps/hub/src/manager-state.postgres.test.ts b/apps/hub/src/manager-state.postgres.test.ts index 882681bc..08697ba3 100644 --- a/apps/hub/src/manager-state.postgres.test.ts +++ b/apps/hub/src/manager-state.postgres.test.ts @@ -4,7 +4,7 @@ import { randomBytes, randomUUID } from "node:crypto"; import { sql } from "drizzle-orm"; import { createDB, runMigrations, schema } from "@intx/db"; import type { Tx } from "./github/admin-github-grant.js"; -import { done, liveListener, recordEvent, take, type PullRequestRef } from "./manager-state.js"; +import { done, liveListener, recordEvent, recordResult, take, type PullRequestRef } from "./manager-state.js"; import { migrateTriageTables } from "./triage-db.js"; const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" }; @@ -238,13 +238,66 @@ test("PostgreSQL: a claim taken again after it ended keeps its history, starts a }); }, 120_000); +test("PostgreSQL: a manager that posted or mailed takes once more, then the claim ends with its reviewer still out, and the reviewer's result is taken by the next manager", async () => { + await withDatabase(async function wakeByResult(db) { + await recordEvent(db, TENANT, GITHUB, "d1", webhook("h1")); + await startManager(db, GITHUB, "run_a"); + expect(await take(db, TENANT, GITHUB, "run_a")).toMatchObject({ outcome: "pass", pass: { head: "h1" } }); + const reviewer = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_a", deadline: new Date(Date.now() + 60_000).toISOString() }; + expect(await done(db, TENANT, GITHUB, "run_a", [reviewer], { wait: true })).toEqual({ outcome: "more", wait: true }); + expect(await take(db, TENANT, GITHUB, "run_a")).toEqual({ outcome: "idle", reviewers: [reviewer] }); + expect(await done(db, TENANT, GITHUB, "run_a", [reviewer])).toEqual({ outcome: "ended" }); + expect(await stateOf(db, CLAIM)).toMatchObject({ status: "ended", reviewers: [reviewer] }); + + await recordEvent(db, TENANT, GITHUB, "run_judge", { type: "reviewer-result", head: "h1", payload: { kind: "judge", failed: "the judge did not finish" } }); + await liveRun(db, "run_b"); + expect(await take(db, TENANT, GITHUB, "run_b")).toMatchObject({ outcome: "pass", pass: { head: "h1", events: ["evt:github:acme/widgets#7:run_judge"] }, reviewers: [reviewer] }); + expect(await take(db, TENANT, GITHUB, "run_a")).toEqual({ outcome: "refused" }); + }); +}, 120_000); + +test("PostgreSQL: a reviewer result is recorded only on the newest head a mail named, judged by the one statement that records it", async () => { + await withDatabase(async function conditional(db) { + const result = { kind: "judge", failed: "the judge did not finish" }; + expect(await recordResult(db, TENANT, GITHUB, "run_first", "h0", result)).toEqual({ recorded: true, current: undefined }); + await recordEvent(db, TENANT, GITHUB, "d1", webhook("h1")); + expect(await recordResult(db, TENANT, GITHUB, "run_stale", "h0", result)).toEqual({ recorded: false, current: "h1" }); + expect(await recordResult(db, TENANT, GITHUB, "run_judge", "h1", result)).toEqual({ recorded: true, current: "h1" }); + expect(await recordResult(db, TENANT, GITHUB, "run_judge", "h1", result)).toEqual({ recorded: false, current: "h1" }); + const rows = await db.execute<{ key: string }>(sql`SELECT key FROM corbits_triage.manager_state WHERE state->>'type' = 'reviewer-result' ORDER BY key`); + expect(rows.map((row) => row.key)).toEqual(["evt:github:acme/widgets#7:run_first", "evt:github:acme/widgets#7:run_judge"]); + }); +}, 120_000); + +test("PostgreSQL: done logs once each reviewer the pass gave up on past its deadline with no report", async () => { + await withDatabase(async function overdue(db) { + await recordEvent(db, TENANT, GITHUB, "d1", webhook("h1")); + await startManager(db, GITHUB, "run_a"); + await take(db, TENANT, GITHUB, "run_a"); + const past = new Date(Date.now() - 1_000).toISOString(); + const silent = { kind: "code-review", head: "h1", runId: "run_silent", deployedBy: "run_a", deadline: past }; + const reported = { ...silent, kind: "judge", runId: "run_reported" }; + const due = { ...silent, runId: "run_due", deadline: new Date(Date.now() + 60_000).toISOString() }; + await done(db, TENANT, GITHUB, "run_a", [silent, reported, due], { wait: true }); + await recordEvent(db, TENANT, GITHUB, "run_reported", { type: "reviewer-result", head: "h1", payload: { kind: "judge", failed: "the judge did not finish" } }); + await take(db, TENANT, GITHUB, "run_a"); + const logged: Record[] = []; + function log(entry: Record): void { + logged.push(entry); + } + await done(db, TENANT, GITHUB, "run_a", [due], { wait: true, log }); + await done(db, TENANT, GITHUB, "run_a", [due], { log }); + expect(logged).toEqual([{ level: "warn", msg: "reviewer_overdue", tenantId: TENANT, ...GITHUB, kind: "code-review", head: "h1", runId: "run_silent", deadline: past }]); + }); +}, 120_000); + test("PostgreSQL: take skips events another transaction holds, keeps the newest head and drops reviewer results for another head", async () => { await withDatabase(async function skips(db) { await recordEvent(db, TENANT, GITHUB, "d1", webhook("h1")); await startManager(db, GITHUB, "run_a"); expect(await take(db, TENANT, GITHUB, "run_a")).toMatchObject({ outcome: "pass", pass: { head: "h1" } }); const reviewer = { kind: "judge", head: "h1", runId: "run_judge", deployedBy: "run_a", deadline: new Date(Date.now() + 60_000).toISOString() }; - expect(await done(db, TENANT, GITHUB, "run_a", [reviewer])).toEqual({ outcome: "more", wait: true }); + expect(await done(db, TENANT, GITHUB, "run_a", [reviewer], { wait: true })).toEqual({ outcome: "more", wait: true }); await recordEvent(db, TENANT, GITHUB, "d2", webhook("h2")); await recordEvent(db, TENANT, GITHUB, "r1", { type: "reviewer-result", head: "h1", payload: { verdict: "pass" } }); diff --git a/apps/hub/src/manager-state.ts b/apps/hub/src/manager-state.ts index aba5929c..f0c54544 100644 --- a/apps/hub/src/manager-state.ts +++ b/apps/hub/src/manager-state.ts @@ -2,12 +2,12 @@ // pull request when no live run holds it, and done checks for pending events under the claim's lock, so an event is // either taken by the live manager or by the next one. Only the run the claim names takes events or finishes a pass. import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNull, like, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, like, or, sql, type SQL } from "drizzle-orm"; import { jsonb, primaryKey, text } from "drizzle-orm/pg-core"; import { type } from "arktype"; import { schema, type DB } from "@intx/db"; import { asRecord } from "@corbits/common"; -import { APPROVE_DO_KIND, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, REPO_SOURCES, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; import type { Tx } from "./github/admin-github-grant.js"; import { resolveLiveDeployments, type LiveDeployment } from "./github/deployment.js"; import { triageSchema } from "./triage-db.js"; @@ -31,6 +31,7 @@ const ClaimState = type({ reviewers: ReviewerState.array(), history: PassState.and({ runId: "string > 0", doneAt: "string.date.iso", "waiting?": "true", "request?": "Record" }).array(), starts: "number.integer >= 0", + "overdueWokenAt?": "string.date.iso", }); const EventPayload = type("Record").narrow(function plainObject(value) { return !Array.isArray(value); @@ -62,6 +63,11 @@ export type Closing = { head: string; startedAt: string; mail?: Record Promise; type DoneResult = { outcome: "refused" } | { outcome: "ended" } | { outcome: "more"; wait: boolean }; +/** + * `wait` when the pass posted to the host or mailed a reviewer, so the manager waits once for what that brings before + * its claim ends. `report` and `record` record what the pass did; `log` names each reviewer it gave up on. + */ +export type DoneOptions = { wait?: boolean; report?: PassReport; record?: RecordDone; log?: (entry: Record) => void }; function pullKey({ source, repo, number }: PullRequestRef): string { return `${source}:${repo}#${number}`; @@ -146,18 +152,43 @@ function passHeadOf(taken: readonly TakenEvent[], history: Claim["history"]): st return mails.findLast((row) => row.event.payload["kind"] !== APPROVE_DO_KIND)?.event.head ?? history.at(-1)?.head ?? mails.at(-1)?.event.head; } -/** Records the event under its delivery id; false when it was recorded before, as on a redelivery. */ -export async function recordEvent(db: DB["db"], tenantId: string, pr: PullRequestRef, deliveryId: string, event: Pick): Promise { +/** The event's row key and state under its delivery id. */ +function newEvent(pr: PullRequestRef, deliveryId: string, event: Pick): { key: string; state: SQL } { if (deliveryId.length === 0) throw new Error(`an event for ${claimKey(pr)} has no delivery id`); const key = eventKey(pr, deliveryId); const fields = NewEvent(asStored({ type: event.type, head: event.head, payload: event.payload })); if (fields instanceof type.errors) throw new Error(`manager_state ${key} is not an event: ${fields.summary}`); // Stamped by the database to the microsecond as it inserts the row, so events sort in the order they were recorded. - const state = sql`${JSON.stringify(fields)}::jsonb || jsonb_build_object('receivedAt', to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"'))`; + return { key, state: sql`${JSON.stringify(fields)}::jsonb || jsonb_build_object('receivedAt', to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"'))` }; +} + +/** Records the event under its delivery id; false when it was recorded before, as on a redelivery. */ +export async function recordEvent(db: DB["db"], tenantId: string, pr: PullRequestRef, deliveryId: string, event: Pick): Promise { + const { key, state } = newEvent(pr, deliveryId, event); const inserted = await db.insert(managerState).values({ tenantId, key, kind: "event", state }).onConflictDoNothing().returning({ key: managerState.key }); return inserted.length === 1; } +const ResultRecorded = type([{ recorded: "0 | 1", current: "string | null" }]); + +/** + * Records a reviewer's result on `head` under its run's id, in one statement only while `head` is the newest a pull + * request mail named, or none named one yet, so a push recorded meanwhile cannot slip between the check and the record. + * `recorded` is false as well when the run reported before; `current` is the newest head the statement saw. + */ +export async function recordResult(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, head: string, payload: Record): Promise<{ recorded: boolean; current: string | undefined }> { + const { key, state } = newEvent(pr, runId, { type: "reviewer-result", head, payload }); + const [row] = ResultRecorded.assert(await db.execute(sql` + WITH newest AS (SELECT ${managerState.state}->>'head' AS head FROM ${managerState} WHERE ${mails(tenantId, pr)} ORDER BY ${newestFirst} LIMIT 1), + inserted AS ( + INSERT INTO ${managerState} (tenant_id, key, kind, state) + SELECT ${tenantId}, ${key}, 'event', ${state} WHERE NOT EXISTS (SELECT 1 FROM newest WHERE head <> ${head}) + ON CONFLICT DO NOTHING RETURNING 1 + ) + SELECT (SELECT count(*) FROM inserted)::int AS recorded, (SELECT head FROM newest) AS current`)); + return { recorded: row.recorded === 1, current: row.current ?? undefined }; +} + /** * How long a just-deployed manager may have no sidecar and still hold the pull request, and a just-deployed listener * still count as live. The listener waits this long for a manager to start. @@ -313,7 +344,7 @@ export async function codeReviews(db: DB["db"], tenantId: string, pr: PullReques const failed = asRecord(reviewer.result)?.["failed"]; const base = { runId: reviewer.runId, head: reviewer.head, at: run.createdAt.toISOString() }; if (typeof failed === "string") records.push({ ...base, state: "failed", failed }); - else if (reviewer.result === undefined && claim.status === "live" && Date.parse(reviewer.deadline) > Date.now()) records.push({ ...base, state: "running" }); + else if (reviewer.result === undefined && Date.parse(reviewer.deadline) > Date.now()) records.push({ ...base, state: "running" }); } } return records.sort((a, b) => b.at.localeCompare(a.at)); @@ -379,12 +410,26 @@ async function closingOf(tx: Tx, tenantId: string, pr: PullRequestRef, runId: st return { head, startedAt: first.takenAt, ...(mail && { mail }) }; } +/** The claim's reviewers that ran past their deadline with no report and that `recorded` no longer waits for. */ +async function givenUp(tx: Tx, tenantId: string, pr: PullRequestRef, claim: Claim, recorded: readonly Reviewer[], now: Date): Promise { + const late = claim.reviewers.filter((reviewer) => reviewer.result === undefined && Date.parse(reviewer.deadline) <= now.getTime() && !recorded.some((kept) => kept.runId === reviewer.runId && kept.result === undefined)); + if (late.length === 0) return []; + const rows = await tx.select({ key: managerState.key }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, inArray(managerState.key, late.map((reviewer) => eventKey(pr, reviewer.runId))))); + const reported = new Set(rows.map((row) => row.key)); + return late.filter((reviewer) => !reported.has(eventKey(pr, reviewer.runId))); +} + +function reviewerKey({ kind, head, runId }: Reviewer): string { + return `${kind}:${head}:${runId}`; +} + /** - * 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. + * Closes the open pass with what it reported and records `reviewers`. The manager has more to do while an event is + * pending, at once, and once more after a pass that asks to `wait`; otherwise the claim ends, and a later event or + * reviewer result starts the next manager. A reviewer the pass gave up on past its deadline with no report is logged. */ -export async function done(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, reviewers: readonly Reviewer[], report?: PassReport, record?: RecordDone): Promise { +export async function done(db: DB["db"], tenantId: string, pr: PullRequestRef, runId: string, reviewers: readonly Reviewer[], { wait = false, report, record, log }: DoneOptions = {}): Promise { return db.transaction(async function finishPass(tx) { const claim = await lockClaim(tx, tenantId, pr); if (!fencedTo(claim, runId)) return { outcome: "refused" }; @@ -393,13 +438,14 @@ export async function done(db: DB["db"], tenantId: string, pr: PullRequestRef, r const now = new Date(); 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 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) { + for (const reviewer of await givenUp(tx, tenantId, pr, claim, recorded, now)) { + log?.({ level: "warn", msg: "reviewer_overdue", tenantId, source: pr.source, repo: pr.repo, number: pr.number, kind: reviewer.kind, head: reviewer.head, runId: reviewer.runId, deadline: reviewer.deadline }); + } + if (waiting || wait) { await writeClaim(tx, tenantId, pr, { ...kept, status: "live", runId }); return { outcome: "more", wait: !waiting }; } @@ -421,15 +467,120 @@ export async function passRequest(db: DB["db"], tenantId: string, pr: PullReques */ export type ManagedPull = { pr: PullRequestRef; history: ClosedPass[]; live?: { runId: string; head?: string; pass?: Pass }; 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 { +/** The pull request's mails that name its head: its webhooks other than a held Do's approval. */ +function mails(tenantId: string, pr: PullRequestRef): SQL | undefined { + return 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}`); +} + +const newestFirst = sql`${managerState.state}->>'receivedAt' DESC, ${managerState.key} DESC`; + +/** The newest head a pull request mail named, which a live manager takes next or has taken; undefined before any. */ +async function newestHead(tx: Tx | DB["db"], 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)) + .where(mails(tenantId, pr)) + .orderBy(newestFirst) .limit(1); return row && eventOf(row.key, row.state).head; } +/** The claim's reviewers still out on a head older than the newest one a pull request mail named, which no manager waits for any more. */ +export async function superseded(db: DB["db"], tenantId: string, pr: PullRequestRef): Promise { + return db.transaction(async function older(tx) { + const claim = await lockClaim(tx, tenantId, pr); + const head = await newestHead(tx, tenantId, pr); + if (!claim || head === undefined) return []; + return claim.reviewers.filter((reviewer) => reviewer.head !== head && reviewer.result === undefined); + }); +} + +/** Takes the reviewer off the claim, once its run was asked to stop. */ +export async function dropReviewer(db: DB["db"], tenantId: string, pr: PullRequestRef, reviewer: Reviewer): Promise { + await db.transaction(async function drop(tx) { + const claim = await lockClaim(tx, tenantId, pr); + if (!claim) return; + const key = reviewerKey(reviewer); + await writeClaim(tx, tenantId, pr, { ...claim, reviewers: claim.reviewers.filter((held) => reviewerKey(held) !== key) }); + }); +} + +const CLAIM_KEY = /^mgr:([^:]+):(.+)#(\d+)$/; + +function pullOf(key: string): PullRequestRef { + const [, source, repo, number] = CLAIM_KEY.exec(key) ?? []; + const known = REPO_SOURCES.find((each) => each === source); + if (!known || repo === undefined || number === undefined) throw new Error(`manager_state ${key} is not a claim key`); + return { source: known, repo, number: Number(number) }; +} + +/** How long an overdue wake no manager has taken the claim since holds off the next, while its manager may still be reaching a sidecar. */ +const OVERDUE_UNSTARTED_GRACE_MS = 30 * 60_000; + +/** Whether the claim's last overdue wake is still within its grace: a manager took the claim since, or none has yet. */ +function recentlyWoken(claim: Claim, now: Date): boolean { + if (claim.overdueWokenAt === undefined) return false; + const woken = Date.parse(claim.overdueWokenAt); + const grace = Date.parse(claim.claimedAt) >= woken ? MANAGER_START_GRACE_MS : OVERDUE_UNSTARTED_GRACE_MS; + return now.getTime() - woken < grace; +} + +/** Records on the claim that the hub woke a manager for an overdue reviewer, so the next reconcile pass holds off. */ +export async function recordOverdueWake(db: DB["db"], tenantId: string, pr: PullRequestRef): Promise { + await db.transaction(async function wokenAt(tx) { + const claim = await lockClaim(tx, tenantId, pr); + if (!claim) throw new Error(`${claimKey(pr)} has no claim to wake`); + await writeClaim(tx, tenantId, pr, { ...claim, overdueWokenAt: new Date().toISOString() }); + }); +} + +/** The tenant's pull requests whose claim holds a reviewer with no result. */ +export async function reviewedPulls(db: DB["db"], tenantId: string): Promise { + const rows = await db.select({ key: managerState.key }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), eq(managerState.kind, "claim"), sql`jsonb_path_exists(${managerState.state}, '$.reviewers[*] ? (!(exists(@.result)))')`)); + return rows.map((row) => pullOf(row.key)); +} + +/** Each claim with a reviewer result recorded that no pass has taken, with its newest such result. */ +async function unreadResults(db: DB["db"], tenantId: string): Promise> { + const rows = await db.select({ key: managerState.key, state: managerState.state }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, sql`${managerState.state}->>'consumedBy' IS NULL`, sql`${managerState.state}->>'type' = 'reviewer-result'`)); + // A result's key is its pull request's event prefix and the reporting run's id, which has no colon. + return new Map(eventsOf(rows).map((row) => [`mgr:${row.key.slice("evt:".length, row.key.lastIndexOf(":"))}`, row.event])); +} + +/** + * The tenant's pull requests whose manager has gone, with the head to wake one on: each with a reviewer result no pass + * has taken, as when waking a manager for it failed, once the manager start grace has passed since that result, so a + * manager its own wake deployed is not deployed again while it starts; and each whose reviewer of its newest head is past + * its deadline with no result recorded, so a manager posts its verdict without the reviewer. Either is skipped while the + * claim's last overdue wake is within its grace. + */ +export async function overduePulls(db: DB["db"], tenantId: string): Promise<{ pr: PullRequestRef; head: string }[]> { + const now = new Date(); + const unread = await unreadResults(db, tenantId); + const late = sql`jsonb_path_exists(${managerState.state}, '$.reviewers[*] ? (!(exists(@.result)) && @.deadline <= $now)', jsonb_build_object('now', ${now.toISOString()}::text))`; + const rows = await db.select({ key: managerState.key, state: managerState.state }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), eq(managerState.kind, "claim"), unread.size === 0 ? late : or(late, inArray(managerState.key, [...unread.keys()])))); + const overdue: { pr: PullRequestRef; head: string }[] = []; + for (const { key, state } of rows) { + const claim = claimOf(key, state); + const pr = pullOf(key); + if (recentlyWoken(claim, now) || (claim.status === "live" && claim.runId !== undefined && !(await runGone(db, claim.runId, now)))) continue; + const head = await newestHead(db, tenantId, pr); + const result = unread.get(key); + if (result !== undefined) { + if (now.getTime() - Date.parse(result.receivedAt) >= MANAGER_START_GRACE_MS) overdue.push({ pr, head: head ?? result.head }); + continue; + } + if (head === undefined) continue; + const pastDeadline = claim.reviewers.filter((reviewer) => reviewer.head === head && reviewer.result === undefined && Date.parse(reviewer.deadline) <= now.getTime()); + if (pastDeadline.length === 0) continue; + const reported = await db.select({ key: managerState.key }).from(managerState) + .where(and(eq(managerState.tenantId, tenantId), sql`${managerState.kind} = 'event'`, inArray(managerState.key, pastDeadline.map((reviewer) => eventKey(pr, reviewer.runId))))); + if (reported.length < pastDeadline.length) overdue.push({ pr, head }); + } + return overdue; +} + /** 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 []; diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index 94980d27..1ee02a4a 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -35,7 +35,7 @@ import { RAILWAY_PLACE } from "@corbits/provisioner-railway"; import { REPO_SOURCES, repoSourceOf, TRIAGE_LISTENERS, TRIAGE_MAIL_SENDER, TRIAGE_MANAGERS, type RepoSource } from "@corbits/triage-contracts"; import { asRecord, errorText } from "@corbits/common"; import { createInterchangeHub } from "./interchange-hub.js"; -import { createManagerIntake } from "./manager-intake.js"; +import { createManagerIntake, createManagerWake, createOverdueWake, createSupersede } from "./manager-intake.js"; import { MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; import { deploymentUp, liveListener, passRequest } from "./manager-state.js"; import { mountTenantRoutes } from "./tenant-routes.js"; @@ -300,6 +300,13 @@ function liveListenerOf(tenantId: string, source: RepoSource): Promise { return (await liveListenerOf(tenantId, source))?.address; } +const { supersede, supersedeAll } = createSupersede({ + db: composition.db, + async cancel(tenantId, runId, reason) { + await composition.workflowLifecycleService.requestCancellation(tenantId, runId, reason); + }, + log: logJson, +}); const toListener = createManagerIntake({ db: composition.db, listener: listenerAddress, @@ -320,9 +327,18 @@ const toListener = createManagerIntake({ deliver: function deliverOf(tenantId, address, payload) { return deliverToListener(tenantId, address, payload); }, + supersede, log: logJson, }); -mountTenantRoutes(composition.app, { db: composition.db, requireGrant, log: logJson, now }); +const wakeManager = createManagerWake({ + db: composition.db, + listener: listenerAddress, + deliver: function deliverOf(tenantId, address, payload) { + return deliverToListener(tenantId, address, payload); + }, +}); +const wakeOverdue = createOverdueWake({ db: composition.db, wake: wakeManager, log: logJson }); +mountTenantRoutes(composition.app, { db: composition.db, requireGrant, log: logJson, wake: wakeManager, supersede, now }); async function tenantDomain(tenantId: string): Promise { const row = await composition.db.query.tenant.findFirst({ where: eq(schema.tenant.id, tenantId), @@ -607,6 +623,8 @@ async function reconcileAll(): Promise { for (const tenant of await reconcileTenants()) { await releaseHostsOf(tenant.id); await watchListeners(tenant.id); + await supersedeAll(tenant.id); + await wakeOverdue(tenant.id); } const outcomes: ReconcileOutcome[] = []; for (const reconcile of reconcilers) outcomes.push(await reconcile()); diff --git a/apps/hub/src/tenant-routes.ts b/apps/hub/src/tenant-routes.ts index 7677e22d..49f040a9 100644 --- a/apps/hub/src/tenant-routes.ts +++ b/apps/hub/src/tenant-routes.ts @@ -6,7 +6,7 @@ import { createArtifactGrants } from "./artifact-grants.js"; import { createManagerRoutes, type ManagerRoutesDeps } from "./manager-routes.js"; /** The hub's own routes under `/api/tenants/:tenantId/`, mounted on the stock app after its tenant resolution. */ -export function mountTenantRoutes(app: Hono, { db, requireGrant, log, now }: ManagerRoutesDeps & { db: ArtifactDb & CronDb; now: () => Date }): void { +export function mountTenantRoutes(app: Hono, { db, requireGrant, log, wake, supersede, now }: ManagerRoutesDeps & { db: ArtifactDb & CronDb; now: () => Date }): void { app.route("/api/tenants/:tenantId", createArtifactRoutes({ db, contentStore: InlineContentStore, @@ -14,5 +14,5 @@ export function mountTenantRoutes(app: Hono, { db, requireGrant, log, no onArtifactCreated: createArtifactGrants(now), })); app.route("/api/tenants/:tenantId/cron", createCronRoutes({ db, requireGrant })); - app.route("/api/tenants/:tenantId", createManagerRoutes({ db, requireGrant, log })); + app.route("/api/tenants/:tenantId", createManagerRoutes({ db, requireGrant, log, wake, supersede })); } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2bbe65d6..f80ef993 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -267,8 +267,8 @@ any other tenant or an unknown sender. - `triage-manager-github` and `triage-manager-bitbucket-datacenter` (`triageManager`): one live run per pull request, which runs only while it has work, and the only workflow that writes to the pull request. It takes - mail from the hub (`tenant`), the listener that deployed it (`invoker`) and - the judge and the code reviews it mailed (`correspondent`). A mail starts a loop of passes. Each pass + mail from the hub (`tenant`) and the listener that deployed it (`invoker`). + A mail starts a loop of passes. Each pass takes, which claims the pull request and its pending events and answers the newest pull request mail the hub recorded. Each held Do approved since runs as pr-triage runs an approval, which skips one whose pull request moved past @@ -284,9 +284,15 @@ any other tenant or an unknown sender. 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. + reports one of these has its own done step. A pass that posted to the host + or mailed a judge or code review asks done to `wait`. Done then answers + `more` when an event is pending, and the next pass runs at once; `more` with + `wait` after a pass that asked, and the next pass runs once 30 seconds have + passed, taking whatever arrived meanwhile; otherwise `ended`. So a manager + ends 30 seconds after it last posted or mailed, never waiting out a judge or + a code review: their results reach the hub, which wakes the pull request's + manager the way a host event does, and the listener deploys a fresh one once + the claim has ended. Once the passes end, fail or run out, or take finds another run holds the claim, the run's last step stops it through the stop-run route under a `manage` grant on its own run, so the run is ended once and never keeps @@ -328,13 +334,37 @@ any other tenant or an unknown sender. decided verdict, so a later pass on that head whose rules still let code review past posts it with the verdict and deploys no other; a review that could not be mailed, reported a failure or ran past its deadline leaves the - verdict to go up alone and is deployed again on the next event. Each reviewer the manager records names - the manager run that deployed it (`deployedBy`). Done stops each judge or - code review this run deployed and no longer waits for, one on an older head - or past its deadline, through the same stop-run route; - one an earlier manager deployed runs out its own deadline. Stock grants the + verdict to go up alone and is deployed again on the next event. Each + reviewer keeps its head and deadline on the claim, so whichever manager + takes the next pass on that head sees a deadline that passed with no + report; it posts the verdict without that reviewer, which for a judge is + Needs Human Review and for code review holds approve and merge for a person, + and done logs `reviewer_overdue`. The triage reconcile pass, every 5 + minutes, wakes a manager through `createManagerWake` for each pull request + whose manager is not up while a reviewer of its newest head is past its + deadline with no result recorded (`createOverdueWake`, + `apps/hub/src/manager-intake.ts`), and logs `manager_woken_overdue`. A + wake that succeeded is recorded on the claim (`overdueWokenAt`), and the + next pass wakes again only once the manager start grace has passed: two + minutes when a manager has taken the claim since, thirty while none has, so + a manager slow to reach a sidecar is never deployed twice. Each pull request + is woken on its own: one whose wake fails is logged as + `manager_overdue_wake_failed`, the rest are still woken, and the next pass + tries it again. The same pass wakes a pull request whose manager is not up + while a reviewer result no pass has taken waits for it, as after the + reviewer-result route could not wake one, once two minutes have passed + since that result was recorded, so a manager the route's own wake deployed + is not deployed again while it starts, and with the same hold after its + own overdue wake. So + within one pass of the deadline that manager posts the verdict without the + reviewer and the pull request needs a person. Each reviewer the manager + records names the manager run that deployed it (`deployedBy`). Done stops a + judge or code review this run deployed that ran past its deadline on the + pass's head, through the same stop-run route. Stock grants the deployer `read` on a run it deployed, and the vendored Interchange carries - `manage` instead (`vendor/interchange/VENDORED.md`). + `manage` instead (`vendor/interchange/VENDORED.md`). A reviewer on an older + head is stopped by the hub, which a later manager could not do: see intake + below. 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 @@ -355,7 +385,7 @@ any other tenant or an unknown sender. sidecar. It takes mail only from that manager and reviews the request. Stock runs a step whose dependencies failed, so however the review ended it records its `StageResult` through the reviewer-result route (a failure as - `failed`), mails the manager a wake-up and stops. It posts nothing to the + `failed`), which wakes the pull request's manager, and stops. It posts nothing to the pull request. Its definition declares only `create` on `triage-manager:result` and `manage` on `workflow-run:self`, sourced from its creator: it can report a result and stop itself, and reach no other @@ -367,12 +397,12 @@ any other tenant or an unknown sender. dependencies failed, so however the judge ended it records a `StageResult` of kind `judge` through the reviewer-result route: the whole verdict evaluate made, its comment, labels, Dos and rule results, or a failure as - `failed`. It mails the manager a wake-up and stops. It posts nothing to the + `failed`, which wakes the pull request's manager, and stops. It posts nothing to the pull request. Its definition declares only `create` on `triage-manager:result` and `manage` on `workflow-run:self`, sourced from its creator. -- A run sends mail only through an agent step's transport, so the listener, - the manager, the judge and the code review mail through one agent whose +- A run sends mail only through an agent step's transport, so the listener + and the manager mail through one agent whose director makes no model call: it runs its one tool, which waits for a just-deployed run to reach a sidecar until a deadline and then mails it as the sending run. @@ -594,7 +624,8 @@ only what is missing, and none once the summary stands on either head. A pull request's latest review (`GET …/code-review`, `code-review-runs.ts`) is read from `manager_state`: the newest of the code reviews its claim still -waits on, shown as running, those the manager gave up on, such as one past its +waits on, shown as running from the reviewer's own record until it reports or +its deadline passes, whether or not a manager is up, those the manager gave up on, such as one past its deadline, and the results its code reviews reported through the reviewer-result route that a pass took or that name a reviewer of the claim. A review that did not finish shows as failed; its run log keeps the error. @@ -813,10 +844,9 @@ read it, and passes the rest of the request to stock. way it answers the reviewers the last done recorded and the newest pull request mail recorded: the pass's own, or for a pass without one the newest `pr` mail. Done - closes the pass and, under the claim's lock, ends the claim only when no - event is pending and no reviewer of the current head is due, so an event - committed before it is taken by this manager and one after it by the - next. A manager run reaches take and done, and a reviewer run records its + closes the pass and, under the claim's lock, ends the claim when no event + is pending and the pass did not ask to `wait`, so an event committed before + it is taken by this manager and one after it by the next. A manager run reaches take and done, and a reviewer run records its result (under its own run id), through `POST /api/tenants/:tenantId/manager/{take,done,reviewer-result}` (`apps/hub/src/manager-routes.ts`). They admit a run token only, as the @@ -835,7 +865,20 @@ read it, and passes the rest of the request to stock. Its body is a `StageResult` (`@corbits/triage-contracts`): the reviewer's kind (`code-review` or `judge`), the head it reviewed and its result, a code review's output or a judge's whole verdict, or `failed` with why it - has none. + has none. The result is recorded by one statement only while its head is + the newest one a pull request mail named, so no push slips between the + check and the record; one for another head is dropped and logged as + `reviewer_result_stale`. Otherwise, once recorded, the route wakes the pull + request's manager as intake does for a host event (`createManagerWake`, + `apps/hub/src/manager-intake.ts`): the live manager the claim names is + mailed, or the host's listener is mailed the head to deploy one; a + workspace with no listener logs `reviewer_result_unrouted`. A wake that + fails leaves the result recorded and nothing else, is logged as + `reviewer_result_wake_failed`, and answers 502 `wake_failed`; the next + reconcile pass wakes a manager for it. Done records the reviewers the + manager reports, then the hub stops each the claim holds on a head older + than the newest, as one the pass deployed while a push arrived, the way + intake does below. - Pull request, waiting, held-Do approval and backlog mail goes to the host's listener (`apps/hub/src/manager-intake.ts`), on both hosts. Each pull request the mail names (a reconciler's `items`, a backlog's open pull @@ -845,7 +888,13 @@ read it, and passes the rest of the request to stock. workflow carries it out. The hub records one webhook event per pull request that passed, with the head's code review request when the policy turns code review on and the App's login on GitHub, then mails the listener the - heads. When the host's head moved past the mail's, the event goes without + heads. Once it records a head, every judge or code review the claim still + waits on for an older head of that pull request is stopped through the + stock run lifecycle with the hub's own authority and then taken off the + claim, each logged once as `reviewer_superseded`. A stop that fails is + logged as `reviewer_supersede_failed` and leaves the reviewer on the claim; + each triage reconcile pass stops the superseded reviewers of every claim + again (`createSupersede`). When the host's head moved past the mail's, the event goes without a review request and the hub logs `code_review_head_moved`; the new head's own mail carries one. An event's id is the webhook's delivery id, which the coalescer's mail carries, so a redelivery records nothing new; any diff --git a/packages/pr-code-review-anthropic/src/directors.ts b/packages/pr-code-review-anthropic/src/directors.ts index 039ac2c3..3f8041fe 100644 --- a/packages/pr-code-review-anthropic/src/directors.ts +++ b/packages/pr-code-review-anthropic/src/directors.ts @@ -3,4 +3,3 @@ import { directors } from "./review-directors.js"; export const readerDirectorFactory = directors.reader.factory; export const consolidateDirectorFactory = directors.consolidate.factory; export const structuredDirectorFactory = directors.structured.factory; -export const mailDirectorFactory = directors.mail.factory; diff --git a/packages/pr-code-review/src/agents.ts b/packages/pr-code-review/src/agents.ts index 66eb4570..31ccbc98 100644 --- a/packages/pr-code-review/src/agents.ts +++ b/packages/pr-code-review/src/agents.ts @@ -1,5 +1,4 @@ import { defineAgent, defineTool } from "@intx/agent"; -import { mailAgentOf } from "@corbits/triage-manager/mail"; import { bitbucketRead } from "@corbits/bitbucket-datacenter-tool"; import { githubRead } from "@corbits/github-tool"; import { repoReadUnder } from "@corbits/repo-read-tool"; @@ -113,7 +112,7 @@ const WRITE_SCHEMA = { * The review's agents, each declaring its step's model on the profile's provider and run by the provider's directors; * the repository finder reads checkouts under the sidecar data dir `dataDir` names. */ -export function codeReviewAgents(profile: CodeReviewProfile, { reader, consolidate, structured, mail }: CodeReviewDirectors, dataDir: () => string) { +export function codeReviewAgents(profile: CodeReviewProfile, { reader, consolidate, structured }: CodeReviewDirectors, dataDir: () => string) { function inference(role: Role) { return { sources: [{ provider: profile.provider, model: profile.models[role] }] }; } @@ -158,8 +157,6 @@ export function codeReviewAgents(profile: CodeReviewProfile, { reader, consolida planAgent: structuredAgent("code-review-plan", "plan", PLAN_SCHEMA), checkAgent: structuredAgent("code-review-check", "check", CHECK_SCHEMA), writeAgent: structuredAgent("code-review-write", "write", WRITE_SCHEMA), - /** Replies to the manager that deployed the review; it makes no model call. */ - mailAgent: mailAgentOf(mail, inference("write"), { tolerate: false }), }; } diff --git a/packages/pr-code-review/src/directors.ts b/packages/pr-code-review/src/directors.ts index bd5adc38..ad3cf7db 100644 --- a/packages/pr-code-review/src/directors.ts +++ b/packages/pr-code-review/src/directors.ts @@ -1,7 +1,5 @@ -import { mailDirectorOf } from "@corbits/triage-manager/mail"; import { consolidateDirectorOf } from "./consolidate-director.js"; import type { CodeReviewProfile } from "./profile.js"; -import { packageName } from "./package-name.js"; import { readerDirectorOf } from "./reader-director.js"; import { structuredDirectorOf } from "./structured-director.js"; @@ -10,7 +8,7 @@ import { structuredDirectorOf } from "./structured-director.js"; * its `interchange.directors` module exports their factories. */ export function codeReviewDirectors(profile: CodeReviewProfile, dataDir: () => string) { - return { reader: readerDirectorOf(profile, dataDir), consolidate: consolidateDirectorOf(profile), structured: structuredDirectorOf(profile), mail: mailDirectorOf(packageName(profile.packageName)) }; + return { reader: readerDirectorOf(profile, dataDir), consolidate: consolidateDirectorOf(profile), structured: structuredDirectorOf(profile) }; } export type CodeReviewDirectors = ReturnType; diff --git a/packages/pr-code-review/src/pr-code-review.ts b/packages/pr-code-review/src/pr-code-review.ts index 766de659..43a7cb1b 100644 --- a/packages/pr-code-review/src/pr-code-review.ts +++ b/packages/pr-code-review/src/pr-code-review.ts @@ -102,7 +102,7 @@ function hubAction(handler: string, input: Selector, after: readonly string[]) { /** * The review a pull request's manager deploys, under `id`, also its mail address: it reviews the request the manager - * mails it, reports the review through the hub, or that it did not finish, replies so the manager collects it, then + * mails it, reports the review through the hub, or that it did not finish, which wakes the pull request's manager, then * stops, whatever failed before. It posts nothing to the pull request. Only the manager that deployed it may mail it. */ export function codeReviewReviewer(agents: CodeReviewAgents) { @@ -113,10 +113,9 @@ export function codeReviewReviewer(agents: CodeReviewAgents) { steps: { reviewRequest: action({ handler: "reviewRequest", input: { from: "trigger.payload" } }), ...codeReviewSteps(agents, sources, { prefix: "", request: { from: "steps.reviewRequest.output.request" }, after: ["reviewRequest"] }), - // Stock runs a step whose dependencies failed, so the report, the reply and the stop run however the review ended. + // Stock runs a step whose dependencies failed, so the report and the stop run however the review ended. report: hubAction("report", { merge: [{ from: "steps" }, { project: { from: "trigger" }, fields: ["payload"] }] }, ["review"]), - reply: step({ agent: agents.mailAgent, input: { from: "steps.report.output.mail" }, after: ["report"] }), - stop: hubAction("stop", { from: "trigger.payload" }, ["reply"]), + stop: hubAction("stop", { from: "trigger.payload" }, ["report"]), }, }); return defineWorkflow({ diff --git a/packages/triage-judge/README.md b/packages/triage-judge/README.md index 75f01b83..3ca5a434 100644 --- a/packages/triage-judge/README.md +++ b/packages/triage-judge/README.md @@ -2,8 +2,8 @@ The judge a pull request's manager deploys (`triageJudge`, `triage-judge-`). The manager mails it the rules output for one head. It runs those chunks through System One, then focused's follow-up questions, then evaluate, and reports -the whole verdict, which the manager posts as it is, through the reviewer-result route. It mails the manager a wake-up -and stops, whether or not the judge finished. It posts nothing to the pull request. +the whole verdict, which the manager posts as it is, through the reviewer-result route, where the hub wakes the pull +request's manager. It then stops, whether or not the judge finished. It posts nothing to the pull request. Only the manager that deployed it may mail it. Its definition declares `create` on `triage-manager:result` and `manage` on its own run. diff --git a/packages/triage-judge/src/actions/report.test.ts b/packages/triage-judge/src/actions/report.test.ts index 3a5453eb..de054d98 100644 --- a/packages/triage-judge/src/actions/report.test.ts +++ b/packages/triage-judge/src/actions/report.test.ts @@ -4,10 +4,9 @@ import { judgeRequest, report, stop } from "./report.js"; const RULES = { items: [{ facts: {} }], batch: false, chunks: [{ item: 0, state: {}, questions: [] }] }; const CHECKOUT = { source: "github", repo: "acme/widgets", sha: "h1" }; -const REPLY = { to: "run_manager", content: "{\"kind\":\"manage\"}" }; const MAIL = { headers: { from: "run_manager@acme.test", to: ["run_judge@acme.test"] }, - parts: [{ contentType: "text/plain", text: JSON.stringify({ kind: "judge", tenantId: "tnt_acme", number: 8, checkout: CHECKOUT, rules: RULES, reply: REPLY }) }], + parts: [{ contentType: "text/plain", text: JSON.stringify({ kind: "judge", tenantId: "tnt_acme", number: 8, checkout: CHECKOUT, rules: RULES }) }], }; const VERDICT = { repo: "acme/widgets", number: 8, headSha: "h1", verdict: "pass", reasons: ["Every rule passed."], checks: [], request: { repo: "acme/widgets", number: 8, comment: "Ready." }, automatic: ["labels"], actions: [] }; @@ -23,12 +22,12 @@ function hubContext() { return { ctx, calls }; } -test("a manager's judge hands over the rules, reports the verdict through the hub, replies to the manager and stops, posting nothing", async () => { +test("a manager's judge hands over the rules, reports the verdict through the hub and stops, posting nothing", async () => { const { ctx, calls } = hubContext(); const signal = new AbortController().signal; expect(await judgeRequest(MAIL)).toEqual(RULES); - expect(await report({ payload: MAIL, evaluate: { output: VERDICT }, judge: { output: [] } }, ctx, signal)).toEqual({ recorded: true, mail: { tenantId: "tnt_acme", ...REPLY } }); - expect(await report({ payload: MAIL }, ctx, signal)).toEqual({ recorded: true, mail: { tenantId: "tnt_acme", ...REPLY } }); + expect(await report({ payload: MAIL, evaluate: { output: VERDICT }, judge: { output: [] } }, ctx, signal)).toEqual({ recorded: true }); + expect(await report({ payload: MAIL }, ctx, signal)).toEqual({ recorded: true }); expect(await stop(MAIL, ctx, signal)).toEqual({ stopped: true }); const reported = { source: "github", repo: "acme/widgets", number: 8, kind: "judge", head: "h1" }; expect(calls).toEqual([ diff --git a/packages/triage-judge/src/actions/report.ts b/packages/triage-judge/src/actions/report.ts index 03b0a269..37bc1ef7 100644 --- a/packages/triage-judge/src/actions/report.ts +++ b/packages/triage-judge/src/actions/report.ts @@ -1,17 +1,15 @@ import type { EffectContext } from "@intx/workflow"; import { type } from "arktype"; import { JudgeVerdict, REPO_SOURCES } from "@corbits/triage-contracts"; -import type { MailInput } from "@corbits/triage-manager/mail"; import { answered, callHub, endRun, mailOf, tenantPath } from "@corbits/triage-manager/actions/hub"; -/** The manager's mail to a judge it deployed: the rules output, the pull request and the mail to reply with. */ +/** The manager's mail to a judge it deployed: the rules output and the pull request. */ const JudgeMail = type({ kind: "'judge'", tenantId: "string > 0", number: "number.integer > 0", checkout: { source: type.enumerated(...REPO_SOURCES), repo: "string > 0", sha: "string > 0" }, rules: { items: "unknown[]", batch: "boolean", chunks: "unknown[]" }, - reply: { to: "string > 0", content: "string > 0" }, }); function judgeMailOf(payload: unknown) { @@ -34,20 +32,19 @@ const NOT_FINISHED = "the judge did not finish"; /** * Reports the verdict to the pull request's manager as a `StageResult` through the hub, or that the judge did not - * finish; the reply then wakes the manager. The verdict goes whole, so the manager posts it as it was made. Posts nothing. + * finish; the hub then wakes the manager. The verdict goes whole, so the manager posts it as it was made. Posts nothing. */ -export async function report(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ recorded: boolean; mail: MailInput }> { +export async function report(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ recorded: boolean }> { const parsed = ReportInput(input); if (parsed instanceof type.errors) throw new Error(`report: ${parsed.summary}`); - const { tenantId, number, checkout, reply } = judgeMailOf(parsed.payload).mail; + const { tenantId, number, checkout } = judgeMailOf(parsed.payload).mail; const verdict = JudgeVerdict(parsed.evaluate?.output); const outcome = parsed.evaluate?.output === undefined || verdict instanceof type.errors ? { failed: NOT_FINISHED } : { result: verdict }; const result = { source: checkout.source, repo: checkout.repo, number, kind: "judge" as const, head: checkout.sha, ...outcome }; - const { recorded } = answered(await callHub(ctx, "report", tenantPath(tenantId, "/manager/reviewer-result"), "POST", result), "report", Recorded); - return { recorded, mail: { tenantId, ...reply } }; + return answered(await callHub(ctx, "report", tenantPath(tenantId, "/manager/reviewer-result"), "POST", result), "report", Recorded); } -/** Stops the judge's run once it has replied, whether or not the verdict, its report or its reply went through. */ +/** Stops the judge's run once it has reported, whether or not the verdict or its report went through. */ export async function stop(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ stopped: true }> { const { mail, runId } = judgeMailOf(input); await endRun(ctx, mail.tenantId, runId); diff --git a/packages/triage-judge/src/directors.ts b/packages/triage-judge/src/directors.ts index cdbd051c..52de3e99 100644 --- a/packages/triage-judge/src/directors.ts +++ b/packages/triage-judge/src/directors.ts @@ -1,13 +1 @@ -import { mailDirectorOf } from "@corbits/triage-manager/mail"; -import { triageJudgeDirector, triageJudgeDirectorFactory } from "@corbits/triage-workflows/directors"; - -/** Set by the workflow build to the published package name; source runs use the workspace package. */ -declare const WORKFLOW_PACKAGE_NAME: string | undefined; - -const packageName = typeof WORKFLOW_PACKAGE_NAME === "undefined" ? "@corbits/triage-judge" : WORKFLOW_PACKAGE_NAME; - -export const mailDirector = mailDirectorOf(packageName); - -export const mailDirectorFactory = mailDirector.factory; - -export { triageJudgeDirector, triageJudgeDirectorFactory }; +export { triageJudgeDirector, triageJudgeDirectorFactory } from "@corbits/triage-workflows/directors"; diff --git a/packages/triage-judge/src/triage-judge.test.ts b/packages/triage-judge/src/triage-judge.test.ts index f9bc9a1c..2895e6bf 100644 --- a/packages/triage-judge/src/triage-judge.test.ts +++ b/packages/triage-judge/src/triage-judge.test.ts @@ -70,7 +70,6 @@ function actionCredentials(hub: (url: URL, init?: RequestInit) => Promise { calls.push({ method: init?.method ?? "GET", path: url.pathname, ...(init?.body !== undefined && { body: JSON.parse(String(init.body)) }) }); return init?.method === "DELETE" ? new Response(null, { status: 202 }) : Response.json({ recorded: true }, { status: 201 }); @@ -81,16 +80,11 @@ async function runJudge(rulesOutput: { chunks: unknown[] }) { headers: { from: "run_manager@triage.test", to: [`${RUN}@triage.test`] }, parts: [{ contentType: "text/plain", text: JSON.stringify({ kind: "judge", tenantId: TENANT, number: 8, checkout: { source: "github", repo: "acme/widgets", sha: SHA }, rules: rulesOutput, - reply: { to: "run_manager", content: JSON.stringify({ kind: "manage", tenantId: TENANT, repo: "acme/widgets", repoSource: "github", prNumber: 8 }) }, }) }], }; const { terminalStatus } = await runLocal(body.inline, { authorize: allow, invokeStep: async function invoke({ agent, input }) { - if (agent.id.startsWith("triage-mail")) { - mailed.push(input); - return { output: JSON.stringify({ mailed: true }) }; - } if (agent.id !== "triage-judge") throw new Error(`no agent but the judge runs, not ${agent.id}`); if (!input || typeof input !== "object" || !("questions" in input) || !Array.isArray(input.questions)) throw new Error("the judge step was not given a chunk"); const text = input.questions.filter(isQuestion).map((question) => answerText(question.id)).join(""); @@ -100,20 +94,19 @@ async function runJudge(rulesOutput: { chunks: unknown[] }) { actionCredentials: actionCredentials(hub), triggerPayload: mail, }).complete; - return { terminalStatus, calls, mailed }; + return { terminalStatus, calls }; } test("the judge runs the chunks the manager mailed, reports the whole verdict and stops, posting nothing", async () => { const rulesOutput = await ruled(); expect(rulesOutput.chunks.length).toBeGreaterThan(0); - const { terminalStatus, calls, mailed } = await runJudge(rulesOutput); + const { terminalStatus, calls } = await runJudge(rulesOutput); expect(terminalStatus).toBe("completed"); const posted = calls.find((call) => call.method === "POST"); expect(posted).toMatchObject({ path: `/api/tenants/${TENANT}/manager/reviewer-result`, body: { kind: "judge", head: SHA, source: "github", repo: "acme/widgets", number: 8 } }); if (posted === undefined) throw new Error("the judge posted nothing"); expect((posted.body as { result?: unknown }).result).toMatchObject({ headSha: SHA, verdict: "pass", request: { header: { head: SHA, verdict: "pass" } }, automatic: [], actions: [] }); expect(calls.some((call) => call.path.includes("github.com"))).toBe(false); - expect(mailed).toHaveLength(1); expect(calls.at(-1)).toEqual({ method: "DELETE", path: `/api/tenants/${TENANT}/workflows/runs/${RUN}` }); }); diff --git a/packages/triage-judge/src/triage-judge.ts b/packages/triage-judge/src/triage-judge.ts index d2943717..974e65dd 100644 --- a/packages/triage-judge/src/triage-judge.ts +++ b/packages/triage-judge/src/triage-judge.ts @@ -2,9 +2,7 @@ import { action, defineWorkflow, map, onTrigger, SELF_RUN_RESOURCE, step, type S import { HUB_CREDENTIAL_HANDLE } from "@intx/types/credential-delivery"; import { MANAGER_RESULT_ACTION, MANAGER_RESULT_RESOURCE } from "@corbits/triage-contracts"; import { HUB_WRITE } from "@corbits/triage-manager/actions/hub"; -import { mailAgentOf } from "@corbits/triage-manager/mail"; import { judgeAgent } from "@corbits/triage-workflows/agents"; -import { mailDirector } from "./directors.js"; const TIMEOUT = 15 * 60_000; const STEP = { retry: { maxAttempts: 3, initialBackoffMs: 5_000, maxBackoffMs: 60_000 }, timeout: TIMEOUT } as const; @@ -23,8 +21,9 @@ function hubAction(handler: string, input: Selector, after: readonly string[]) { /** * The judge a pull request's manager deploys, under `id`, also its mail address. It runs the model checks the machine - * rules produced, reports the verdict through the hub, or that it did not finish, replies so the manager collects it, - * then stops, whatever failed before. It posts nothing to the pull request. Only the manager that deployed it may mail it. + * rules produced, reports the verdict through the hub, or that it did not finish, which wakes the pull request's + * manager, then stops, whatever failed before. It posts nothing to the pull request. Only the manager that deployed it + * may mail it. */ export function triageJudge({ id }: { id: string }) { const body = defineWorkflow({ @@ -36,10 +35,9 @@ export function triageJudge({ id }: { id: string }) { parts: action({ handler: "parts", input: { merge: [RULED, JUDGED] }, after: ["judge"] }), judgeParts: map({ over: { from: "steps.parts.output.chunks" }, step: step({ agent: judgeAgent, input: { from: "trigger.payload" }, ...STEP }), after: ["parts"] }), evaluate: action({ handler: "evaluate", input: { merge: [RULED, BOTH_JUDGED] }, after: ["judgeParts"] }), - // Stock runs a step whose dependencies failed, so the report, the reply and the stop run however the judge ended. + // Stock runs a step whose dependencies failed, so the report and the stop run however the judge ended. report: hubAction("report", { merge: [{ from: "steps" }, { project: { from: "trigger" }, fields: ["payload"] }] }, ["evaluate"]), - reply: step({ agent: mailAgentOf(mailDirector, judgeAgent.inference, { tolerate: false }), input: { from: "steps.report.output.mail" }, after: ["report"] }), - stop: hubAction("stop", { from: "trigger.payload" }, ["reply"]), + stop: hubAction("stop", { from: "trigger.payload" }, ["report"]), }, }); return defineWorkflow({ diff --git a/packages/triage-manager/README.md b/packages/triage-manager/README.md index 3442f184..5050c66a 100644 --- a/packages/triage-manager/README.md +++ b/packages/triage-manager/README.md @@ -10,7 +10,10 @@ A workspace's listener and a pull request's manager. request and its pending events; each held Do approved since runs as pr-triage runs an approval; then pr-triage's own steps run on the newest mail the hub recorded for the head. A policy of only deterministic rules posts on that 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 next pass runs at once while an event is pending, or every 30 seconds while it waits for a reviewer. With a + up. The next pass runs at once while an event is pending; after a pass that posted to the host or mailed a reviewer, + the next runs once 30 seconds later, and a manager whose pass after that has nothing to do ends. It never waits out + a reviewer: a reviewer's result reaches the hub, which wakes the live manager or has the listener deploy a fresh + one. With a published judge, a pass deploys [`@corbits/triage-judge`](../triage-judge) once the host's checks have finished and mails it the rules output; while they run the head waits. The pass that collects its report posts the judge's verdict as the judge made it while the rules still need the judge, and so does every later pass on that head once @@ -18,8 +21,9 @@ A workspace's listener and a pull request's manager. head's request; the pass that collects its report mirrors the verdict and posts the review together, and a later pass on that head posts the stored review again without deploying another. A reviewer that could not be mailed, failed or ran past its one-hour deadline leaves the verdict to go up alone and is deployed again on the next event; - a judge the release does not publish posts nothing. `done` stops a reviewer this run deployed that is on an older - head or past its deadline. The run's last step is its only end: it stops the run once `done` ends the pull + a judge the release does not publish posts nothing. `done` stops a reviewer this run deployed that ran past its + deadline; the hub stops one on an older head when it records the newer head or when `done` reports it, and wakes a manager once a deadline + passes with no manager up. The run's last step is its only end: it stops the run once `done` ends the pull request's claim, take is refused because another run holds it, or the passes fail or run out. What each deploys next, a published workflow's source asset at one commit and the offerings it runs on (`Release`), is @@ -29,8 +33,8 @@ code review only when the code review's is set, and the judge only when the judg A run mails only through an agent step, so both mail through `mailAgentOf`, whose director makes no model call. The code review a manager deploys is `codeReviewReviewer` in [`@corbits/pr-code-review`](../pr-code-review), and the judge is -`triageJudge` in [`@corbits/triage-judge`](../triage-judge). Each reports through its own `report`, whether or not it -finished, and posts nothing. The routes, the claim and the mail are in +`triageJudge` in [`@corbits/triage-judge`](../triage-judge). Each reports through its own `report` to the hub, whether or +not it finished, and posts nothing; the hub then wakes the pull request's manager. The routes, the claim and the mail are in [ARCHITECTURE.md](../../docs/ARCHITECTURE.md#workflows). ## Test diff --git a/packages/triage-manager/src/actions/judge.ts b/packages/triage-manager/src/actions/judge.ts index 9d2519df..bba0c244 100644 --- a/packages/triage-manager/src/actions/judge.ts +++ b/packages/triage-manager/src/actions/judge.ts @@ -7,7 +7,7 @@ import { hostChecksPending } from "@corbits/triage-workflows/logic/render"; import { decided } from "@corbits/triage-workflows/logic/settled"; import { Mailed, type MailInput } from "../mail.js"; import { publishedFor } from "../release.js"; -import { deploy, managerMailOf, manageMail } from "./hub.js"; +import { deploy, managerMailOf } from "./hub.js"; import { Reviewer, TakeOutput } from "./manager.js"; /** How long the manager waits for a judge it deployed, from its deploy to its report. */ @@ -81,8 +81,8 @@ export async function judgeScope(input: unknown): Promise<{ deploy: boolean; pos const DeployInput = type({ payload: "unknown", take: { output: TakeOutput }, rules: { output: "unknown" } }); /** - * Deploys the judge from the host's published release. It is mailed the rules output and, to reply with, the mail that - * wakes this manager. The mail waits for the new run until the judge's deadline. + * Deploys the judge from the host's published release. It is mailed the rules output, and reports its verdict to the + * hub, which wakes the pull request's manager. The mail waits for the new run until the judge's deadline. */ export async function deployJudge(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ reviewer: Reviewer; mail: MailInput }> { const parsed = DeployInput(input); @@ -96,7 +96,7 @@ export async function deployJudge(input: unknown, ctx: EffectContext, _signal: A if (!release) throw new Error("deployJudge: release.json publishes no judge for the host"); const judgeRun = await deploy(ctx, "deployJudge", tenantId, release); const deadline = new Date(Date.now() + JUDGE_DEADLINE_MS).toISOString(); - const content = JSON.stringify({ kind: "judge", tenantId, number: pullRequest.number, checkout: { source, repo: pullRequest.repo, sha: head }, rules, reply: { to: runId, content: manageMail(tenantId, pullRequest) } }); + const content = JSON.stringify({ kind: "judge", tenantId, number: pullRequest.number, checkout: { source, repo: pullRequest.repo, sha: head }, rules }); return { reviewer: { kind: "judge", head, runId: judgeRun, deployedBy: runId, deadline }, mail: { tenantId, to: judgeRun, content, readyBy: deadline } }; } diff --git a/packages/triage-manager/src/actions/manager.ts b/packages/triage-manager/src/actions/manager.ts index 676212ea..b3ac4e4b 100644 --- a/packages/triage-manager/src/actions/manager.ts +++ b/packages/triage-manager/src/actions/manager.ts @@ -203,21 +203,18 @@ function reviewersAfter(take: TakeOutput, mailed: Reviewer | undefined): Reviewe } /** - * The reviewers this run deployed that the pass stops waiting for without a report: one on an older head, or past its - * deadline. One an earlier manager deployed runs out its own deadline, since only its deployer may stop it. + * The reviewers of the pass's head this run deployed that ran past their deadline with no report. The hub stops one on + * an older head when it records the newer one; one an earlier manager deployed runs out on its own, since only its + * deployer may stop it. */ function abandoned({ head, reviewers }: TakeOutput, runId: string): Reviewer[] { - if (head === undefined) return []; - return reviewers.filter(function abandon(reviewer) { - if ((reviewer.kind !== "code-review" && reviewer.kind !== "judge") || reviewer.deployedBy !== runId) return false; - if (reviewer.head !== head) return true; - return reviewer.result === undefined && Date.parse(reviewer.deadline) <= Date.now(); - }); + return reviewers.filter((reviewer) => reviewer.deployedBy === runId && reviewer.head === head && reviewer.result === undefined && Date.parse(reviewer.deadline) <= Date.now()); } /** - * 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. + * Closes the pass with what it ran and stops the reviewers it no longer waits for. A pass that posted to the host or + * mailed a reviewer asks the hub to `wait`. `more` runs the next pass at once, or with `wait` after waiting for what the + * pass brought; `ended` ends the run, and a reviewer's result or a later event starts the next one. */ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ outcome: "ended" | "refused" } | { outcome: "more"; wait: boolean }> { const parsed = DoneInput(input); @@ -226,7 +223,8 @@ export async function done(input: unknown, ctx: EffectContext, _signal: AbortSig const mail = managerMailOf(parsed.payload); const mailed = parsed.judgeMailed?.output.reviewer ?? parsed.reviewMailed?.output.reviewer; const report = doneReport(parsed); - const body = { ...mail.pullRequest, reviewers: reviewersAfter(parsed.take.output, mailed), ...(report && { report }), ...(parsed.approvals && { approvals: parsed.approvals.output }) }; + const wait = mailed !== undefined || report !== undefined || (parsed.approvals?.output.results.length ?? 0) > 0; + const body = { ...mail.pullRequest, reviewers: reviewersAfter(parsed.take.output, mailed), wait, ...(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); diff --git a/packages/triage-manager/src/actions/report.test.ts b/packages/triage-manager/src/actions/report.test.ts index 3f469187..e86ed475 100644 --- a/packages/triage-manager/src/actions/report.test.ts +++ b/packages/triage-manager/src/actions/report.test.ts @@ -3,8 +3,7 @@ import type { EffectContext } from "@intx/workflow"; import { report, reviewRequest, stop } from "./report.js"; const REQUEST = { kind: "code-review", number: 8, title: "Add the flag", description: "", diff: "diff --git a/f b/f", checkout: { source: "github", repo: "acme/widgets", sha: "h1" }, diffHash: "hash1" }; -const REPLY = { to: "run_manager", content: "{\"kind\":\"manage\"}" }; -const MAIL = { headers: { from: "run_manager@acme.test", to: ["run_review@acme.test"] }, parts: [{ contentType: "text/plain", text: JSON.stringify({ ...REQUEST, tenantId: "tnt_acme", reply: REPLY }) }] }; +const MAIL = { headers: { from: "run_manager@acme.test", to: ["run_review@acme.test"] }, parts: [{ contentType: "text/plain", text: JSON.stringify({ ...REQUEST, tenantId: "tnt_acme" }) }] }; const REVIEW = { summary: "No bugs.", comments: [], budgetExhausted: false }; /** Each effect runs once against a hub that records its calls and records the result. */ @@ -19,11 +18,11 @@ function hubContext() { return { ctx, calls }; } -test("a manager's code review hands its request to the review, reports the review and its diff's hash through the hub, replies to the manager and stops, posting nothing", async () => { +test("a manager's code review hands its request to the review, reports the review and its diff's hash through the hub and stops, posting nothing", async () => { const { ctx, calls } = hubContext(); - expect(await reviewRequest(MAIL)).toEqual({ request: { ...REQUEST, tenantId: "tnt_acme", reply: REPLY } }); - expect(await report({ payload: MAIL, review: { output: REVIEW } }, ctx, new AbortController().signal)).toEqual({ recorded: true, mail: { tenantId: "tnt_acme", ...REPLY } }); - expect(await report({ payload: MAIL }, ctx, new AbortController().signal)).toEqual({ recorded: true, mail: { tenantId: "tnt_acme", ...REPLY } }); + expect(await reviewRequest(MAIL)).toEqual({ request: { ...REQUEST, tenantId: "tnt_acme" } }); + expect(await report({ payload: MAIL, review: { output: REVIEW } }, ctx, new AbortController().signal)).toEqual({ recorded: true }); + expect(await report({ payload: MAIL }, ctx, new AbortController().signal)).toEqual({ recorded: true }); expect(await stop(MAIL, ctx, new AbortController().signal)).toEqual({ stopped: true }); const reported = { source: "github", repo: "acme/widgets", number: 8, kind: "code-review", head: "h1" }; expect(calls).toEqual([ diff --git a/packages/triage-manager/src/actions/report.ts b/packages/triage-manager/src/actions/report.ts index 0e1dd69c..1e28efd3 100644 --- a/packages/triage-manager/src/actions/report.ts +++ b/packages/triage-manager/src/actions/report.ts @@ -1,17 +1,15 @@ import type { EffectContext } from "@intx/workflow"; import { type } from "arktype"; import { CodeReviewOutput, REPO_SOURCES } from "@corbits/triage-contracts"; -import type { MailInput } from "../mail.js"; import { answered, callHub, endRun, mailOf, tenantPath } from "./hub.js"; -/** The manager's mail to a code review it deployed: the review request, the tenant and the mail to reply with. */ +/** The manager's mail to a code review it deployed: the review request and the tenant. */ const ReviewMail = type({ kind: "'code-review'", tenantId: "string > 0", number: "number.integer > 0", checkout: { source: type.enumerated(...REPO_SOURCES), repo: "string > 0", sha: "string > 0" }, "diffHash?": "string > 0", - reply: { to: "string > 0", content: "string > 0" }, }); function reviewMailOf(payload: unknown) { @@ -34,20 +32,19 @@ const NOT_FINISHED = "the review did not finish"; /** * Reports the review to the pull request's manager as a `StageResult` through the hub, or that it did not finish; the - * reply then wakes the manager. Posts nothing. Its input is every step's output beside the mail. + * hub then wakes the manager. Posts nothing. Its input is every step's output beside the mail. */ -export async function report(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ recorded: boolean; mail: MailInput }> { +export async function report(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ recorded: boolean }> { const parsed = ReportInput(input); if (parsed instanceof type.errors) throw new Error(`report: ${parsed.summary}`); - const { tenantId, number, checkout, diffHash, reply } = reviewMailOf(parsed.payload).mail; + const { tenantId, number, checkout, diffHash } = reviewMailOf(parsed.payload).mail; const review = CodeReviewOutput(parsed.review?.output); const outcome = review instanceof type.errors ? { failed: NOT_FINISHED } : { result: review, ...(diffHash && { diffHash }) }; const result = { source: checkout.source, repo: checkout.repo, number, kind: "code-review", head: checkout.sha, ...outcome }; - const { recorded } = answered(await callHub(ctx, "report", tenantPath(tenantId, "/manager/reviewer-result"), "POST", result), "report", Recorded); - return { recorded, mail: { tenantId, ...reply } }; + return answered(await callHub(ctx, "report", tenantPath(tenantId, "/manager/reviewer-result"), "POST", result), "report", Recorded); } -/** Stops the review's run once it has replied, whether or not the review, its report or its reply went through. */ +/** Stops the review's run once it has reported, whether or not the review or its report went through. */ export async function stop(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ stopped: true }> { const { mail, runId } = reviewMailOf(input); await endRun(ctx, mail.tenantId, runId); diff --git a/packages/triage-manager/src/actions/review.ts b/packages/triage-manager/src/actions/review.ts index 0f543aca..539aedf8 100644 --- a/packages/triage-manager/src/actions/review.ts +++ b/packages/triage-manager/src/actions/review.ts @@ -7,7 +7,7 @@ import { call, hostWrite, toolName } from "@corbits/triage-workflows/logic/host" import { codeReviewSummary } from "@corbits/triage-workflows/logic/render"; import { Mailed, type MailInput } from "../mail.js"; import { publishedFor } from "../release.js"; -import { deploy, managerMailOf, manageMail } from "./hub.js"; +import { deploy, managerMailOf } from "./hub.js"; import { Reviewer, storedReview, TakeOutput } from "./manager.js"; /** How long the manager waits for a code review it deployed, from its deploy to its report. */ @@ -126,8 +126,8 @@ export async function reviewScope(input: unknown): Promise<{ deploy: boolean; po const DeployInput = type({ payload: "unknown", take: { output: TakeOutput } }); /** - * Deploys code review from the host's published release; it is mailed the head's review request and, to reply with, the mail that wakes - * this manager. The mail waits for the new run until the review's deadline. + * Deploys code review from the host's published release; it is mailed the head's review request, and reports its result + * to the hub, which wakes the pull request's manager. The mail waits for the new run until the review's deadline. */ export async function deployReview(input: unknown, ctx: EffectContext, _signal: AbortSignal): Promise<{ reviewer: Reviewer; mail: MailInput }> { const parsed = DeployInput(input); @@ -140,7 +140,7 @@ export async function deployReview(input: unknown, ctx: EffectContext, _signal: const again = asRecord(taken["codeReview"])?.["again"]; const reviewRun = await deploy(ctx, "deploy", tenantId, release); const deadline = new Date(Date.now() + REVIEW_DEADLINE_MS).toISOString(); - const content = JSON.stringify({ ...asRecord(taken["codeReviewRequest"]), tenantId, reply: { to: runId, content: manageMail(tenantId, pullRequest) } }); + const content = JSON.stringify({ ...asRecord(taken["codeReviewRequest"]), tenantId }); return { reviewer: { kind: "code-review", head, runId: reviewRun, deployedBy: runId, deadline, ...(typeof again === "string" && { again }) }, mail: { tenantId, to: reviewRun, content, readyBy: deadline } }; } diff --git a/packages/triage-manager/src/triage-manager.test.ts b/packages/triage-manager/src/triage-manager.test.ts index 19de9be9..4b65778a 100644 --- a/packages/triage-manager/src/triage-manager.test.ts +++ b/packages/triage-manager/src/triage-manager.test.ts @@ -4,7 +4,7 @@ import { createActionCredentialsFactory, createCredentialProviderRegistry, creat import type { GrantRule } from "@intx/types/authz"; import { HUB_CREDENTIAL_HANDLE } from "@intx/types/credential-delivery"; import { runLocal, type ActionHandler, type LoopFn, type OnTriggerPrimitive, type WorkflowAuthorizeFn, type WorkflowDefinition } from "@intx/workflow"; -import { ACTION_KINDS, APPROVE_DO_KIND, emptyPack, WAITING_KIND, WAITING_LABEL, type ActionKind, type CheckPack } from "@corbits/triage-contracts"; +import { ACTION_KINDS, APPROVE_DO_KIND, emptyPack, VERDICT_LABELS, WAITING_KIND, WAITING_LABEL, type ActionKind, type CheckPack } from "@corbits/triage-contracts"; import type { Snapshot } from "@corbits/triage-workflows/evals/cases"; import { fixtureGithub, triggerMail } from "@corbits/triage-workflows/evals/serve"; import { evaluate } from "@corbits/triage-workflows/actions/evaluate"; @@ -28,8 +28,6 @@ const END = { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${RUN}` }; const DEPLOY = { method: "POST", path: `${TENANT_PATH}/workflows/deployments` }; const DEPLOYED = "run_deployed"; const STOP_REVIEW = { method: "DELETE", path: `${TENANT_PATH}/workflows/runs/${DEPLOYED}` }; -/** A code review an earlier manager deployed, which only that manager may stop. */ -const INHERITED = "run_inherited"; /** The eval fixture's head. */ const HEAD = "e0a1".repeat(10); @@ -82,7 +80,7 @@ const allow: WorkflowAuthorizeFn = async function allow() { type HubCall = { method: string; path: string; body?: unknown }; type HubAnswer = Record | typeof REFUSED; -/** The hub's manager routes, answering each take and done in turn, its deploy route and its stop-run route, which refuses to stop an inherited review. */ +/** The hub's manager routes, answering each take and done in turn, its deploy route and its stop-run route. */ function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[] }) { const calls: HubCall[] = []; function answer(next: HubAnswer | undefined): Response { @@ -94,7 +92,6 @@ function fakeHub({ takes, dones = [] }: { takes: HubAnswer[]; dones?: HubAnswer[ calls.push({ method, path: url.pathname, ...(init?.body !== undefined && { body: JSON.parse(String(init.body)) }) }); if (url.pathname === `${TENANT_PATH}/manager/take`) return answer(takes.shift()); if (url.pathname === `${TENANT_PATH}/manager/done`) return answer(dones.shift()); - if (method === "DELETE" && url.pathname === `${TENANT_PATH}/workflows/runs/${INHERITED}`) return Response.json({ error: { code: "forbidden" } }, { status: 403 }); if (method === "DELETE" && url.pathname.startsWith(`${TENANT_PATH}/workflows/runs/`)) return new Response(null, { status: 202 }); if (method === DEPLOY.method && url.pathname === DEPLOY.path) return Response.json({ id: DEPLOYED }, { status: 201 }); return Response.json({ error: { code: "not_found" } }, { status: 404 }); @@ -118,11 +115,19 @@ function actionCredentials(bindings: Bindings, host: Fetch, hub: (url: URL, init const loopFns = (ref: string) => (loops as Record)[ref]!; +/** A clock a minute further on at every reading, so a pass's wait has passed by the time it is timed. */ +function hurried() { + let readings = 0; + return function clock() { + return new Date(Date.now() + ++readings * 60_000); + }; +} + /** * Runs one mail's body; the mail agent's director runs its one tool, so the mail it would send is recorded and answered * `mailed` in the `{ reply, turn }` a live agent step outputs. */ -async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetch, hub: ReturnType["hub"], mailed: unknown[] = [], delivers = true) { +async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetch, hub: ReturnType["hub"], mailed: unknown[] = [], delivers = true, clock?: () => Date) { const body = (definition.steps.events as OnTriggerPrimitive).body; if (!("inline" in body)) throw new Error("the body is not inline"); const { terminalStatus, outputs } = await runLocal(body.inline, { @@ -138,6 +143,7 @@ async function runMail(definition: WorkflowDefinition, mail: unknown, host: Fetc actionCredentials: actionCredentials((definition.credentialBindings ?? []) as Bindings, host, hub), loopFns, triggerPayload: mail, + ...(clock && { clock }), }).complete; const passes = outputs["passes"] as { iterations: number; final: Record } | undefined; const last = passes?.final ?? outputs; @@ -242,9 +248,19 @@ describe("triage-manager", () => { expect(live.calls).toEqual([TAKE, DONE, TAKE, DONE, END]); }); + test("a pass that posted asks the hub to wait, and the pass after the wait, which runs nothing, ends the run", async () => { + const { host } = githubHost(); + const hub = fakeHub({ takes: [PASS, IDLE], dones: [{ outcome: "more", wait: true }, ENDED] }); + const { passes, ran } = await runMail(github, wake(), host, hub.hub, [], true, hurried()); + expect(passes).toBe(2); + expect(ran).toEqual(["approvals", "done", "paused", "poll", "ran", "take"]); + expect(hub.calls.filter((call) => call.path === DONE.path).map((call) => (call.body as { wait: boolean }).wait)).toEqual([true, false]); + expect(hub.calls.at(-1)).toEqual(END); + }); + test("passes that run out are stopped once, by the run's last step", async () => { const { host } = githubHost(); - const passes = 280; + const passes = 40; const hub = fakeHub({ takes: Array.from({ length: passes }, () => IDLE), dones: Array.from({ length: passes }, () => MORE) }); expect(await runMail(github, wake(), host, hub.hub)).toMatchObject({ terminalStatus: "completed", stop: { stopped: true }, passes }); expect(hub.calls.filter((call) => call.method === "DELETE")).toEqual([END]); @@ -270,20 +286,13 @@ describe("triage-manager", () => { 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 () => { + test("an old review's reply finds nothing for its head, runs no stage 1 and leaves the superseded review for the hub to stop", 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(["approvals", "done", "paused", "ran", "take"]); expect(writes).toEqual([]); - expect(hub.calls).toEqual([TAKE, DONE, STOP_REVIEW, END]); - }); - - test("a review an earlier manager deployed is left to run out its own deadline", async () => { - const { host } = githubHost(); - const inherited = { ...reviewer(), head: "older", runId: INHERITED, deployedBy: "run_earlier" }; - const hub = fakeHub({ takes: [{ ...IDLE, request: { ...REQUEST, headSha: "newer" }, reviewers: [inherited] }], dones: [ENDED] }); - expect((await runMail(github, wake(), host, hub.hub)).terminalStatus).toBe("completed"); expect(hub.calls).toEqual([TAKE, DONE, END]); + expect(doneBody(hub.calls)["wait"]).toBe(false); }); test("a pass of an approval alone, before any pull request mail was recorded, runs it and triages nothing", async () => { @@ -368,7 +377,8 @@ describe("triage-manager with code review", () => { expect(first.calls[2]).toMatchObject({ ...DONE, body: { reviewers: [{ kind: "code-review", head: HEAD, runId: DEPLOYED, deployedBy: RUN }] } }); const [mail] = mailed as [{ tenantId: string; to: string; content: string; readyBy: string }]; expect(Date.parse(mail.readyBy)).toBeGreaterThan(Date.now()); - expect({ ...mail, content: JSON.parse(mail.content) }).toMatchObject({ tenantId: TENANT, to: DEPLOYED, content: { ...REVIEW_REQUEST.codeReviewRequest, tenantId: TENANT, reply: { to: RUN, content: manageMail(TENANT, PULL) } } }); + expect({ ...mail, content: JSON.parse(mail.content) }).toEqual({ tenantId: TENANT, to: DEPLOYED, readyBy: mail.readyBy, content: { ...REVIEW_REQUEST.codeReviewRequest, tenantId: TENANT } }); + expect(doneBody(first.calls)["wait"]).toBe(true); const second = fakeHub({ takes: [reported({ result: REVIEW }, ON)], dones: [ENDED] }); const { ran } = await runMail(reviewed, wake(), host, second.hub, mailed); @@ -601,10 +611,11 @@ describe("triage-manager with a judge", () => { expect(first.calls[1]).toMatchObject({ ...DEPLOY, body: { source: { kind: "asset", assetId: "ast_judge", package: { format: "source", commitSha: "c0ffee", packageName: JUDGE_RELEASE.workflow.packageName } }, entry: JUDGE_RELEASE.workflow.entry, sourceOfferingIds: ["off_d"], defaultSourceOfferingId: "off_d" } }); expect(first.calls[2]).toMatchObject({ ...DONE, body: { reviewers: [{ kind: "judge", head: HEAD, runId: DEPLOYED, deployedBy: RUN }] } }); const [mail] = mailed as [{ tenantId: string; to: string; content: string; readyBy: string }]; - const content = JSON.parse(mail.content) as { kind: string; rules: MailedRules; reply: { to: string } }; + const content = JSON.parse(mail.content) as { kind: string; rules: MailedRules }; expect(content.kind).toBe("judge"); expect(content.rules.chunks.length).toBeGreaterThan(0); - expect(content.reply.to).toBe(RUN); + expect(content).not.toHaveProperty("reply"); + expect(doneBody(first.calls)["wait"]).toBe(true); expect(Date.parse(mail.readyBy)).toBeGreaterThan(Date.now()); const verdict = await judgeVerdict(content.rules); @@ -718,7 +729,7 @@ describe("triage-manager with a judge", () => { const { ran } = await runMail(github, wake(), host, hub.hub); expect(ran).toContain("mirrorRules"); expect(ran).not.toContain("deployJudge"); - expect(writes.length).toBeGreaterThan(0); + expect(writes).toContainEqual(expect.objectContaining({ path: `/repos/${PR.repo}/issues/${PR.number}/labels`, body: { labels: [VERDICT_LABELS.unsure] } })); expect(writes).not.toContainEqual(JUDGE_POSTED); expect(hub.calls).toEqual(calls); }); diff --git a/packages/triage-manager/src/triage-manager.ts b/packages/triage-manager/src/triage-manager.ts index 203d4a1d..e8a4fb7d 100644 --- a/packages/triage-manager/src/triage-manager.ts +++ b/packages/triage-manager/src/triage-manager.ts @@ -5,17 +5,15 @@ import { credentialBindingsFor, judgeAgent, type Sources } from "@corbits/triage import { WRITE_PULL_REQUEST } from "@corbits/triage-workflows/logic/host"; import { hostAction, triageSteps } from "@corbits/triage-workflows/triage-steps"; import { HUB_WRITE } from "./actions/hub.js"; -import { JUDGE_DEADLINE_MS } from "./actions/judge.js"; -import { REVIEW_DEADLINE_MS } from "./actions/review.js"; import { mailDirector } from "./directors.js"; import { mailAgentOf } from "./mail.js"; import { deployGrants, type Release } from "./release.js"; -/** How often a manager waiting for its code review asks the hub whether the review reported. */ +/** How long a manager whose pass posted to the host or mailed a reviewer waits for what that brings before it ends. */ const POLL_MS = 30_000; -/** Enough passes to wait out a judge and then a code review, and the events that arrive meanwhile. */ -const PASSES = Math.ceil((JUDGE_DEADLINE_MS + REVIEW_DEADLINE_MS) / POLL_MS) + 40; +/** Enough passes for the events that arrive while the manager runs; it waits only after a pass that posted or mailed. */ +const PASSES = 40; const PAYLOAD = { project: { from: "trigger" }, fields: ["payload"] } as const; @@ -84,7 +82,7 @@ function judgeSteps(): Record { * One pass: take claims the pull request and its pending events. Each held Do approved since runs; then, when the head * is triaged, pr-triage's own steps run on the newest mail the hub recorded for it. A head held for CI is marked * waiting. A policy of only deterministic rules is triaged on this pass. One that asks the judge or for code review - * posts once both have reported or given up. A pass after one that waits for a reviewer first waits `POLL_MS`. + * posts once both have reported or given up. A pass after one that posted or mailed a reviewer first waits `POLL_MS`. */ function passOf(id: string, sources: Sources) { const { waiting, held, markWaiting, unready, rules, mirrorRules, executeRules } = triageSteps(sources); @@ -124,10 +122,11 @@ function passOf(id: string, sources: Sources) { /** * A pull request's manager under `id`, also its mail address, for the given hosts: the only workflow that writes to the * pull request, one live run per pull request, which runs only while it has work. A mail wakes it to run passes until - * done ends the claim or another run holds the claim, after which its last step stops it. When the host's release - * publishes a code review or a judge, each runs as its own deployment, which reports back; `codeReview` and `judge`, the - * ones published as the definition loads, name the assets it declares the grants to deploy. Mail comes from the hub, - * the listener that deployed it, and the judge and the reviews it mailed. + * done ends the claim or another run holds the claim, after which its last step stops it; once a pass posted to the + * host or mailed a reviewer, it waits `POLL_MS` for what that brings and then ends. When the host's release publishes a + * code review or a judge, each runs as its own deployment, which reports its result to the hub; the hub wakes the + * manager then, or has the listener deploy a new one. `codeReview` and `judge`, the ones published as the definition + * loads, name the assets it declares the grants to deploy. Mail comes from the hub and the listener that deployed it. */ export function triageManager({ id, sources, codeReview, judge }: { id: string; sources: Sources; codeReview: Release | undefined; judge?: Release }) { const perMail = defineWorkflow({ @@ -158,7 +157,7 @@ export function triageManager({ id, sources, codeReview, judge }: { id: string; ...(codeReview ? deployGrants(codeReview) : []), ...(judge ? deployGrants(judge) : []), ], - mailAccept: { invoker: true, correspondent: true, tenant: true }, + mailAccept: { invoker: true, tenant: true }, steps: { events: onTrigger({ on: { type: "mail", to: id }, body: perMail, onBodyFailure: "tolerate" }), }, diff --git a/tests/chain/chain.test.ts b/tests/chain/chain.test.ts index 6dc2153b..acdd150e 100644 --- a/tests/chain/chain.test.ts +++ b/tests/chain/chain.test.ts @@ -24,6 +24,7 @@ const JUDGED = `${OWNER}/judged`; const WAITING = `${OWNER}/waiting`; const PUSHED = `${OWNER}/pushed`; const HELD = `${OWNER}/held`; +const WOKEN = `${OWNER}/woken`; // Interchange's `liveWorkflowRunStatuses`: an anchor is born `deployed` and flips to `running` on its first trigger. const LIVE_DEPLOYMENT_STATUSES = new Set(["deployed", "running"]); @@ -61,6 +62,7 @@ beforeAll(async () => { github.addRepo(WAITING, policy([CI_GREEN])); github.addRepo(PUSHED, policy([MAX_FILES])); github.addRepo(HELD, policy([MAX_FILES], ASSIGN_ON_PASS)); + github.addRepo(WOKEN, policy([MAX_FILES, FOCUSED])); const [created] = await Promise.all([createDatabase(), buildPortal(portalDir)]); database = created; hub = await startHub(database, github, portalDir); @@ -323,3 +325,40 @@ test("a held Do approved through the hub runs once on the pull request", async ( await settled(HELD, 1, 2); expect(github.writesTo(HELD, 1).filter((write) => write.path.endsWith("/assignees"))).toEqual([expect.objectContaining({ method: "POST", body: { assignees: ["octocat"] } })]); }); + +const woken = once(async function woken() { + const head = sha("woken-1"); + await after(approved); + const [before, asked, missed] = [(await managers()).length, model.decisions.length, github.misses.length]; + const release = model.hold(); + let releasedAt: Date; + let ended: Claim; + let first: Awaited>; + try { + await pushed(WOKEN, pullOf(1, head)); + // The judge's answer is held, so its manager mails it, waits out its 30 seconds and ends with the judge still out. + ended = await settled(WOKEN, 1); + first = (await managers()).slice(before); + } finally { + releasedAt = new Date(); + release(); + } + const claim = await settled(WOKEN, 1, ended.history.length + 1); + return { head, ended, first, releasedAt, claim, deployed: (await managers()).slice(before), asked: model.decisions.slice(asked).map((row) => row.id), writes: github.writesTo(WOKEN, 1), misses: github.misses.slice(missed) }; +}); + +test("a manager ends after it mails the judge, and the judge's report wakes a fresh manager through the hub, which posts the verdict", async () => { + const { head, ended, first, releasedAt, claim, deployed, asked, writes, misses } = await woken(); + expect(misses).toEqual([]); + expect(ended.status).toBe("ended"); + expect(first).toHaveLength(1); + expect(isLiveDeployment(first[0]!.status)).toBe(false); + expect(deployed).toHaveLength(2); + expect(deployed[1]!.createdAt.getTime()).toBeGreaterThanOrEqual(releasedAt.getTime() - 1_000); + expect(claim).toMatchObject({ status: "ended" }); + expect(claim.history.at(-1)).toMatchObject({ head, runId: deployed[1]!.id }); + expect(asked).toContain("focused"); + expect(comments(writes)).toEqual([expect.stringContaining("**Triage: Pass**")]); + expect(labelled(writes)).toEqual([VERDICT_LABELS.pass]); + expect(await verdictsOf(WOKEN, 1)).toEqual([expect.objectContaining({ run_id: deployed[1]!.id, head_sha: head, run_status: "completed" })]); +}); diff --git a/tests/chain/model.ts b/tests/chain/model.ts index 780621bc..ac28907f 100644 --- a/tests/chain/model.ts +++ b/tests/chain/model.ts @@ -1,12 +1,24 @@ // The workspace's decision model, served locally on System One's wire: it answers each question from the test's script -// and keeps what it was asked. +// and keeps what it was asked. While held, it answers nothing until released. export type Question = { type: "noul" | "choice" | "score"; instructions: string; criteria?: unknown }; export type Answer = { type: "noul"; noul: number } | { type: "choice"; choice: string; probabilities: Record; confidence: number }; export function stubModel(decide: (id: string, question: Question) => Answer) { const decisions: { id: string; question: Question }[] = []; + let held: Promise | undefined; + + /** Holds every answer until the returned release is called. */ + function hold(): () => void { + const { promise, resolve } = Promise.withResolvers(); + held = promise; + return function release() { + held = undefined; + resolve(); + }; + } async function systemOne(req: Request): Promise { + await held; const { questions } = await req.json() as { questions: Record }; const answers = Object.fromEntries(Object.entries(questions).map(function answer([id, question]) { decisions.push({ id, question }); @@ -24,6 +36,7 @@ export function stubModel(decide: (id: string, question: Question) => Answer) { return { origin: `http://127.0.0.1:${server.port}`, decisions, + hold, stop(): void { server.stop(true); },