diff --git a/apps/hub/src/github/deployment-heal.postgres.test.ts b/apps/hub/src/github/deployment-heal.postgres.test.ts index 103268bb3..c1854987c 100644 --- a/apps/hub/src/github/deployment-heal.postgres.test.ts +++ b/apps/hub/src/github/deployment-heal.postgres.test.ts @@ -145,6 +145,7 @@ function harness(db: ReturnType["db"]) { rotationKey: "rotation", tenants: async () => [{ id: TENANT, domain: DOMAIN, config: CONFIG }], liveDeployments, + listener: async () => undefined, latestDeployments: (tenantId, limit) => latestDeployments(db, tenantId, WORKFLOW, limit), rotation: { afterRuns: 1_000, diff --git a/apps/hub/src/github/interleave.postgres.test.ts b/apps/hub/src/github/interleave.postgres.test.ts index 30694a450..2374c7d4d 100644 --- a/apps/hub/src/github/interleave.postgres.test.ts +++ b/apps/hub/src/github/interleave.postgres.test.ts @@ -51,6 +51,7 @@ function route(store: TriageStateStore, deliver: Deliver) { pullHead: async () => "abc123", readCheckPack: async () => ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }), liveDeployments: async () => [DEPLOYMENT], + listener: async () => undefined, observeRuns: noRuns, managerRuns: noRuns, store, @@ -67,6 +68,7 @@ function reconciler(store: TriageStateStore, heads: number[], deliver: Deliver, rotationKey: "rotation", tenants: async () => [{ id: TENANT, domain: "tenant.example", config: CONFIG }], liveDeployments: async () => [DEPLOYMENT], + listener: async () => undefined, latestDeployments: async () => [], rotation: { afterRuns: 15, redeploy: async () => ({ status: "skipped", reason: "unused" }), sameSource: async () => false, release: async () => {}, claim: async () => false, clear: async () => {} }, openHeadsFor: async () => async () => heads.map((number) => ({ number, headSha: number === 8 ? "abc123" : `sha${number}`, updatedAt: LONG_AGO, draft: false })), diff --git a/apps/hub/src/github/pr-triage.test.ts b/apps/hub/src/github/pr-triage.test.ts index 055a25395..01e28a312 100644 --- a/apps/hub/src/github/pr-triage.test.ts +++ b/apps/hub/src/github/pr-triage.test.ts @@ -29,7 +29,7 @@ function stubDb(repo: Record) { } as never; } -function observed(runs: ObservedRun[]): GithubPrTriageDeps["observeRuns"] { +function observed(runs: ObservedRun[]): GithubPrTriageDeps["observeRuns"] & GithubPrTriageDeps["managerRuns"] { return async function observeRuns() { return { byRepo: new Map([[REPO, new Map([["8@abc123", runs]])]]), runCount: runs.length, verdicts: [], approved: [] }; }; @@ -73,6 +73,7 @@ function handler(repo: Record, s: Stub, overrides: Partial (number === 8 ? "abc123" : undefined), readCheckPack: async () => ({ status: "ok", pack: PACK, automatic: [], source: { repo: { id: "art_pack", version: 1 } } }), liveDeployments: async () => [ANCHOR], + listener: async () => undefined, observeRuns: observed([]), managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }), store: stored([], s.saved), @@ -166,6 +167,17 @@ describe("createGithubPrTriage", () => { expect(s.delivered).toHaveLength(1); }); + test("with the host's listener live, mails with no pr-triage deployment and refuses a pull request whose manager is up", async () => { + const s = stub(); + const listening: Partial = { listener: async () => ({ ...ANCHOR, runId: "listener-1", address: "listener-1@tenant.example" }), liveDeployments: async () => [] }; + expect((await handler(ENABLED, s, listening)(request(), MEMBER)).status).toBe(202); + expect(s.delivered).toHaveLength(1); + const managed = stub(); + const running: ObservedRun = { runId: "run-manager", source: "github", status: "running", startedAt: NOW.toISOString() }; + await refused(await handler(ENABLED, managed, { ...listening, managerRuns: observed([running]) })(request(), MEMBER), 409, "already_running"); + expect(managed.delivered).toEqual([]); + }); + test("refuses a head still running on a replaced deployment or stored as running, but not one stored running past the stuck time", async () => { const live: ObservedRun = { runId: "run-live", source: "github", status: "running", startedAt: new Date(NOW.getTime() - 60_000).toISOString() }; const replaced = { ...ANCHOR, runId: "anchor-0", address: "anchor-0@tenant.example" }; diff --git a/apps/hub/src/github/pr-triage.ts b/apps/hub/src/github/pr-triage.ts index f8f3e43cd..24aa365fa 100644 --- a/apps/hub/src/github/pr-triage.ts +++ b/apps/hub/src/github/pr-triage.ts @@ -1,6 +1,7 @@ -// Queues one open pull request's head for the live pr-triage deployment at a -// maintainer's request. The row is marked queued before the mail goes out so -// repeated clicks and reconcile passes see it; a failed mail puts it back. +// Queues one open pull request's head at a maintainer's request, for the host's +// listener while one is live and the live pr-triage deployment otherwise. The +// row is marked queued before the mail goes out so repeated clicks and reconcile +// passes see it; a failed mail puts it back. // Attempts are left alone: a head the hub gave up on is only ever retried here. import { eq } from "drizzle-orm"; import { type } from "arktype"; @@ -32,6 +33,8 @@ export type GithubPrTriageDeps = PortalCredentialDeps & { readCheckPack: PolicyReader; /** The named workflow's, newest first; mail goes to the first, and a head may still be running on a replaced one. */ liveDeployments: (tenantId: string, workflow: string) => Promise; + /** The workspace's live listener for the host, if it has one; its managers then run the pull requests. */ + listener: (tenantId: string, source: RepoSource) => Promise; observeRuns: ObserveRuns; managerRuns: ManagerRuns; store: TriageStateStore; @@ -94,9 +97,9 @@ export function createGithubPrTriage(deps: GithubPrTriageDeps) { if (ready instanceof Response) return ready; const { tenant, record, policy, pack } = ready; const triage = triageWorkflowFor(sourceOf(record)); - const deployments = await deps.liveDeployments(tenantId, triage.id); - const deployment = newestLive(deployments); - if (!deployment) return failure(503, "service_not_running", `The ${triage.id} service is not running.`); + const listening = (await deps.listener(tenantId, sourceOf(record))) !== undefined; + const deployments = listening ? [] : await deps.liveDeployments(tenantId, triage.id); + if (!listening && !newestLive(deployments)) return failure(503, "service_not_running", `The ${triage.id} service is not running.`); let openHead: string | undefined; try { diff --git a/apps/hub/src/github/triage-reconciler.test.ts b/apps/hub/src/github/triage-reconciler.test.ts index e71bbb53f..dd5c8485d 100644 --- a/apps/hub/src/github/triage-reconciler.test.ts +++ b/apps/hub/src/github/triage-reconciler.test.ts @@ -5,7 +5,7 @@ import { formatRunAddress } from "@intx/types"; import { workflowRunRepoIdForAddress, type RepoId, type WorkflowRunEvent, type WorkflowRunReader } from "@intx/hub-sessions"; import { emptyPack, stockTriggerMail, type ActionKind, type CheckPack, type PrTriageRow } from "@corbits/triage-contracts"; import type { CheckRun } from "@corbits/github-reads"; -import { DEFAULT_RECONCILE_POLICY, planTenant, type ObservedRun, type OpenPr } from "./reconcile-plan.js"; +import { DEFAULT_RECONCILE_POLICY, planTenant, runKey, type ObservedRun, type OpenPr } from "./reconcile-plan.js"; import { triageReconcilePolicy } from "../env.js"; import { TriageStateConflictError, type TriageStateStore } from "./triage-state-store.js"; import { createRunTriggerDeliverer } from "@corbits/webhooks"; @@ -34,6 +34,8 @@ const LONG_AGO = "2026-10-07T00:00:00.000Z"; const VERSION = "0.1.0-sha-11111111"; const OFFERING = "off_decision"; const LIVE = deployment("run_live", new Date(LONG_AGO)); +/** A listener names no workflow version or offering. */ +const LISTENER = { runId: "run_listener", address: formatRunAddress("run_listener", DOMAIN), createdAt: new Date(LONG_AGO), cancelling: false }; const MINUTE = 60_000; const BASE_MOVE_WINDOW = DEFAULT_RECONCILE_POLICY.baseMoveWindowMs / MINUTE; @@ -188,6 +190,9 @@ type HarnessOptions = { rotateAfterRuns?: number; now?: () => Date; deployments?: LiveDeployment[]; + listener?: LiveDeployment; + /** The managers' runs by head, read on each pass. */ + managed?: () => Record; whileRedeploying?: (deployments: LiveDeployment[]) => void; rotation?: RotationRecord; sameSource?: boolean; @@ -259,6 +264,7 @@ function harness(logs: Logs, prs: OpenPr[] | Record, options: return [tenant()]; }, liveDeployments, + listener: async () => options.listener, latestDeployments: async () => [], rotation: { afterRuns: options.rotateAfterRuns ?? 1_000, @@ -295,7 +301,7 @@ function harness(logs: Logs, prs: OpenPr[] | Record, options: }, activity: options.activity ?? { stale: () => true }, observeTenant: observer.observe, - managerRuns: async () => ({ byRepo: new Map(), runCount: 0, verdicts: [], approved: [] }), + managerRuns: async () => ({ byRepo: new Map(options.managed ? [[REPO, new Map(Object.entries(options.managed()))]] : []), runCount: 0, verdicts: [], approved: [] }), headChecks: async () => options.headChecks ?? [], verdicts: { latest: async () => options.recorded ?? [] }, store: store(), @@ -347,6 +353,24 @@ function rotationLogs(logged: ReadonlyArray>): unknown[] } describe("triage reconciler", () => { + test("a host whose listener is live is caught up with no pr-triage deployment", async () => { + const run = harness({}, [head(1)], { deployments: [], listener: LISTENER }); + await run.reconcile(); + expect(mailedNumbers(run.delivered)).toEqual([1]); + }); + + test("a head mailed through the listener is mailed once while its manager runs and once it reported", async () => { + let minutes = 0; + let managed: Record = {}; + const run = harness({}, [head(1)], { deployments: [], listener: LISTENER, now: () => at(minutes), managed: () => managed }); + await run.reconcile(); + managed = { [runKey(1, "sha1")]: [{ runId: "run_manager", source: "github", status: "running", startedAt: at(1).toISOString() }] }; + for (minutes = 1; minutes <= 3; minutes++) await run.reconcile(); + managed = { [runKey(1, "sha1")]: [{ runId: "run_manager", source: "github", status: "completed", startedAt: at(1).toISOString() }] }; + for (minutes = 4; minutes <= 60; minutes += 14) await run.reconcile(); + expect(mailedNumbers(run.delivered)).toEqual([1]); + }); + test("a host without build webhooks: a head whose verdict waits on CI is queued again once its builds stop pending", async () => { const waiting = { ...current(1), repo: REPO, number: 1, state: "awaiting-review", actor: "system", humanGated: false, checks: [{ check: "ci-green", kind: "machine", result: "unconfirmed" }] }; const logs: Logs = { [repoKey(LIVE.runId)]: { run_1: [started(1), rendered(waiting), completed] } }; diff --git a/apps/hub/src/github/triage-reconciler.ts b/apps/hub/src/github/triage-reconciler.ts index 8f7f02e9f..2d6ac692f 100644 --- a/apps/hub/src/github/triage-reconciler.ts +++ b/apps/hub/src/github/triage-reconciler.ts @@ -1,7 +1,8 @@ // Brings every open pull request on an enabled repository to a triaged head // without relying on any webhook, sidecar or portal visit having survived: // reads GitHub's open pull requests, the live deployments' run logs and the -// stored state, then queues what the plan says, each head to a worker of its own. +// stored state, then queues what the plan says, each head to a worker of its own, +// or to the host's listener while one is live. // A head waiting for CI is released here once its CI finished or its wait ran // out, on hosts and pull requests whose CI results send no webhook. import { dependsOnCi, hasEnabledRules, marksWaiting, repoPolicy, type PrTriageRow, type RepoPolicy, type RepoSource, type VerdictDetail } from "@corbits/triage-contracts"; @@ -43,6 +44,12 @@ export type TriageReconcilerDeps = { rotationKey: string; tenants: () => Promise; liveDeployments: (tenantId: string) => Promise; + /** + * The host's live listener, whose managers run its pull requests: the pass then needs no pr-triage deployment, and + * neither rotates nor heals one, so a rotation under way is abandoned. A listener names no workflow version, so no + * head starts over for one. + */ + listener: (tenantId: string, source: RepoSource) => Promise; latestDeployments: (tenantId: string, limit: number) => Promise; rotation: DeploymentRotation; /** Undefined when the tenant holds no credential for any pull request host. */ @@ -456,9 +463,10 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { async function reconcileTenant(tenant: ReconcileTenant): Promise { const repos = enabledRepos(tenant.config, deps.source); if (repos.length === 0) return false; + const listener = await deps.listener(tenant.id, deps.source); // Only live deployments' and recent workers' logs are read; heads triaged under a released one stay settled in the stored state until the live deployment runs another workflow version. const { deployments, observed, workers } = await deps.observeTenant(tenant.id, tenant.domain); - if (deployments.length === 0) { + if (!listener && deployments.length === 0) { deps.log({ level: "warn", msg: "triage_reconcile_skipped", tenantId: tenant.id, reason: "no_live_deployment" }); return heal(tenant.id); } @@ -475,11 +483,13 @@ export function createTriageReconciler(deps: TriageReconcilerDeps) { return false; } const runs = mergeObservedRuns([...observed, ...workers]); - let deployment = newestLive(deployments); - try { - deployment = await route(tenant, deployments, observed); - } catch (err) { - deps.log({ level: "error", msg: "triage_rotation_failed", tenantId: tenant.id, error: String(err) }); + let deployment = listener ?? newestLive(deployments); + if (!listener) { + try { + deployment = await route(tenant, deployments, observed); + } catch (err) { + deps.log({ level: "error", msg: "triage_rotation_failed", tenantId: tenant.id, error: String(err) }); + } } if (!deployment) { deps.log({ level: "warn", msg: "triage_reconcile_skipped", tenantId: tenant.id, reason: "no_live_deployment" }); diff --git a/apps/hub/src/integrations/app.ts b/apps/hub/src/integrations/app.ts index bbdf1c9d7..ea1854785 100644 --- a/apps/hub/src/integrations/app.ts +++ b/apps/hub/src/integrations/app.ts @@ -397,7 +397,7 @@ export function createIntegrationsApp(guards: IntegrationGuards, handlers: Integ ...invalid, ...failures, 502: json("`source_failed`: the pull request could not be read; `hub_unavailable`: it could not be queued", Failure), - 503: json("`service_not_running`: the pr-triage service is not running; `github_rate_limited`: GitHub's rate limit is reached, `Retry-After` and `resetAt` say when it lifts", Failure), + 503: json("`service_not_running`: neither the listener nor the pr-triage service is running; `github_rate_limited`: GitHub's rate limit is reached, `Retry-After` and `resetAt` say when it lifts", Failure), }, }), member, diff --git a/apps/hub/src/manager-state.postgres.test.ts b/apps/hub/src/manager-state.postgres.test.ts index 5abaddb51..127cbfd0f 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, recordEvent, take, type PullRequestRef } from "./manager-state.js"; +import { done, liveListener, recordEvent, 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" }; @@ -396,6 +396,20 @@ test("PostgreSQL: a manager deployed inside the startup window, or one with a si }); }, 120_000); +test("PostgreSQL: a listener counts as live as a manager does, and one deployed without a sidecar past the startup window is released", async () => { + await withDatabase(async function listener(db) { + await db.insert(schema.workflowDefinition).values({ id: "wfd_listener", tenantId: TENANT, name: "triage-listener-github" }); + await db.insert(schema.workflowRun).values({ id: "run_listener", anchorRunId: "run_listener", definitionId: "wfd_listener", tenantId: TENANT, status: "deployed", createdAt: new Date() }); + expect(await liveListener(db, TENANT, "github")).toMatchObject({ runId: "run_listener" }); + expect(await liveListener(db, TENANT, "bitbucket-datacenter")).toBeUndefined(); + await db.update(schema.workflowRun).set({ createdAt: new Date(Date.now() - 3 * MINUTE) }).where(sql`${schema.workflowRun.id} = 'run_listener'`); + expect(await liveListener(db, TENANT, "github")).toBeUndefined(); + const [left] = await db.select({ cancellationRequestedAt: schema.workflowRun.cancellationRequestedAt, cancellationReason: schema.workflowRun.cancellationReason }).from(schema.workflowRun).where(sql`${schema.workflowRun.id} = 'run_listener'`); + expect(left?.cancellationRequestedAt).toBeInstanceOf(Date); + expect(left?.cancellationReason).toBe("listener deployed without a sidecar"); + }); +}, 120_000); + test("PostgreSQL: a malformed event payload or reviewer is refused", async () => { await withDatabase(async function badPayload(db) { await startManager(db, GITHUB, "run_a"); diff --git a/apps/hub/src/manager-state.ts b/apps/hub/src/manager-state.ts index 63ad6b52b..39040206b 100644 --- a/apps/hub/src/manager-state.ts +++ b/apps/hub/src/manager-state.ts @@ -6,8 +6,9 @@ import { and, desc, eq, inArray, isNull, like, 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 { APPROVE_DO_KIND, type RepoSource } from "@corbits/triage-contracts"; +import { APPROVE_DO_KIND, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; import type { Tx } from "./github/admin-github-grant.js"; +import { resolveLiveDeployment, type LiveDeployment } from "./github/deployment.js"; import { triageSchema } from "./triage-db.js"; const managerState = triageSchema.table("manager_state", { @@ -155,12 +156,15 @@ export async function recordEvent(db: DB["db"], tenantId: string, pr: PullReques return inserted.length === 1; } -/** How long a just-deployed manager may have no sidecar and still hold the pull request. The listener waits this long for it to start. */ +/** + * 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. + */ const MANAGER_START_GRACE_MS = 2 * 60_000; type ManagerRun = { status: string; createdAt: Date; allocationStatus: string | null; sidecarId: string | null }; -async function managerRun(tx: Tx, runId: string): Promise { +async function managerRun(tx: Tx | DB["db"], runId: string): Promise { const [run] = await tx .select({ status: schema.workflowRun.status, @@ -182,18 +186,36 @@ function runIsUp(run: ManagerRun | undefined, now: Date): boolean { return now.getTime() - run.createdAt.getTime() < MANAGER_START_GRACE_MS; } -async function runGone(tx: Tx, runId: string, now: Date): Promise { +async function runGone(tx: Tx | DB["db"], runId: string, now: Date): Promise { return !runIsUp(await managerRun(tx, runId), now); } -/** Asks the sweep to stop a deploy that still counts as live but never became a process, so it cannot start after its claim has moved on. */ -async function releaseStuck(tx: Tx, runId: string, now: Date): Promise { +/** Whether a deployment is up as the hub counts a listener or a manager: running, on its sidecar, or just deployed. */ +export async function deploymentUp(db: DB["db"], runId: string): Promise { + return !(await runGone(db, runId, new Date())); +} + +/** + * Asks the sweep to stop a deploy that still counts as live but never became a process, so a manager cannot start after + * its claim has moved on, and a listener is deployed again by the next Upgrade. + */ +async function releaseStuck(tx: Tx | DB["db"], runId: string, now: Date, what: "manager" | "listener"): Promise { await tx .update(schema.workflowRun) - .set({ cancellationRequestedAt: now, cancellationDeadline: now, cancellationReason: "manager deployed without a sidecar" }) + .set({ cancellationRequestedAt: now, cancellationDeadline: now, cancellationReason: `${what} deployed without a sidecar` }) .where(and(eq(schema.workflowRun.id, runId), inArray(schema.workflowRun.status, [...schema.liveWorkflowRunStatuses]), isNull(schema.workflowRun.cancellationRequestedAt))); } +/** The workspace's newest listener for the host while it is up, as a manager is; one that never got a sidecar is released. */ +export async function liveListener(db: DB["db"], tenantId: string, source: RepoSource): Promise { + const listener = await resolveLiveDeployment(db, tenantId, TRIAGE_LISTENERS[source]); + if (!listener) return undefined; + const now = new Date(); + if (runIsUp(await managerRun(db, listener.runId), now)) return listener; + await releaseStuck(db, listener.runId, now, "listener"); + return undefined; +} + /** The run id of the pull request's manager while that run is up. */ export async function liveManager(db: DB["db"], tenantId: string, pr: PullRequestRef): Promise { return db.transaction(async function findLive(tx) { @@ -218,7 +240,7 @@ async function claimFor(tx: Tx, tenantId: string, pr: PullRequestRef, runId: str if (!claim) throw new Error(`manager_state ${key} vanished under its lock`); if (claim.status === "live" && claim.runId === runId) return claim; if (claim.status === "live" && claim.runId !== undefined && !(await runGone(tx, claim.runId, now))) return undefined; - if (claim.runId !== undefined && claim.runId !== runId) await releaseStuck(tx, claim.runId, now); + if (claim.runId !== undefined && claim.runId !== runId) await releaseStuck(tx, claim.runId, now, "manager"); const taken = { ...claim, ...fresh, starts: claim.starts + 1 }; await writeClaim(tx, tenantId, pr, taken); return taken; diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index 59fba8c67..c583b72ba 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -30,11 +30,11 @@ import { } from "@corbits/cron"; import { E2B_PLACE } from "@corbits/provisioner-e2b"; import { RAILWAY_PLACE } from "@corbits/provisioner-railway"; -import { TRIAGE_LISTENERS, TRIAGE_MAIL_SENDER, type RepoSource } from "@corbits/triage-contracts"; +import { TRIAGE_MAIL_SENDER, type RepoSource } from "@corbits/triage-contracts"; import { createInterchangeHub } from "./interchange-hub.js"; import { createManagerIntake } from "./manager-intake.js"; import { MANAGER_RUN_TOKEN_ROUTES } from "./manager-routes.js"; -import { passRequest } from "./manager-state.js"; +import { deploymentUp, liveListener, passRequest } from "./manager-state.js"; import { mountTenantRoutes } from "./tenant-routes.js"; import { connectDockerSidecars, dockerPlaces, sweepDockerSidecarsOf } from "./sidecar/docker.js"; import { connectE2bSidecars, tendE2bSidecarsOf, type E2bSidecarProvisioner } from "./sidecar/e2b.js"; @@ -75,7 +75,7 @@ import { createCoalescer, type Coalescer } from "./github/coalescer.js"; import { createCoordinator, type Coordinator, type TriageJob } from "./github/coordinator.js"; import { createWorkerSpawner, releaseOrphanedWorkers } from "./github/worker-spawn.js"; import { DeliveryCache } from "./github/dedupe.js"; -import { activeTenants, createDeploymentRotation, latestDeployments, NoLiveDeploymentError, readyMaterializer, recentWorkers, systemSenderMaterializer, resolveLiveDeployment, resolveLiveDeployments, withWorkflowVersions } from "./github/deployment.js"; +import { activeTenants, createDeploymentRotation, latestDeployments, NoLiveDeploymentError, readyMaterializer, recentWorkers, systemSenderMaterializer, resolveLiveDeployment, resolveLiveDeployments, withWorkflowVersions, type LiveDeployment } from "./github/deployment.js"; import { createGithubOpenPulls, createMemberOpenPulls, createOpenPullsCache, keptOpenPrs } from "./github/open-pulls.js"; import { createRepoReads } from "./github/repo-reads.js"; import { createGithubPrActions } from "./github/pr-actions.js"; @@ -313,12 +313,15 @@ const requireGrant = createRequireGrant({ conditionRegistry: HUB_CONDITION_REGISTRY, }); /** A workspace with a host's listener live runs that host's pull requests in managers. */ -async function liveListener(tenantId: string, source: RepoSource): Promise { - return (await resolveLiveDeployment(composition.db, tenantId, TRIAGE_LISTENERS[source]))?.address; +function liveListenerOf(tenantId: string, source: RepoSource): Promise { + return liveListener(composition.db, tenantId, source); +} +async function listenerAddress(tenantId: string, source: RepoSource): Promise { + return (await liveListenerOf(tenantId, source))?.address; } const toListener = createManagerIntake({ db: composition.db, - listener: liveListener, + listener: listenerAddress, readPull: function readPull(tenantId, record, number) { return pullRequests.withDiff(tenantId, record, number); }, @@ -496,7 +499,7 @@ async function inPrWorkflow(tenantId: string, repo: string, source: RepoSource): } // Repositories left switched on send nothing while code review has nowhere to run, nor one the PR workflow or a manager reviews. async function reviewedSources(tenantId: string, record: RepoRecord) { - if (!(await codeReviewPlaced(tenantId)) || (await inPrWorkflow(tenantId, record.name, sourceOf(record))) || (await liveListener(tenantId, sourceOf(record)))) return []; + if (!(await codeReviewPlaced(tenantId)) || (await inPrWorkflow(tenantId, record.name, sourceOf(record))) || (await listenerAddress(tenantId, sourceOf(record)))) return []; return codeReviewSources(composition.db, tenantId); } /** The hosts whose PR workflow deploys: a connected repository's policy file needs it, and code review has a place to run. */ @@ -654,6 +657,7 @@ function triageStack(triage: TriageWorkflow) { rotationKey: triage.rotationKey, tenants: reconcileTenants, liveDeployments, + listener: liveListenerOf, headChecks: tenantHeadChecks, verdicts: verdictRecordStore, latestDeployments: function latestTriageDeployments(tenantId, limit) { @@ -718,6 +722,7 @@ const githubPrTriage = createGithubPrTriage({ }, readCheckPack, liveDeployments: liveTriageDeployments, + listener: liveListenerOf, observeRuns: observeTriageRuns, managerRuns, store: triageStateStore, @@ -953,7 +958,13 @@ const integrations = createIntegrationsApp({ db: composition.db, getSession: com workflows: [...TRIAGE_WORKFLOWS, ...PR_WORKFLOW_VARIANTS].map((workflow) => workflow.id), }), workflowDeploy: createWorkflowDeploy({ db: composition.db, allocation: composition.workflowAllocationService, onDeployed: onSourcesChanged, log: logJson }), - workflowVersions: createWorkflowVersions({ db: composition.db, launchSpecs }), + workflowVersions: createWorkflowVersions({ + db: composition.db, + launchSpecs, + up: function upOf(anchorRunId) { + return deploymentUp(composition.db, anchorRunId); + }, + }), codeReview: createCodeReviewSwitch({ db: composition.db, authorize: authorizePortal, onChanged: onSourcesChanged, place: placeCodeReview, inPrWorkflow }), codeReviewPlace: createCodeReviewPlace({ authorize: authorizePortal, place: placeCodeReview, prWorkflowSources, probes: [...dockerPlaces(docker), { guide: E2B_PLACE, found: e2b }, { guide: RAILWAY_PLACE, found: railway }] }), codeReviewRun: createCodeReviewRun({ db: composition.db, authorize: authorizePortal, feed: codeReviewFeed, place: placeCodeReview, inPrWorkflow }), diff --git a/apps/hub/src/workflow-versions.test.ts b/apps/hub/src/workflow-versions.test.ts index 9756cc2ba..ec6aed6a8 100644 --- a/apps/hub/src/workflow-versions.test.ts +++ b/apps/hub/src/workflow-versions.test.ts @@ -52,7 +52,7 @@ function db() { } function handler() { - return createWorkflowVersions({ db: db(), launchSpecs: { get: async (anchorRunId) => (SPECS[anchorRunId.replace("_worker", "")] ?? null) as never } }); + return createWorkflowVersions({ db: db(), up: async (anchorRunId) => anchorRunId !== "run_source", launchSpecs: { get: async (anchorRunId) => (SPECS[anchorRunId.replace("_worker", "")] ?? null) as never } }); } describe("createWorkflowVersions", () => { @@ -61,9 +61,9 @@ describe("createWorkflowVersions", () => { expect(res.status).toBe(200); expect(await res.json()).toEqual({ deployments: [ - { id: "run_tarball", workflow: "pr-triage", createdAt: CREATED_AT.toISOString(), cancelling: false, worker: false, package: WORKFLOW, tools: [TOOL] }, - { id: "run_source", workflow: "pr-triage-historical", createdAt: CREATED_AT.toISOString(), cancelling: true, worker: false, package: SOURCE_WORKFLOW, tools: [] }, - { id: "run_tarball_worker", workflow: "pr-triage", createdAt: CREATED_AT.toISOString(), cancelling: false, worker: true, package: WORKFLOW, tools: [TOOL] }, + { id: "run_tarball", workflow: "pr-triage", createdAt: CREATED_AT.toISOString(), cancelling: false, live: true, worker: false, package: WORKFLOW, tools: [TOOL] }, + { id: "run_source", workflow: "pr-triage-historical", createdAt: CREATED_AT.toISOString(), cancelling: true, live: false, worker: false, package: SOURCE_WORKFLOW, tools: [] }, + { id: "run_tarball_worker", workflow: "pr-triage", createdAt: CREATED_AT.toISOString(), cancelling: false, live: true, worker: true, package: WORKFLOW, tools: [TOOL] }, ], }); }); diff --git a/apps/hub/src/workflow-versions.ts b/apps/hub/src/workflow-versions.ts index 10b619602..d65b780b0 100644 --- a/apps/hub/src/workflow-versions.ts +++ b/apps/hub/src/workflow-versions.ts @@ -10,6 +10,8 @@ export const WORKFLOW_VERSIONS_PATH = "/api/integrations/workflow-versions"; export type WorkflowVersionsDeps = { db: DB["db"]; launchSpecs: Pick; + /** Whether the deployment is up as the hub counts a listener or a manager. */ + up: (anchorRunId: string) => Promise; }; export function createWorkflowVersions(deps: WorkflowVersionsDeps) { @@ -28,6 +30,7 @@ export function createWorkflowVersions(deps: WorkflowVersionsDeps) { workflow: live.workflow, createdAt: live.createdAt.toISOString(), cancelling: live.cancelling, + live: await deps.up(live.runId), worker: kind === "worker", package: versions?.workflow ?? null, tools: versions?.tools ?? [], diff --git a/apps/web/src/components/settings/TriageSection.tsx b/apps/web/src/components/settings/TriageSection.tsx index 300770041..1b33f82f6 100644 --- a/apps/web/src/components/settings/TriageSection.tsx +++ b/apps/web/src/components/settings/TriageSection.tsx @@ -1,6 +1,6 @@ import { useMemo, type ReactNode } from "react"; import { Link } from "react-router-dom"; -import { presetFor, type RepoSource } from "@corbits/triage-contracts"; +import { presetFor, TRIAGE_LISTENERS, type RepoSource } from "@corbits/triage-contracts"; import { useCheckPacks } from "../../lib/check-packs.ts"; import { modelSecrets } from "../../lib/credential-actions.ts"; import { DeniedNotice } from "../../lib/denied.tsx"; @@ -17,9 +17,9 @@ import { KeyRow } from "./KeyRow.tsx"; import { PolicyFiles } from "./PolicyFiles.tsx"; import { TriageUpgrade } from "./TriageUpgrade.tsx"; -/** The connected hosts whose pr-triage workflow has no deployment. */ +/** The connected hosts with neither a listener up nor a pr-triage workflow deployed. */ function undeployedSources(deployments: readonly DeployedWorkflow[], sources: readonly RepoSource[]): RepoSource[] { - return sources.filter((source) => currentDeployment(deployments, `pr-triage-${source}`) === undefined); + return sources.filter((source) => currentDeployment(deployments, TRIAGE_LISTENERS[source])?.live !== true && currentDeployment(deployments, `pr-triage-${source}`) === undefined); } type Status = { sources: RepoSource[]; undeployed: RepoSource[]; model: boolean; triaging: number; total: number }; diff --git a/apps/web/src/lib/triage-upgrade.test.ts b/apps/web/src/lib/triage-upgrade.test.ts index a64c97fb3..531158efa 100644 --- a/apps/web/src/lib/triage-upgrade.test.ts +++ b/apps/web/src/lib/triage-upgrade.test.ts @@ -29,7 +29,7 @@ function fakeHub() { } if (path === "/api/tenants/t/workflows/deployments") return deployments as T; if (path === "/api/integrations/workflow-versions/t") { - return { deployments: deployments.filter((row) => row.status === "deployed").map((row) => ({ id: row.id, workflow: workflows[row.id], createdAt: row.createdAt, cancelling: false, worker: false, package: null, tools: [] })) } as T; + return { deployments: deployments.filter((row) => row.status === "deployed").map((row) => ({ id: row.id, workflow: workflows[row.id], createdAt: row.createdAt, cancelling: false, live: true, worker: false, package: null, tools: [] })) } as T; } if (method === "DELETE" && path.startsWith("/api/tenants/t/workflows/runs/")) { deployments.find((row) => path.endsWith(row.id))!.status = "cancelled"; @@ -74,7 +74,7 @@ function sharedHub() { } if (path === "/api/tenants/t/workflows/deployments") return deployments.map((row) => ({ ...row })) as T; if (path === "/api/integrations/workflow-versions/t") { - return { deployments: deployments.filter((row) => row.status === "deployed").map((row) => ({ id: row.id, workflow: "triage-listener-github", createdAt: row.createdAt, cancelling: false, worker: false, package: null, tools: [] })) } as T; + return { deployments: deployments.filter((row) => row.status === "deployed").map((row) => ({ id: row.id, workflow: "triage-listener-github", createdAt: row.createdAt, cancelling: false, live: true, worker: false, package: null, tools: [] })) } as T; } if (method === "DELETE" && path.startsWith("/api/tenants/t/workflows/runs/")) { deployments.find((row) => path.endsWith(row.id))!.status = "cancelled"; diff --git a/apps/web/src/lib/workflow-deploy.test.ts b/apps/web/src/lib/workflow-deploy.test.ts index 9b20e7eed..2158e876f 100644 --- a/apps/web/src/lib/workflow-deploy.test.ts +++ b/apps/web/src/lib/workflow-deploy.test.ts @@ -45,11 +45,12 @@ function fakeHub(tarballSeed: Record = {}, unconnected: readonly workflow: assets.find((asset) => asset.id === row.definitionAssetId)?.name, createdAt: row.createdAt, cancelling: false, + live: true, worker: false, package: packageOf(pins.get(row.id)!), tools: [{ name: INDEX[0]!.name, version: INDEX[0]!.version }], })); - const running = workers.map((row, n) => ({ ...row, id: `worker_${n}`, workflow: "pr-triage-github", createdAt: "2026-10-09T00:00:00.000Z", cancelling: false, worker: true })); + const running = workers.map((row, n) => ({ ...row, id: `worker_${n}`, workflow: "pr-triage-github", createdAt: "2026-10-09T00:00:00.000Z", cancelling: false, live: true, worker: true })); return { deployments: [...listeners, ...running] } as T; } if (method === "GET" && url.pathname === "/api/tenants/t/assets") { diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2a02d290a..370dbadb2 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -88,7 +88,14 @@ GitHub ──webhook──► Hub ──spawn──► Sidecar ──► pr-tria workflow version and decision-model offering. It also replaces a pr-triage deployment once its run log grows past `TRIAGE_ROTATE_AFTER_RUNS`, and redeploys one Interchange failed, after two minutes and at most three - times an hour. + times an hour. While the host's listener is live, a pass needs no + pr-triage deployment, neither replaces nor redeploys one (a replacement + under way is abandoned, its copy left running), carries no answers, and + mails its heads through the manager intake. A listener counts as live as + a manager does: running, on its sidecar, or deployed within the last two + minutes; one deployed without a sidecar past that is released, so the next + Upgrade deploys it again. The workflow-versions route reports each + deployment's `live` the same way, which Settings reads. - **Coordinator**: every triage mail (webhook, retry, reconciler) and every code review mail is a job in one in-memory queue keyed per lane (triage or code review), host and pull request. Each job runs in a worker deployment diff --git a/packages/triage-contracts/src/hub-responses.ts b/packages/triage-contracts/src/hub-responses.ts index 06c8d2fc9..0e972c5f4 100644 --- a/packages/triage-contracts/src/hub-responses.ts +++ b/packages/triage-contracts/src/hub-responses.ts @@ -200,6 +200,8 @@ export const DeployedVersions = type({ workflow: "string", createdAt: "string", cancelling: "boolean", + /** Up as the hub counts a listener or a manager: running, on its sidecar, or just deployed. */ + live: "boolean", /** A worker runs one triage job on a copy of a listener; it still pins that listener's versions. */ worker: "boolean", package: PackageVersion.or("null"),