Public: anyone can read this; the crew can change it.
Throughput: 70 open crew pulls are waiting on a builder or a reviewer. #11725
Check · Review · Land · Live done — Landed and live in production.
torque/forge:torque/11647-throughput-70-open-crew-pulls-are-waitin → main 2 commits fork 1 thread · 1 unresolved
4 files
changed +66
−22
docs/features/keeper.md +1 −1
| @@ -4,7 +4,7 @@ The forge watches its own landing rate, names the one thing binding it, and acts | ||
| 4 | 4 | |
| 5 | 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 | 6 | 2. **Find** (`keeper_cause.ts`): the first of `ceiling`, `usage` ("resets at HH:MM UTC"), `unassigned`, `empty_pillar`, `stale_reds`, `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 (`stale_pile`'s issue lists the ten oldest waiting pulls, red and conflicted ones left out, so a builder knows where to start). `ceiling`, `usage`, `capacity` and `stale_reds` (the crew's red-pull sweep is already at it) are only said. | |
| 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 (`stale_pile`'s issue lists the ten oldest waiting pulls, red and conflicted ones left out, each with what it waits on: a reviewer, a builder or the merge). `ceiling`, `usage`, `capacity` and `stale_reds` (the crew's red-pull sweep is already at it) are only said. | |
| 8 | 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 | 9 | |
| 10 | 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`. |
src/flow/keeper.ts +3 −3
| @@ -17,7 +17,7 @@ import { createLog } from "../kernel/log.ts"; | ||
| 17 | 17 | import { ownRepoPath } from "../kernel/own_repo.ts"; |
| 18 | 18 | import { KEEPER_CAUSES, pickCause, readKeeperFacts, type KeeperCause, type KeeperMove } from "./keeper_cause.ts"; |
| 19 | 19 | import { isLow, landedBetween, readRate, type Rate } from "./keeper_rate.ts"; |
| 20 | import { oldestWaiting } from "./pile.ts"; | |
| 20 | import { oldestWaitingOn } from "./pile.ts"; | |
| 21 | 21 | import { during } from "../perf/jobs.ts"; |
| 22 | 22 | |
| 23 | 23 | const log = createLog("keeper"); |
| @@ -117,8 +117,8 @@ export function fileOnce(app: App, found: { cause: KeeperCause; says: string }, | ||
| 117 | 117 | const [owner, name] = ownRepoPath().split("/"); |
| 118 | 118 | const repo = owner && name ? getRepo(app.db, owner, name) : undefined; |
| 119 | 119 | if (!repo) return log.warn("keeper cannot file: the forge's own repository is missing", { path: ownRepoPath() }); |
| 120 | const oldest = found.cause === "stale_pile" ? oldestWaiting(app.db, 10, repo.visibility !== "private") : []; | |
| 121 | const list = oldest.length === 0 ? "" : `\n\nOldest waiting, start here:\n${oldest.map((s) => `- ${s}`).join("\n")}`; | |
| 120 | const oldest = found.cause === "stale_pile" ? oldestWaitingOn(app.db, 10, repo.visibility !== "private") : []; | |
| 121 | const list = oldest.length === 0 ? "" : `\n\nOldest waiting, start here:\n${oldest.map((s) => `- ${s.ref} (waits on ${s.on})`).join("\n")}`; | |
| 122 | 122 | const made = createIssue(app.db, { |
| 123 | 123 | repoId: repo.id, |
| 124 | 124 | authorId: repo.ownerId, |
src/flow/pile.ts +29 −9
| @@ -64,16 +64,27 @@ export function pileSays(pile: Pile): string { | ||
| 64 | 64 | return `${pile.total} open crew ${pile.total === 1 ? "pull" : "pulls"}, each counted once in the first pile it fits: ${pile.red} red, ${pile.conflicted} conflicted, ${pile.changesAsked} changes asked, ${pile.approved} approved and waiting, ${pile.awaitingReview} awaiting review.`; |
| 65 | 65 | } |
| 66 | 66 | |
| 67 | /** What a waiting pull waits on: its latest verdict decides (a bare comment is no verdict). */ | |
| 68 | export type WaitsOn = "a reviewer" | "a builder" | "the merge"; | |
| 69 | ||
| 70 | export interface Waiting { | |
| 71 | readonly ref: string; | |
| 72 | readonly on: WaitsOn; | |
| 73 | } | |
| 74 | ||
| 67 | 75 | /** |
| 68 | 76 | * The oldest open crew pulls that wait on a person or a reviewer (not red, |
| 69 | * not conflicted), as `owner/repo#n`: what a filed "waiting" issue names so a | |
| 70 | * builder can start at the first one. Private repositories are left out | |
| 71 | * unless the caller says `publicOnly` is false: the safe value is the default. | |
| 77 | * not conflicted), each with what it waits on: no verdict waits on a | |
| 78 | * reviewer, changes requested on a builder, approved on the merge. A filed | |
| 79 | * "waiting" issue names them so whoever picks it up knows which act is theirs. | |
| 80 | * Private repositories are left out unless `publicOnly` is false. | |
| 72 | 81 | */ |
| 73 | export function oldestWaiting(db: Database, limit: number, publicOnly = true): string[] { | |
| 82 | export function oldestWaitingOn(db: Database, limit: number, publicOnly = true): Waiting[] { | |
| 74 | 83 | const rows = db |
| 75 | .query<{ repo: string; number: number }, [number, number]>( | |
| 76 | `SELECT o.username || '/' || r.name AS repo, i.number AS number | |
| 84 | .query<{ repo: string; number: number; verdict: string | null }, [number]>( | |
| 85 | `SELECT o.username || '/' || r.name AS repo, i.number AS number, | |
| 86 | (SELECT v.state FROM review v WHERE v.issue_id = i.id AND v.state IN ('approved','changes_requested') | |
| 87 | ORDER BY v.created_at DESC, v.id DESC LIMIT 1) AS verdict | |
| 77 | 88 | FROM issue i JOIN pull p ON p.issue_id = i.id JOIN repo r ON r.id = i.repo_id |
| 78 | 89 | JOIN user o ON o.id = r.owner_id JOIN user a ON a.id = i.author_id |
| 79 | 90 | WHERE i.state = 'open' AND p.merged_at IS NULL AND (?1 = 0 OR r.visibility = 'public') |
| @@ -81,9 +92,18 @@ export function oldestWaiting(db: Database, limit: number, publicOnly = true): s | ||
| 81 | 92 | AND EXISTS (SELECT 1 FROM repo_crew c WHERE c.repo_id = r.id AND c.user_id = a.id) |
| 82 | 93 | AND NOT COALESCE((SELECT c.state FROM run c WHERE c.kind = 'check' AND c.issue_id = i.id |
| 83 | 94 | AND c.commit_oid = p.head_oid ORDER BY c.id DESC LIMIT 1) = 'failed', 0) |
| 84 | ORDER BY i.created_at ASC, i.id ASC LIMIT ?2`, | |
| 95 | ORDER BY i.created_at ASC, i.id ASC`, | |
| 85 | 96 | ) |
| 86 | .all(publicOnly ? 1 : 0, Math.max(0, Math.floor(limit))); | |
| 97 | .all(publicOnly ? 1 : 0); | |
| 87 | 98 | const conflicted = new Set(conflictedPulls(db, publicOnly).map((p) => `${p.repo}#${p.number}`)); |
| 88 | return rows.map((r) => `${r.repo}#${r.number}`).filter((s) => !conflicted.has(s)); | |
| 99 | const cap = Math.max(0, Math.floor(limit)); | |
| 100 | const out: Waiting[] = []; | |
| 101 | if (cap === 0) return out; | |
| 102 | for (const r of rows) { | |
| 103 | const ref = `${r.repo}#${r.number}`; | |
| 104 | if (conflicted.has(ref)) continue; | |
| 105 | out.push({ ref, on: r.verdict === "approved" ? "the merge" : r.verdict === "changes_requested" ? "a builder" : "a reviewer" }); | |
| 106 | if (out.length >= cap) break; | |
| 107 | } | |
| 108 | return out; | |
| 89 | 109 | } |
test/flow-pile-oldest.test.ts +33 −9
| @@ -9,7 +9,7 @@ import { createRepo } from "../src/core/repos.ts"; | ||
| 9 | 9 | import { createUser } from "../src/core/users.ts"; |
| 10 | 10 | import { id } from "../src/kernel/id.ts"; |
| 11 | 11 | import { fileOnce } from "../src/flow/keeper.ts"; |
| 12 | import { oldestWaiting } from "../src/flow/pile.ts"; | |
| 12 | import { oldestWaitingOn } from "../src/flow/pile.ts"; | |
| 13 | 13 | |
| 14 | 14 | let root: string; |
| 15 | 15 | let app: App; |
| @@ -35,6 +35,8 @@ afterAll(() => { | ||
| 35 | 35 | rmSync(root, { recursive: true, force: true }); |
| 36 | 36 | }); |
| 37 | 37 | |
| 38 | const refs = (limit: number, publicOnly = true): string[] => oldestWaitingOn(app.db, limit, publicOnly).map((w) => w.ref); | |
| 39 | ||
| 38 | 40 | function pull(number: number, createdAt: string): string { |
| 39 | 41 | const issue = id(); |
| 40 | 42 | app.db |
| @@ -50,15 +52,15 @@ test("oldest first, red pulls left out, capped at the limit", () => { | ||
| 50 | 52 | pull(3, "2026-11-01T02:00:00.000Z"); |
| 51 | 53 | pull(4, "2026-11-01T03:00:00.000Z"); |
| 52 | 54 | app.db.query(`INSERT INTO run (id, repo_id, kind, issue_id, commit_oid, state, reason) VALUES ('0001x', ?1, 'check', ?2, 'h', 'failed', '')`).run(repoId, red); |
| 53 | expect(oldestWaiting(app.db, 2)).toEqual(["trevato/open#1", "trevato/open#3"]); | |
| 54 | expect(oldestWaiting(app.db, 10)).toEqual(["trevato/open#1", "trevato/open#3", "trevato/open#4"]); | |
| 55 | expect(oldestWaiting(app.db, 0)).toEqual([]); | |
| 55 | expect(refs(2)).toEqual(["trevato/open#1", "trevato/open#3"]); | |
| 56 | expect(refs(10)).toEqual(["trevato/open#1", "trevato/open#3", "trevato/open#4"]); | |
| 57 | expect(refs(0)).toEqual([]); | |
| 56 | 58 | }); |
| 57 | 59 | |
| 58 | 60 | test("a conflicted pull is left out", () => { |
| 59 | 61 | const c = pull(5, "2026-11-01T04:00:00.000Z"); |
| 60 | 62 | app.db.query(`INSERT INTO run (id, repo_id, kind, issue_id, state, reason, created_at) VALUES ('0002x', ?1, 'agent', ?2, 'queued', 'conflict', '2026-11-02T00:00:00.000Z')`).run(repoId, c); |
| 61 | expect(oldestWaiting(app.db, 10)).toEqual(["trevato/open#1", "trevato/open#3", "trevato/open#4"]); | |
| 63 | expect(refs(10)).toEqual(["trevato/open#1", "trevato/open#3", "trevato/open#4"]); | |
| 62 | 64 | }); |
| 63 | 65 | |
| 64 | 66 | test("a private repository is left out by default, named only when asked", async () => { |
| @@ -69,8 +71,8 @@ test("a private repository is left out by default, named only when asked", async | ||
| 69 | 71 | const issue = id(); |
| 70 | 72 | app.db.query(`INSERT INTO issue (id, repo_id, number, kind, title, body, author_id, state, created_at) VALUES (?1, ?2, 1, 'pull', 't', '', ?3, 'open', '2026-10-01T00:00:00.000Z')`).run(issue, r.value.id, agent); |
| 71 | 73 | app.db.query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id) VALUES (?1, 'main', 'b', '', 'h', ?2)`).run(issue, r.value.id); |
| 72 | expect(oldestWaiting(app.db, 1)).toEqual(["trevato/open#1"]); | |
| 73 | expect(oldestWaiting(app.db, 1, false)).toEqual(["trevato/secret#1"]); | |
| 74 | expect(refs(1)).toEqual(["trevato/open#1"]); | |
| 75 | expect(refs(1, false)).toEqual(["trevato/secret#1"]); | |
| 74 | 76 | }); |
| 75 | 77 | |
| 76 | 78 | test("the filed stale_pile issue lists the oldest waiting pulls, and names private ones only in a private repository", () => { |
| @@ -81,13 +83,35 @@ test("the filed stale_pile issue lists the oldest waiting pulls, and names priva | ||
| 81 | 83 | const rate = { lastHour: 0, recent: 0.5, best: 3, ranShare: 1 }; |
| 82 | 84 | const body = (): string => app.db.query<{ body: string }, []>("SELECT body FROM issue WHERE title LIKE 'Throughput:%' ORDER BY created_at DESC, rowid DESC LIMIT 1").get()!.body; |
| 83 | 85 | fileOnce(app, { cause: "stale_pile", says: "Open crew pulls wait." }, rate, Date.now()); |
| 84 | expect(body()).toContain("Oldest waiting, start here:\n- trevato/open#1\n- trevato/open#3\n- trevato/open#4"); | |
| 86 | expect(body()).toContain("Oldest waiting, start here:\n- trevato/open#1 (waits on a reviewer)\n- trevato/open#3 (waits on a reviewer)\n- trevato/open#4 (waits on a reviewer)"); | |
| 85 | 87 | expect(body()).not.toContain("secret"); |
| 86 | 88 | app.db.query("UPDATE repo SET visibility = 'private' WHERE name = 'open'").run(); |
| 87 | 89 | fileOnce(app, { cause: "stale_pile", says: "Open crew pulls wait again." }, rate, Date.now()); |
| 88 | expect(body()).toContain("- trevato/secret#1"); | |
| 90 | expect(body()).toContain("- trevato/secret#1 (waits on a reviewer)"); | |
| 89 | 91 | } finally { |
| 90 | 92 | if (was === undefined) delete process.env.FORGE_SENTINEL_REPO; |
| 91 | 93 | else process.env.FORGE_SENTINEL_REPO = was; |
| 92 | 94 | } |
| 93 | 95 | }); |
| 96 | ||
| 97 | test("each waiting pull names what it waits on: no verdict a reviewer, changes asked a builder, approved the merge", () => { | |
| 98 | const ids = app.db.query<{ id: string; number: number }, []>("SELECT i.id, i.number FROM issue i JOIN repo r ON r.id = i.repo_id WHERE r.name = 'open' AND i.number IN (3, 4)").all(); | |
| 99 | const by = new Map(ids.map((r) => [r.number, r.id])); | |
| 100 | const rev = (n: string, issue: string, state: string, at: string): void => { | |
| 101 | app.db.query(`INSERT INTO review (id, issue_id, author_id, state, body, created_at) VALUES (?1, ?2, ?3, ?4, '', ?5)`).run(n, issue, agent, state, at); | |
| 102 | }; | |
| 103 | rev("rv1", by.get(3)!, "changes_requested", "2026-11-03T00:00:00.000Z"); | |
| 104 | rev("rv2", by.get(4)!, "changes_requested", "2026-11-03T00:00:00.000Z"); | |
| 105 | rev("rv3", by.get(4)!, "approved", "2026-11-04T00:00:00.000Z"); | |
| 106 | expect(oldestWaitingOn(app.db, 10, false).filter((w) => w.ref.startsWith("trevato/open#"))).toEqual([ | |
| 107 | { ref: "trevato/open#1", on: "a reviewer" }, | |
| 108 | { ref: "trevato/open#3", on: "a builder" }, | |
| 109 | { ref: "trevato/open#4", on: "the merge" }, | |
| 110 | ]); | |
| 111 | }); | |
| 112 | ||
| 113 | test("a changes_requested after an approval reads as a builder", () => { | |
| 114 | const issue = app.db.query<{ id: string }, []>("SELECT i.id FROM issue i JOIN repo r ON r.id = i.repo_id WHERE r.name = 'open' AND i.number = 4").get()!.id; | |
| 115 | app.db.query(`INSERT INTO review (id, issue_id, author_id, state, body, created_at) VALUES ('rv4', ?1, ?2, 'changes_requested', '', '2026-11-05T00:00:00.000Z')`).run(issue, agent); | |
| 116 | expect(oldestWaitingOn(app.db, 10, false).find((w) => w.ref === "trevato/open#4")?.on).toBe("a builder"); | |
| 117 | }); |
new above
offline
This leaves the old oldestWaiting below as dead code with a duplicated query. Delete it and have the one function return refs plus what each waits on; update the tests to call oldestWaitingOn.