fix(hub): route every pull request through the listener

.env.example

@@ -46,8 +46,6 @@ HUB_DATA_DIR=./tmp/hub-data
 # Milliseconds between passes that re-queue open pull requests whose head is not triaged. Default 300000.
 # TRIAGE_RECONCILE_INTERVAL_MS=
 
-# Runs the pr-triage deployment's log may hold before the hub deploys a fresh copy of it. Default 15.
-# TRIAGE_ROTATE_AFTER_RUNS=
 # Pull request heads the reconciler lets a tenant have queued or running at once, each in a sidecar of its own. Default 5.
 # TRIAGE_MAX_IN_FLIGHT=
 # Milliseconds after a head is capped by runs that never started before it gets a fresh set of attempts. Default 21600000.

apps/hub/src/code-review-feed.test.ts

@@ -41,16 +41,12 @@ test("each new non-draft head is sent to its host's code review once; Review aga
     source: "event",
     payload: { kind: "code-review", number: 7, title: "Fix", description: "Body", diff: PULL.diff, diffHash: diffHash(PULL.diff), checkout: { source: "github", repo: "acme/widgets", sha: "sha1" } },
   }]);
-  expect(await feed.again("t", RECORD, 7)).toBe("sent");
-  expect(sent.map((job) => job.source)).toEqual(["event", "retry"]);
 });
 
-test("a head that takes the newest review over is not sent, and Review again sends it anyway", async () => {
+test("a head that takes the newest review over is not sent", async () => {
   const { feed, sent } = harness(["github"], PULL, true);
   await feed.offer("t", RECORD, mail("synchronize"));
   expect(sent).toEqual([]);
-  expect(await feed.again("t", RECORD, 7)).toBe("sent");
-  expect(sent.map((job) => job.source)).toEqual(["retry"]);
 });
 
 test("a diff's hash changes with any blob id, and a diff with none, as GitHub rebuilds from the file list, has none and is sent unhashed", async () => {
@@ -64,9 +60,11 @@ test("a diff's hash changes with any blob id, and a diff with none, as GitHub re
 });
 
 test("nothing is sent while code review is off, the key is missing or the pull request turned draft", async () => {
-  expect(await harness([]).feed.again("t", RECORD, 7)).toBe("not_reviewed");
-  expect(await harness().feed.again("t", { ...RECORD, codeReview: false }, 7)).toBe("not_reviewed");
-  expect(await harness(["github"], { ...PULL, draft: true }).feed.again("t", RECORD, 7)).toBe("draft");
+  for (const [reviewed, record, pull] of [[[], RECORD, PULL], [["github"], { ...RECORD, codeReview: false }, PULL], [["github"], RECORD, { ...PULL, draft: true }]] as const) {
+    const { feed, sent } = harness([...reviewed], pull);
+    await feed.offer("t", record, mail("synchronize"));
+    expect(sent).toEqual([]);
+  }
 });
 
 test("a diff over the limit is cut at a line boundary and names the files it cut part-way or left out", () => {
@@ -95,7 +93,7 @@ test("a cut names files as git writes them with or without prefixes and quotes,
 
 test("a host answer that is not a unified diff is refused, not reviewed", async () => {
   const { feed, sent } = harness(["github"], { ...PULL, diff: "{\"url\":\"…\"}" });
-  await expect(feed.again("t", RECORD, 7)).rejects.toThrow("the host returned no unified diff");
+  await feed.offer("t", RECORD, mail("synchronize"));
   expect(sent).toEqual([]);
 });
 

apps/hub/src/code-review-feed.ts

@@ -42,7 +42,7 @@ export type CodeReviewFeedDeps = {
   log: (entry: Record<string, unknown>) => void;
 };
 
-export type Offered = "sent" | "not_reviewed" | "draft" | "not_open";
+type Unreviewed = "not_reviewed" | "draft" | "not_open";
 
 /**
  * The diff's hash, its `index` lines' blob ids included, so a base change to a file the diff touches changes it; undefined
@@ -172,7 +172,7 @@ export function createCodeReviewFeed(deps: CodeReviewFeedDeps) {
   }
 
   /** The pull request to review and its request, or why it is not reviewed. */
-  async function toReview(tenantId: string, record: RepoRecord, number: number): Promise<{ pull: PullRequestToReview; request: ReviewRequest } | Exclude<Offered, "sent">> {
+  async function toReview(tenantId: string, record: RepoRecord, number: number): Promise<{ pull: PullRequestToReview; request: ReviewRequest } | Unreviewed> {
     if (record.codeReview !== true || !(await deps.reviewed(tenantId, record)).includes(sourceOf(record))) return "not_reviewed";
     const pull = await deps.read(tenantId, record, number);
     if (!pull.open) return "not_open";
@@ -181,15 +181,15 @@ export function createCodeReviewFeed(deps: CodeReviewFeedDeps) {
   }
 
   /** Sends the reserved head, which is let go when the send fails. */
-  async function send(tenantId: string, record: RepoRecord, number: number, request: ReviewRequest, again: boolean): Promise<void> {
+  async function send(tenantId: string, record: RepoRecord, number: number, request: ReviewRequest): Promise<void> {
     const headSha = request.checkout.sha;
     try {
-      await deps.send({ tenantId, workflow: codeReviewWorkflowFor(sourceOf(record)).id, repo: record.name, number, headSha, source: again ? "retry" : "event", payload: request });
+      await deps.send({ tenantId, workflow: codeReviewWorkflowFor(sourceOf(record)).id, repo: record.name, number, headSha, source: "event", payload: request });
     } catch (err) {
       sent.delete(keyOf(tenantId, record, number, headSha));
       throw err;
     }
-    deps.log({ level: "info", msg: "code_review_sent", tenantId, repo: record.name, pr: number, headSha, again, ...(request.cut && { cut: request.cut }) });
+    deps.log({ level: "info", msg: "code_review_sent", tenantId, repo: record.name, pr: number, headSha, ...(request.cut && { cut: request.cut }) });
   }
 
   /** A new head is sent once, unless it takes the newest review over. */
@@ -202,7 +202,7 @@ export function createCodeReviewFeed(deps: CodeReviewFeedDeps) {
       deps.log({ level: "info", msg: "code_review_carried", tenantId, repo: record.name, pr: number, headSha: pull.headSha });
       return;
     }
-    await send(tenantId, record, number, request, false);
+    await send(tenantId, record, number, request);
   }
 
   return {
@@ -217,14 +217,6 @@ export function createCodeReviewFeed(deps: CodeReviewFeedDeps) {
         deps.log({ level: "error", msg: "code_review_send_failed", tenantId, repo: record.name, pr: mail.prNumber, error: String(err) });
       }
     },
-    /** "Review again": the pull request's current head, even when it was reviewed already or took a review over. */
-    async again(tenantId: string, record: RepoRecord, number: number): Promise<Offered> {
-      const found = await toReview(tenantId, record, number);
-      if (typeof found === "string") return found;
-      reserve(keyOf(tenantId, record, number, found.pull.headSha));
-      await send(tenantId, record, number, found.request, true);
-      return "sent";
-    },
     /** A queued review whose worker could not run, so a later event for its head sends it again. */
     forget(job: TriageJob): void {
       const host = CODE_REVIEW_WORKFLOWS.find((workflow) => workflow.id === job.workflow)?.source;

apps/hub/src/code-review.postgres.test.ts

@@ -6,9 +6,9 @@ import { createDB, runMigrations, schema } from "@intx/db";
 import { createSidecarPluginRegistry, type SidecarProvisioner } from "@intx/hub-sessions";
 import type { SidecarCapabilityDeclaration } from "@intx/types";
 import { DOCKER_PLACE, DOCKER_REMOTE_PLACE } from "@corbits/provisioner-docker";
-import type { CodeReviewPlace } from "@corbits/triage-contracts";
+import { emptyPack, type CodeReviewPlace } from "@corbits/triage-contracts";
 import { createCodeReviewPlace, createCodeReviewPlacement, createCodeReviewRun, createCodeReviewSwitch } from "./code-review.js";
-import type { Offered } from "./code-review-feed.js";
+import { NoLiveDeploymentError } from "./github/deployment.js";
 import { codeReviewSources } from "./github/source-workflow.js";
 import { repoRecords, triageNs } from "./github/tenant-config.js";
 
@@ -102,32 +102,38 @@ test("PostgreSQL: code review turns on only with an isolated place to run, the h
     expect(repoRecords(triageNs(tenant?.config)).map((record) => [record.name, record.codeReview])).toEqual([["acme/widgets", false], ["TRI/widgets", undefined], ["acme/gone", undefined]]);
     expect(changed).toEqual([TENANT, TENANT]);
 
-    const outcomes: Array<Offered | Error> = ["sent", "draft", new Error("github GET /repos/acme/widgets/pulls/7 -> 404")];
-    const runCodeReview = createCodeReviewRun({
+    await db.update(schema.tenant).set({ config: { corbitsTriage: { repos: repos.map((record) => ({ ...record, enabled: true })) } } }).where(eq(schema.tenant.id, TENANT));
+    const heads: Array<{ open: boolean; draft: boolean } | Error> = [{ open: true, draft: false }, { open: true, draft: true }, { open: false, draft: false }, new Error("github GET /repos/acme/widgets/pulls/7 -> 404")];
+    const sent: unknown[] = [];
+    const runDeps = {
       db,
       authorize: async () => true,
-      feed: {
-        async again() {
-          const next = outcomes.shift()!;
-          if (next instanceof Error) throw next;
-          return next;
-        },
-      },
       place,
-      inPrWorkflow: async () => false,
-    });
-    const runNowhere = createCodeReviewRun({ db, authorize: async () => true, feed: { again: async () => "sent" }, place: nowhere, inPrWorkflow: async () => false });
-    const runDecided = createCodeReviewRun({ db, authorize: async () => true, feed: { again: async () => "sent" }, place, inPrWorkflow: async () => true });
-    expect(((await (await runDecided({ repo: "acme/widgets", number: 7 }, MEMBER)).json()) as { error: { code: string } }).error.code).toBe("policy_file_decides");
+      readCheckPack: async () => ({ status: "ok" as const, pack: { ...emptyPack("acme/widgets"), checks: { "no-merge-conflicts": { enabled: true } } }, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }),
+      async readHead() {
+        const next = heads.shift()!;
+        if (next instanceof Error) throw next;
+        return { title: "Fix", description: "", headSha: "sha1", ...next };
+      },
+      async send(_tenantId: string, payload: unknown) {
+        sent.push(payload);
+      },
+    };
+    const runCodeReview = createCodeReviewRun(runDeps);
+    const runNowhere = createCodeReviewRun({ ...runDeps, place: nowhere });
+    const runUnlistened = createCodeReviewRun({ ...runDeps, readHead: async () => ({ title: "Fix", description: "", headSha: "sha1", open: true, draft: false }), send: async () => Promise.reject(new NoLiveDeploymentError("triage-listener-github")) });
     expect((await runNowhere({ repo: "acme/widgets", number: 7 }, MEMBER)).status).toBe(409);
-    async function run(repo: string) {
-      const response = await runCodeReview({ repo, number: 7 }, MEMBER);
+    async function run(repo: string, runner = runCodeReview) {
+      const response = await runner({ repo, number: 7 }, MEMBER);
       return [response.status, ((await response.json()) as { status?: string; error?: { code: string } }).error?.code ?? "queued"];
     }
     expect(await run("acme/missing")).toEqual([404, "repo_not_found"]);
     expect(await run("acme/widgets")).toEqual([202, "queued"]);
+    expect(sent).toEqual([expect.objectContaining({ kind: "pr", prNumber: 7, headSha: "sha1", noWait: true, codeReview: { on: true, for: ["everyone"], triage: ["pass", "fail", "unsure"], rules: "any", again: expect.any(String) } })]);
     expect(await run("acme/widgets")).toEqual([409, "draft"]);
+    expect(await run("acme/widgets")).toEqual([409, "not_open"]);
     expect(await run("acme/widgets")).toEqual([502, "code_review_unavailable"]);
+    expect(await run("acme/widgets", runUnlistened)).toEqual([503, "service_not_running"]);
   } finally {
     await handle?.close();
     await admin.db.execute(sql.raw(`drop database if exists ${database} with (force)`));

apps/hub/src/code-review.ts

@@ -1,14 +1,19 @@
+import { randomUUID } from "node:crypto";
 import { type } from "arktype";
 import { resolveTenantSidecarCapabilityPolicies, type DB } from "@intx/db";
 import type { SidecarPluginRegistry } from "@intx/hub-sessions";
 import { CODE_REVIEW_PLACEMENT } from "@corbits/pr-code-review/pr-code-review";
-import { REPO_SOURCES, type CodeReviewPlace, type CodeReviewPlaceOption, type PlaceGuide, type RepoSource } from "@corbits/triage-contracts";
+import { errorText } from "@corbits/common";
+import { EVERYONE, REPO_SOURCES, triageStopped, VERDICTS, type CodeReviewPlace, type CodeReviewPlaceOption, type PlaceGuide, type RepoSource } from "@corbits/triage-contracts";
 import { githubLimited } from "./github/github-refusal.js";
 import { failure } from "./github/portal-credential.js";
 import { connectedSources, hasCodeReviewKey } from "./github/source-workflow.js";
-import { tenantConfigOf } from "./github/bridge.js";
+import { mailPayload, tenantConfigOf } from "./github/bridge.js";
 import { patchCorbitsTriage, repoRecords, sourceOf, triageNs, type RepoRecord } from "./github/tenant-config.js";
-import type { CodeReviewFeed, Offered } from "./code-review-feed.js";
+import type { PullRequestHead } from "./code-review-feed.js";
+import type { PolicyReader } from "./github/check-pack-store.js";
+import { NoLiveDeploymentError } from "./github/deployment.js";
+import { triageInputs } from "./github/pr-triage.js";
 import type { Member } from "./integrations/app.js";
 
 export const CODE_REVIEW_PATH = "/api/integrations/code-review";
@@ -56,12 +61,6 @@ export function createCodeReviewPlace(deps: CodeReviewPlaceDeps) {
   };
 }
 
-const REFUSED: Record<Exclude<Offered, "sent">, { code: string; message: string }> = {
-  not_reviewed: { code: "code_review_off", message: "Code review is off for this repository, or the Anthropic key is missing." },
-  draft: { code: "draft", message: "The pull request is a draft; code review starts once it is ready." },
-  not_open: { code: "not_open", message: "The pull request is not open." },
-};
-
 type CodeReviewDeps = {
   db: DB["db"];
   authorize: (principalId: string, tenantId: string, resource: string, action: string) => Promise<boolean>;
@@ -116,26 +115,42 @@ export function createCodeReviewSwitch(deps: CodeReviewDeps) {
   };
 }
 
-type CodeReviewRunDeps = Pick<CodeReviewDeps, "db" | "authorize" | "place" | "inPrWorkflow"> & { feed: Pick<CodeReviewFeed, "again"> };
+type CodeReviewRunDeps = Pick<CodeReviewDeps, "db" | "authorize" | "place"> & {
+  readCheckPack: PolicyReader;
+  /** The pull request as its host shows it now, without its diff. */
+  readHead: (tenantId: string, record: RepoRecord, number: number) => Promise<PullRequestHead>;
+  /** Hands the mail to the host's listener, whose manager deploys the review. */
+  send: (tenantId: string, payload: unknown) => Promise<void>;
+};
+
+/** Code review as "Review again" asks for it: on, whatever the policy's own scope says, and named as a review of its own. */
+function reviewNow() {
+  return { on: true, for: [EVERYONE], triage: [...VERDICTS], rules: "any", again: randomUUID() };
+}
 
-/** "Review again": reviews the pull request's current head once more. */
+/** "Review again": triages the pull request's current head with code review on, so its manager reviews it once more. */
 export function createCodeReviewRun(deps: CodeReviewRunDeps) {
   return async function runCodeReview(body: typeof CodeReviewRunBody.infer, { tenantId, principalId }: Member): Promise<Response> {
     if (!(await deps.authorize(principalId, tenantId, `tenant:${tenantId}`, "manage"))) {
       return failure(403, "forbidden", "You do not have permission to manage this workspace.");
     }
     if ((await deps.place(tenantId)) === null) return noPlace();
+    if (!(await hasCodeReviewKey(deps.db, tenantId))) return failure(409, "anthropic_key_missing", "Save an Anthropic key in Settings before reviewing again.");
     const source = body.source ?? "github";
     const record = repoRecords(triageNs(await tenantConfigOf(deps, tenantId))).find((row) => row.name === body.repo && sourceOf(row) === source && row.connected);
     if (!record) return failure(404, "repo_not_found", "The workspace has no such connected repository.");
-    if (await deps.inPrWorkflow(tenantId, record.name, source)) return fileDecides("Press Triage again to review the latest commit.");
-    let offered: Offered;
+    const inputs = await triageInputs(deps, tenantId, record);
+    if (inputs instanceof Response) return inputs;
     try {
-      offered = await deps.feed.again(tenantId, record, body.number);
+      const pull = await deps.readHead(tenantId, record, body.number);
+      if (!pull.open) return failure(409, "not_open", "The pull request is not open.");
+      if (pull.draft) return failure(409, "draft", "The pull request is a draft; code review starts once it is ready.");
+      const payload = mailPayload(record, inputs.policy, inputs.pack, { prNumber: body.number, headSha: pull.headSha, noWait: true });
+      await deps.send(tenantId, { ...payload, codeReview: reviewNow() });
     } catch (err) {
-      return githubLimited(err) ?? failure(502, "code_review_unavailable", `Could not send the pull request to code review: ${err instanceof Error ? err.message : String(err)}`);
+      if (err instanceof NoLiveDeploymentError) return failure(503, "service_not_running", triageStopped(source));
+      return githubLimited(err) ?? failure(502, "code_review_unavailable", `Could not send the pull request to code review: ${errorText(err)}`);
     }
-    if (offered !== "sent") return failure(409, REFUSED[offered].code, REFUSED[offered].message);
     return Response.json({ status: "queued" }, { status: 202 });
   };
 }

apps/hub/src/env.ts

@@ -70,7 +70,6 @@ const HubEnvSchema = type({
   "RAILWAY_GIT_REPO_NAME?": "string",
   "RAILWAY_ENVIRONMENT_ID?": "string",
   "TRIAGE_RECONCILE_INTERVAL_MS?": positiveInteger,
-  "TRIAGE_ROTATE_AFTER_RUNS?": positiveInteger,
   "TRIAGE_MAX_IN_FLIGHT?": positiveInteger,
   "TRIAGE_CAPPED_RETRY_MS?": positiveInteger,
   "TRIAGE_COALESCE_QUIET_MS?": positiveInteger,
@@ -255,15 +254,6 @@ export function triageReconcileIntervalMs(env: HubEnv): number {
   return env.TRIAGE_RECONCILE_INTERVAL_MS === undefined ? 5 * 60_000 : Number(env.TRIAGE_RECONCILE_INTERVAL_MS);
 }
 
-/**
- * The stock hub checks every commit pushed to a deployment's run log against
- * the whole log, so the hub replaces the pr-triage deployment with a fresh copy
- * once its log holds more runs than this.
- */
-export function triageRotateAfterRuns(env: HubEnv): number {
-  return numberOr(env.TRIAGE_ROTATE_AFTER_RUNS, 15);
-}
-
 /** Heads the reconciler lets a tenant have queued or running at once; each runs in a worker sidecar of its own. */
 export function triageReconcilePolicy(env: Pick<HubEnv, "TRIAGE_MAX_IN_FLIGHT" | "TRIAGE_CAPPED_RETRY_MS">): ReconcilePolicy {
   return {