trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns3g rEventsFlows

Flow keeper: stuck, conflicted and wrongly held pulls, rebase loops, lulls and forge-down, acted on instead of alarmed #11255

Merged scoutcrew opened this in answer to trevatoowner

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

scout/forge:scout/7556-flow-keeper-stuck-conflicted-and-wrongly → main 6 commits fork

  • Merged. scout/7556-flow-keeper-stuck-conflicted-and-wrongly was merged into main as befa93c.

Review

Sign in to review this change.
23 files changed +1009 −17
docs/features/flow-keeper.md added +19
@@ -0,0 +1,19 @@
1 # Flow keeper (#7556)
2
3 A maintenance duty, `flow-keeper` (`src/maintain/duties/flow_keeper.ts`), runs once a minute and leaves a `maintenance_run` row (see `/maintenance`) with its findings, each with its probe, and what it did. It reads the database and spends no model tokens.
4
5 | Part | Module (`src/maintain/keeper/`) | What it does |
6 | --- | --- | --- |
7 | Stuck | `stuck.ts` | Finding: an approved, open, unheld pull with no run, review or push for 30 min that is not landable (conflicted, red, unchecked). Green on its head is queue wait. |
8 | Conflicted, rebase loops | `rebase.ts` | Assigns a conflicted approved pull to the builder who can push its branch (`ownBuilder`, #7006), at most 6 per run. After 2 assignments with no new head it stops and says so once on the pull. A person's assignee or hold is never moved. |
9 | Wrongly held | `holds.ts` | Lifts a crew hold when a check went green on the head after it, or it is over 6 h old. A person's hold is never read. |
10 | Lulls | `lull.ts` | Builders idle while startable issues exist, or checks idle while pulls wait, two samples in a row: one `keeper_lull` start event with the cause (`crew/idle_reason.ts` words: ceiling reached, no subscription with headroom, hot file), one end event. Nothing while nothing is startable. |
11 | Forge down | `down.ts` | Health failing twice from two vantage points: a `forge_down` event, then `forge_up`. Alarm only; the release keeper owns rollback. Vantages: `FORGE_HEALTH_VANTAGES=home=https://…/healthz,far=https://…/healthz`; with fewer than two, nothing is sampled. |
12 | Closed, handed off | `handed.ts` | An in-repo crew pull closed with no reason, or assigned to a person, is reported once, as a comment on the pull. |
13 | Alarms | `words.ts` | Each firing flow alarm gets a finding and an action line saying what the keeper did (or that it only reports). |
14
15 Conflicted pulls are the ones the crew's runs and the finishing pass already mark (`flow/conflicted.ts`); the keeper does not ask git. Comments the keeper writes are posted under the name of whoever turned autonomy on, with a line saying so, as the finishing pass's hand-over does (`crew/finishing.ts`): the repository has no steward identity of its own to post as, and the line on each comment says the forge wrote it.
16
17 Terminal door: the keeper's acts (`keeper_rebase`, `keeper_release`, `keeper_lull`, `forge_down`/`forge_up`) are also the `actions` of each `flow-keeper` row, and the terminal already lists those rows (the maintenance runs screen, `src/tui/screens/maintainruns.ts`, and `/maintenance`), so it has no screen of its own. The raw events are admin-only (`event_audience.ts`).
18
19 Tests: `test/keeper-stuck.test.ts`, `keeper-rebase-loop`, `keeper-holds`, `keeper-lull`, `keeper-down-two-vantages`, `keeper-handed` (handed-off pulls, alarm words, forge_down through the default runner).
src/core/event_audience.ts +4
@@ -220,6 +220,10 @@ export function classifiedAudience(kind: string): EventAudience | undefined {
220220 kind.startsWith("clog.") ||
221221 // What the throughput keeper did to lift a brake is the owner's too (`flow/keeper.ts`).
222222 kind.startsWith("keeper.") ||
223 // The flow keeper's acts on pulls, lulls and forge-down are the owner's too (`maintain/keeper/`, #7556).
224 kind.startsWith("keeper_") ||
225 kind === "forge_down" ||
226 kind === "forge_up" ||
223227 // Who is trusted to fold in, and the switch that stops it, are the owner's (`actions/trusted_sources.ts`).
224228 kind.startsWith("trust.") ||
225229 kind.startsWith("foldin.") ||
src/crew/idle_reason.ts +22 −14
@@ -13,7 +13,7 @@
1313
1414 import type { App } from "../app.ts";
1515 import { listCrewRoles } from "../core/agents.ts";
16 import { listAutonomousRepos } from "../core/repos.ts";
16 import { listAutonomousRepos, type Repo } from "../core/repos.ts";
1717 import { listActive } from "../core/runs.ts";
1818 import { standings } from "../core/standing.ts";
1919 import { grantedAccounts } from "../core/subscriptions.ts";
@@ -27,6 +27,25 @@ export interface IdleReason {
2727 readonly href: string;
2828 }
2929
30 /** The ceiling or the subscriptions, when one of them is why this repository starts nothing; null when neither is. */
31 export function whatBlocks(app: App, repo: Repo, now: Date): IdleReason | null {
32 const { db } = app;
33 if (atCeiling(repo, spentToday(db, repo.id, now))) return { word: "ceiling reached", href: "/meter" };
34
35 const granted = grantedAccounts(db, repo.id, repo.ownerId);
36 if (granted.length > 0) {
37 const stand = standings(db, granted.map((a) => a.id), now);
38 const accounts = granted.flatMap((a) => {
39 const s = stand.get(a.id);
40 return s === undefined ? [] : [{ name: a.name, standing: s }];
41 });
42 // Whether a window has room does not depend on what a run costs.
43 if (pace(accounts, repo.autonomySlots, 1, now).slots === 0)
44 return { word: "no subscription with headroom", href: "/subscriptions" };
45 }
46 return null;
47 }
48
3049 /** Why no builder is starting anything now; null while a builder slot is busy or something can start. */
3150 export function idleReason(app: App, now: Date): IdleReason | null {
3251 const { db } = app;
@@ -39,19 +58,8 @@ export function idleReason(app: App, now: Date): IdleReason | null {
3958 const active = listActive(db, "agent").filter((r) => r.repoId === repo.id);
4059 if (slottedRuns(active, crew).length >= repo.autonomySlots) return null;
4160
42 if (atCeiling(repo, spentToday(db, repo.id, now))) return { word: "ceiling reached", href: "/meter" };
43
44 const granted = grantedAccounts(db, repo.id, repo.ownerId);
45 if (granted.length > 0) {
46 const stand = standings(db, granted.map((a) => a.id), now);
47 const accounts = granted.flatMap((a) => {
48 const s = stand.get(a.id);
49 return s === undefined ? [] : [{ name: a.name, standing: s }];
50 });
51 // Whether a window has room does not depend on what a run costs.
52 if (pace(accounts, repo.autonomySlots, 1, now).slots === 0)
53 return { word: "no subscription with headroom", href: "/subscriptions" };
54 }
61 const blocked = whatBlocks(app, repo, now);
62 if (blocked !== null) return blocked;
5563
5664 const held = heldNow(app, repo);
5765 if (held.startable.length > 0) return null;
src/flow/alarm_stuck.ts +2 −2
@@ -43,7 +43,7 @@ interface PullRow {
4343 head_branch: string;
4444 head_oid: string;
4545 }
46 interface Last {
46 export interface Last {
4747 at: string;
4848 what: string;
4949 }
@@ -52,7 +52,7 @@ interface Last {
5252 const later = (a: Last | undefined, b: Last): Last =>
5353 a === undefined || b.at > a.at ? b : a;
5454
55 function lastEvent(db: Database, p: PullRow): Last | undefined {
55 export function lastEvent(db: Database, p: Pick<PullRow, "id" | "repo_id" | "head_repo_id" | "head_branch">): Last | undefined {
5656 let last: Last | undefined;
5757 const review = db
5858 .query<{ at: string }, [string]>(
src/maintain/duties/flow_keeper.ts added +102
@@ -0,0 +1,102 @@
1 /**
2 * The flow keeper (#7556): what the steward's unstick, lull-watch and alarm scripts did by hand.
3 * Once a minute it reads the open pulls and acts: re-assigns conflicted approved pulls to a
4 * builder who can push, lifts holds whose reason is gone, reports pulls closed or handed to a
5 * person, samples for lulls and for the forge being down, and answers each flow alarm with what it
6 * did. A person's hold is never touched. Each part is a module under `../keeper/`.
7 */
8
9 import type { Database } from "bun:sqlite";
10 import { firedAlarms } from "../../flow/alarms.ts";
11 import { conflictedPulls } from "../../flow/conflicted.ts";
12 import type { Duty, Finding } from "../duty.ts";
13 import { sampleHealth, vantagesFromEnv, type Vantage } from "../keeper/down.ts";
14 import { liftStale } from "../keeper/holds.ts";
15 import { reportHanded } from "../keeper/handed.ts";
16 import { readBuilders, readChecks, sample, type Reading } from "../keeper/lull.ts";
17 import { nameOf, openPulls } from "../keeper/pulls.ts";
18 import { rebasePass } from "../keeper/rebase.ts";
19 import { probe } from "../keeper/record.ts";
20 import { stuckPulls } from "../keeper/stuck.ts";
21 import { sentence } from "../keeper/words.ts";
22
23 export interface Deps {
24 /** `owner/name#N` of each pull main has moved out from under; default: the conflicted readings the crew keeps. */
25 readonly conflicted?: (db: Database) => ReadonlySet<string>;
26 readonly vantages?: () => readonly Vantage[];
27 /** The builders' reading; default asks the running forge. */
28 readonly builders?: (now: Date) => Reading | undefined;
29 }
30
31 const fromRuns = (db: Database): ReadonlySet<string> => new Set(conflictedPulls(db).map((p) => `${p.repo}#${p.number}`));
32
33 export function flowKeeper(deps: Deps = {}): Duty {
34 return {
35 name: "flow-keeper",
36 intervalMs: 60_000,
37 async run({ db, now, app }) {
38 const findings: Finding[] = [];
39 const actions: string[] = [];
40 const at = now.toISOString();
41 const pulls = openPulls(db);
42 const conflicted = (deps.conflicted ?? fromRuns)(db);
43
44 const stuck = stuckPulls(db, now, pulls, conflicted);
45 for (const s of stuck) {
46 findings.push({
47 summary: `${nameOf(s.pull)} is approved and stuck: ${s.why}, quiet for ${Math.round(s.quietMs / 60_000)} min`,
48 probe: probe("minutes since the pull's last run, review or push", Math.round(s.quietMs / 60_000), now),
49 });
50 }
51
52 const moves = rebasePass(db, pulls, conflicted);
53 for (const m of moves.reassigned) actions.push(`re-assigned ${m}`);
54 for (const m of moves.stopped) actions.push(`stopped re-assigning ${m}: the head did not move; said on the pull`);
55 if (moves.waiting.length > 0) {
56 findings.push({
57 summary: `${moves.waiting.length} conflicted ${moves.waiting.length === 1 ? "pull waits" : "pulls wait"} for the next run (cap per run)`,
58 probe: probe("conflicted approved pulls past the per-run cap", moves.waiting.length, now),
59 });
60 }
61
62 const lifted = liftStale(db, now, pulls);
63 for (const line of lifted) actions.push(line);
64 for (const line of reportHanded(db, now)) actions.push(line);
65
66 const readings: Reading[] = [];
67 const builders = deps.builders ? deps.builders(now) : app ? readBuilders(app, now) : undefined;
68 if (builders) readings.push(builders);
69 readings.push(readChecks(db, pulls));
70 for (const r of readings) {
71 const line = sample(db, r);
72 if (line !== undefined) {
73 actions.push(line);
74 findings.push({ summary: line, probe: probe(`${r.kind} waiting while idle`, r.waiting, now) });
75 }
76 }
77
78 const down = await sampleHealth(db, (deps.vantages ?? vantagesFromEnv)());
79 if (down !== undefined) {
80 actions.push(down);
81 findings.push({ summary: down, probe: probe("health vantages failing twice", 2, now) });
82 }
83
84 // Each alarm gets a response: what the keeper did about it, or that it only reports.
85 let fired: ReturnType<typeof firedAlarms> = [];
86 try {
87 fired = firedAlarms(db, now, { dataDir: null });
88 } catch (e) {
89 const why = e instanceof Error ? e.message : String(e);
90 findings.push({ summary: `alarms unreadable: ${why}`, probe: probe("flow alarms that could not be read", 1, now) });
91 }
92 for (const a of fired) {
93 const did = sentence(a.key, { stuck: stuck.length, reassigned: moves.reassigned.length, lifted: lifted.length, down: down !== undefined });
94 findings.push({ summary: `alarm ${a.key}: ${a.says}`, probe: { kind: "measurement", what: `alarm ${a.key} firing since ${a.since}`, number: 1, at } });
95 actions.push(`alarm ${a.key}: ${did}`);
96 }
97 return { findings, actions };
98 },
99 };
100 }
101
102 export const duty: Duty = flowKeeper();
src/maintain/keeper/down.ts added +64
@@ -0,0 +1,64 @@
1 /**
2 * Forge down (#7556): the health check fails twice in a row from two vantage points. One bad
3 * vantage is a bad network, not an outage. The keeper writes an event and does nothing else; the
4 * release keeper owns rollback.
5 */
6
7 import type { Database } from "bun:sqlite";
8 import { note } from "./record.ts";
9
10 export interface Vantage {
11 readonly name: string;
12 /** True when the forge answered health from here. Never throws. */
13 readonly ok: () => Promise<boolean>;
14 }
15
16 /** Failures in a row, per vantage, per database. */
17 const fails = new WeakMap<Database, Map<string, number>>();
18
19 /** Vantages from `FORGE_HEALTH_VANTAGES`: `name=https://url,name=https://url`. Fewer than two means none can confirm. */
20 export function vantagesFromEnv(env: string | undefined = process.env.FORGE_HEALTH_VANTAGES): Vantage[] {
21 const out: Vantage[] = [];
22 for (const part of (env ?? "").split(",")) {
23 const [name, url] = part.trim().split(/=(.*)/s);
24 if (!name || !url || !/^https?:\/\//.test(url)) continue;
25 out.push({
26 name,
27 ok: async () => {
28 try {
29 return (await fetch(url, { signal: AbortSignal.timeout(5_000), redirect: "manual" })).ok;
30 } catch {
31 return false;
32 }
33 },
34 });
35 }
36 return out.length >= 2 ? out : [];
37 }
38
39 function downNow(db: Database): boolean {
40 const last = db
41 .query<{ kind: string }, []>("SELECT kind FROM event WHERE kind IN ('forge_down','forge_up') ORDER BY created_at DESC, rowid DESC LIMIT 1")
42 .get();
43 return last?.kind === "forge_down";
44 }
45
46 /** One sampling: every vantage asked at once; the line to report when an event was written. */
47 export async function sampleHealth(db: Database, vantages: readonly Vantage[]): Promise<string | undefined> {
48 if (vantages.length < 2) return undefined;
49 let mine = fails.get(db);
50 if (mine === undefined) fails.set(db, (mine = new Map()));
51 const answers = await Promise.all(vantages.map(async (v) => [v.name, await v.ok().catch(() => false)] as const));
52 for (const [name, ok] of answers) mine.set(name, ok ? 0 : (mine.get(name) ?? 0) + 1);
53 const failing = vantages.filter((v) => (mine.get(v.name) ?? 0) >= 2).map((v) => v.name);
54 const down = downNow(db);
55 if (failing.length >= 2 && !down) {
56 note(db, "forge_down", undefined, { vantages: failing });
57 return `forge down: health failed twice from ${failing.join(" and ")}; alarm only`;
58 }
59 if (down && answers.every(([, ok]) => ok)) {
60 note(db, "forge_up", undefined, {});
61 return "forge answers health again from every vantage";
62 }
63 return undefined;
64 }
src/maintain/keeper/handed.ts added +46
@@ -0,0 +1,46 @@
1 /**
2 * Closed or handed-off pulls (#7556): an in-repo pull the crew wrote that was closed without a
3 * reason, or handed to a person, is reported once, on the pull. Nothing is moved.
4 */
5
6 import type { Database } from "bun:sqlite";
7 import { sayOnce } from "./record.ts";
8
9 const DAY_MS = 86_400_000;
10
11 interface Row {
12 id: string;
13 repo_id: string;
14 repo: string;
15 number: number;
16 state: string;
17 who: string | null;
18 }
19
20 export function reportHanded(db: Database, now: Date): string[] {
21 const since = new Date(now.getTime() - DAY_MS).toISOString();
22 const rows = db
23 .query<Row, [string]>(
24 `SELECT issue.id, issue.repo_id, o.username || '/' || r.name AS repo, issue.number, issue.state, asg.username AS who
25 FROM issue JOIN pull ON pull.issue_id = issue.id
26 JOIN repo r ON r.id = issue.repo_id JOIN user o ON o.id = r.owner_id
27 JOIN user au ON au.id = issue.author_id
28 LEFT JOIN user asg ON asg.id = issue.assignee_id AND asg.is_agent = 0
29 WHERE issue.kind = 'pull' AND pull.merged_at IS NULL AND au.is_agent = 1
30 AND r.autonomy != 'off' AND (pull.head_repo_id IS NULL OR pull.head_repo_id = issue.repo_id)
31 AND ((issue.state = 'closed' AND issue.close_reason IS NULL AND issue.closed_at > ?1)
32 OR (issue.state = 'open' AND asg.id IS NOT NULL))`,
33 )
34 .all(since);
35 const said: string[] = [];
36 for (const r of rows) {
37 const closed = r.state === "closed";
38 const line = closed
39 ? "this pull was closed with no reason given and nothing merged. Reopen it if the work should land."
40 : `this pull is assigned to ${r.who}, a person, so the crew leaves it to them.`;
41 if (sayOnce(db, { id: r.id, repoId: r.repo_id }, `${closed ? "closed" : "handed"}:${r.id}`, line)) {
42 said.push(`${r.repo}#${r.number}: ${closed ? "closed" : `handed to ${r.who}`}, said on the pull`);
43 }
44 }
45 return said;
46 }
src/maintain/keeper/holds.ts added +30
@@ -0,0 +1,30 @@
1 /**
2 * Wrongly held pulls (#7556): a hold the crew placed whose reason is gone is lifted: a check went
3 * green on the head after the hold, or it is older than six hours. A person's hold is never read.
4 */
5
6 import type { Database } from "bun:sqlite";
7 import { crewHoldsPlacedBefore, releaseHold } from "../../core/holds.ts";
8 import { checkOn, nameOf, type KPull } from "./pulls.ts";
9 import { note } from "./record.ts";
10
11 export const HOLD_MAX_MS = 6 * 3_600_000;
12
13 export function liftStale(db: Database, now: Date, pulls: readonly KPull[]): string[] {
14 const byId = new Map(pulls.map((p) => [p.id, p]));
15 const lifted: string[] = [];
16 // `crewHoldsPlacedBefore` reads only holds an agent placed: a person's never reaches this loop.
17 for (const hold of crewHoldsPlacedBefore(db, now.toISOString())) {
18 const pull = byId.get(hold.pullId);
19 if (pull === undefined) continue;
20 const check = checkOn(db, pull);
21 const green = check?.state === "ok" && check.finishedAt !== null && check.finishedAt > hold.heldAt;
22 const old = now.getTime() - Date.parse(hold.heldAt) > HOLD_MAX_MS;
23 if (!green && !old) continue;
24 if (!releaseHold(db, hold.pullId, hold.heldBy)) continue;
25 const reason = green ? "a check went green on its head after the hold" : "the hold was older than 6 hours";
26 note(db, "keeper_release", pull.repoId, { pull: pull.id, number: pull.number, by: hold.heldByName, why: hold.why, reason });
27 lifted.push(`${nameOf(pull)}: lifted ${hold.heldByName}'s hold (${reason})`);
28 }
29 return lifted;
30 }
src/maintain/keeper/lull.ts added +94
@@ -0,0 +1,94 @@
1 /**
2 * Lulls (#7556): builders idle while startable issues exist, or checks idle while pulls wait,
3 * two samples in a row. The first such sample pair writes one `keeper_lull` start event naming
4 * the cause, in the words home's builders tile uses (`crew/idle_reason.ts`); the first sample
5 * after writes one end event. Nothing is written while nothing is startable.
6 */
7
8 import type { Database } from "bun:sqlite";
9 import type { App } from "../../app.ts";
10 import { listCrewRoles } from "../../core/agents.ts";
11 import { listAutonomousRepos } from "../../core/repos.ts";
12 import { listActive } from "../../core/runs.ts";
13 import { slottedRuns } from "../../crew/autonomy.ts";
14 import { heldNow } from "../../crew/held.ts";
15 import { whatBlocks } from "../../crew/idle_reason.ts";
16 import { checkOn, type KPull } from "./pulls.ts";
17 import { note } from "./record.ts";
18
19 export type LullKind = "builders" | "checks";
20
21 export interface Reading {
22 readonly kind: LullKind;
23 /** Work waits and nothing is taking it. */
24 readonly idle: boolean;
25 /** How much waits; the probe's number. */
26 readonly waiting: number;
27 /** Why, in the builders tile's words; read only when idle. */
28 readonly cause: string;
29 }
30
31 /** Samples in a row that make a lull. */
32 export const SAMPLES = 2;
33
34 /** Per database: consecutive idle samples by kind (a restart forgets them; a lull is then seen two samples later). */
35 const streaks = new WeakMap<Database, Map<LullKind, number>>();
36
37 /** The builders' reading: startable issues that no builder is taking, in the autonomous repositories with a crew. */
38 export function readBuilders(app: App, now: Date): Reading {
39 let waiting = 0;
40 let idle = false;
41 let cause = "no cause found";
42 for (const repo of listAutonomousRepos(app.db)) {
43 if (repo.autonomy !== "autonomous" || repo.autonomyBy === null || repo.archivedAt !== null) continue;
44 const crew = listCrewRoles(app.db, repo.id);
45 if (crew.length === 0) continue;
46 const held = heldNow(app, repo);
47 const startable = held.startable.length;
48 if (startable === 0) continue;
49 waiting += startable;
50 const active = listActive(app.db, "agent").filter((r) => r.repoId === repo.id);
51 if (slottedRuns(active, crew).length >= repo.autonomySlots) continue;
52 if (!idle) {
53 idle = true;
54 cause = whatBlocks(app, repo, now)?.word ?? (held.by.hot_file.length > 0 ? "hot file" : "no cause found");
55 }
56 }
57 return { kind: "builders", idle, waiting, cause };
58 }
59
60 /** The checks' reading: open pulls with no check on their head while no check or deploy is queued or running. */
61 export function readChecks(db: Database, pulls: readonly KPull[]): Reading {
62 const waiting = pulls.filter((p) => p.hold === undefined && checkOn(db, p) === undefined).length;
63 const busy = listActive(db, "check").length + listActive(db, "deploy").length;
64 return { kind: "checks", idle: waiting > 0 && busy === 0, waiting, cause: "no check is queued or running" };
65 }
66
67 /** Whether the last lull event of this kind began one that has not ended. */
68 export function lullOpen(db: Database, kind: LullKind): boolean {
69 const last = db
70 .query<{ phase: string | null }, [string]>(
71 `SELECT json_extract(payload, '$.phase') AS phase FROM event
72 WHERE kind = 'keeper_lull' AND json_extract(payload, '$.lull') = ?1 ORDER BY created_at DESC, rowid DESC LIMIT 1`,
73 )
74 .get(kind);
75 return last?.phase === "start";
76 }
77
78 /** One sample in; the line to report when it wrote an event, else undefined. */
79 export function sample(db: Database, reading: Reading): string | undefined {
80 let mine = streaks.get(db);
81 if (mine === undefined) streaks.set(db, (mine = new Map()));
82 const open = lullOpen(db, reading.kind);
83 if (!reading.idle) {
84 mine.set(reading.kind, 0);
85 if (!open) return undefined;
86 note(db, "keeper_lull", undefined, { lull: reading.kind, phase: "end", waiting: reading.waiting });
87 return `${reading.kind} lull ended`;
88 }
89 const streak = (mine.get(reading.kind) ?? 0) + 1;
90 mine.set(reading.kind, streak);
91 if (streak < SAMPLES || open) return undefined;
92 note(db, "keeper_lull", undefined, { lull: reading.kind, phase: "start", cause: reading.cause, waiting: reading.waiting });
93 return `${reading.kind} lull began: ${reading.cause}`;
94 }
src/maintain/keeper/pulls.ts added +82
@@ -0,0 +1,82 @@
1 /** The open pulls the flow keeper reads (#7556): one query, plain rows, and the few facts every part asks of them. */
2
3 import type { Database } from "bun:sqlite";
4 import { holdsFor, type Hold } from "../../core/holds.ts";
5 import { reviewSummariesFor } from "../../core/reviews.ts";
6
7 export interface KPull {
8 readonly id: string;
9 readonly repoId: string;
10 /** `owner/name`. */
11 readonly repo: string;
12 readonly number: number;
13 readonly authorId: string;
14 readonly assigneeId: string | null;
15 readonly headRepoId: string | null;
16 readonly headBranch: string;
17 readonly headOid: string;
18 readonly approved: boolean;
19 readonly hold: Hold | undefined;
20 }
21
22 interface Row {
23 id: string;
24 repo_id: string;
25 repo: string;
26 number: number;
27 author_id: string;
28 assignee_id: string | null;
29 head_repo_id: string | null;
30 head_branch: string;
31 head_oid: string;
32 }
33
34 /** Open, unmerged pulls in repositories the crew works on its own. */
35 export function openPulls(db: Database): KPull[] {
36 const rows = db
37 .query<Row, []>(
38 `SELECT issue.id, issue.repo_id, o.username || '/' || r.name AS repo, issue.number, issue.author_id,
39 issue.assignee_id, pull.head_repo_id, pull.head_branch, pull.head_oid
40 FROM issue JOIN pull ON pull.issue_id = issue.id
41 JOIN repo r ON r.id = issue.repo_id JOIN user o ON o.id = r.owner_id
42 WHERE issue.kind = 'pull' AND issue.state = 'open' AND pull.merged_at IS NULL
43 AND r.autonomy != 'off' AND r.archived_at IS NULL
44 ORDER BY issue.created_at, issue.id`,
45 )
46 .all();
47 const verdicts = reviewSummariesFor(db, rows.map((p) => ({ id: p.id, headOid: p.head_oid })));
48 const holds = holdsFor(db, rows.map((p) => p.id));
49 return rows.map((p) => {
50 const v = verdicts.get(p.id);
51 return {
52 id: p.id,
53 repoId: p.repo_id,
54 repo: p.repo,
55 number: p.number,
56 authorId: p.author_id,
57 assigneeId: p.assignee_id,
58 headRepoId: p.head_repo_id,
59 headBranch: p.head_branch,
60 headOid: p.head_oid,
61 approved: v !== undefined && v.approvals > 0 && v.changesRequested === 0,
62 hold: holds.get(p.id),
63 };
64 });
65 }
66
67 /** The newest check on the pull's head, if one has finished or is running. */
68 export function checkOn(db: Database, pull: Pick<KPull, "id" | "headOid">): { state: string; finishedAt: string | null } | undefined {
69 return (
70 db
71 .query<{ state: string; finishedAt: string | null }, [string, string]>(
72 `SELECT state, finished_at AS finishedAt FROM run
73 WHERE issue_id = ?1 AND commit_oid = ?2 AND kind IN ('check','deploy')
74 ORDER BY created_at DESC, id DESC LIMIT 1`,
75 )
76 .get(pull.id, pull.headOid) ?? undefined
77 );
78 }
79
80
81 /** The pull as its repository's words name it. */
82 export const nameOf = (p: Pick<KPull, "repo" | "number">): string => `${p.repo}#${p.number}`;
src/maintain/keeper/rebase.ts added +69
@@ -0,0 +1,69 @@
1 /**
2 * Conflicted approved pulls (#7556): each goes to the builder who can push to its branch
3 * (`ownBuilder`, the push check #7006 made), at most `CAP` per run. After `ATTEMPTS` re-assignments
4 * with no new head the keeper stops and says so on the pull, once per head.
5 */
6
7 import type { Database } from "bun:sqlite";
8 import { listCrewRoles, type Agent } from "../../core/agents.ts";
9 import { getIssueById, setAssignee } from "../../core/issues.ts";
10 import { getRepoById } from "../../core/repos.ts";
11 import { listActive } from "../../core/runs.ts";
12 import { ownBuilder } from "../../crew/finishing.ts";
13 import { nameOf, type KPull } from "./pulls.ts";
14 import { countNotes, note, sayOnce } from "./record.ts";
15
16 export const CAP = 6;
17 export const ATTEMPTS = 2;
18
19 export interface Moves {
20 readonly reassigned: string[];
21 readonly stopped: string[];
22 /** Conflicted pulls past the cap, left for the next run. */
23 readonly waiting: string[];
24 }
25
26 export function rebasePass(db: Database, pulls: readonly KPull[], conflicted: ReadonlySet<string>): Moves {
27 const out: Moves = { reassigned: [], stopped: [], waiting: [] };
28 const load = new Map<string, number>();
29 for (const run of listActive(db, "agent")) if (run.actorId !== null) load.set(run.actorId, (load.get(run.actorId) ?? 0) + 1);
30 const crews = new Map<string, Agent[]>();
31 for (const pull of pulls) {
32 if (!pull.approved || pull.hold !== undefined || !conflicted.has(nameOf(pull))) continue;
33 const issue = getIssueById(db, pull.id);
34 const repo = getRepoById(db, pull.repoId);
35 if (issue === undefined || repo === undefined) continue;
36 let builders = crews.get(pull.repoId);
37 if (builders === undefined) {
38 builders = listCrewRoles(db, pull.repoId).filter((c) => c.role === "builder").map((c) => c.agent);
39 crews.set(pull.repoId, builders);
40 }
41 // A person's name on it is a person holding it: reported by `handed.ts`, never moved.
42 if (issue.assigneeId !== null && !builders.some((b) => b.user.id === issue.assigneeId) && isPerson(db, issue.assigneeId)) continue;
43 const tried = countNotes(db, "keeper_rebase", "pull", pull.id, pull.headOid);
44 if (tried >= ATTEMPTS) {
45 const said = sayOnce(
46 db,
47 pull,
48 `rebase-stop:${pull.id}:${pull.headOid}`,
49 `this pull is approved and conflicts with ${repo.defaultBranch}; I assigned it to a builder ${tried} times and the head did not move, so I stop. Merge ${repo.defaultBranch} into it, or push a new head and I will look again.`,
50 );
51 if (said) out.stopped.push(nameOf(pull));
52 continue;
53 }
54 const builder = ownBuilder(issue, builders, load, pull.headRepoId === null ? undefined : getRepoById(db, pull.headRepoId)?.ownerId);
55 if (builder === undefined || builder.user.id === issue.assigneeId) continue;
56 if (out.reassigned.length >= CAP) {
57 out.waiting.push(nameOf(pull));
58 continue;
59 }
60 setAssignee(db, pull.id, builder.user.id);
61 note(db, "keeper_rebase", pull.repoId, { pull: pull.id, number: pull.number, head: pull.headOid, builder: builder.user.username });
62 load.set(builder.user.id, (load.get(builder.user.id) ?? 0) + 1);
63 out.reassigned.push(`${nameOf(pull)} to ${builder.user.username}`);
64 }
65 return out;
66 }
67
68 const isPerson = (db: Database, userId: string): boolean =>
69 db.query<{ n: number }, [string]>("SELECT count(*) AS n FROM user WHERE id = ?1 AND is_agent = 0").get(userId)!.n > 0;
src/maintain/keeper/record.ts added +41
@@ -0,0 +1,41 @@
1 /** What the flow keeper writes: events of its own kind, a probe for each finding, a line on a pull said once. */
2
3 import type { Database } from "bun:sqlite";
4 import { addComment } from "../../core/issues.ts";
5 import { EventBus } from "../../core/events.ts";
6 import type { Probe } from "../duty.ts";
7
8 export const probe = (what: string, number: number, now: Date): Probe => ({ kind: "measurement", what, number, at: now.toISOString() });
9
10 /** The keeper's event: durable, on the repository when there is one. */
11 export function note(db: Database, kind: string, repoId: string | undefined, payload: Record<string, unknown>): void {
12 new EventBus(db).append({ kind, repoId, payload });
13 }
14
15 /** How many events of `kind` carry `field` = `value` (and, when given, `head`). */
16 export function countNotes(db: Database, kind: string, field: string, value: string, head?: string): number {
17 return db
18 .query<{ n: number }, [string, string, string, string | null]>(
19 `SELECT count(*) AS n FROM event WHERE kind = ?1 AND json_extract(payload, '$.' || ?2) = ?3
20 AND (?4 IS NULL OR json_extract(payload, '$.head') = ?4)`,
21 )
22 .get(kind, field, value, head ?? null)!.n;
23 }
24
25 /** Who the forge speaks for on a pull: the person who turned autonomy on, else the owner (the pattern `crew/finishing.ts` uses for a hand-over). */
26 export function speaker(db: Database, repoId: string): string {
27 return db
28 .query<{ who: string }, [string]>("SELECT COALESCE(autonomy_by, owner_id) AS who FROM repo WHERE id = ?1")
29 .get(repoId)!.who;
30 }
31
32 export const SAID_UNDER = " (The forge posts this under your name; you did not write it.)";
33
34 /** One comment on a pull, tagged by `key`, written once however often it is asked. True when this call wrote it. */
35 export function sayOnce(db: Database, pull: { id: string; repoId: string }, key: string, body: string): boolean {
36 if (countNotes(db, "keeper_said", "key", key) > 0) return false;
37 const said = addComment(db, { issueId: pull.id, authorId: speaker(db, pull.repoId), body: `Flow keeper: ${body}${SAID_UNDER}` });
38 if (!said.ok) return false;
39 note(db, "keeper_said", pull.repoId, { key, pull: pull.id });
40 return true;
41 }
src/maintain/keeper/stuck.ts added +31
@@ -0,0 +1,31 @@
1 /**
2 * Stuck pulls (#7556): approved, open, no run or review event for 30 minutes, and not
3 * landable. Green on its head and approved is queue wait, not stuck (as `alarm_stuck.ts`).
4 */
5
6 import type { Database } from "bun:sqlite";
7 import { lastEvent, STUCK_MINUTES } from "../../flow/alarm_stuck.ts";
8 import { checkOn, nameOf, type KPull } from "./pulls.ts";
9
10 export interface Stuck {
11 readonly pull: KPull;
12 /** Why it cannot land. */
13 readonly why: "conflicted" | "red" | "unchecked";
14 readonly quietMs: number;
15 }
16
17 export function stuckPulls(db: Database, now: Date, pulls: readonly KPull[], conflicted: ReadonlySet<string>): Stuck[] {
18 const out: Stuck[] = [];
19 for (const pull of pulls) {
20 if (!pull.approved || pull.hold !== undefined) continue;
21 const check = checkOn(db, pull);
22 const isConflicted = conflicted.has(nameOf(pull));
23 if (check?.state === "ok" && !isConflicted) continue;
24 const last = lastEvent(db, { id: pull.id, repo_id: pull.repoId, head_repo_id: pull.headRepoId, head_branch: pull.headBranch });
25 if (last === undefined) continue;
26 const quietMs = now.getTime() - Date.parse(last.at);
27 if (quietMs < STUCK_MINUTES * 60_000) continue;
28 out.push({ pull, why: isConflicted ? "conflicted" : check?.state === "failed" ? "red" : "unchecked", quietMs });
29 }
30 return out.sort((a, b) => b.quietMs - a.quietMs);
31 }
src/maintain/keeper/words.ts added +21
@@ -0,0 +1,21 @@
1 /** The keeper's answer to each flow alarm (`flow/alarms.ts`), in one sentence. */
2
3 export interface Did {
4 readonly stuck: number;
5 readonly reassigned: number;
6 readonly lifted: number;
7 readonly down: boolean;
8 }
9
10 export function sentence(key: string, did: Did): string {
11 switch (key) {
12 case "stuck":
13 return `${did.stuck} stuck ${did.stuck === 1 ? "pull" : "pulls"} read; re-assigned ${did.reassigned} conflicted, lifted ${did.lifted} stale ${did.lifted === 1 ? "hold" : "holds"}.`;
14 case "idle_crew":
15 return "lull sampler watching; it writes a start event with the cause after two idle samples.";
16 case "unreachable":
17 return did.down ? "event written; the release keeper owns rollback." : "health sampled from two vantage points; no event until both fail twice.";
18 default:
19 return "reported only; the keeper acts on nothing here.";
20 }
21 }
src/maintain/registry.ts +2 −1
@@ -3,5 +3,6 @@
33 import type { Duty } from "./duty.ts";
44 import { duty as pullsWaiting } from "./duties/stale_pulls.ts";
55 import { duty as efficiencyWatcher } from "./duties/efficiency_watcher.ts";
6 import { duty as flowKeeper } from "./duties/flow_keeper.ts";
67
7 export const duties: readonly Duty[] = [pullsWaiting, efficiencyWatcher];
8 export const duties: readonly Duty[] = [pullsWaiting, efficiencyWatcher, flowKeeper];
test/event-audience-coverage.test.ts +1
@@ -20,6 +20,7 @@ import { classifiedAudience, eventAudience } from "../src/core/event_audience.ts
2020 */
2121 const COMPUTED: Record<string, readonly string[]> = {
2222 "src/actions/runs.ts kind": ["run.queued", "run.started", "run.finished"],
23 "src/maintain/keeper/record.ts kind": ["forge_down", "forge_up", "keeper_lull", "keeper_rebase", "keeper_release", "keeper_said"],
2324 };
2425
2526 /** The values of `${issue.kind}`: a thread is an issue or a pull. */
test/helpers/keeper-seed.ts added +86
@@ -0,0 +1,86 @@
1 /** A bare forge for the flow keeper's tests: rows written straight, so each test names the one fact it is about. */
2 import type { Database } from "bun:sqlite";
3 import { mkdtemp } from "node:fs/promises";
4 import { tmpdir } from "node:os";
5 import { join } from "node:path";
6 import { createApp, type App } from "../../src/app.ts";
7
8 export const ago = (now: Date, minutes: number): string => new Date(now.getTime() - minutes * 60_000).toISOString();
9
10 export interface Seed {
11 readonly app: App;
12 readonly db: Database;
13 readonly repoId: string;
14 readonly owner: string;
15 /** Builders on the crew, by name. */
16 readonly builders: Record<string, string>;
17 readonly person: string;
18 user(name: string, agent: boolean): string;
19 pull(o: { number: number; author: string; assignee?: string | null; head?: string; approved?: boolean; createdAt: string; headRepoId?: string }): string;
20 run(o: { pull: string; kind: string; state: string; head: string; at: string; reason?: string }): void;
21 hold(pull: string, by: string, at: string, untilGreen?: boolean): void;
22 }
23
24 let n = 0;
25 const nid = (p: string): string => `${p}${(++n).toString().padStart(6, "0")}`;
26
27 export async function seed(names: string[] = ["ada", "bo"]): Promise<Seed> {
28 const dir = await mkdtemp(join(tmpdir(), "forge-keeper-seed-"));
29 const app = createApp({ dataDir: dir, port: 0 });
30 const db = app.db;
31 const user = (name: string, agent: boolean): string => {
32 const id = nid("u");
33 db.query("INSERT INTO user (id, username, email, password_hash, is_agent) VALUES (?1, ?2, ?3, 'x', ?4)").run(id, name, `${name}@forge.test`, agent ? 1 : 0);
34 if (agent) db.query("INSERT INTO agent (user_id, roles) VALUES (?1, 'builder')").run(id);
35 return id;
36 };
37 const owner = user("owner", false);
38 const person = user("pat", false);
39 const repoId = nid("r");
40 db.query("INSERT INTO repo (id, owner_id, name, visibility, autonomy, autonomy_by) VALUES (?1, ?2, 'app', 'public', 'autonomous', ?2)").run(repoId, owner);
41 const builders: Record<string, string> = {};
42 for (const name of names) {
43 builders[name] = user(name, true);
44 db.query("INSERT INTO repo_crew (repo_id, user_id, role) VALUES (?1, ?2, 'builder')").run(repoId, builders[name]);
45 }
46 let number = 0;
47 return {
48 app,
49 db,
50 repoId,
51 owner,
52 builders,
53 person,
54 user,
55 pull(o) {
56 const id = nid("i");
57 const head = o.head ?? `head${o.number}`;
58 number = Math.max(number, o.number);
59 db.query("INSERT INTO issue (id, repo_id, number, kind, title, author_id, assignee_id, created_at, updated_at) VALUES (?1, ?2, ?3, 'pull', ?4, ?5, ?6, ?7, ?7)").run(
60 id, repoId, o.number, `pull ${o.number}`, o.author, o.assignee ?? null, o.createdAt,
61 );
62 db.query("INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id) VALUES (?1, 'main', ?2, 'base', ?3, ?4)").run(
63 id, `branch-${o.number}`, head, o.headRepoId ?? null,
64 );
65 if (o.approved !== false) {
66 db.query("INSERT INTO review (id, issue_id, author_id, state, commit_oid, created_at) VALUES (?1, ?2, ?3, 'approved', ?4, ?5)").run(nid("v"), id, owner, head, o.createdAt);
67 }
68 return id;
69 },
70 run(o) {
71 db.query("INSERT INTO run (id, repo_id, kind, issue_id, commit_oid, state, reason, created_at, started_at, finished_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8, ?8)").run(
72 nid("n"), repoId, o.kind, o.pull, o.head, o.state, o.reason ?? "", o.at,
73 );
74 },
75 hold(pull, by, at, untilGreen = false) {
76 db.query("INSERT INTO pull_hold (id, issue_id, held_by, why, held_at, until_green) VALUES (?1, ?2, ?3, 'wait', ?4, ?5)").run(nid("h"), pull, by, at, untilGreen ? 1 : 0);
77 },
78 };
79 }
80
81 /** One keeper run over the seed. */
82 export async function runKeeper(s: Seed, now: Date, deps: import("../../src/maintain/duties/flow_keeper.ts").Deps & { withApp?: boolean } = {}) {
83 const { flowKeeper } = await import("../../src/maintain/duties/flow_keeper.ts");
84 const { withApp, ...rest } = deps;
85 return flowKeeper({ builders: () => undefined, ...(rest.vantages === undefined && !withApp ? { vantages: () => [] } : {}), ...rest }).run({ db: s.db, now, ...(withApp ? { app: s.app } : {}) });
86 }
test/keeper-down-two-vantages.test.ts added +40
@@ -0,0 +1,40 @@
1 /** Forge down (#7556): health failing twice from two vantage points is an event; one bad vantage is not. */
2 import { expect, test } from "bun:test";
3 import { vantagesFromEnv, type Vantage } from "../src/maintain/keeper/down.ts";
4 import { runKeeper, seed } from "./helpers/keeper-seed.ts";
5
6 const now = new Date("2026-11-11T12:00:00Z");
7 const kinds = (s: Awaited<ReturnType<typeof seed>>) =>
8 s.db.query<{ kind: string }, []>("SELECT kind FROM event WHERE kind IN ('forge_down','forge_up') ORDER BY created_at, rowid").all().map((e) => e.kind);
9 const vantage = (name: string, up: { v: boolean }): Vantage => ({ name, ok: async () => up.v });
10
11 test("both vantages failing twice writes one forge_down; recovery writes forge_up", async () => {
12 const s = await seed();
13 const a = { v: false };
14 const b = { v: false };
15 const deps = { vantages: () => [vantage("home", a), vantage("far", b)] };
16 await runKeeper(s, now, deps);
17 expect(kinds(s)).toEqual([]);
18 const out = await runKeeper(s, now, deps);
19 expect(kinds(s)).toEqual(["forge_down"]);
20 expect(out.actions[0]).toBe("forge down: health failed twice from home and far; alarm only");
21 await runKeeper(s, now, deps);
22 expect(kinds(s)).toEqual(["forge_down"]);
23 a.v = true;
24 b.v = true;
25 await runKeeper(s, now, deps);
26 expect(kinds(s)).toEqual(["forge_down", "forge_up"]);
27 });
28
29 test("one vantage failing is a bad network, not an outage", async () => {
30 const s = await seed();
31 const deps = { vantages: () => [vantage("home", { v: false }), vantage("far", { v: true })] };
32 for (let i = 0; i < 4; i++) await runKeeper(s, now, deps);
33 expect(kinds(s)).toEqual([]);
34 });
35
36 test("fewer than two vantages configured can confirm nothing", () => {
37 expect(vantagesFromEnv("home=https://a.test/healthz")).toEqual([]);
38 expect(vantagesFromEnv("")).toEqual([]);
39 expect(vantagesFromEnv("a=https://a.test/h,b=ftp://x,c=https://c.test/h").map((v) => v.name)).toEqual(["a", "c"]);
40 });
test/keeper-handed.test.ts added +80
@@ -0,0 +1,80 @@
1 /** The flow keeper reports closed-without-reason and person-assigned crew pulls once, on the pull (#7556); and answers alarms in words. */
2 import { expect, test } from "bun:test";
3 import { sentence } from "../src/maintain/keeper/words.ts";
4 import { vantagesFromEnv } from "../src/maintain/keeper/down.ts";
5 import { ago, runKeeper, seed } from "./helpers/keeper-seed.ts";
6
7 const now = new Date("2026-11-11T12:00:00Z");
8
9 const comments = (s: Awaited<ReturnType<typeof seed>>, issue: string): string[] =>
10 s.db.query<{ body: string }, [string]>("SELECT body FROM comment WHERE issue_id = ?1 ORDER BY rowid").all(issue).map((c) => c.body);
11
12 function close(s: Awaited<ReturnType<typeof seed>>, id: string, reason: string | null): void {
13 s.db.query("UPDATE issue SET state = 'closed', close_reason = ?2, closed_at = ?3 WHERE id = ?1").run(id, reason, ago(now, 30));
14 }
15
16 test("a crew pull closed with no reason is said once on the pull", async () => {
17 const s = await seed();
18 const pull = s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 90) });
19 close(s, pull, null);
20 const first = await runKeeper(s, now);
21 expect(first.actions).toContain("owner/app#1: closed, said on the pull");
22 const second = await runKeeper(s, now);
23 expect(second.actions.filter((a) => a.includes("closed, said"))).toEqual([]);
24 const said = comments(s, pull);
25 expect(said).toHaveLength(1);
26 expect(said[0]).toContain("closed with no reason given");
27 });
28
29 test("a crew pull assigned to a person is said once, and stays theirs", async () => {
30 const s = await seed();
31 const pull = s.pull({ number: 2, author: s.builders.ada!, assignee: s.person, createdAt: ago(now, 90) });
32 const first = await runKeeper(s, now);
33 expect(first.actions).toContain("owner/app#2: handed to pat, said on the pull");
34 await runKeeper(s, now);
35 expect(comments(s, pull)).toHaveLength(1);
36 expect(s.db.query<{ a: string }, [string]>("SELECT assignee_id AS a FROM issue WHERE id = ?1").get(pull)!.a).toBe(s.person);
37 });
38
39 test("a closed pull with a reason says nothing", async () => {
40 const s = await seed();
41 const pull = s.pull({ number: 3, author: s.builders.ada!, createdAt: ago(now, 90) });
42 close(s, pull, "dropped");
43 const out = await runKeeper(s, now);
44 expect(out.actions.filter((a) => a.includes("said on the pull"))).toEqual([]);
45 expect(comments(s, pull)).toEqual([]);
46 });
47
48 test("a firing alarm yields a finding and an action line from words.ts", async () => {
49 const s = await seed();
50 s.db.query("INSERT INTO issue (id, repo_id, number, kind, title, author_id, created_at, updated_at) VALUES ('ready1', ?1, 9, 'issue', 'ready', ?2, ?3, ?3)").run(s.repoId, s.person, ago(now, 120));
51 const out = await runKeeper(s, now);
52 expect(out.findings.some((f) => f.summary.startsWith("alarm idle_crew:"))).toBe(true);
53 expect(out.actions).toContain(`alarm idle_crew: ${sentence("idle_crew", { stuck: 0, reassigned: 0, lifted: 0, down: false })}`);
54 });
55
56 test("words answer idle_crew and unreachable, with and without an outage", () => {
57 const did = { stuck: 0, reassigned: 0, lifted: 0, down: false };
58 expect(sentence("idle_crew", did)).toContain("lull sampler");
59 expect(sentence("unreachable", did)).toContain("no event until both fail twice");
60 expect(sentence("unreachable", { ...did, down: true })).toContain("release keeper owns rollback");
61 expect(sentence("other", did)).toContain("reported only");
62 });
63
64 test("forge_down is reached through the default runner path: env vantages, app passed", async () => {
65 const server = Bun.serve({ port: 0, hostname: "127.0.0.1", fetch: () => new Response("no", { status: 500 }) });
66 const old = process.env.FORGE_HEALTH_VANTAGES;
67 process.env.FORGE_HEALTH_VANTAGES = `home=http://127.0.0.1:${server.port}/h,far=http://127.0.0.1:${server.port}/h`;
68 try {
69 expect(vantagesFromEnv()).toHaveLength(2);
70 const s = await seed();
71 await runKeeper(s, now, { withApp: true });
72 const out = await runKeeper(s, now, { withApp: true });
73 expect(out.actions.some((a) => a.startsWith("forge down: health failed twice"))).toBe(true);
74 expect(s.db.query<{ n: number }, []>("SELECT count(*) AS n FROM event WHERE kind = 'forge_down'").get()!.n).toBe(1);
75 } finally {
76 if (old === undefined) delete process.env.FORGE_HEALTH_VANTAGES;
77 else process.env.FORGE_HEALTH_VANTAGES = old;
78 server.stop(true);
79 }
80 });
test/keeper-holds.test.ts added +35
@@ -0,0 +1,35 @@
1 /** Wrongly held pulls (#7556): a stale or answered crew hold is lifted; a person's never. */
2 import { expect, test } from "bun:test";
3 import { getHold } from "../src/core/holds.ts";
4 import { ago, runKeeper, seed } from "./helpers/keeper-seed.ts";
5
6 const now = new Date("2026-11-11T12:00:00Z");
7
8 test("a person's hold stays however old; a crew hold older than 6 hours is lifted", async () => {
9 const s = await seed();
10 const mine = s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 900) });
11 s.hold(mine, s.person, ago(now, 800));
12 const stale = s.pull({ number: 2, author: s.builders.ada!, createdAt: ago(now, 900) });
13 s.hold(stale, s.builders.bo!, ago(now, 400));
14 const fresh = s.pull({ number: 3, author: s.builders.ada!, createdAt: ago(now, 900) });
15 s.hold(fresh, s.builders.bo!, ago(now, 60));
16 const out = await runKeeper(s, now);
17 expect(getHold(s.db, mine)).toBeDefined();
18 expect(getHold(s.db, stale)).toBeUndefined();
19 expect(getHold(s.db, fresh)).toBeDefined();
20 expect(out.actions.filter((a) => a.includes("lifted bo"))).toEqual(["owner/app#2: lifted bo's hold (the hold was older than 6 hours)"]);
21 });
22
23 test("a crew hold with a check green on its head after the hold is lifted", async () => {
24 const s = await seed();
25 const pull = s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 120) });
26 s.hold(pull, s.builders.bo!, ago(now, 60));
27 s.run({ pull, kind: "check", state: "ok", head: "head1", at: ago(now, 20) });
28 const before = s.pull({ number: 2, author: s.builders.ada!, createdAt: ago(now, 120) });
29 s.hold(before, s.builders.bo!, ago(now, 10));
30 s.run({ pull: before, kind: "check", state: "ok", head: "head2", at: ago(now, 50) }); // green came before the hold
31 const out = await runKeeper(s, now);
32 expect(getHold(s.db, pull)).toBeUndefined();
33 expect(getHold(s.db, before)).toBeDefined();
34 expect(out.actions.join("\n")).toContain("a check went green on its head after the hold");
35 });
test/keeper-lull.test.ts added +43
@@ -0,0 +1,43 @@
1 /** Lulls (#7556): two idle samples write one start event with its cause; the first busy one writes the end. */
2 import { expect, test } from "bun:test";
3 import type { Reading } from "../src/maintain/keeper/lull.ts";
4 import { ago, runKeeper, seed } from "./helpers/keeper-seed.ts";
5
6 const now = new Date("2026-11-11T12:00:00Z");
7 const events = (s: Awaited<ReturnType<typeof seed>>) =>
8 s.db.query<{ payload: string }, []>("SELECT payload FROM event WHERE kind = 'keeper_lull' ORDER BY created_at, rowid").all().map((e) => JSON.parse(e.payload));
9 const builders = (idle: boolean, cause = "ceiling reached"): Reading => ({ kind: "builders", idle, waiting: idle ? 3 : 0, cause });
10
11 test("a lull writes one start event with its cause and one end event", async () => {
12 const s = await seed();
13 const run = (idle: boolean) => runKeeper(s, now, { builders: () => builders(idle) });
14 await run(true);
15 expect(events(s)).toEqual([]);
16 const second = await run(true);
17 expect(events(s)).toEqual([{ lull: "builders", phase: "start", cause: "ceiling reached", waiting: 3 }]);
18 expect(second.actions).toContain("builders lull began: ceiling reached");
19 await run(true);
20 expect(events(s).length).toBe(1);
21 await run(false);
22 expect(events(s).map((e) => e.phase)).toEqual(["start", "end"]);
23 await run(false);
24 expect(events(s).length).toBe(2);
25 });
26
27 test("nothing is written while nothing is startable, and one idle sample is no lull", async () => {
28 const s = await seed();
29 for (let i = 0; i < 4; i++) await runKeeper(s, now, { builders: () => builders(false) });
30 expect(events(s)).toEqual([]);
31 await runKeeper(s, now, { builders: () => builders(true) });
32 await runKeeper(s, now, { builders: () => builders(false) });
33 await runKeeper(s, now, { builders: () => builders(true) });
34 expect(events(s)).toEqual([]);
35 });
36
37 test("checks idle while a pull waits for one is a lull of its own", async () => {
38 const s = await seed();
39 s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 5) });
40 await runKeeper(s, now);
41 await runKeeper(s, now);
42 expect(events(s)).toEqual([{ lull: "checks", phase: "start", cause: "no check is queued or running", waiting: 1 }]);
43 });
test/keeper-rebase-loop.test.ts added +58
@@ -0,0 +1,58 @@
1 /** Conflicted approved pulls go to a builder who can push; two tries with no new head stop (#7556). */
2 import { expect, test } from "bun:test";
3 import { ago, runKeeper, seed } from "./helpers/keeper-seed.ts";
4
5 const now = new Date("2026-11-11T12:00:00Z");
6 const conflicted = (...n: number[]) => ({ conflicted: () => new Set(n.map((x) => `owner/app#${x}`)) });
7 const assignee = (s: Awaited<ReturnType<typeof seed>>, pull: string) =>
8 s.db.query<{ a: string | null }, [string]>("SELECT assignee_id AS a FROM issue WHERE id = ?1").get(pull)!.a;
9 const comments = (s: Awaited<ReturnType<typeof seed>>, pull: string) =>
10 s.db.query<{ n: number }, [string]>("SELECT count(*) AS n FROM comment WHERE issue_id = ?1 AND body LIKE 'Flow keeper:%'").get(pull)!.n;
11
12 test("a conflicted approved pull is re-assigned in one run to the builder who authored it, and not a third time on the same head", async () => {
13 const s = await seed();
14 const { ada, bo } = s.builders;
15 const pull = s.pull({ number: 1, author: ada!, assignee: bo!, createdAt: ago(now, 60) });
16 const first = await runKeeper(s, now, conflicted(1));
17 expect(first.actions).toContain("re-assigned owner/app#1 to ada");
18 expect(assignee(s, pull)).toBe(ada!);
19
20 s.db.query("UPDATE issue SET assignee_id = ?2 WHERE id = ?1").run(pull, bo!); // still stuck: someone moved it again
21 const second = await runKeeper(s, now, conflicted(1));
22 expect(second.actions).toContain("re-assigned owner/app#1 to ada");
23
24 s.db.query("UPDATE issue SET assignee_id = ?2 WHERE id = ?1").run(pull, bo!);
25 const third = await runKeeper(s, now, conflicted(1));
26 expect(third.actions.some((a) => a.startsWith("re-assigned"))).toBe(false);
27 expect(third.actions.some((a) => a.startsWith("stopped re-assigning owner/app#1"))).toBe(true);
28 expect(assignee(s, pull)).toBe(bo!);
29 expect(comments(s, pull)).toBe(1);
30
31 const fourth = await runKeeper(s, now, conflicted(1));
32 expect(fourth.actions.some((a) => a.startsWith("stopped"))).toBe(false);
33 expect(comments(s, pull)).toBe(1);
34
35 // A new head is a new chance.
36 s.db.query("UPDATE pull SET head_oid = 'newhead' WHERE issue_id = ?1").run(pull);
37 s.db.query("UPDATE review SET commit_oid = 'newhead' WHERE issue_id = ?1").run(pull);
38 const fifth = await runKeeper(s, now, conflicted(1));
39 expect(fifth.actions).toContain("re-assigned owner/app#1 to ada");
40 });
41
42 test("at most six per run; the rest wait for the next", async () => {
43 const s = await seed();
44 for (let i = 1; i <= 8; i++) s.pull({ number: i, author: s.builders.ada!, assignee: s.builders.bo!, createdAt: ago(now, 60) });
45 const out = await runKeeper(s, now, conflicted(1, 2, 3, 4, 5, 6, 7, 8));
46 expect(out.actions.filter((a) => a.startsWith("re-assigned")).length).toBe(6);
47 expect(out.findings.some((f) => f.summary.startsWith("2 conflicted pulls wait"))).toBe(true);
48 });
49
50 test("a pull a person holds is left alone, and so is one a person is assigned", async () => {
51 const s = await seed();
52 const held = s.pull({ number: 1, author: s.builders.ada!, assignee: s.builders.bo!, createdAt: ago(now, 60) });
53 s.hold(held, s.person, ago(now, 30));
54 const handed = s.pull({ number: 2, author: s.builders.ada!, assignee: s.person, createdAt: ago(now, 60) });
55 const out = await runKeeper(s, now, conflicted(1, 2));
56 expect(out.actions.some((a) => a.startsWith("re-assigned"))).toBe(false);
57 expect(assignee(s, handed)).toBe(s.person);
58 });
test/keeper-stuck.test.ts added +37
@@ -0,0 +1,37 @@
1 /** The flow keeper reads stuck pulls (#7556): approved, quiet 30 minutes, not landable; and answers the stuck alarm. */
2 import { expect, test } from "bun:test";
3 import { ago, runKeeper, seed } from "./helpers/keeper-seed.ts";
4
5 const now = new Date("2026-11-11T12:00:00Z");
6
7 test("an approved pull with a red check and no event for 40 minutes is stuck, and the alarm gets an answer", async () => {
8 const s = await seed();
9 const pull = s.pull({ number: 1, author: s.builders.ada!, assignee: s.builders.ada!, createdAt: ago(now, 40) });
10 s.run({ pull, kind: "check", state: "failed", head: "head1", at: ago(now, 40) });
11 const out = await runKeeper(s, now);
12 expect(out.findings.map((f) => f.summary)).toContain("owner/app#1 is approved and stuck: red, quiet for 40 min");
13 expect(out.actions.some((a) => a.startsWith("alarm stuck: 1 stuck pull read"))).toBe(true);
14 expect(out.findings.every((f) => f.probe !== undefined)).toBe(true);
15 });
16
17 test("conflicted and unchecked say so", async () => {
18 const s = await seed();
19 s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 90) });
20 s.pull({ number: 2, author: s.builders.ada!, createdAt: ago(now, 90) });
21 const out = await runKeeper(s, now, { conflicted: () => new Set(["owner/app#1"]) });
22 const says = out.findings.map((f) => f.summary);
23 expect(says).toContain("owner/app#1 is approved and stuck: conflicted, quiet for 90 min");
24 expect(says).toContain("owner/app#2 is approved and stuck: unchecked, quiet for 90 min");
25 });
26
27 test("green and approved on its head is queue wait; a recent event, a hold or no approval is not stuck", async () => {
28 const s = await seed();
29 const green = s.pull({ number: 1, author: s.builders.ada!, createdAt: ago(now, 90) });
30 s.run({ pull: green, kind: "check", state: "ok", head: "head1", at: ago(now, 80) });
31 s.pull({ number: 2, author: s.builders.ada!, createdAt: ago(now, 10) });
32 const held = s.pull({ number: 3, author: s.builders.ada!, createdAt: ago(now, 90) });
33 s.hold(held, s.person, ago(now, 80));
34 s.pull({ number: 4, author: s.builders.ada!, createdAt: ago(now, 90), approved: false });
35 const out = await runKeeper(s, now);
36 expect(out.findings.filter((f) => f.summary.includes("approved and stuck"))).toEqual([]);
37 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs