trevato/forge

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

More
Forks1g fMapg aRuns13g rEventsFlows

The throughput keeper: the forge watches its own landing rate, finds what binds it, and acts #7800

Merged scoutcrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

scout/forge:scout/7645-the-throughput-keeper-the-forge-watches → main 10 commits fork

  • Merged. scout/7645-the-throughput-keeper-the-forge-watches was merged into main as 0fa9759.

Review

Sign in to review this change.
20 files changed +1038 −8
README.md +4
@@ -400,6 +400,10 @@ days drawn beside it, on its page, in the terminal's Milestones screen and in
400400 the MCP's `milestones` tool (`closedPerDay`, `paceDate`); `docs/PLANNING.md`
401401 has the rule.
402402
403 ### The throughput keeper
404
405 Once a minute the forge measures its landing rate, names the one cause binding it, and lifts Flow only or unsticks assigned issues when that is its to do; if landings do not rise in 2 h it puts the switch back and asks the owner one question with one button. It shows as a line on home and `/boards` (a site admin sees repository names; everyone else a repository-free cause), in the MCP `flow` report, in the terminal's Flow dialog (`l` answers) and as "Forge per day" on the crew scorecard. No model, no dependency. See `docs/features/keeper.md`.
406
403407 ## Where the rest is
404408
405409 - [`docs/DEVELOP.md`](docs/DEVELOP.md) — developing the forge: tests, checks, the gate, lint and types.
docs/features/keeper.md added +14
@@ -0,0 +1,14 @@
1 # The throughput keeper (#7645)
2
3 The forge watches its own landing rate, names the one thing binding it, and acts where the act is reversible and its own. `src/flow/keeper.ts` ticks once a minute. It reads the database and asks no model, so it still speaks when the subscription is out.
4
5 1. **Measure** (`keeper_rate.ts`): pulls landed per hour over the last 3 h against the best 3 h rate of the week before, and the share of the last 24 hours that ran (a landing or a builder run started). Low is under half the best and under one an hour.
6 2. **Find** (`keeper_cause.ts`): the first of `ceiling`, `usage` ("resets at HH:MM UTC"), `unassigned`, `empty_pillar`, `red_checks`, `stale_pile`, `capacity`. It asks the dispatcher's own questions (`heldNow`, `atCeiling`, `pace`).
7 3. **Act**: `unassigned` sweeps assigned issues with no run. `empty_pillar`, `red_checks` and `stale_pile` become one filed issue in the forge's own repository, once a week per cause. `ceiling`, `usage` and `capacity` are only said.
8 4. **Check** after 2 h: landings rose and the act stands, or it is undone, the cause is left alone for 6 h and the owner is asked one question with one button (`POST /flow/keeper/answer`).
9
10 Where it shows: one line on home and on the boards panel, `keeper` in the MCP `flow` report, the terminal's Flow dialog (the line, the question, and `l` for its one button after a confirm) and its crew scorecard ("Forge per day" rows), and "Forge per day" (hours running, landed) on the crew scorecard and in `scorecard` (`keeperDays`). Every attempt is a row in `keeper_attempt`.
11
12 Not done: DORA change-fail rate, revert rate, queue-wait p95 and rework per agent from the issue discussion are not measured yet.
13
14 Who reads what: a site admin reads the sentence with repository names and the question with its button. Everyone else, guests and agents included, reads the cause with no repository in it (`readKeeper(db, full)`). The Flow only switch is retired (#7771), so no act asks a question today; one already on the books is answered with its button, and the answer is a `keeper.answered` event carrying the owner's id.
src/actions/scorecard.ts +4
@@ -15,6 +15,7 @@ import { modelDays, type ModelDay } from "../core/model_days.ts";
1515 import { readReview } from "./model_review.ts";
1616 import { wastedDays, type WastedDay } from "../core/wasted_days.ts";
1717 import { listAgents } from "../core/agents.ts";
18 import { keeperDays, type KeeperDay } from "../flow/keeper_rate.ts";
1819 import type { User } from "../core/users.ts";
1920
2021 export interface AgentScorecard {
@@ -47,6 +48,7 @@ export function readScorecards(
4748 agents: AgentScorecard[];
4849 modelDays: ModelDay[];
4950 wastedDays: WastedDay[];
51 keeperDays: KeeperDay[];
5052 modelReview: ReturnType<typeof readReview>;
5153 } | undefined {
5254 if (actor?.isAdmin !== true) return undefined;
@@ -58,6 +60,8 @@ export function readScorecards(
5860 const key = (p: SplitScorecard) => `${p.model}\u0000${p.promptVersion}`;
5961 return {
6062 modelDays: modelDays(app.db),
63 // Hours the forge ran and pulls landed, per UTC day of the last week (#7645).
64 keeperDays: keeperDays(app.db, new Date()),
6165 modelReview: readReview(app),
6266 wastedDays: wastedDays(app.db),
6367 agents: listAgents(app.db).map((agent) => {
src/core/event_audience.ts +2
@@ -161,6 +161,8 @@ export function classifiedAudience(kind: string): EventAudience | undefined {
161161 kind.startsWith("crew.model_") ||
162162 // A clog is the whole forge's flow, the owner's to hear (`flow/clog_tick.ts`).
163163 kind.startsWith("clog.") ||
164 // What the throughput keeper did to lift a brake is the owner's too (`flow/keeper.ts`).
165 kind.startsWith("keeper.") ||
164166 // Events the retired FLOW ONLY switch wrote stay the admins' to hear.
165167 kind === "flow_only.cleared" ||
166168 // A report from outside waits for a person who decides; its arrival is
src/core/migration-pins/20261101130000-keeper.pin added +1
@@ -0,0 +1,1 @@
1 e10ae5d8c6a6c7afdd71f28231f6e26c25db781f10ea3138fac5793accb5d3fb
src/core/migrations/20261101130000-keeper.ts added +24
@@ -0,0 +1,24 @@
1 // The throughput keeper (#7645): what it found binding now (one row), and
2 // each attempt it made, asked or filed (`flow/keeper.ts`). An attempt waits
3 // `pending` until its effect is checked, then is `worked` or `undone`; `asked`
4 // is a question for the owner with one button; `filed` is an issue it opened
5 // once for a cause it cannot act on.
6 export default `
7 CREATE TABLE IF NOT EXISTS keeper_state (
8 id INTEGER PRIMARY KEY CHECK (id = 1),
9 cause TEXT NOT NULL DEFAULT '',
10 says TEXT NOT NULL DEFAULT '',
11 at INTEGER NOT NULL DEFAULT 0
12 );
13 CREATE TABLE IF NOT EXISTS keeper_attempt (
14 id INTEGER PRIMARY KEY AUTOINCREMENT,
15 at INTEGER NOT NULL,
16 cause TEXT NOT NULL,
17 action TEXT NOT NULL,
18 repo_id TEXT NOT NULL DEFAULT '',
19 says TEXT NOT NULL DEFAULT '',
20 outcome TEXT NOT NULL CHECK (outcome IN ('pending','worked','undone','asked','answered','filed')),
21 check_at INTEGER NOT NULL DEFAULT 0
22 );
23 CREATE INDEX IF NOT EXISTS keeper_attempt_at ON keeper_attempt(at DESC);
24 `;
src/flow/keeper.ts added +255
@@ -0,0 +1,255 @@
1 /**
2 * The throughput keeper (#7645): once a minute it measures the landing rate
3 * (`keeper_rate.ts`); when it is low it names the one cause binding it
4 * (`keeper_cause.ts`) and acts only where the act is reversible and its own:
5 * unsticking assigned issues with no run. After `EFFECT_MS` it checks whether
6 * landings rose; if not it undoes the act, remembers the cause as tried and
7 * asks the owner one question with one button. A cause it cannot act on is
8 * filed as one issue, once. It reads the database and spends no model tokens,
9 * so it still speaks when the subscription is out.
10 */
11
12 import type { Database } from "bun:sqlite";
13 import type { App } from "../app.ts";
14 import { createIssue } from "../core/issues.ts";
15 import { getRepo } from "../core/repos.ts";
16 import { createLog } from "../kernel/log.ts";
17 import { ownRepoPath } from "../kernel/own_repo.ts";
18 import { KEEPER_CAUSES, pickCause, readKeeperFacts, type KeeperCause, type KeeperMove } from "./keeper_cause.ts";
19 import { isLow, landedBetween, readRate, type Rate } from "./keeper_rate.ts";
20
21 const log = createLog("keeper");
22 const HOUR_MS = 3_600_000;
23 /** How long after an act landings are compared with the same span before it. */
24 export const EFFECT_MS = 2 * HOUR_MS;
25 /** A cause that did not help is left alone this long. */
26 export const TRIED_MS = 6 * HOUR_MS;
27 /** A cause filed as an issue is not filed again this long. */
28 export const FILE_ONCE_MS = 7 * 24 * HOUR_MS;
29
30 type Outcome = "pending" | "worked" | "undone" | "asked" | "answered" | "filed";
31
32 export interface Attempt {
33 readonly id: number;
34 readonly at: number;
35 readonly cause: KeeperCause;
36 readonly action: string;
37 readonly repoId: string;
38 readonly says: string;
39 readonly outcome: Outcome;
40 readonly checkAt: number;
41 }
42
43 interface AttemptRow {
44 id: number;
45 at: number;
46 cause: string;
47 action: string;
48 repo_id: string;
49 says: string;
50 outcome: string;
51 check_at: number;
52 }
53
54 const toAttempt = (r: AttemptRow): Attempt => ({
55 id: r.id,
56 at: r.at,
57 cause: r.cause as KeeperCause,
58 action: r.action,
59 repoId: r.repo_id,
60 says: r.says,
61 outcome: r.outcome as Outcome,
62 checkAt: r.check_at,
63 });
64
65 const COLS = "id, at, cause, action, repo_id, says, outcome, check_at";
66
67 export function listAttempts(db: Database, limit = 20): Attempt[] {
68 return db.query<AttemptRow, [number]>(`SELECT ${COLS} FROM keeper_attempt ORDER BY id DESC LIMIT ?1`).all(limit).map(toAttempt);
69 }
70
71 function record(db: Database, a: Omit<Attempt, "id">): number {
72 return Number(
73 db
74 .query("INSERT INTO keeper_attempt (at, cause, action, repo_id, says, outcome, check_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)")
75 .run(a.at, a.cause, a.action, a.repoId, a.says, a.outcome, a.checkAt).lastInsertRowid,
76 );
77 }
78
79 const setOutcome = (db: Database, id: number, outcome: Outcome): void => {
80 db.query("UPDATE keeper_attempt SET outcome = ?2 WHERE id = ?1").run(id, outcome);
81 };
82
83 /** Causes that were undone lately, or filed within the week: not tried again. */
84 function skipped(db: Database, now: number): Set<KeeperCause> {
85 const rows = db
86 .query<{ cause: string; outcome: string }, [number, number]>(
87 `SELECT cause, outcome FROM keeper_attempt
88 WHERE (outcome IN ('undone','asked') AND at >= ?1) OR (outcome = 'filed' AND at >= ?2)`,
89 )
90 .all(now - TRIED_MS, now - FILE_ONCE_MS);
91 return new Set(rows.map((r) => r.cause).filter((c): c is KeeperCause => (KEEPER_CAUSES as readonly string[]).includes(c)));
92 }
93
94 /** Checks every act whose time is up: landings rose and it stands, or it is undone and the owner is asked. */
95 function settle(app: App, now: number): void {
96 const { db } = app;
97 const due = db
98 .query<AttemptRow, [number]>(`SELECT ${COLS} FROM keeper_attempt WHERE outcome = 'pending' AND check_at <= ?1`)
99 .all(now)
100 .map(toAttempt);
101 for (const a of due) {
102 const after = landedBetween(db, a.at, now);
103 const before = landedBetween(db, a.at - (now - a.at), a.at);
104 if (after > before) {
105 setOutcome(db, a.id, "worked");
106 continue;
107 }
108 setOutcome(db, a.id, "undone");
109 app.bus.append({ kind: "keeper.undone", repoId: a.repoId, payload: { cause: a.cause, action: a.action } });
110 }
111 }
112
113 /** Files one issue for a cause the keeper cannot act on, in the forge's own repository. */
114 function fileOnce(app: App, found: { cause: KeeperCause; says: string }, rate: Rate, now: number): void {
115 const [owner, name] = ownRepoPath().split("/");
116 const repo = owner && name ? getRepo(app.db, owner, name) : undefined;
117 if (!repo) return log.warn("keeper cannot file: the forge's own repository is missing", { path: ownRepoPath() });
118 const made = createIssue(app.db, {
119 repoId: repo.id,
120 authorId: repo.ownerId,
121 title: `Throughput: ${found.says}`.slice(0, 200),
122 body: `The throughput keeper found this binding landings (${rate.recent.toFixed(1)} an hour against a best of ${rate.best.toFixed(1)}) and cannot act on it itself.\n\n${found.says}\n\nFiles: docs/features/keeper.md`,
123 });
124 if (!made.ok) return;
125 app.bus.append({ kind: "issue.opened", repoId: repo.id, payload: { number: made.value.number } });
126 record(app.db, { at: now, cause: found.cause, action: "file", repoId: repo.id, says: `Filed #${made.value.number}`, outcome: "filed", checkAt: 0 });
127 }
128
129 function saveState(db: Database, cause: string, says: string, now: number): void {
130 db.query(
131 `INSERT INTO keeper_state (id, cause, says, at) VALUES (1, ?1, ?2, ?3)
132 ON CONFLICT (id) DO UPDATE SET cause = ?1, says = ?2, at = ?3`,
133 ).run(cause, says, now);
134 }
135
136 export interface KeeperTick {
137 readonly cause: KeeperCause | null;
138 readonly acted: KeeperMove | null;
139 }
140
141 /** One tick. `onLift` runs after a brake is lifted (a sweep of assigned issues). */
142 export function tickKeeper(app: App, now = new Date(), onLift: () => void = () => log.warn("keeper lifted a brake with no sweep to run")): KeeperTick {
143 const { db } = app;
144 const t = now.getTime();
145 settle(app, t);
146 const rate = readRate(db, now);
147 if (!isLow(rate)) {
148 saveState(db, "", "", t);
149 return { cause: null, acted: null };
150 }
151 const found = pickCause(readKeeperFacts(app, now), skipped(db, t));
152 if (!found) {
153 saveState(db, "", "", t);
154 return { cause: null, acted: null };
155 }
156 // One act at a time: an act waiting on its check is left to show its effect, and its cause stays named.
157 const waiting = db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM keeper_attempt WHERE outcome = 'pending'").get();
158 if ((waiting?.n ?? 0) > 0) return { cause: found.cause, acted: null };
159 saveState(db, found.cause, found.says, t);
160 if (found.move === "unstick") {
161 onLift();
162 record(db, { at: t, cause: found.cause, action: found.move, repoId: found.repoId, says: found.says, outcome: "pending", checkAt: t + EFFECT_MS });
163 app.bus.append({ kind: "keeper.acted", repoId: found.repoId, payload: { cause: found.cause, action: found.move, says: found.says } });
164 return { cause: found.cause, acted: found.move };
165 }
166 if (found.move === "file") {
167 fileOnce(app, found, rate, t);
168 return { cause: found.cause, acted: found.move };
169 }
170 return { cause: found.cause, acted: null };
171 }
172
173 /** Ticks once a minute; a tick that throws is logged and the next one runs. Returns the stop. */
174 export function startKeeper(app: App, onLift: () => void, everyMs = 60_000, now: () => number = Date.now): () => void {
175 const timer = setInterval(() => {
176 try {
177 tickKeeper(app, new Date(now()), onLift);
178 } catch (error) {
179 log.warn("keeper tick failed", { error: String(error) });
180 }
181 }, everyMs);
182 return () => clearInterval(timer);
183 }
184
185 export interface KeeperAsk {
186 readonly id: number;
187 readonly question: string;
188 readonly button: string;
189 }
190
191 export interface KeeperRead {
192 /** The cause binding landings now, one sentence; null while landings are fine. */
193 readonly cause: string | null;
194 readonly name: KeeperCause | null;
195 /** The last thing it did, one sentence; null before it did anything. */
196 readonly last: string | null;
197 readonly ask: KeeperAsk | null;
198 }
199
200 const ACTION_WORDS: Record<string, string> = {
201 unstick: "unstuck assigned issues",
202 file: "filed an issue",
203 };
204
205 const clock = (ms: number): string => new Date(ms).toISOString().slice(11, 16);
206
207 /** The cause with no repository named in it: what a reader who may not see every repository is told. */
208 const GENERIC: Record<KeeperCause, string> = {
209 ceiling: "A daily ceiling is reached: builders start again at 00:00 UTC.",
210 usage: "Usage is the limit: no subscription has headroom.",
211 unassigned: "Ready issues wait and no builder is running.",
212 empty_pillar: "Nothing is ready: no pillar has an issue a builder can start.",
213 red_checks: "Open crew pulls have a red check.",
214 stale_pile: "Open crew pulls are waiting on a builder or a reviewer.",
215 capacity: "Every builder slot is busy.",
216 };
217
218 /**
219 * What home, the pipeline and the MCP show: cheap, three small reads. Only
220 * `full` (a site admin) reads the sentence that names repositories and the
221 * question; anyone else reads the cause with no repository in it.
222 */
223 export function readKeeper(db: Database, full: boolean): KeeperRead {
224 const state = db.query<{ cause: string; says: string }, []>("SELECT cause, says FROM keeper_state WHERE id = 1").get();
225 const acts = db
226 .query<AttemptRow, []>(`SELECT ${COLS} FROM keeper_attempt WHERE outcome IN ('pending','worked','undone','filed') ORDER BY id DESC LIMIT 1`)
227 .all()
228 .map(toAttempt)[0];
229 const asked = db.query<AttemptRow, []>(`SELECT ${COLS} FROM keeper_attempt WHERE outcome = 'asked' ORDER BY id DESC LIMIT 1`).all().map(toAttempt)[0];
230 const last = acts
231 ? `At ${clock(acts.at)} UTC it ${ACTION_WORDS[acts.action] ?? acts.action}${acts.outcome === "undone" ? ": landings did not rise" : acts.outcome === "worked" ? ": landings rose" : acts.outcome === "pending" ? ", checking the effect" : ""}.`
232 : null;
233 return {
234 cause: state && state.cause !== "" ? (full ? state.says : (GENERIC[state.cause as KeeperCause] ?? "")) || null : null,
235 name: state && state.cause !== "" ? (state.cause as KeeperCause) : null,
236 last,
237 // No act asks a question since FLOW ONLY left (#7771); a question already on the books is still answerable.
238 ask: full && asked ? { id: asked.id, question: asked.says, button: "Got it" } : null,
239 };
240 }
241
242 /** The line for a page: the cause and the last act, in one sentence run; empty when there is neither. */
243 export function keeperLine(read: KeeperRead): string {
244 return [read.cause === null ? "" : `Binding landings: ${read.cause}`, read.last ?? ""].filter((s) => s !== "").join(" ");
245 }
246
247 /** The owner answers a question with its one button: the act it proposed, now for good. */
248 export function answerKeeper(app: App, id: number, actorId?: string): { done: boolean; message: string } {
249 const row = app.db.query<AttemptRow, [number]>(`SELECT ${COLS} FROM keeper_attempt WHERE id = ?1 AND outcome = 'asked'`).get(id);
250 if (!row) return { done: false, message: "That question is already answered." };
251 const a = toAttempt(row);
252 setOutcome(app.db, a.id, "answered");
253 app.bus.append({ kind: "keeper.answered", repoId: a.repoId, ...(actorId === undefined ? {} : { actorId }), payload: { action: a.action } });
254 return { done: true, message: "Done." };
255 }
src/flow/keeper_cause.ts added +158
@@ -0,0 +1,158 @@
1 /**
2 * What binds the forge's landings right now (#7645): one cause from a closed
3 * list, in the order the dispatcher meets them. `pickCause` is plain data in,
4 * plain data out; `readKeeperFacts` reads the same questions the dispatcher
5 * asks (`heldNow`, `atCeiling`, `pace`) so the keeper and the builders cannot
6 * disagree. No model is asked anything.
7 */
8
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 { standings } from "../core/standing.ts";
14 import { grantedAccounts } from "../core/subscriptions.ts";
15 import { slottedRuns, spentToday } from "../crew/autonomy.ts";
16 import { atCeiling } from "../crew/dispatch.ts";
17 import { heldNow } from "../crew/held.ts";
18 import { pace } from "../crew/pacing.ts";
19 import { readPile } from "./pile.ts";
20
21 export const KEEPER_CAUSES = [
22 "ceiling",
23 "usage",
24 "unassigned",
25 "empty_pillar",
26 "red_checks",
27 "stale_pile",
28 "capacity",
29 ] as const;
30
31 export type KeeperCause = (typeof KEEPER_CAUSES)[number];
32
33 /** What the keeper does about a cause: unstick, file once, or only say. */
34 export type KeeperMove = "unstick" | "file" | "say";
35
36 export interface RepoFacts {
37 readonly repoId: string;
38 readonly slug: string;
39 /** Ready issues a builder could start now. */
40 readonly startable: number;
41 /** Builder runs in flight, and the slots they fill. */
42 readonly running: number;
43 readonly slots: number;
44 readonly ceilingUsd: number;
45 readonly atCeiling: boolean;
46 readonly noHeadroom: boolean;
47 /** When the soonest spent window resets (ISO), if a subscription is out. */
48 readonly resetsAt: string | null;
49 }
50
51 export interface KeeperFacts {
52 readonly repos: readonly RepoFacts[];
53 readonly red: number;
54 readonly stale: number;
55 }
56
57 export interface Found {
58 readonly cause: KeeperCause;
59 readonly move: KeeperMove;
60 /** One sentence naming it. */
61 readonly says: string;
62 readonly repoId: string;
63 }
64
65 /** The pile that is not moving counts as stale from this many pulls; red from this many. */
66 export const STALE_PULLS = 20;
67 export const RED_PULLS = 5;
68
69 const hhmm = (iso: string): string => `${iso.slice(11, 16)} UTC`;
70
71 /** The first cause that binds, skipping those already tried; null when none does. */
72 export function pickCause(f: KeeperFacts, skip: ReadonlySet<KeeperCause> = new Set()): Found | null {
73 const hit = (found: Found): Found | null => (skip.has(found.cause) ? null : found);
74 for (const r of f.repos) {
75 const waiting = r.startable;
76 if (r.atCeiling && waiting > 0 && !skip.has("ceiling"))
77 return { cause: "ceiling", move: "say", repoId: r.repoId, says: `The daily ceiling ($${r.ceilingUsd}) is reached in ${r.slug}: builders start again at 00:00 UTC.` };
78 }
79 for (const r of f.repos) {
80 if (r.noHeadroom && r.startable > 0 && !skip.has("usage"))
81 return {
82 cause: "usage",
83 move: "say",
84 repoId: r.repoId,
85 says: `Usage is the limit: no subscription has headroom${r.resetsAt ? `, resets at ${hhmm(r.resetsAt)}` : ""}.`,
86 };
87 }
88 for (const r of f.repos) {
89 if (r.startable > 0 && r.running === 0 && !r.atCeiling && !r.noHeadroom) {
90 const found = hit({
91 cause: "unassigned",
92 move: "unstick",
93 repoId: r.repoId,
94 says: `${r.startable} ready issue${r.startable === 1 ? "" : "s"} in ${r.slug} and no builder is running.`,
95 });
96 if (found) return found;
97 }
98 }
99 const first = f.repos[0];
100 if (first && f.repos.every((r) => r.startable === 0)) {
101 const found = hit({ cause: "empty_pillar", move: "file", repoId: first.repoId, says: "Nothing is ready: no pillar has an issue a builder can start." });
102 if (found) return found;
103 }
104 if (first && f.red >= RED_PULLS) {
105 const found = hit({ cause: "red_checks", move: "file", repoId: first.repoId, says: `${f.red} open crew pulls have a red check.` });
106 if (found) return found;
107 }
108 if (first && f.stale >= STALE_PULLS) {
109 const found = hit({ cause: "stale_pile", move: "file", repoId: first.repoId, says: `${f.stale} open crew pulls are waiting on a builder or a reviewer.` });
110 if (found) return found;
111 }
112 for (const r of f.repos) {
113 if (r.running >= r.slots && r.slots > 0 && !skip.has("capacity"))
114 return { cause: "capacity", move: "say", repoId: r.repoId, says: `All ${r.slots} builder slot${r.slots === 1 ? "" : "s"} in ${r.slug} are busy.` };
115 }
116 return null;
117 }
118
119 /** The facts, read from the dispatcher's own questions. */
120 export function readKeeperFacts(app: App, now: Date): KeeperFacts {
121 const { db } = app;
122 const repos: RepoFacts[] = [];
123 for (const repo of listAutonomousRepos(db)) {
124 if (repo.autonomy !== "autonomous" || repo.autonomyBy === null || repo.archivedAt !== null) continue;
125 const crew = listCrewRoles(db, repo.id);
126 if (crew.length === 0) continue;
127 const held = heldNow(app, repo);
128 const active = listActive(db, "agent").filter((r) => r.repoId === repo.id);
129 let noHeadroom = false;
130 let resetsAt: string | null = null;
131 const granted = grantedAccounts(db, repo.id, repo.ownerId);
132 if (granted.length > 0) {
133 const stand = standings(db, granted.map((a) => a.id), now);
134 const accounts = granted.flatMap((a) => {
135 const s = stand.get(a.id);
136 return s === undefined ? [] : [{ name: a.name, standing: s }];
137 });
138 noHeadroom = pace(accounts, repo.autonomySlots, 1, now).slots === 0;
139 if (noHeadroom)
140 for (const a of accounts)
141 for (const w of a.standing.windows)
142 if (w.resetsAt && (resetsAt === null || w.resetsAt.value < resetsAt)) resetsAt = w.resetsAt.value;
143 }
144 repos.push({
145 repoId: repo.id,
146 slug: `${repo.ownerName}/${repo.name}`,
147 startable: held.startable.length,
148 running: slottedRuns(active, crew).length,
149 slots: repo.autonomySlots,
150 ceilingUsd: repo.autonomyCeilingUsd,
151 atCeiling: atCeiling(repo, spentToday(db, repo.id, now)),
152 noHeadroom,
153 resetsAt,
154 });
155 }
156 const pile = readPile(db);
157 return { repos, red: pile.red, stale: pile.awaitingReview + pile.changesAsked + pile.approved };
158 }
src/flow/keeper_rate.ts added +109
@@ -0,0 +1,109 @@
1 /**
2 * What the throughput keeper measures (#7645), from pulls and runs the forge
3 * already keeps: landed per hour against the best the forge has recently
4 * sustained, and the share of hours it ran. An hour ran when a pull landed in
5 * it or a builder started a run in it. No model, no table of its own.
6 */
7
8 import type { Database } from "bun:sqlite";
9
10 const HOUR_MS = 3_600_000;
11 const DAY_MS = 24 * HOUR_MS;
12 /** Hours of history the best is taken over. */
13 export const BEST_HOURS = 7 * 24;
14 /** The span a rate is averaged over, in hours. */
15 const SPAN = 3;
16
17 /** Pulls merged into a default branch in [from, to], by ISO time. */
18 export function landedBetween(db: Database, fromMs: number, toMs: number): number {
19 const row = db
20 .query<{ n: number }, [string, string]>(
21 `SELECT COUNT(*) AS n FROM pull p JOIN issue i ON i.id = p.issue_id JOIN repo r ON r.id = i.repo_id
22 WHERE p.base_branch = r.default_branch AND p.merged_at >= ?1 AND p.merged_at <= ?2`,
23 )
24 .get(new Date(fromMs).toISOString(), new Date(toMs).toISOString());
25 return row?.n ?? 0;
26 }
27
28 /** Per hour back from `now` (index 0 the last hour): landings, and whether a builder started a run. */
29 export function hourly(db: Database, now: number, hours: number): { landed: number[]; ran: boolean[] } {
30 const since = new Date(now - hours * HOUR_MS).toISOString();
31 const until = new Date(now).toISOString();
32 const landed = new Array<number>(hours).fill(0);
33 const ran = new Array<boolean>(hours).fill(false);
34 const at = (iso: string): number => Math.floor((now - Date.parse(iso)) / HOUR_MS);
35 const merged = db
36 .query<{ at: string }, [string, string]>(
37 `SELECT p.merged_at AS at FROM pull p JOIN issue i ON i.id = p.issue_id JOIN repo r ON r.id = i.repo_id
38 WHERE p.base_branch = r.default_branch AND p.merged_at >= ?1 AND p.merged_at <= ?2`,
39 )
40 .all(since, until);
41 for (const m of merged) {
42 const h = at(m.at);
43 if (h >= 0 && h < hours) {
44 landed[h]! += 1;
45 ran[h] = true;
46 }
47 }
48 const started = db
49 .query<{ at: string }, [string, string]>(
50 "SELECT started_at AS at FROM run WHERE kind = 'agent' AND started_at >= ?1 AND started_at <= ?2",
51 )
52 .all(since, until);
53 for (const s of started) {
54 const h = at(s.at);
55 if (h >= 0 && h < hours) ran[h] = true;
56 }
57 return { landed, ran };
58 }
59
60 export interface Rate {
61 /** Landed in the last hour. */
62 readonly lastHour: number;
63 /** Landed per hour over the last three. */
64 readonly recent: number;
65 /** The best three-hour rate in the week before those three; 0 with no history. */
66 readonly best: number;
67 /** Share of the last 24 hours that ran, 0 to 1. */
68 readonly ranShare: number;
69 }
70
71 export function readRate(db: Database, now: Date): Rate {
72 const { landed, ran } = hourly(db, now.getTime(), BEST_HOURS);
73 const sum = (from: number): number => landed.slice(from, from + SPAN).reduce((a, b) => a + b, 0) / SPAN;
74 let best = 0;
75 for (let h = SPAN; h + SPAN <= landed.length; h++) best = Math.max(best, sum(h));
76 return {
77 lastHour: landed[0] ?? 0,
78 recent: sum(0),
79 best,
80 ranShare: ran.slice(0, 24).filter(Boolean).length / 24,
81 };
82 }
83
84 /** Landings well under what the forge has sustained: under half the best, and under one an hour. */
85 export function isLow(rate: Rate): boolean {
86 return rate.recent < Math.max(1, rate.best / 2);
87 }
88
89 export interface KeeperDay {
90 readonly day: string;
91 readonly hoursRunning: number;
92 readonly landed: number;
93 }
94
95 /** The last `days` UTC days, oldest first: hours that ran and pulls landed, for the scorecard. */
96 export function keeperDays(db: Database, now: Date, days = 7): KeeperDay[] {
97 const start = Math.floor(now.getTime() / DAY_MS) * DAY_MS - (days - 1) * DAY_MS;
98 const hours = Math.ceil((now.getTime() - start) / HOUR_MS);
99 const { landed, ran } = hourly(db, now.getTime(), hours);
100 const out = new Map<string, { hoursRunning: number; landed: number }>();
101 for (let d = 0; d < days; d++) out.set(new Date(start + d * DAY_MS).toISOString().slice(0, 10), { hoursRunning: 0, landed: 0 });
102 for (let h = 0; h < hours; h++) {
103 const slot = out.get(new Date(now.getTime() - (h + 1) * HOUR_MS + 1).toISOString().slice(0, 10));
104 if (!slot) continue;
105 slot.landed += landed[h]!;
106 if (ran[h]) slot.hoursRunning += 1;
107 }
108 return [...out].map(([day, v]) => ({ day, ...v }));
109 }
src/http/server.ts +2
@@ -283,6 +283,7 @@ import {
283283 asksMute,
284284 asksPage,
285285 } from "../web/studio/asks_page.ts";
286 import { keeperAnswer } from "../web/studio/keeper_row.ts";
286287 import { homeArchive, homeArchived, homeArchiveUndo } from "../web/studio/home_archive.ts";
287288 import { liveFeedFragment } from "../web/studio/feed.ts";
288289 import { serviceWorker } from "../web/sw.ts";
@@ -751,6 +752,7 @@ export function buildRouter(): Router<Handler> {
751752 router.add("GET", "/explore", explorePage);
752753 router.add("POST", "/front", frontUpdate);
753754 router.add("POST", "/home/archive", homeArchive);
755 router.add("POST", "/flow/keeper/answer", keeperAnswer);
754756 router.add("GET", "/home/archived/:op", homeArchived);
755757 router.add("POST", "/home/archived/:op/undo", homeArchiveUndo);
756758 router.add("GET", "/signin", signinPage);
src/main.ts +5 −1
@@ -51,7 +51,7 @@ import { failAbandonedRuns } from "./core/runs.ts";
5151 import { DRAIN_LINGER_MAX_S, drainLingerMs, startDraining } from "./http/health.ts";
5252 import { takePresence, warmBehind } from "./apps/presence.ts";
5353 import { STALE_AFTER_MS } from "./runs/lease.ts";
54 import { startCrew } from "./crew/dispatch.ts";
54 import { startCrew, sweepAssigned } from "./crew/dispatch.ts";
5555 import { CREW_OFF_IN_PREVIEW, crewOffInPreview } from "./crew/preview.ts";
5656 import { typeSafeDecider } from "./crew/decide.ts";
5757 import { graphInstagram } from "./crew/instagram.ts";
@@ -59,6 +59,7 @@ import { peopleOnCrews } from "./core/agents.ts";
5959 import { startNotifier } from "./crew/notify.ts";
6060 import { expireCrewHolds } from "./actions/holds.ts";
6161 import { startClogTick } from "./flow/clog_tick.ts";
62 import { startKeeper } from "./flow/keeper.ts";
6263 import { sweepStaleAsks } from "./crew/stale_asks.ts";
6364 import { recordAskLedger } from "./web/studio/asks.ts";
6465 import { startPromptRoutine } from "./crew/prompts.ts";
@@ -593,6 +594,8 @@ const processTick = (): void => {
593594 };
594595 // The clog watch (#6524, #6757) reads the flow every minute; its state is the database's.
595596 const stopClogTick = startClogTick(app.db, app.bus, (error) => app.log.warn("clog tick failed", { error: String(error) }), 60_000);
597 // The throughput keeper (#7645) names what binds landings each minute and lifts what is its to lift.
598 const stopKeeper = startKeeper(app, () => sweepAssigned(app));
596599 // An agent's hold that has stood 24 h is released each minute (#6918).
597600 const holdTimer = setInterval(() => {
598601 try {
@@ -750,6 +753,7 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
750753 clearInterval(staleTimer);
751754 clearInterval(followTimer);
752755 stopClogTick();
756 stopKeeper();
753757 clearInterval(processTimer);
754758 clearInterval(holdTimer);
755759 clearInterval(modelReviewTimer);
src/system/telemetry.ts +3
@@ -11,6 +11,7 @@ import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../
1111 import { firedAlarms } from "../flow/alarms.ts";
1212 import { readVitals, vitalsFor, vitalsLine, vitalsMisses } from "../flow/vitals.ts";
1313 import { parkedSays, readParked } from "../flow/parked.ts";
14 import { readKeeper } from "../flow/keeper.ts";
1415 import { readPipeline, verdict } from "../flow/pipeline.ts";
1516 import { type QueueReading, readQueues } from "../flow/queue.ts";
1617 import { reviewersBusy, reviewingSays } from "../flow/reviewing.ts";
@@ -39,6 +40,8 @@ export function flowFor(app: App, reader: Reader, now = new Date()) {
3940 const parked = readParked(app.db, now, reader.kind === "agent");
4041 return {
4142 says: verdict(stages, now.getTime()),
43 // The cause binding landings now and the keeper's last act (#7645).
44 keeper: readKeeper(app.db, reader.kind === "owner"),
4245 // How long each main has gone unverified after landing (#7786); private repositories for the owner only.
4346 verified: readVerified(app, reader.kind === "owner").map((row) => ({ ...row, says: verifiedSays(row, now.getTime()) })),
4447 // Issues the crew parked in the last 24 h and 7 d, the newest three named (#6471).
src/tui/forge.ts +5 −1
@@ -56,6 +56,7 @@ import { capacityScreen, mayCapacity } from "./screens/capacity.ts";
5656 import { forgeCeiling } from "../flow/ceiling.ts";
5757 import { flowLines, flowScreen } from "./screens/flow.ts";
5858 import { readHeld } from "../flow/held.ts";
59 import { answerKeeper, readKeeper } from "../flow/keeper.ts";
5960 import { readVerified } from "../flow/verified.ts";
6061 import { memoryScreen } from "./screens/system.ts";
6162 import { noteRead, telemetryReader } from "../actions/telemetry.ts";
@@ -293,6 +294,8 @@ export function forgeCommands(session: Session): (ctx: Ctx) => Command[] {
293294 hint: "the pipeline's bottleneck and throughput",
294295 run: () => {
295296 if (session.user) noteRead(session.app, session.user, flowReader, "flow");
297 const keeper = readKeeper(session.db, flowReader.kind === "owner");
298 const user = session.user;
296299 ctx.push(flowScreen(ctx, flowLines(
297300 session.db,
298301 new Date(),
@@ -301,7 +304,8 @@ export function forgeCommands(session: Session): (ctx: Ctx) => Command[] {
301304 readHeld(session.app, flowReader.kind === "owner"),
302305 forgeCeiling(session.app.config),
303306 readVerified(session.app, flowReader.kind === "owner"),
304 )));
307 keeper,
308 ), keeper.ask && user ? { ask: keeper.ask, answer: () => answerKeeper(session.app, keeper.ask!.id, user.id) } : undefined));
305309 },
306310 });
307311 // The forge's capacity and guards, as MCP `capacity` gates them: a forge admin who is a person.
src/tui/screens/flow.ts +29 −3
@@ -24,6 +24,7 @@ import { pullCounts } from "../../crew/hot_files.ts";
2424 import { warmingLine } from "../../crew/hot_history.ts";
2525 import { ago } from "../../feed/rank.ts";
2626 import type { HeldRow } from "../../flow/held.ts";
27 import { keeperLine, type KeeperAsk, type KeeperRead } from "../../flow/keeper.ts";
2728 import { verifiedSays, type VerifiedRow } from "../../flow/verified.ts";
2829 import { conflictedSays, rebaseLaneSays } from "../../flow/conflicted.ts";
2930 import { IDLE_LIMIT, type IdlePull } from "../../flow/idle_pulls.ts";
@@ -47,6 +48,7 @@ import { binding } from "../app/keys.ts";
4748 import { pad, truncate, wrap } from "../term/width.ts";
4849 import { dialog } from "../ui/widgets.ts";
4950 import { live } from "../../system/live.ts";
51 import { confirmScreen } from "./common.ts";
5052 import { duplicateLines } from "./system.ts";
5153
5254 const WIDTH = 64;
@@ -64,6 +66,8 @@ export function flowLines(
6466 ceilingUsd: number | null = null,
6567 /** How long each main has gone unverified after landing (`readVerified`, already cut to what the reader may see). */
6668 verified?: readonly VerifiedRow[],
69 /** The keeper's cause and last act (`readKeeper`, already cut to what the reader may see). */
70 keeper?: KeeperRead,
6771 ): string[] {
6872 const stages = readPipeline(db, now);
6973 const flow = readThroughput(db, stages.map((s) => s.key), now);
@@ -85,6 +89,8 @@ export function flowLines(
8589 const ids = repoIds(db, queues.map((q) => q.repo));
8690 return [
8791 ...wrap(verdict(stages, now.getTime()), inner),
92 ...(keeper === undefined || keeperLine(keeper) === "" ? [] : ["", ...wrap(keeperLine(keeper), inner)]),
93 ...(keeper?.ask == null ? [] : wrap(`${keeper.ask.question} Press l to ${keeper.ask.button.toLowerCase()}.`, inner)),
8894 "",
8995 ...wrap(vitalsLine(vitalsFor(readVitals(db, now, agent !== undefined), agent === undefined)), inner),
9096 ...(agent === undefined ? clogLines(readClog(db, now), now.getTime()).flatMap((l) => wrap(l, inner)) : []),
@@ -226,10 +232,15 @@ export function heldLines(rows: readonly HeldRow[], inner: number): string[] {
226232 ];
227233 }
228234
229 export function flowScreen(ctx: Ctx, lines: readonly string[]): Screen {
230 return {
235 export function flowScreen(
236 ctx: Ctx,
237 lines: readonly string[],
238 /** The keeper's open question (a site admin's); `answer` presses its one button and says what happened. */
239 keeper?: { ask: KeeperAsk; answer: () => { done: boolean; message: string } },
240 ): Screen {
241 const self: Screen = {
231242 title: "Flow",
232 keys: [binding(["enter", "esc", "q"], "close")],
243 keys: [binding(["enter", "esc", "q"], "close"), ...(keeper?.ask ? [binding(["l"], keeper.ask.button.toLowerCase())] : [])],
233244 render(area, s) {
234245 const buttons = dialog(s, area, ctx.theme.c, {
235246 title: "Flow",
@@ -242,7 +253,22 @@ export function flowScreen(ctx: Ctx, lines: readonly string[]): Screen {
242253 },
243254 key(e) {
244255 if (e.key === "enter" || e.key === "esc" || e.key === "q") ctx.pop();
256 else if (e.key === "l" && keeper)
257 ctx.push(
258 confirmScreen(ctx, self, {
259 title: keeper.ask.button,
260 body: wrap(keeper.ask.question, WIDTH - 6),
261 actions: [{ label: "Cancel", variant: "ghost" }, { label: keeper.ask.button }],
262 onChoose: (i) => {
263 if (i !== 1) return;
264 const answered = keeper.answer();
265 ctx.toast(answered.message, answered.done ? "success" : "danger");
266 ctx.pop();
267 },
268 }),
269 );
245270 return true;
246271 },
247272 };
273 return self;
248274 }
src/tui/screens/scorecard.ts +10 −1
@@ -21,6 +21,7 @@ import {
2121 type Scorecard,
2222 type SplitScorecard,
2323 } from "../../core/scorecard.ts";
24 import { keeperDays, type KeeperDay } from "../../flow/keeper_rate.ts";
2425 import type { Agent } from "../../core/agents.ts";
2526 import { listTeams, type Team } from "../../core/teams.ts";
2627 import type { Ctx, Screen } from "../app/app.ts";
@@ -136,6 +137,14 @@ export function scorecardRows(
136137 return [...agentRows, ...teamRows];
137138 }
138139
140 /** "Forge per day" as the page's table has it: hours the forge ran and pulls it landed, newest day first (#7645). */
141 export function forgeDayRows(days: readonly KeeperDay[]): ListRow[] {
142 return [...days].reverse().map((d) => ({
143 title: `Forge ${d.day}`,
144 secondary: `${d.hoursRunning} hour${d.hoursRunning === 1 ? "" : "s"} running · ${d.landed} landed`,
145 }));
146 }
147
139148 type Review = ReturnType<typeof readReview>;
140149
141150 /**
@@ -195,7 +204,7 @@ export function scorecardScreen(f: Session, ctx: Ctx): Screen {
195204 const names = new Map((listed?.ok === true ? listed.value : []).map((a) => [a.user.id, a.user.username]));
196205 // Re-read after a decision, so the button's result shows at once.
197206 let review = allowed && cards.length > 0 ? reviewRows(readReview(f.app), names) : { rows: [], changeAt: [] };
198 const rows = (): ListRow[] => [...cards, ...review.rows];
207 const rows = (): ListRow[] => [...cards, ...review.rows, ...(allowed ? forgeDayRows(keeperDays(f.db, new Date())) : [])];
199208 const waiting = (): string | undefined => review.changeAt[state.index - cards.length];
200209 const decide = (take: boolean): void => {
201210 const id = waiting();
src/web/pages/crew.ts +31 −1
@@ -35,6 +35,7 @@ import {
3535 type Scorecard,
3636 type SplitScorecard,
3737 } from "../../core/scorecard.ts";
38 import { keeperDays, type KeeperDay } from "../../flow/keeper_rate.ts";
3839 import { decideChange, readReview } from "../../actions/model_review.ts";
3940 import { wastedDays, type WastedDay } from "../../core/wasted_days.ts";
4041 import { modelDays, type ModelDay } from "../../core/model_days.ts";
@@ -440,6 +441,7 @@ function loadScorecards(ctx: Ctx) {
440441 byModel: agentSplits(ctx.app.db, "model"),
441442 byPrompt: agentSplits(ctx.app.db, "prompt_version"),
442443 byModelDay: modelDays(ctx.app.db),
444 keeperDays: keeperDays(ctx.app.db, new Date()),
443445 wasted: wastedDays(ctx.app.db),
444446 };
445447 }
@@ -581,7 +583,7 @@ function crewView(
581583 splitScorecards,
582584 recentSplitScorecards,
583585 } = data.scorecards;
584 const { scorecards, byModel, byPrompt, byModelDay, wasted } = data.scorecards;
586 const { scorecards, byModel, byPrompt, byModelDay, wasted, keeperDays: forgeDays } = data.scorecards;
585587 const { spending } = data;
586588 const usernameError =
587589 error !== undefined && /username/i.test(error) ? error : undefined;
@@ -694,6 +696,7 @@ function crewView(
694696 () => html`<span class="muted">default</span>`,
695697 )}
696698 ${modelDaysCard(byModelDay)}
699 ${keeperDaysCard(forgeDays)}
697700 ${wastedDaysCard(wasted)}
698701 ${splitsCard(
699702 "Per prompt version",
@@ -783,6 +786,33 @@ function drillCard(drill: CrewHandover["drill"]): Html {
783786 );
784787 }
785788
789 /** Hours the forge ran and pulls it landed, per day of the last week (#7645). */
790 function keeperDaysCard(rows: readonly KeeperDay[]): Html {
791 return card(
792 html`<div class="table-scroll">
793 <table class="table" id="crew-keeper-days">
794 <thead>
795 <tr>
796 <th>Day</th>
797 <th>Hours running</th>
798 <th>Landed</th>
799 </tr>
800 </thead>
801 <tbody>
802 ${rows.map(
803 (r) => html`<tr>
804 <td class="mono">${r.day}</td>
805 <td class="num">${r.hoursRunning}</td>
806 <td class="num">${r.landed}</td>
807 </tr>`,
808 )}
809 </tbody>
810 </table>
811 </div>`,
812 { pad: false, head: html`<h2>Forge per day</h2>` },
813 );
814 }
815
786816 /** Spend and landed pulls per model per day: the escalation, measured. */
787817 function modelDaysCard(rows: readonly ModelDay[]): Html {
788818 if (rows.length === 0) return html``;
src/web/pages/flow-panel.ts +2 −1
@@ -46,6 +46,7 @@ import { readHeld } from "../../flow/held.ts";
4646 import { readVerified, verifiedSays, type VerifiedRow } from "../../flow/verified.ts";
4747 import type { App } from "../../app.ts";
4848 import { html, type Html } from "../html.ts";
49 import { keeperRow } from "../studio/keeper_row.ts";
4950 import { gatesCard } from "./boards-gates.ts";
5051 import { heldCard } from "./boards-held.ts";
5152 import { badge, card, meter } from "../ui.ts";
@@ -113,7 +114,7 @@ export function flowPanel(app: App, reader: Reader, url: URL, base: string): Htm
113114 const newest = parked.newest[0];
114115 const listed = asked(url);
115116 const shown = listed === null ? undefined : out.get(listed.key);
116 return html`${card(
117 return html`${keeperRow(app.db, reader.kind === "owner")}${card(
117118 html`<p id="bottleneck" data-stage="${worst?.key ?? ""}">
118119 <strong>${verdict(stages, now)}</strong>
119120 </p>
src/web/studio/home.ts +2
@@ -53,6 +53,7 @@ import { needsYouSection } from "./asks_page.ts";
5353 import { flowCards } from "./flow_card.ts";
5454 import { composer, feedJs, feedNote, feedRegion } from "./feed.ts";
5555 import { alarmLines, cachedAlarms } from "./alarm_lines.ts";
56 import { keeperRow } from "./keeper_row.ts";
5657 import { clogRow, kpiRow, vitalsRow, waitingLine } from "./kpi_row.ts";
5758 import { feedCandidateCache } from "../../feed/feed.ts";
5859 import { notePhase, notePhaseRest, takeSessionTime, timePhase, timePhaseAsync } from "../../core/route_timing.ts";
@@ -295,6 +296,7 @@ export async function studioHome(
295296 : liveLine(ctx)}
296297 ${figures ? timePhase(ctx.request, "vitals", () => vitalsRow(ctx.app.db)) : null}
297298 ${figures ? clogRow(ctx.app.db) : null}
299 ${figures ? keeperRow(ctx.app.db, true) : null}
298300 ${figures ? alarmLines(cachedAlarms(ctx.app.db, new Date(), ceilingUsd)) : null}
299301 ${timePhase(ctx.request, "waiting", () => waitingLine(ctx.app.db, user), () => waitingCache(ctx.app.db))}
300302 ${figures
src/web/studio/keeper_row.ts added +45
@@ -0,0 +1,45 @@
1 /**
2 * The throughput keeper on a page (#7645): one muted line, the cause binding
3 * landings and the last thing it did, and, when it asks, the question with
4 * its one button. Home and the pipeline panel both draw it; `POST
5 * /flow/keeper/answer` is the button (a site admin's, same-origin).
6 */
7
8 import type { Database } from "bun:sqlite";
9 import { answerKeeper, keeperLine, readKeeper } from "../../flow/keeper.ts";
10 import type { Ctx } from "../../http/context.ts";
11 import { signInFirst } from "../chrome.ts";
12 import { readForm } from "../form.ts";
13 import { html, type Html } from "../html.ts";
14 import { redirect } from "../layout.ts";
15 import { btn } from "../ui.ts";
16
17 export function keeperRow(db: Database, full: boolean): Html {
18 const read = readKeeper(db, full);
19 const line = keeperLine(read);
20 if (line === "" && read.ask === null) return html``;
21 return html`${line === "" ? null : html`<p class="small muted break" id="keeper">${line}</p>`}
22 ${read.ask === null
23 ? null
24 : html`<form method="post" action="/flow/keeper/answer" id="keeper-ask" class="bar small">
25 <span>${read.ask.question}</span>
26 <input type="hidden" name="id" value="${String(read.ask.id)}" />
27 ${btn(read.ask.button, { type: "submit" })}
28 </form>`}`;
29 }
30
31 // POST /flow/keeper/answer — the one button on the keeper's question.
32 export async function keeperAnswer(ctx: Ctx): Promise<Response> {
33 if (!ctx.auth.user) return signInFirst(ctx);
34 const form = await readForm(ctx, "admin");
35 if (form instanceof Response) return form;
36 if (!ctx.auth.user.isAdmin) return new Response("forbidden", { status: 403 });
37 const raw = form.get("id");
38 const id = typeof raw === "string" && /^\d{1,12}$/.test(raw) ? Number(raw) : 0;
39 const answered = id > 0 ? answerKeeper(ctx.app, id, ctx.auth.user.id) : { done: false, message: "That question is already answered." };
40 return redirect("/", {
41 url: ctx.url,
42 forgeUrl: ctx.app.config.url,
43 flash: { tone: answered.done ? "success" : "danger", message: answered.message },
44 });
45 }
test/flow-keeper.test.ts added +333
@@ -0,0 +1,333 @@
1 /** The throughput keeper (#7645): it names what binds landings, unsticks what is its to unstick, checks, undoes, and files once. */
2 import { afterAll, beforeAll, expect, test } from "bun:test";
3 import { mkdtemp, rm } 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 import { addToCrew, createAgent } from "../src/actions/crew.ts";
8 import { openIssue } from "../src/actions/issues.ts";
9 import { commitToBranch, createRepo, updateRepoSettings } from "../src/actions/repos.ts";
10 import { readScorecards } from "../src/actions/scorecard.ts";
11 import { getRepoById, type Repo } from "../src/core/repos.ts";
12 import { createSession } from "../src/core/sessions.ts";
13 import { noteLimit } from "../src/core/standing.ts";
14 import { addSubscription, EVERYONE, setGrants } from "../src/core/subscriptions.ts";
15 import type { User } from "../src/core/users.ts";
16 import { isLow, keeperDays, readRate, type Rate } from "../src/flow/keeper_rate.ts";
17 import { createUser } from "../src/core/users.ts";
18 import { createForgeApp } from "../src/tui/forge.ts";
19 import { flowLines, flowScreen } from "../src/tui/screens/flow.ts";
20 import { scorecardScreen } from "../src/tui/screens/scorecard.ts";
21 import { headlessTerminal } from "../src/tui/term/terminal.ts";
22 import { settleApp } from "./helpers/tui.ts";
23 import { answerKeeper, EFFECT_MS, listAttempts, readKeeper, tickKeeper } from "../src/flow/keeper.ts";
24 import { pickCause, type KeeperFacts, type RepoFacts } from "../src/flow/keeper_cause.ts";
25 import { flowFor } from "../src/system/telemetry.ts";
26 import { serve, type Servers } from "../src/http/server.ts";
27 import { seedAdmin } from "./helpers/admin.ts";
28 import { quietLogs } from "./helpers/globals.ts";
29
30 quietLogs();
31
32 let dir: string;
33 let app: App;
34 let servers: Servers;
35 let admin: User;
36 let repo: Repo;
37 let was: string | undefined;
38 const fresh = (): Repo => getRepoById(app.db, repo.id)!;
39 const cookie = (): Record<string, string> => ({ cookie: `forge_session=${createSession(app.db, admin.id).token}; forge_mode=studio` });
40 const filed = (): number => listAttempts(app.db, 50).filter((a) => a.outcome === "filed").length;
41
42 beforeAll(async () => {
43 was = process.env.FORGE_SENTINEL_REPO;
44 process.env.FORGE_SENTINEL_REPO = "alice/engine";
45 dir = await mkdtemp(join(tmpdir(), "forge-keeper-"));
46 app = createApp({ dataDir: dir, port: 0 });
47 servers = await serve(app);
48 const a = await seedAdmin(app, { username: "alice", email: "[email protected]", password: "a-long-passphrase" });
49 if (!a.ok) throw new Error("signup failed");
50 admin = a.value;
51 const made = await createRepo(app, admin, { name: "engine", visibility: "private" });
52 if (!made.ok) throw new Error(String(made.error));
53 repo = made.value;
54 const seeded = await commitToBranch(app, admin, repo, { message: "Start", files: [{ path: "README.md", content: "# engine\n" }] });
55 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
56 const agent = await createAgent(app, admin, { username: "mason" });
57 if (!agent.ok) throw new Error(String(agent.error));
58 expect(addToCrew(app, admin, repo, "mason").ok).toBe(true);
59 const on = await updateRepoSettings(app, admin, repo, {
60 autonomy: "autonomous",
61 autonomyDirection: "Build.",
62 autonomyCeilingUsd: 10,
63 autonomySlots: 2,
64 autonomyBy: admin.id,
65 });
66 expect(on.ok).toBe(true);
67 }, 60_000);
68
69 afterAll(async () => {
70 if (was === undefined) delete process.env.FORGE_SENTINEL_REPO;
71 else process.env.FORGE_SENTINEL_REPO = was;
72 servers.stop();
73 app.close();
74 await rm(dir, { recursive: true, force: true });
75 });
76
77 test("2026-11-01: ready issues wait and no builder runs: it names the cause and unsticks", async () => {
78 for (let n = 1; n <= 55; n++) {
79 const o = openIssue(app, admin, fresh(), { title: `Work ${n}` });
80 if (!o.ok) throw new Error(String(o.error));
81 }
82 let swept = 0;
83 const now = new Date();
84 const tick = tickKeeper(app, now, () => void (swept += 1));
85 expect(tick).toEqual({ cause: "unassigned", acted: "unstick" });
86 expect(swept).toBe(1);
87 const read = readKeeper(app.db, true);
88 expect(read.cause).toContain("55 ready issues in alice/engine and no builder is running");
89 expect(read.last).toContain("unstuck assigned issues");
90 // Not again while its check is pending.
91 expect(tickKeeper(app, new Date(now.getTime() + 60_000), () => void (swept += 1)).acted).toBeNull();
92 expect(swept).toBe(1);
93 });
94
95 test("the cause and last act show on home, the boards panel, and the MCP flow report", async () => {
96 const home = await (await fetch(`${servers.url}/`, { headers: cookie() })).text();
97 expect(home).toContain('id="keeper"');
98 expect(home).toContain("Binding landings:");
99 const boards = await (await fetch(`${servers.url}/boards`, { headers: cookie() })).text();
100 expect(boards).toContain("unstuck assigned issues");
101 const report = flowFor(app, { kind: "owner" });
102 expect(report.keeper.cause).toContain("no builder is running");
103 expect(report.keeper.last).toContain("unstuck assigned issues");
104 });
105
106 test("no rise in landings after the check time: the act is undone and the next tick does not try it again", () => {
107 const later = new Date(Date.now() + EFFECT_MS + 60_000);
108 tickKeeper(app, later);
109 expect(listAttempts(app.db).map((a) => a.outcome)).toEqual(["undone"]);
110 expect(readKeeper(app.db, true).ask).toBeNull();
111 expect(tickKeeper(app, new Date(later.getTime() + 60_000)).acted).toBeNull();
112 });
113
114 test("a rise in landings keeps the act", async () => {
115 app.db.query("DELETE FROM keeper_attempt").run();
116 const t = new Date(Date.now() + 10 * EFFECT_MS);
117 expect(tickKeeper(app, t, () => {}).acted).toBe("unstick");
118 const issue = openIssue(app, admin, fresh(), { title: "Landed" });
119 if (!issue.ok) throw new Error(String(issue.error));
120 app.db
121 .query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id, merged_at) VALUES (?1, ?2, 'b/1', '', '', ?3, ?4)`)
122 .run(issue.value.id, fresh().defaultBranch, repo.id, new Date(t.getTime() + 30 * 60_000).toISOString());
123 tickKeeper(app, new Date(t.getTime() + EFFECT_MS + 1000));
124 expect(listAttempts(app.db).some((a) => a.action === "unstick" && a.outcome === "worked")).toBe(true);
125 });
126
127 test("a cause it cannot act on is filed once, not once per tick", () => {
128 const before = filed();
129 // Nothing is ready any more: every open issue is closed.
130 app.db.query("UPDATE issue SET state = 'closed' WHERE repo_id = ?1").run(repo.id);
131 const t = Date.now() + 100 * EFFECT_MS;
132 for (let i = 0; i < 4; i++) tickKeeper(app, new Date(t + i * 60_000));
133 expect(filed()).toBe(before + 1);
134 const issues = app.db.query<{ title: string }, []>("SELECT title FROM issue WHERE title LIKE 'Throughput:%'").all();
135 expect(issues.length).toBe(1);
136 expect(issues[0]!.title).toContain("Nothing is ready");
137 });
138
139 test("usage is the limit: it says when it resets, and spends nothing to say it", () => {
140 const sub = addSubscription(app.db, dir, { name: "espo", token: "sk-ant-oat01-" + "x".repeat(24) });
141 if (!sub.ok) throw new Error(sub.error);
142 setGrants(app.db, repo.id, EVERYONE, [sub.value.id]);
143 const now = Date.now() + 200 * EFFECT_MS;
144 const resets = new Date(now + 3_600_000);
145 noteLimit(app.db, sub.value.id, { window: "5h", resetsAt: resets.toISOString(), observedAt: new Date(now).toISOString(), said: "limit" });
146 app.db.query("UPDATE issue SET state = 'open' WHERE repo_id = ?1 AND title = 'Work 1'").run(repo.id);
147 const tick = tickKeeper(app, new Date(now));
148 expect(tick).toEqual({ cause: "usage", acted: null });
149 expect(readKeeper(app.db, true).cause).toBe(`Usage is the limit: no subscription has headroom, resets at ${resets.toISOString().slice(11, 16)} UTC.`);
150 });
151
152 test("the closed list: the ceiling outranks usage, and a tried cause gives way to the next", () => {
153 const repoFacts: RepoFacts = {
154 repoId: "r", slug: "a/b", startable: 2, running: 0, slots: 1,
155 ceilingUsd: 10, atCeiling: true, noHeadroom: true, resetsAt: null,
156 };
157 const facts: KeeperFacts = { repos: [repoFacts], red: 0, stale: 0 };
158 expect(pickCause(facts)?.cause).toBe("ceiling");
159 expect(pickCause(facts, new Set(["ceiling"]))?.cause).toBe("usage");
160 const open = repoFacts;
161 expect(pickCause({ repos: [open], red: 0, stale: 0 })?.cause).toBe("ceiling");
162 expect(pickCause({ repos: [open], red: 0, stale: 0 }, new Set(["ceiling"]))?.cause).toBe("usage");
163 expect(pickCause({ repos: [{ ...repoFacts, atCeiling: false, noHeadroom: false }], red: 0, stale: 0 })?.cause).toBe("unassigned");
164 expect(pickCause({ repos: [], red: 6, stale: 0 })).toBeNull();
165 });
166
167 test("the scorecard carries hours running and landed per day for a week", () => {
168 const card = readScorecards(app, admin);
169 expect(card?.keeperDays.length).toBe(7);
170 const before = card!.keeperDays.reduce((n, d) => n + d.landed, 0);
171 const issue = openIssue(app, admin, fresh(), { title: "Landed a minute ago" });
172 if (!issue.ok) throw new Error(String(issue.error));
173 app.db
174 .query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id, merged_at) VALUES (?1, ?2, 'b/recent', '', '', ?3, ?4)`)
175 .run(issue.value.id, fresh().defaultBranch, repo.id, new Date(Date.now() - 5 * 60_000).toISOString());
176 const after = readScorecards(app, admin)!.keeperDays;
177 expect(after.reduce((n, d) => n + d.landed, 0)).toBe(before + 1);
178 expect(card!.keeperDays.every((d) => /^\d{4}-\d\d-\d\d$/.test(d.day))).toBe(true);
179 });
180
181 const post = (headers: Record<string, string>, body: string) =>
182 fetch(`${servers.url}/flow/keeper/answer`, {
183 method: "POST",
184 headers: { origin: servers.url, "content-type": "application/x-www-form-urlencoded", ...headers },
185 body,
186 redirect: "manual",
187 });
188 const ask = (): number => {
189 app.db.query("DELETE FROM keeper_attempt").run();
190 return Number(
191 app.db
192 .query("INSERT INTO keeper_attempt (at, cause, action, repo_id, says, outcome, check_at) VALUES (?1, 'unassigned', 'unstick', ?2, 'Unsticking did not help. Leave it?', 'asked', 0)")
193 .run(Date.now(), repo.id).lastInsertRowid,
194 );
195 };
196
197 test("a guest, a non-admin and an agent never read a private repository's name from the keeper", async () => {
198 ask();
199 app.db.query("DELETE FROM keeper_state").run();
200 app.db
201 .query("INSERT INTO keeper_state (id, cause, says, at) VALUES (1, 'unassigned', '55 ready issues in alice/engine and no builder is running.', ?1)")
202 .run(Date.now());
203 const guest = await (await fetch(`${servers.url}/boards`, { redirect: "follow" })).text();
204 expect(guest).not.toContain("alice/engine");
205 const bob = await createUser(app.db, { origin: "seed", trusted: true, username: "bob", email: "[email protected]", password: "a-long-passphrase" });
206 if (!bob.ok) throw new Error("bob");
207 const bobCookie = { cookie: `forge_session=${createSession(app.db, bob.value.id).token}` };
208 for (const path of ["/boards", "/"]) {
209 const page = await (await fetch(`${servers.url}${path}`, { headers: bobCookie })).text();
210 expect(page).not.toContain("alice/engine");
211 expect(page).not.toContain('id="keeper-ask"');
212 }
213 const agentRead = flowFor(app, { kind: "agent" }).keeper;
214 expect(agentRead.cause).not.toBeNull();
215 expect(JSON.stringify(agentRead)).not.toContain("alice/engine");
216 expect(agentRead.ask).toBeNull();
217 const owner = flowFor(app, { kind: "owner" }).keeper;
218 expect(owner.cause).toContain("alice/engine");
219 expect(owner.ask).not.toBeNull();
220 });
221
222 test("POST /flow/keeper/answer: a guest signs in first, a non-admin is refused, a bad or stale id says it is gone", async () => {
223 const id = ask();
224 const guest = await post({}, `id=${id}`);
225 expect(guest.status).toBe(303);
226 expect(guest.headers.get("location")).toContain("sign");
227 const bobCookie = {
228 cookie: `forge_session=${createSession(app.db, app.db.query<{ id: string }, []>("SELECT id FROM user WHERE username = 'bob'").get()!.id).token}`,
229 };
230 expect((await post(bobCookie, `id=${id}`)).status).toBe(403);
231 for (const body of ["id=0", "id=abc", "id=", "", `id=${id + 1000}`]) {
232 const gone = await post(cookie(), body);
233 expect(gone.status).toBe(303);
234 expect(gone.headers.get("set-cookie") ?? "").toContain("already");
235 }
236 expect(answerKeeper(app, id + 1000)).toEqual({ done: false, message: "That question is already answered." });
237 });
238
239 test("a question for any action is answered with Done", () => {
240 const id = Number(
241 app.db
242 .query("INSERT INTO keeper_attempt (at, cause, action, repo_id, says, outcome, check_at) VALUES (?1, 'red_checks', 'file', ?2, 'Other', 'asked', 0)")
243 .run(Date.now(), repo.id).lastInsertRowid,
244 );
245 expect(answerKeeper(app, id).message).toBe("Done.");
246 });
247
248 test("the owner's answer is on the audit trail with the owner as actor", () => {
249 app.db.query("DELETE FROM event WHERE kind = 'keeper.answered'").run();
250 const id = ask();
251 expect(answerKeeper(app, id, admin.id).done).toBe(true);
252 const row = app.db.query<{ actor_id: string | null }, []>("SELECT actor_id FROM event WHERE kind = 'keeper.answered'").get();
253 expect(row?.actor_id).toBe(admin.id);
254 });
255
256 test("unstick: the sweep runs, the row is pending, and it is undone when landings do not rise", async () => {
257 app.db.query("DELETE FROM keeper_attempt").run();
258 app.db.query("UPDATE issue SET state = 'open' WHERE repo_id = ?1 AND title LIKE 'Work %'").run(repo.id);
259 let swept = 0;
260 const t = new Date(Date.now() + 400 * EFFECT_MS);
261 expect(tickKeeper(app, t, () => void (swept += 1))).toEqual({ cause: "unassigned", acted: "unstick" });
262 expect(swept).toBe(1);
263 const [row] = listAttempts(app.db);
264 expect(row).toMatchObject({ action: "unstick", outcome: "pending", cause: "unassigned" });
265 tickKeeper(app, new Date(t.getTime() + EFFECT_MS + 1000), () => void (swept += 1));
266 expect(listAttempts(app.db).find((a) => a.action === "unstick")?.outcome).toBe("undone");
267 });
268
269 test("isLow, readRate and keeperDays bucket pulls and runs by hour and UTC day", () => {
270 const T = Date.UTC(2040, 0, 10, 12, 30);
271 let n = 0;
272 const land = (iso: string): void => {
273 const issue = openIssue(app, admin, fresh(), { title: `Seed ${(n += 1)}` });
274 if (!issue.ok) throw new Error(String(issue.error));
275 app.db
276 .query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id, merged_at) VALUES (?1, ?2, ?3, '', '', ?4, ?5)`)
277 .run(issue.value.id, fresh().defaultBranch, `seed/${n}`, repo.id, iso);
278 };
279 for (let i = 0; i < 3; i++) land("2040-01-10T11:45:00.000Z");
280 land("2040-01-09T23:50:00.000Z");
281 for (let i = 0; i < 12; i++) land("2040-01-08T10:30:00.000Z");
282 app.db
283 .query("INSERT INTO run (id, repo_id, kind, state, started_at) VALUES ('keeper-seed-run', ?1, 'agent', 'ok', '2040-01-10T09:10:00.000Z')")
284 .run(repo.id);
285 const rate = readRate(app.db, new Date(T));
286 expect(rate.lastHour).toBe(3);
287 expect(rate.recent).toBe(1);
288 expect(rate.best).toBe(4);
289 expect(rate.ranShare).toBeCloseTo(3 / 24);
290 expect(isLow(rate)).toBe(true);
291 const rated = (recent: number, best: number): Rate => ({ lastHour: 0, recent, best, ranShare: 0 });
292 expect(isLow(rated(0.5, 0))).toBe(true);
293 expect(isLow(rated(1, 1.5))).toBe(false);
294 expect(isLow(rated(1.2, 5))).toBe(true);
295 expect(isLow(rated(2.5, 5))).toBe(false);
296 const days = keeperDays(app.db, new Date(T));
297 expect(days.map((d) => d.day)).toEqual(["2040-01-04", "2040-01-05", "2040-01-06", "2040-01-07", "2040-01-08", "2040-01-09", "2040-01-10"]);
298 const byDay = Object.fromEntries(days.map((d) => [d.day, d]));
299 expect(byDay["2040-01-10"]).toEqual({ day: "2040-01-10", hoursRunning: 2, landed: 3 });
300 expect(byDay["2040-01-09"]).toEqual({ day: "2040-01-09", hoursRunning: 1, landed: 1 });
301 expect(byDay["2040-01-08"]).toEqual({ day: "2040-01-08", hoursRunning: 1, landed: 12 });
302 expect(byDay["2040-01-07"]).toEqual({ day: "2040-01-07", hoursRunning: 0, landed: 0 });
303 });
304
305 test("the terminal shows the keeper line and its one button on Flow, and Forge per day on the scorecard", async () => {
306 const id = ask();
307 const keeper = readKeeper(app.db, true);
308 expect(keeper.ask?.id).toBe(id);
309 const { app: tui, session } = createForgeApp(
310 app,
311 { sshUsername: "alice", remoteAddress: "127.0.0.1", user: admin },
312 { terminal: headlessTerminal(120, 40), exitHooks: false },
313 );
314 tui.start();
315 await settleApp(tui);
316 const key = async (k: string): Promise<string> => {
317 tui.dispatch({ type: "key", key: k, ctrl: false, alt: false, shift: false });
318 await settleApp(tui);
319 return tui.screen.toText();
320 };
321 tui.push(flowScreen(tui.ctx, flowLines(app.db, new Date(), undefined, undefined, undefined, null, undefined, keeper), { ask: keeper.ask!, answer: () => answerKeeper(app, id, admin.id) }));
322 await settleApp(tui);
323 expect(tui.screen.toText()).toContain("Leave it?");
324 expect(tui.screen.toText()).toContain("Press l");
325 await key("l");
326 await key("right");
327 await key("enter");
328 expect(readKeeper(app.db, true).ask).toBeNull();
329 tui.push(scorecardScreen(session, tui.ctx));
330 await settleApp(tui);
331 expect(tui.screen.toText()).toContain("Forge 20");
332 expect(tui.screen.toText()).toContain("running");
333 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs