trevato/forge

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

More
Forks1g fMapg aRuns14g rEventsFlows

/flow: the forge shows its own bottleneck — every stage from ready to live with its queue, wait and capacity, the knob beside it, machine use, and 24 h throughput #1729

Merged masoncrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

mason/forge:mason/1554-flow-the-forge-shows-its-own-bottleneck → main 6 commits fork 5 threads · 5 unresolved

  • Merged. mason/1554-flow-the-forge-shows-its-own-bottleneck was merged into main as cdd641d.

Review

Sign in to review this change.
12 files changed +690 −2
README.md +1
@@ -443,6 +443,7 @@ from before this record say nothing).
443443 mechanical rather than advisory.
444444
445445 `docs/FLOW.md` is where every issue stands: `GET /api/v1/repos/:owner/:name/flow` returns each open issue's stage, since, stalled and unblock, plus what went live in the last 24 h.
446 Where the whole forge waits is a forge admin's: `/flow` shows each stage from ready to live with what waits, the oldest wait and its capacity, under one sentence naming the bottleneck; the terminal's Flow command and the MCP `flow` tool say the same sentence.
446447
447448 `docs/PLANNING.md` is how issues depend on issues: the edge, the map, and
448449 what "ready now" means to a person and to the crew. The map also shows each
docs/FLOW.md +13
@@ -13,3 +13,16 @@ pull 1 h, check 30 min, review 12 h, land 2 h; live never. A pull closed without
1313 merging puts the issue back at `issue`; a re-opened pull counts from the re-open;
1414 with several open pulls the furthest along speaks. An issue is linked to a pull
1515 by `Closes #N` in the pull's body, as the merge itself reads it.
16
17 ## The forge's pipeline
18
19 `src/flow/pipeline.ts` is the same road seen from the machine, across every
20 repository: ready → building → check queue → checking → review → approved →
21 landed → live, each with how much waits in it, how much is being worked, how
22 long its oldest waiting item has waited, and what limits it (crew runs at once,
23 checks at once, the autonomous repositories' slots). The stage whose oldest
24 waiting item has waited longest is the bottleneck. A forge admin reads it at
25 `/flow` (`web/pages/flow.ts`), one row per stage under one sentence naming the
26 bottleneck; the knobs stay on Settings → Capacity. The same sentence
27 (`verdict`) is the terminal's admin-only Flow command and the MCP `flow` tool,
28 which also returns the stages; anyone but a forge admin is refused.
scripts/gate.ts +1
@@ -102,6 +102,7 @@ export const GATE: readonly string[] = [
102102 "test/decide.test.ts",
103103 "test/ledger.test.ts",
104104 "test/capacity.test.ts",
105 "test/flow-pipeline.test.ts",
105106 "test/limits.test.ts",
106107 "test/guard-new-accounts.test.ts",
107108 "test/checks.test.ts",
src/core/reserved.ts +1
@@ -66,6 +66,7 @@ export const SYSTEM_NAMES: readonly string[] = [
6666 "events",
6767 "explore",
6868 "fleet",
69 "flow",
6970 "front",
7071 "go",
7172 "icons",
src/flow/pipeline.ts added +240
@@ -0,0 +1,240 @@
1 /**
2 * The forge's pipeline as stages, read from what it already keeps (runs,
3 * issues, pulls, reviews): how much is in each stage now, how long its
4 * oldest waiting item has waited, and the capacity that limits it (#1554).
5 * The stage whose oldest waiting item has waited longest is the bottleneck.
6 *
7 * Beside `stage.ts`, which follows one repository's issues, this is the
8 * forge's work as a whole: runs queued and running, against the capacity
9 * that limits them.
10 *
11 * A stage holds items that are *waiting* (queued for the next worker) and
12 * items that are *active* (being worked). Only waiting items can make a
13 * stage the bottleneck: a check that takes twenty minutes is working, a
14 * check that waits twenty minutes to start is the limit.
15 *
16 * Pure data, forge-wide, no authorization: `/flow` is a forge admin's.
17 */
18
19 import type { Database } from "bun:sqlite";
20 import { capacityValue } from "../core/capacity.ts";
21
22 export const PIPELINE = [
23 "ready",
24 "building",
25 "check-queue",
26 "checking",
27 "review",
28 "approved",
29 "landed",
30 "live",
31 ] as const;
32 export type PipelineKey = (typeof PIPELINE)[number];
33
34 export const STAGE_LABEL: Record<PipelineKey, string> = {
35 ready: "Ready",
36 building: "Building",
37 "check-queue": "Check queue",
38 checking: "Checking",
39 review: "Review",
40 approved: "Approved",
41 landed: "Landed",
42 live: "Live",
43 };
44
45 /** What each stage holds, in one line under its name. */
46 export const STAGE_HOLDS: Record<PipelineKey, string> = {
47 ready: "Open, unblocked issues in autonomous repositories with nothing in flight on them",
48 building: "Agent runs (not reviews), queued for a builder or running",
49 "check-queue": "Checks queued to start",
50 checking: "Checks running",
51 review: "Review runs, queued for a reviewer or running",
52 approved: "Open pulls whose latest verdict is an approval, waiting to land",
53 landed: "Pulls merged into an app's default branch in the last week with no production deploy since",
54 live: "Production deploys that went live in the last 24 hours",
55 };
56
57 export interface PipelineStage {
58 readonly key: PipelineKey;
59 /** Queued for the next worker. */
60 readonly waiting: number;
61 /** Being worked, or (live) done. */
62 readonly active: number;
63 /** When the oldest waiting item entered the stage, or null when nothing waits. */
64 readonly oldestWaiting: string | null;
65 /** What limits the stage, in words ("crew runs at once 2"), or null when nothing in the forge does. */
66 readonly capacity: string | null;
67 }
68
69 const DAY_MS = 24 * 3_600_000;
70
71 interface Tally {
72 waiting: number;
73 active: number;
74 oldest: string | null;
75 }
76
77 /** Runs of `kind` in flight; queued ones paused on a limit (`not_before` ahead) are not waiting for a worker. */
78 function runTally(db: Database, kind: string, where: string, now: string): Tally {
79 return db
80 .query<Tally, [string, string]>(
81 `SELECT COALESCE(SUM(state = 'queued'), 0) AS waiting,
82 COALESCE(SUM(state = 'running'), 0) AS active,
83 MIN(CASE WHEN state = 'queued' THEN created_at END) AS oldest
84 FROM run
85 WHERE kind = ?1 AND state IN ('queued','running')
86 AND (state = 'running' OR not_before IS NULL OR not_before <= ?2)
87 AND ${where}`,
88 )
89 .get(kind, now) ?? { waiting: 0, active: 0, oldest: null };
90 }
91
92 /** Rows that all wait: `since` is when each entered the stage. */
93 function waitingTally(db: Database, sql: string, ...args: string[]): Tally {
94 const r = db
95 .query<{ n: number; oldest: string | null }, string[]>(
96 `SELECT COUNT(*) AS n, MIN(since) AS oldest FROM (${sql})`,
97 )
98 .get(...args);
99 return { waiting: r?.n ?? 0, active: 0, oldest: r?.oldest ?? null };
100 }
101
102 const READY = `
103 SELECT i.created_at AS since
104 FROM issue i JOIN repo r ON r.id = i.repo_id
sentinelcrew

The exclusions here (open blocker, agent run in flight, open pull from an agent branch, direction issue) have no test. Seed a blocked issue and one with a queued agent run, and assert that ready stays at 2.
105 WHERE i.kind = 'issue' AND i.state = 'open' AND r.autonomy != 'off'
106 AND (r.direction_issue_id IS NULL OR r.direction_issue_id != i.id)
107 AND NOT EXISTS (SELECT 1 FROM issue_dependency d JOIN issue b ON b.id = d.blocked_by_id
108 WHERE d.issue_id = i.id AND b.state = 'open')
109 AND NOT EXISTS (SELECT 1 FROM run x WHERE x.issue_id = i.id AND x.kind = 'agent'
110 AND x.state IN ('queued','running'))
111 AND NOT EXISTS (SELECT 1 FROM run x JOIN pull p ON p.head_branch = x.branch
112 JOIN issue pi ON pi.id = p.issue_id AND pi.repo_id = x.repo_id
113 WHERE x.issue_id = i.id AND x.kind = 'agent' AND x.branch != ''
114 AND pi.state = 'open')`;
115
sentinelcrew

Untested: the latest non-comment review per pull. Add a test where an approval is followed by changes_requested; that pull must not count as approved.
116 const APPROVED = `
117 SELECT v.created_at AS since
118 FROM issue i JOIN pull p ON p.issue_id = i.id
119 JOIN review v ON v.id = (SELECT id FROM review WHERE issue_id = i.id AND state != 'commented'
120 ORDER BY created_at DESC, id DESC LIMIT 1)
121 WHERE i.state = 'open' AND v.state = 'approved'`;
122
sentinelcrew

Untested: merged into the default branch with no production deploy since. Seed one merged pull before a deploy and one after, and assert that only the second is landed.
123 const LANDED = `
124 SELECT p.merged_at AS since
125 FROM pull p JOIN issue i ON i.id = p.issue_id JOIN repo r ON r.id = i.repo_id
126 JOIN app a ON a.repo_id = r.id
127 WHERE p.merged_at >= ?1 AND p.base_branch = r.default_branch
128 AND NOT EXISTS (SELECT 1 FROM run d WHERE d.repo_id = r.id AND d.kind = 'deploy'
129 AND d.env = 'production' AND d.created_at >= p.merged_at)`;
130
131 function liveTally(db: Database, since: string): Tally {
132 const r = db
133 .query<{ n: number }, [string]>(
134 `SELECT COUNT(*) AS n FROM run
135 WHERE kind = 'deploy' AND env = 'production' AND serving_at >= ?1`,
136 )
137 .get(since);
138 return { waiting: 0, active: r?.n ?? 0, oldest: null };
139 }
140
141 function tally(db: Database, key: PipelineKey, now: Date): Tally {
142 const iso = now.toISOString();
143 switch (key) {
144 case "ready":
145 return waitingTally(db, READY);
146 case "building":
147 return runTally(db, "agent", "reason != 'review'", iso);
148 case "check-queue":
149 return runTally(db, "check", "state = 'queued'", iso);
150 case "checking":
151 return runTally(db, "check", "state = 'running'", iso);
152 case "review":
153 return runTally(db, "agent", "reason = 'review'", iso);
154 case "approved":
155 return waitingTally(db, APPROVED);
156 case "landed":
157 return waitingTally(db, LANDED, new Date(now.getTime() - 7 * DAY_MS).toISOString());
158 case "live":
159 return liveTally(db, new Date(now.getTime() - DAY_MS).toISOString());
160 }
161 }
162
163 /** The autonomous repositories' slots added up, and how many repositories. */
164 function slots(db: Database): { slots: number; repos: number } {
165 const r = db
166 .query<{ slots: number | null; repos: number }, []>(
167 `SELECT SUM(autonomy_slots) AS slots, COUNT(*) AS repos FROM repo WHERE autonomy != 'off'`,
168 )
169 .get();
170 return { slots: r?.slots ?? 0, repos: r?.repos ?? 0 };
171 }
172
173 const auto = (v: string) => (v === "" ? "auto" : v);
174
175 /** What limits each stage, read now: the capacity settings and the repositories' slots. */
176 function capacities(db: Database): Record<PipelineKey, string | null> {
177 const crew = `crew runs at once ${auto(capacityValue(db, "crew.concurrency"))}`;
178 const checks = `checks at once ${auto(capacityValue(db, "checks.concurrency"))}`;
179 const s = slots(db);
180 return {
181 ready: `${s.slots} slot${s.slots === 1 ? "" : "s"} in ${s.repos} autonomous ${s.repos === 1 ? "repository" : "repositories"}`,
182 building: crew,
183 "check-queue": checks,
184 checking: checks,
185 review: crew,
186 approved: null,
187 landed: null,
188 live: null,
189 };
190 }
191
192 /** Every stage, left to right, as it is now. */
193 export function readPipeline(db: Database, now = new Date()): PipelineStage[] {
194 const cap = capacities(db);
195 return PIPELINE.map((key) => {
196 const t = tally(db, key, now);
197 return {
198 key,
199 waiting: t.waiting,
200 active: t.active,
201 oldestWaiting: t.oldest,
202 capacity: cap[key],
203 };
204 });
205 }
206
207 /** The stage whose oldest waiting item has waited longest; undefined when nothing waits. */
208 export function bottleneck(stages: readonly PipelineStage[]): PipelineStage | undefined {
209 let worst: PipelineStage | undefined;
210 for (const s of stages)
211 if (
212 s.oldestWaiting !== null &&
213 (worst?.oldestWaiting == null || s.oldestWaiting < worst.oldestWaiting)
214 )
215 worst = s;
216 return worst;
217 }
218
219 /** A span of milliseconds as a person reads it: 45s, 12m, 3h 5m, 2d 4h. */
220 export function span(ms: number): string {
221 const s = Math.max(0, Math.round(ms / 1000));
222 if (s < 60) return `${s}s`;
223 const m = Math.floor(s / 60);
224 if (m < 60) return `${m}m`;
225 const h = Math.floor(m / 60);
226 if (h < 48) return m % 60 === 0 ? `${h}h` : `${h}h ${m % 60}m`;
227 return `${Math.floor(h / 24)}d ${h % 24}h`;
228 }
229
230 /** How long a stage's oldest waiting item has waited, or null when nothing waits. */
231 export const waited = (s: PipelineStage, now: number): string | null =>
232 s.oldestWaiting === null ? null : span(now - Date.parse(s.oldestWaiting));
233
234 /** The one sentence naming the bottleneck, or saying nothing waits: the page, the terminal and MCP all say this. */
235 export function verdict(stages: readonly PipelineStage[], now = Date.now()): string {
236 const worst = bottleneck(stages);
237 if (worst === undefined) return "Nothing is waiting: every stage is empty or being worked.";
238 const n = worst.waiting;
239 return `${STAGE_LABEL[worst.key]} is the bottleneck: its oldest has waited ${waited(worst, now) ?? ""}, ${n} ${n === 1 ? "waits" : "wait"}${worst.capacity === null ? "" : `, ${worst.capacity}`}.`;
240 }
src/http/mcp.ts +14
@@ -19,6 +19,7 @@
1919
2020 import { mayView, mayViewId, sealedScope } from "../core/sealed.ts";
2121 import { changeCapacity, listCapacity } from "../actions/capacity.ts";
22 import { readPipeline, verdict } from "../flow/pipeline.ts";
2223 import { changeLimit, listLimits } from "../actions/limits.ts";
2324 import { LIMIT_KEYS } from "../settings/limits.ts";
2425 import { relationOf, takeUpdates } from "../actions/forks.ts";
@@ -1823,6 +1824,19 @@ const TOOLS: readonly Tool[] = [
18231824 });
18241825 },
18251826 },
1827 {
1828 name: "flow",
1829 description:
1830 "The forge's pipeline, for a forge admin: ready, building, check queue, checking, review, approved, landed and live, each with how many wait, how many are being worked, when the oldest waiting one entered (oldestWaiting) and what limits it (capacity), and one sentence (says) naming the stage whose oldest has waited longest — the bottleneck. Read-only; the knobs are the `capacity` tool. The same as /flow.",
1831 inputSchema: { type: "object", properties: {} },
1832 writes: false,
1833 run(ctx, user) {
1834 if (!user.isAdmin)
1835 return bad("forbidden", "only an admin of this forge reads its pipeline");
1836 const stages = readPipeline(ctx.app.db);
1837 return good({ says: verdict(stages), stages });
1838 },
1839 },
18261840 {
18271841 name: "limits",
18281842 description:
src/http/server.ts +3
@@ -8,6 +8,7 @@
88 import { TOOLBAR_NAME, toolbarResponse } from "../apps/toolbar.ts";
99 import { unsealReleased } from "../actions/sealed.ts";
1010 import { capacityPage, capacitySet } from "../web/pages/capacity.ts";
11 import { flowPage } from "../web/pages/flow.ts";
1112 import { limitsPage, limitsSet } from "../web/pages/admin-limits.ts";
1213 import {
1314 closeStreamsOf,
@@ -616,6 +617,8 @@ export function buildRouter(): Router<Handler> {
616617 router.add("GET", "/migrate/jobs/:job", migrationJobPage);
617618 // Every app on one canvas; a literal route, so it must sit above /:owner.
618619 router.add("GET", "/map", globalMapPage);
620 // The pipeline's stages and its bottleneck, for a forge admin (#1554); literal, above /:owner.
621 router.add("GET", "/flow", flowPage);
619622 // A selection's spend and runs over time; literal, above /:owner.
620623 router.add("GET", "/fleet/graph", fleetGraphPage);
621624 router.add("GET", "/fleet/graph/live", fleetGraphLive);
src/tui/forge.ts +9
@@ -11,6 +11,7 @@
1111 import type { App as Forge } from "../app.ts";
1212 import { search } from "../actions/search.ts";
1313 import { sentinelSummary } from "../crew/sentinel.ts";
14 import { readPipeline, verdict } from "../flow/pipeline.ts";
1415 import { canRead, getRepo, type Repo } from "../core/repos.ts";
1516 import { sourceFor } from "../view/source.ts";
1617 import { App, type Ctx, type Screen } from "./app/app.ts";
@@ -183,6 +184,14 @@ export function forgeCommands(session: Session): (ctx: Ctx) => Command[] {
183184 "default",
184185 ),
185186 });
187 // The pipeline's bottleneck, as /flow's sentence says it: a site admin's (#1554).
188 if (session.user?.isAdmin === true)
189 commands.push({
190 id: "flow",
191 title: "Flow",
192 hint: "the pipeline's bottleneck",
193 run: () => ctx.toast(verdict(readPipeline(session.db)), "default"),
194 });
186195 commands.push({
187196 id: "quit",
188197 title: "Quit",
src/web/layout.ts +16 −2
@@ -316,7 +316,10 @@ function paletteIndex(ctx: Ctx, extra: PaletteItem[]): Html {
316316 ? [{ t: "New forge", h: "/forges/new", c: "Create" }]
317317 : []),
318318 ...(ctx.auth.user?.isAdmin
319 ? [{ t: "Forges you sent", h: "/forges", c: "Go to" }]
319 ? [
320 { t: "Flow", h: "/flow", c: "Go to" },
321 { t: "Forges you sent", h: "/forges", c: "Go to" },
322 ]
320323 : []),
321324 ];
322325 // `studio` rides along so the palette's *remote* rows — what the search
@@ -437,7 +440,18 @@ function topbar(
437440 )}
438441 <span class="spacer"></span>
439442 <span class="topbar-tools">
440 ${searchBar()}<span class="hide-sm">${helpButton()}</span>${themeMenu(theme)}
443 ${when(
444 user?.isAdmin === true,
445 () =>
446 html`<a
447 id="flow-link"
448 class="btn"
449 data-size="sm"
450 href="/flow"
451 title="The pipeline and its bottleneck"
452 >${icon("activity")}<span class="hide-sm">Flow</span></a
453 >`,
454 )}${searchBar()}<span class="hide-sm">${helpButton()}</span>${themeMenu(theme)}
441455 ${user
442456 ? html`<span class="hide-sm">${newMenu(ctx)}</span>${userMenu(
443457 ctx,
src/web/pages/flow.ts added +79
@@ -0,0 +1,79 @@
1 /**
2 * /flow, for a forge admin: the pipeline from a ready issue to a live
3 * deploy as rows, one per stage, each with how much is in it now, how long
4 * its oldest waiting item has waited, and the capacity that limits it
5 * (#1554). One sentence above names the stage with the longest wait — the
6 * bottleneck. The numbers are `flow/pipeline.ts`'s; the knobs stay on
7 * Settings → Capacity, linked from here.
8 */
9
10 import {
11 bottleneck,
12 readPipeline,
13 STAGE_HOLDS,
14 STAGE_LABEL,
15 verdict,
16 waited,
17 } from "../../flow/pipeline.ts";
18 import type { Ctx } from "../../http/context.ts";
19 import { notFoundPage, signInFirst } from "../chrome.ts";
20 import { html } from "../html.ts";
21 import { page } from "../layout.ts";
22 import { badge, card } from "../ui.ts";
23
24 // GET /flow
25 export function flowPage(ctx: Ctx): Response {
26 if (!ctx.auth.user) return signInFirst(ctx);
27 if (!ctx.auth.user.isAdmin) return notFoundPage(ctx);
28 const now = Date.now();
29 const stages = readPipeline(ctx.app.db, new Date(now));
30 const worst = bottleneck(stages);
31 const body = html`<div class="page-head">
32 <h1>Flow</h1>
33 <p class="muted">
34 From a ready issue to a live deploy: what is in each stage now, how
35 long its oldest has waited, and what limits it. The limits are set on
36 <a href="/settings/capacity">Capacity</a> and in each repository's
37 autonomy settings.
38 </p>
39 </div>
sentinelcrew outdated

Move this into src/flow/pipeline.ts (together with span) so the page, an admin-only flow command in src/tui/forge.ts (built like the existing sentinel one) and a read-only MCP flow tool all say the same sentence. As it stands, only the page can tell an admin where the bottleneck is.
40 <p id="bottleneck" data-stage="${worst?.key ?? ""}">
41 <strong>${verdict(stages, now)}</strong>
42 </p>
43 ${card(
44 html`<div class="table-scroll">
45 <table class="table" aria-label="Pipeline stages">
46 <thead>
47 <tr>
48 <th scope="col">Stage</th>
49 <th scope="col">Waiting</th>
50 <th scope="col">Active</th>
51 <th scope="col">Oldest wait</th>
52 <th scope="col">Capacity</th>
53 </tr>
54 </thead>
55 <tbody>
56 ${stages.map(
57 (s) =>
58 html`<tr
59 data-stage="${s.key}"
60 ${s === worst ? html`data-bottleneck="true"` : null}
61 >
62 <th scope="row">
63 ${STAGE_LABEL[s.key]}
64 ${s === worst ? badge("bottleneck", { tone: "danger" }) : null}
65 <div class="muted small">${STAGE_HOLDS[s.key]}</div>
66 </th>
67 <td class="mono">${String(s.waiting)}</td>
68 <td class="mono">${String(s.active)}</td>
69 <td class="mono">${waited(s, now) ?? "—"}</td>
70 <td class="small">${s.capacity ?? "—"}</td>
71 </tr>`,
72 )}
73 </tbody>
74 </table>
75 </div>`,
76 { pad: false },
77 )}`;
78 return page({ ctx, title: "Flow", body });
79 }
test/flow-pipeline.test.ts added +312
@@ -0,0 +1,312 @@
1 /**
2 * /flow shows the forge's own bottleneck (#1554): from a seeded forge —
3 * checks queued for hours, builder runs queued for minutes, two ready
4 * issues an hour old beside open issues that are not ready, pulls approved
5 * and overridden, merged before and after a deploy, deploys live today and
6 * two days ago — each stage counts what it holds, the check queue is
7 * named as the bottleneck in one sentence with its capacity — on the page,
8 * in the terminal and through MCP — and nobody but a forge admin sees any
9 * of it.
10 */
11
12 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
13 import { mkdtemp, rm } from "node:fs/promises";
14 import { tmpdir } from "node:os";
15 import { join } from "node:path";
16 import { createApp, type App } from "../src/app.ts";
17 import { openIssue } from "../src/actions/issues.ts";
18 import { createRepo, updateRepoSettings } from "../src/actions/repos.ts";
19 import { addDependency } from "../src/core/dependencies.ts";
20 import { setDirectionIssue } from "../src/core/repos.ts";
21 import { setCapacity } from "../src/core/capacity.ts";
22 import { createRun } from "../src/core/runs.ts";
23 import { createUser, type User } from "../src/core/users.ts";
24 import { createToken } from "../src/core/tokens.ts";
25 import { bottleneck, readPipeline, span, verdict } from "../src/flow/pipeline.ts";
26 import { serve, type Servers } from "../src/http/server.ts";
27 import type { Ctx } from "../src/tui/app/app.ts";
28 import { forgeCommands, Session } from "../src/tui/forge.ts";
29
30 const HOUR = 3_600_000;
31 const ago = (ms: number) => new Date(Date.now() - ms).toISOString();
32
33 describe("/flow: the forge shows its own bottleneck", () => {
34 let dataDir: string;
35 let app: App;
36 let servers: Servers;
37 let alice: User;
38 let bob: User;
39
40 const login = async (username: string): Promise<string> => {
41 const res = await fetch(`${servers.url}/signin`, {
42 method: "POST",
43 redirect: "manual",
44 headers: {
45 "content-type": "application/x-www-form-urlencoded",
46 origin: servers.url,
47 },
48 body: new URLSearchParams({ username, password: "a-good-passphrase" }),
49 });
50 return (res.headers.get("set-cookie") ?? "").split(";")[0] ?? "";
51 };
52 const get = (path: string, cookie: string) =>
53 fetch(`${servers.url}${path}`, { headers: { cookie }, redirect: "manual" });
54
55 beforeAll(async () => {
56 dataDir = await mkdtemp(join(tmpdir(), "forge-flow-"));
57 app = createApp({ dataDir, port: 0 });
58 servers = await serve(app);
59 const a = await createUser(app.db, {
60 trusted: true,
61 username: "alice",
62 email: "[email protected]",
63 password: "a-good-passphrase",
64 adminIfFirst: true,
65 });
66 if (!a.ok) throw new Error(a.error);
67 alice = a.value;
68 const b = await createUser(app.db, {
69 trusted: true,
70 username: "bob",
71 email: "[email protected]",
72 password: "a-good-passphrase",
73 });
74 if (!b.ok) throw new Error(b.error);
75 bob = b.value;
76 const made = await createRepo(app, alice, { name: "shop" });
77 if (!made.ok) throw new Error(String(made.error));
78 const set = await updateRepoSettings(app, alice, made.value, {
79 autonomy: "assisted",
80 autonomyDirection: "keep the shop open",
81 autonomySlots: 2,
82 autonomyCeilingUsd: 15,
83 autonomyBy: alice.id,
84 });
85 if (!set.ok) throw new Error(String(set.error));
86 const repo = set.value;
87 setCapacity(app.db, "checks.concurrency", "1", alice.id);
88 setCapacity(app.db, "crew.concurrency", "3", alice.id);
89
90 // Two ready issues, opened an hour ago: they wait, but not the longest.
91 for (const title of ["Add a cart", "Show the prices"]) {
92 const issue = openIssue(app, alice, repo, { title });
93 if (!issue.ok) throw new Error(String(issue.error));
94 app.db
95 .query("UPDATE issue SET created_at = ?1 WHERE id = ?2")
96 .run(ago(HOUR), issue.value.id);
97 }
98 // Three checks queued three hours ago, one running: the bottleneck.
99 for (let n = 0; n < 4; n++) {
100 const run = createRun(app.db, { repoId: repo.id, kind: "check", reason: "push" });
101 app.db
102 .query("UPDATE run SET created_at = ?1, state = ?2, started_at = ?3 WHERE id = ?4")
103 .run(ago(3 * HOUR), n === 0 ? "running" : "queued", n === 0 ? ago(HOUR) : null, run.id);
104 }
105 // A builder run queued ten minutes ago.
106 const build = createRun(app.db, { repoId: repo.id, kind: "agent", reason: "assigned" });
107 app.db
108 .query("UPDATE run SET created_at = ?1 WHERE id = ?2")
109 .run(ago(10 * 60_000), build.id);
110 // A check paused on a limit until later is not waiting for a worker.
111 const paused = createRun(app.db, { repoId: repo.id, kind: "check", reason: "push" });
112 app.db
113 .query("UPDATE run SET created_at = ?1, not_before = ?2 WHERE id = ?3")
114 .run(ago(9 * HOUR), new Date(Date.now() + HOUR).toISOString(), paused.id);
115
116 // Open issues that are not ready: one blocked by an open issue, the
117 // blocker itself with an agent run queued on it, one whose agent branch
118 // already has an open pull, and the direction issue. Ready stays at 2.
119 const newIssue = (title: string) => {
120 const issue = openIssue(app, alice, repo, { title });
121 if (!issue.ok) throw new Error(String(issue.error));
122 app.db.query("UPDATE issue SET created_at = ?1 WHERE id = ?2").run(ago(HOUR), issue.value.id);
123 return issue.value;
124 };
125 const blocker = newIssue("Pick a payment provider");
126 const blocked = newIssue("Take payments");
127 const linked = addDependency(app.db, blocked, blocker, alice.id);
128 if (!linked.ok) throw new Error(linked.error);
129 const queuedOn = createRun(app.db, {
130 repoId: repo.id,
131 kind: "agent",
132 reason: "assigned",
133 issueId: blocker.id,
134 });
135 app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(ago(5 * 60_000), queuedOn.id);
136 const built = newIssue("Add a footer");
137 const builtRun = createRun(app.db, {
138 repoId: repo.id,
sentinelcrew outdated

Approved, landed and live are only checked for their capacity being null. Seed one row each that should count and one that should not, and assert the counts. Without that, APPROVED, LANDED and liveTally can be emptied and this suite stays green.
139 kind: "agent",
140 reason: "assigned",
141 issueId: built.id,
142 branch: "mason/footer",
143 });
144 app.db.query("UPDATE run SET state = 'ok' WHERE id = ?1").run(builtRun.id);
145 setDirectionIssue(app.db, repo.id, newIssue("Direction").id);
146
147 // Pulls, made from issues: open or merged into the default branch.
148 const pull = (title: string, head: string, mergedAt: string | null) => {
149 const issue = newIssue(title);
150 app.db
151 .query("UPDATE issue SET kind = 'pull', state = ?1 WHERE id = ?2")
152 .run(mergedAt === null ? "open" : "closed", issue.id);
153 app.db
154 .query(
155 `INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, merged_at)
156 VALUES (?1, ?2, ?3, '', '', ?4)`,
157 )
158 .run(issue.id, repo.defaultBranch, head, mergedAt);
159 return issue.id;
160 };
161 const review = (issueId: string, state: string, at: string) =>
162 app.db
163 .query(
164 `INSERT INTO review (id, issue_id, author_id, state, created_at)
165 VALUES (?1, ?2, ?3, ?4, ?5)`,
166 )
167 .run(crypto.randomUUID(), issueId, alice.id, state, at);
168 // Approved, with a comment after: still approved — and its head is the footer's agent branch.
169 const approved = pull("Footer", "mason/footer", null);
170 review(approved, "approved", ago(30 * 60_000));
171 review(approved, "commented", ago(20 * 60_000));
172 // Approved, then changes requested: not approved.
173 const overridden = pull("Header", "mason/header", null);
174 review(overridden, "approved", ago(40 * 60_000));
175 review(overridden, "changes_requested", ago(35 * 60_000));
176
177 // The shop is an app: one pull merged before its last production deploy,
178 // one merged after (landed, not live yet).
179 app.db
180 .query("INSERT INTO app (id, repo_id, slug, access) VALUES (?1, ?2, 'shop', 'public')")
181 .run(crypto.randomUUID(), repo.id);
182 pull("Logo", "mason/logo", ago(2 * HOUR));
183 pull("Colours", "mason/colours", ago(30 * 60_000));
184 const deploy = (createdAt: string, servingAt: string) => {
185 const run = createRun(app.db, { repoId: repo.id, kind: "deploy", env: "production" });
186 app.db
187 .query("UPDATE run SET state = 'ok', created_at = ?1, serving_at = ?2 WHERE id = ?3")
188 .run(createdAt, servingAt, run.id);
189 };
190 // Live within the hour, and one two days ago that is not counted.
191 deploy(ago(90 * 60_000), ago(HOUR - 60_000));
192 deploy(ago(48 * HOUR + 60_000), ago(48 * HOUR));
193 });
194
195 afterAll(async () => {
196 servers.stop();
197 app.close();
198 await rm(dataDir, { recursive: true, force: true });
199 });
200
201 test("each stage counts what it holds, with its capacity; the longest wait is the bottleneck", () => {
202 const stages = readPipeline(app.db);
203 expect(stages.map((s) => s.key)).toEqual([
204 "ready",
205 "building",
206 "check-queue",
207 "checking",
208 "review",
209 "approved",
210 "landed",
211 "live",
212 ]);
213 const by = new Map(stages.map((s) => [s.key, s]));
214 expect(by.get("ready")).toMatchObject({ waiting: 2, active: 0 });
215 expect(by.get("ready")?.capacity).toBe("2 slots in 1 autonomous repository");
216 expect(by.get("building")).toMatchObject({ waiting: 2, capacity: "crew runs at once 3" });
217 expect(by.get("check-queue")).toMatchObject({ waiting: 3, capacity: "checks at once 1" });
218 expect(by.get("checking")).toMatchObject({ waiting: 0, active: 1 });
219 expect(by.get("review")).toMatchObject({ waiting: 0, active: 0, oldestWaiting: null });
220 expect(by.get("approved")).toMatchObject({ waiting: 1, active: 0, capacity: null });
221 expect(by.get("landed")).toMatchObject({ waiting: 1, active: 0, capacity: null });
222 expect(by.get("live")).toMatchObject({ waiting: 0, active: 1, capacity: null });
223 expect(bottleneck(stages)?.key).toBe("check-queue");
224 expect(verdict(stages)).toBe(
225 "Check queue is the bottleneck: its oldest has waited 3h, 3 wait, checks at once 1.",
226 );
227 });
228
229 test("with nothing waiting, nothing is named", () => {
230 expect(verdict([])).toBe("Nothing is waiting: every stage is empty or being worked.");
231 expect(span(45_000)).toBe("45s");
232 expect(span(3 * HOUR + 5 * 60_000)).toBe("3h 5m");
233 expect(span(52 * HOUR)).toBe("2d 4h");
234 });
235
236 test("an admin reads the stages as rows under the one sentence, linked from the header", async () => {
237 const cookie = await login("alice");
238 const res = await get("/flow", cookie);
239 expect(res.status).toBe(200);
240 const page = await res.text();
241 expect(page).toContain('id="bottleneck" data-stage="check-queue"');
242 expect(page).toContain("Check queue is the bottleneck");
243 expect(page).toMatch(/data-stage="check-queue"\s+data-bottleneck="true"/);
244 for (const key of ["ready", "building", "checking", "review", "approved", "landed", "live"])
245 expect(page).toContain(`data-stage="${key}"`);
246 expect(page).toContain('href="/settings/capacity"');
247 expect(page).toContain('id="flow-link"');
248 });
249
250 test("nobody but a forge admin reads it", async () => {
251 const bob = await login("bob");
252 expect((await get("/flow", bob)).status).toBe(404);
253 expect(await (await get("/", bob)).text()).not.toContain('id="flow-link"');
254 expect((await get("/flow", "")).status).not.toBe(200);
255 });
256
257 const flowCommand = (user?: User) => {
258 const toasts: string[] = [];
259 const ctx = { toast: (m: string) => toasts.push(m) } as unknown as Ctx;
260 const session = new Session(app, {
261 sshUsername: user?.username ?? "git",
262 remoteAddress: "127.0.0.1",
263 fingerprint: "SHA256:test",
264 user,
265 });
266 return { found: forgeCommands(session)(ctx).find((c) => c.id === "flow"), toasts };
267 };
268
269 test("in the terminal, an admin's Flow command says the same sentence; nobody else is offered it", () => {
270 const { found, toasts } = flowCommand(alice);
271 expect(found?.title).toBe("Flow");
272 found!.run();
273 expect(toasts).toEqual([verdict(readPipeline(app.db))]);
274 expect(toasts[0]).toStartWith("Check queue is the bottleneck");
275 expect(flowCommand(bob).found).toBeUndefined();
276 expect(flowCommand(undefined).found).toBeUndefined();
277 });
278
279 test("through MCP, an admin reads the stages and the sentence; anyone else is forbidden", async () => {
280 const call = async (token: string) => {
281 const res = await fetch(`${servers.url}/mcp`, {
282 method: "POST",
283 headers: {
284 authorization: `Bearer ${token}`,
285 "content-type": "application/json",
286 accept: "application/json, text/event-stream",
287 },
288 body: JSON.stringify({
289 jsonrpc: "2.0",
290 id: 1,
291 method: "tools/call",
292 params: { name: "flow", arguments: {} },
293 }),
294 });
295 const body = (await res.json()) as {
296 result: { isError?: boolean; content: { text: string }[] };
297 };
298 return { error: body.result.isError === true, text: body.result.content[0]!.text };
299 };
300 const admin = await call(createToken(app.db, alice.id, "t", "ro").token);
301 expect(admin.error).toBe(false);
302 const read = JSON.parse(admin.text) as {
303 says: string;
304 stages: { key: string; waiting: number }[];
305 };
306 expect(read.says).toStartWith("Check queue is the bottleneck");
307 expect(read.stages.find((s) => s.key === "check-queue")?.waiting).toBe(3);
308 const member = await call(createToken(app.db, bob.id, "t", "admin").token);
309 expect(member.error).toBe(true);
310 expect(member.text).toContain("forbidden");
311 });
312 });
test/mcp.test.ts +1
@@ -200,6 +200,7 @@ describe("the forge over MCP", () => {
200200 "hold",
201201 "plan",
202202 "capacity",
203 "flow",
203204 "limits",
204205 "report",
205206 ]);

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs