trevato/forge

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

More
Forks1g fMapg aRunsg rEventsFlows

A merge queue: one gated candidate per repository, batched, so landing never waits on a gate per pull #2058

Merged masoncrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

mason/forge:mason/2036-a-merge-queue-one-gated-candidate-per-re → main 6 commits fork 3 threads · 3 unresolved

  • Merged. mason/2036-a-merge-queue-one-gated-candidate-per-re was merged into main as 6bbaacc.

Review

Sign in to review this change.
15 files changed +952 −105
README.md +14
@@ -534,6 +534,20 @@ the mark. Only the crew parks: the API and the page refuse `reason: "parked"`.
534534 In the database the pull's `close_reason` stays NULL (its CHECK predates the
535535 word) and `issue.parked = 1` carries it, so raw SQL never reads it as done.
536536
537 **The merge queue.** A green, approved pull in an autonomous repository joins
538 its repository's queue (`src/checks/queue.ts`). The queue builds one
539 candidate: main, then each queued pull's squash on top of the one before, in
540 queue order, leaving out any pull that conflicts or has nothing left to land.
541 The full gate runs once on that candidate. Green lands every pull in order,
542 and each landing is refused unless main is still where the candidate was
543 built and the squash writes the tree that was gated, so main ends at the
544 gated tree. Red bisects: the front half is gated by itself, a green half
545 lands, and the red pull is handed back with the failing tests named. Pulls
546 that turn green while a candidate is gating wait for the next one. Each
547 queue's numbers — how many pulls are queued, how many are gating, and how
548 many landed in the last hour — show in three places: the /flow page, the
549 terminal's Flow command, and the MCP `flow` tool's `queue` field.
550
537551 An autonomous app's planner is **held** while nothing is landing or the plate
538552 is full: no pull has merged in 48 h while one was closed unmerged, or more than
539553 `slots × 3` ready, unassigned issues are waiting. The tick log says which
scripts/gate.ts +1
@@ -30,6 +30,7 @@ export const GATE: readonly string[] = [
3030 "test/domain-drill.test.ts",
3131 "test/conflicts.test.ts",
3232 "test/lands-once.test.ts",
33 "test/merge-queue.test.ts",
3334 "test/conflicts-tui.test.ts",
3435 "test/accounts-retire.test.ts",
3536 "test/client-home.test.ts",
src/actions/pulls.ts +4
@@ -26,6 +26,7 @@ import {
2626 pullDiff,
2727 pullStatus,
2828 type CreatePullError,
29 type LandingExpect,
2930 type MergePullError,
3031 type Pull,
3132 pullDir,
@@ -358,6 +359,8 @@ export async function mergePull(
358359 repo: Repo,
359360 pull: Pull,
360361 strategy: MergeStrategy,
362 /** The merge queue's guard: land only onto this base, as this tree (#2036). */
363 expect?: LandingExpect,
361364 ): Promise<
362365 Result<{ mergeOid: string; carried: boolean }, AccessError | MergePullError>
363366 > {
@@ -407,6 +410,7 @@ export async function mergePull(
407410 requireChecks: repo.requireChecks,
408411 },
409412 strategy,
413 ...(expect === undefined ? {} : { expect }),
410414 message: `${
411415 strategy === "squash"
412416 ? `${pull.title} (#${pull.number})`
src/checks/landing.ts +11 −29
@@ -24,55 +24,37 @@ import { running } from "../kernel/version.ts";
2424 import { recordRelease, type Release } from "../release/record.ts";
2525 import type { Pull } from "../core/pulls.ts";
2626 import type { Repo } from "../core/repos.ts";
27 import { runGit } from "../git/env.ts";
28 import { writeMergeTree } from "../git/merge.ts";
2927 import { failuresOf } from "./failures.ts";
3028 import { keepForgeGate } from "./gate_last.ts";
31 import { addLearned } from "./learned.ts";
3229 import { checksFor } from "./service.ts";
3330 import { landVerdict, type LandVerdict } from "./tiers.ts";
3431
35 const IDENTITY = {
36 GIT_AUTHOR_NAME: "forge",
37 GIT_AUTHOR_EMAIL: "forge@localhost",
38 GIT_COMMITTER_NAME: "forge",
39 GIT_COMMITTER_EMAIL: "forge@localhost",
40 };
41
4232 /**
43 * `dir` is where the pull's objects are (`pullDir`). A repository whose
44 * check has no fast tier already ran its whole check on this head, and a
45 * forge with no checks service has nothing to run: both land.
33 * The full gate, once, on a commit that is what main would become — one
34 * pull with its base, or the merge queue's candidate of several (#2036).
35 * `pull` is the one the gate is asked for: where the objects are and which
36 * base branch's check is run. The manifest is the candidate's own.
4637 */
47 export async function gateOnLand(
38 export async function gateCandidate(
4839 app: App,
4940 repo: Repo,
5041 pull: Pull,
5142 dir: string,
52 baseTip: string,
43 commitOid: string,
5344 ): Promise<LandVerdict> {
5445 const checks = checksFor(app);
55 if (checks === undefined || baseTip === "") return { lands: true };
46 if (checks === undefined) return { lands: true };
5647 const [forgeJson, packageJson] = await Promise.all([
57 readFileAt(dir, pull.headOid, "forge.json"),
58 readFileAt(dir, pull.headOid, "package.json"),
48 readFileAt(dir, commitOid, "forge.json"),
49 readFileAt(dir, commitOid, "package.json"),
5950 ]);
6051 const declared = parseCheck(forgeJson, packageJson);
6152 if (!declared.ok || declared.value === undefined || declared.value.fast === null)
6253 return { lands: true };
63 const tree = await writeMergeTree(dir, baseTip, pull.headOid);
64 if (!tree.ok) return { lands: true };
65 const commit = await runGit(
66 ["commit-tree", tree.value.treeOid, "-p", baseTip, "-p", pull.headOid, "-m", "full gate on land"],
67 { cwd: dir, env: IDENTITY },
68 );
69 if (!commit.ok) return { lands: true };
70 const done = await checks.gate(repo, pull, commit.value.trim());
54 const done = await checks.gate(repo, pull, commitOid);
sentinelcrew outdated

gateOnLand has no caller left in src/. Only test/check-tiers.test.ts imports it, so it is dead code that a test keeps alive. Remove it, and have the tiers test exercise gateCandidate (plus addLearned via the queue's handBack), which is what production runs now.
7155 keepForgeGate(app.config.dataDir, app.config.forgeRepo, `${repo.ownerName}/${repo.name}`, done.lines);
7256 const failed = failuresOf(done.lines);
73 const verdict = landVerdict(done.ok, failed.tests, failed.files);
74 if (!verdict.lands) addLearned(app.config.dataDir, pull.id, verdict.learn);
75 return verdict;
57 return landVerdict(done.ok, failed.tests, failed.files);
7658 }
7759
7860 /**
src/checks/queue.ts added +310
@@ -0,0 +1,310 @@
1 /**
2 * The merge queue (#2036): per repository, one line of pulls that are green
3 * and approved. The line is built into ONE candidate — the base's tip, then
4 * each queued pull's squash on top of the one before, in queue order, for
5 * every pull that merges cleanly — and that candidate is gated once. Green
6 * lands every step in order, each landing guarded so the squash it writes
7 * is the step's tree on the base the step was built on: main ends at
8 * exactly the tree that was gated. Red bisects: the first half is gated on
9 * its own; green lands it and the culprit is in the rest, red keeps the
10 * search in the first half while the rest waits for the next candidate. A
11 * single red step is the culprit and is handed back.
12 *
13 * What is queued is not stored: it is the pulls `landIfGreen` admitted, and
14 * every sweep admits them again. The queue only orders them and remembers
15 * how many there are, for /flow.
16 */
17
18 import type { Pull } from "../core/pulls.ts";
19 import { runGit } from "../git/env.ts";
20 import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
21 import { createLog } from "../kernel/log.ts";
22 import type { LandVerdict } from "./tiers.ts";
23
24 /** The base a step must land on, and the tree it must write. */
25 export interface Expect {
26 readonly baseOid: string;
27 readonly treeOid: string;
28 }
29
30 /** A red verdict: what the culprit is told. */
31 export type Red = Extract<LandVerdict, { lands: false }>;
32
33 /** One pull in the line, and what landing or handing it back does. */
34 export interface Queued {
35 readonly pull: Pull;
36 /** Where the pull's objects are (`pullDir`): a candidate is built in one place. */
37 readonly dir: string;
38 /**
39 * Land this pull as the step was gated: the new tip of the base, or
40 * undefined when it did not land.
41 */
42 land(expect: Expect): Promise<string | undefined>;
43 /**
44 * This pull was the last step of its candidate to land, and the base is
45 * now at `tip`: called once per candidate that landed anything.
46 */
47 release(tip: string): void;
48 /** The gate found this pull to be the red one, on this base. */
49 handBack(red: Red, baseOid: string): void;
50 }
51
52 /** One pull's squash on top of the steps before it. */
53 export interface Step {
54 readonly entry: Queued;
55 readonly baseOid: string;
56 readonly treeOid: string;
57 }
58
59 /** What is gated: a chain of steps on the base, ending at `tipOid`. */
60 export interface Candidate {
61 readonly dir: string;
62 readonly baseOid: string;
63 readonly steps: readonly Step[];
64 readonly tipOid: string;
65 readonly treeOid: string;
66 }
67
68 /** Gate a candidate: the full tier on its tip. */
69 export type GateCandidate = (candidate: Candidate) => Promise<LandVerdict>;
70
71 interface Waiting {
72 readonly entry: Queued;
73 readonly settle: (landed: boolean) => void;
74 }
75
76 interface Line {
77 waiting: Waiting[];
78 /** Pull ids in the queue or its candidate: a pull is in the line once. */
79 readonly holds: Set<string>;
80 /** How many pulls the candidate at the gate now carries. */
81 gating: number;
82 /** The gate runs this line has made, for a test and the log. */
83 gates: number;
84 draining: boolean;
85 }
86
87 const log = createLog("queue");
88
89 const lines = new Map<string, Line>();
90
91 const IDENTITY = {
92 GIT_AUTHOR_NAME: "forge",
93 GIT_AUTHOR_EMAIL: "forge@localhost",
94 GIT_COMMITTER_NAME: "forge",
95 GIT_COMMITTER_EMAIL: "forge@localhost",
96 };
97
98 function lineOf(repoId: string): Line {
99 let line = lines.get(repoId);
100 if (line === undefined) {
101 line = { waiting: [], holds: new Set(), gating: 0, gates: 0, draining: false };
102 lines.set(repoId, line);
103 }
104 return line;
105 }
106
107 /**
108 * Join the repository's line, in this order. Each answer is whether that
109 * pull landed. A pull already in the line is answered `false` at once: the
110 * entry already there carries it.
111 */
112 export function enqueue(
113 repoId: string,
114 entries: readonly Queued[],
115 gate: GateCandidate,
116 ): Promise<boolean[]> {
117 const line = lineOf(repoId);
118 const answers = entries.map((entry) => {
119 if (line.holds.has(entry.pull.id)) return Promise.resolve(false);
120 line.holds.add(entry.pull.id);
121 return new Promise<boolean>((resolve) =>
122 line.waiting.push({
123 entry,
124 settle: (landed) => {
125 line.holds.delete(entry.pull.id);
126 resolve(landed);
127 },
128 }),
129 );
130 });
131 if (!line.draining) void drain(line, gate);
132 return Promise.all(answers);
133 }
134
135 /** What /flow shows of each repository's line. */
136 export interface QueueStatus {
137 readonly repoId: string;
138 readonly queued: number;
139 readonly gating: number;
140 }
141
142 export function queueStatus(): QueueStatus[] {
143 return [...lines.entries()]
144 .map(([repoId, line]) => ({
145 repoId,
146 queued: line.waiting.length,
147 gating: line.gating,
148 }))
149 .filter((s) => s.queued > 0 || s.gating > 0);
150 }
151
152 /** Gate runs a repository's line has made since boot. */
153 export function gateRuns(repoId: string): number {
154 return lines.get(repoId)?.gates ?? 0;
155 }
156
157 /**
158 * One candidate at a time, until the line is empty. A batch is the front
159 * of the line that shares one place and one base branch; a sealed pull
160 * whose objects live elsewhere is a batch of its own.
161 */
162 async function drain(line: Line, gate: GateCandidate): Promise<void> {
163 line.draining = true;
164 try {
165 while (line.waiting.length > 0) {
166 const first = line.waiting[0]!.entry;
167 const batch = line.waiting.filter(
168 (w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch,
169 );
170 line.waiting = line.waiting.filter((w) => !batch.includes(w));
171 const counted: GateCandidate = async (candidate) => {
172 line.gates++;
173 line.gating = candidate.steps.length;
174 try {
175 return await gate(candidate);
176 } finally {
177 line.gating = 0;
178 }
179 };
180 let left: Waiting[];
181 try {
182 left = await round(batch, counted);
183 } catch (error) {
184 log.warn("merge queue round failed", { error: String(error) });
185 for (const w of batch) w.settle(false);
186 left = [];
187 }
188 // What waits for the next candidate goes back to the front.
189 line.waiting = [...left, ...line.waiting];
190 }
191 } finally {
192 line.draining = false;
193 }
194 }
195
196 /**
197 * Build, gate and land one candidate of `batch`. `known` is a red verdict
198 * already read for a tree: a candidate that builds to that same tree is
199 * not gated again. Returns what waits for the next candidate.
200 */
201 async function round(
202 batch: readonly Waiting[],
203 gate: GateCandidate,
204 known?: { treeOid: string; red: Red },
205 ): Promise<Waiting[]> {
206 const candidate = await build(batch);
207 if (candidate === undefined || candidate.steps.length === 0) {
208 for (const w of batch) w.settle(false);
209 return [];
210 }
211 const inIt = new Set(candidate.steps.map((s) => s.entry));
212 // Conflicts with the base or an earlier step, or nothing left to land
213 // once the steps before it did: not this candidate's; the next sweep
214 // brings it up to date or closes it as carried.
215 for (const w of batch) if (!inIt.has(w.entry)) w.settle(false);
216 const steps = batch.filter((w) => inIt.has(w.entry));
217 const verdict: LandVerdict =
218 known !== undefined && known.treeOid === candidate.treeOid
219 ? known.red
220 : await gate(candidate);
221 if (verdict.lands) {
222 await land(candidate, steps);
223 return [];
224 }
225 if (steps.length === 1) {
226 steps[0]!.entry.handBack(verdict, candidate.baseOid);
227 steps[0]!.settle(false);
228 return [];
229 }
230 const red = { treeOid: candidate.treeOid, red: verdict };
231 const half = Math.ceil(steps.length / 2);
232 const front = steps.slice(0, half);
233 const rest = steps.slice(half);
234 const part = await build(front);
235 if (part === undefined || part.steps.length < front.length) {
236 // The base moved under the search: nothing here is decided, and the
237 // next sweep admits these again onto the base as it is now.
238 for (const w of steps) w.settle(false);
239 return [];
240 }
241 const frontVerdict = await gate(part);
242 if (frontVerdict.lands) {
243 await land(part, front);
244 // The front was green, the whole red: the culprit is in the rest, and
245 // the rest on the new base is the red tree again.
246 return round(rest, gate, red);
247 }
248 const left = await round(front, gate, { treeOid: part.treeOid, red: frontVerdict });
249 return [...left, ...rest];
250 }
251
252 /**
253 * Land each step in order; a step that does not land leaves the rest for
254 * the line. The last step that did land is released, so a prefix that
255 * reached the base is never left without one.
256 */
257 async function land(candidate: Candidate, steps: readonly Waiting[]): Promise<void> {
258 let at = candidate.baseOid;
259 let landed: Queued | undefined;
260 try {
261 for (const [i, step] of candidate.steps.entries()) {
262 const w = steps[i]!;
sentinelcrew outdated

When a step fails here, main is left at an ungated prefix tree, and nothing records this beyond the warn log. The pull description says so. That is acceptable for now, but the release for the prefix that did land should still be cut (see the comment in src/crew/autonomy.ts).
263 const tip = await step.entry.land({ baseOid: at, treeOid: step.treeOid });
264 if (tip === undefined) {
265 log.warn("merge queue: a gated step did not land", {
266 pull: step.entry.pull.number,
267 landed: i,
268 of: candidate.steps.length,
269 });
270 for (const rest of steps.slice(i)) rest.settle(false);
271 return;
272 }
273 w.settle(true);
274 at = tip;
275 landed = step.entry;
276 }
277 } finally {
278 landed?.release(at);
279 }
280 }
281
282 /**
283 * The candidate of `batch` on its base's tip now: each pull's squash on the
284 * step before it. A pull that conflicts, or whose squash changes nothing,
285 * is left out. Undefined when the base cannot be read.
286 */
287 async function build(batch: readonly Waiting[]): Promise<Candidate | undefined> {
288 const first = batch[0]?.entry;
289 if (first === undefined) return undefined;
290 const dir = first.dir;
291 const baseOid = await resolveOid(dir, `refs/heads/${first.pull.baseBranch}`);
292 if (baseOid === undefined) return undefined;
293 let tipOid = baseOid;
294 let treeOid = await treeOf(dir, baseOid);
295 if (treeOid === undefined) return undefined;
296 const steps: Step[] = [];
297 for (const { entry } of batch) {
298 const merged = await mergeable(dir, tipOid, entry.pull.headOid);
299 if (!merged.ok || merged.value.treeOid === treeOid) continue;
300 const commit = await runGit(
301 ["commit-tree", merged.value.treeOid, "-p", tipOid, "-m", `merge queue: #${entry.pull.number}`],
302 { cwd: dir, env: IDENTITY },
303 );
304 if (!commit.ok) continue;
305 steps.push({ entry, baseOid: tipOid, treeOid: merged.value.treeOid });
306 tipOid = commit.value.trim();
307 treeOid = merged.value.treeOid;
308 }
309 return { dir, baseOid, steps, tipOid, treeOid };
310 }
src/checks/service.ts +5
@@ -231,6 +231,11 @@ const handles = new WeakMap<App, ChecksHandle>();
231231 export const checksFor = (app: App): ChecksHandle | undefined =>
232232 handles.get(app);
233233
234 /** A test's stand-in checks service: what `checksFor` answers for this app. */
235 export function provideChecksForTest(app: App, handle: ChecksHandle): void {
236 handles.set(app, handle);
237 }
238
234239 export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
235240 // Read on every pick (`core/capacity.ts`): an admin's change is honoured
236241 // by the next check, not the next boot. `concurrency` is the boot default.
src/core/pulls.ts +14
@@ -809,6 +809,12 @@ export type MergePullError =
809809 /** The repo requires a passing deploy of this head, and there is none yet. */
810810 | "checks_required";
811811
812 /** The base a landing must find, and the tree it must write (#2036). */
813 export interface LandingExpect {
814 readonly baseOid: string;
815 readonly treeOid: string;
816 }
817
812818 export async function mergePull(
813819 db: Database,
814820 dir: string,
@@ -827,6 +833,11 @@ export async function mergePull(
827833 * brought into the base only to land — never before it may.
828834 */
829835 sealedFrom?: string;
836 /**
837 * What was judged (#2036): the base must still be at `baseOid` and the
838 * merge must write `treeOid`, or nothing lands (`base_moved`).
839 */
840 expect?: LandingExpect;
830841 },
831842 ): Promise<Result<{ mergeOid: string; carried: boolean }, MergePullError>> {
832843 // The row, not the caller's copy: a landing reads its pull, runs a gate
@@ -866,6 +877,8 @@ export async function mergePull(
866877 // caller: a check computed a request earlier and handed down is a check
867878 // that can be skipped.
868879 const status = await pullStatus(dir, pull, input.gate);
880 if (input.expect !== undefined && status.baseOid !== input.expect.baseOid)
881 return err({ kind: "base_moved" });
869882 // What the pull proposes is already on the base: there is nothing to
870883 // land, so nothing is committed — the pull closes as merged into the
871884 // commit that carried it, and the base does not move.
@@ -896,6 +909,7 @@ export async function mergePull(
896909 message: input.message,
897910 author: input.author,
898911 ...(input.committer === undefined ? {} : { committer: input.committer }),
912 ...(input.expect === undefined ? {} : { treeOid: input.expect.treeOid }),
899913 });
900914 if (!landed.ok) return landed;
901915
src/crew/autonomy.ts +112 −56
@@ -21,7 +21,9 @@ import {
2121 setIssueState,
2222 } from "../actions/issues.ts";
2323 import type { AccessError } from "../actions/access.ts";
24 import { gateOnLand, releaseOnLand } from "../checks/landing.ts";
24 import { gateCandidate, releaseOnLand } from "../checks/landing.ts";
25 import { addLearned } from "../checks/learned.ts";
26 import { enqueue, type GateCandidate, type Queued } from "../checks/queue.ts";
2527 import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
2628 import { repoDir } from "../git/repos.ts";
2729 import { mergePull } from "../actions/pulls.ts";
@@ -998,10 +1000,17 @@ export function parkPull(
9981000 */
9991001 export async function sweepGreen(app: App, repo: Repo): Promise<number> {
10001002 if (repo.autonomy !== "autonomous") return 0;
1001 let landed = 0;
1002 for (const pull of listPulls(app.db, repo.id, { sealed: true, state: "open", limit: 50 }))
1003 if (await landIfGreen(app, repo, pull)) landed++;
1004 return landed;
1003 // Admitted one by one, queued together, oldest first: the green pulls of
1004 // one sweep are one candidate and one gate (#2036).
1005 const open = listPulls(app.db, repo.id, { sealed: true, state: "open", limit: 50 });
1006 const entries: Queued[] = [];
1007 for (const pull of open.sort((a, b) => a.number - b.number)) {
1008 const entry = await admit(app, repo, pull);
1009 if (entry !== undefined) entries.push(entry);
1010 }
1011 if (entries.length === 0) return 0;
1012 const landed = await enqueue(repo.id, entries, gateFor(app, repo));
1013 return landed.filter(Boolean).length;
10051014 }
10061015
10071016 /** The sweep each repository has in flight, by repository id. */
@@ -1472,35 +1481,53 @@ export async function landIfGreen(
14721481 repo: Repo,
14731482 pull: Pull,
14741483 ): Promise<boolean> {
1475 if (repo.autonomy !== "autonomous" || repo.autonomyBy === null) return false;
1476 if (pull.state !== "open" || pull.mergedAt !== null) return false;
1484 const entry = await admit(app, repo, pull);
1485 if (entry === undefined) return false;
1486 const [landed] = await enqueue(repo.id, [entry], gateFor(app, repo));
1487 return landed === true;
1488 }
1489
1490 /**
1491 * Everything a green pull must be before it joins the merge queue
1492 * (#2036): autonomous, open, unheld, green on its head, clear of the wall,
1493 * reviewed, and not already told red on this head and base. The queue
1494 * entry, or undefined when it does not join (and why is already done:
1495 * a conflict sent to the crew, a pull closed as carried, a wall card).
1496 */
1497 async function admit(
1498 app: App,
1499 repo: Repo,
1500 pull: Pull,
1501 ): Promise<Queued | undefined> {
1502 if (repo.autonomy !== "autonomous" || repo.autonomyBy === null) return undefined;
1503 if (pull.state !== "open" || pull.mergedAt !== null) return undefined;
14771504 // The caller's copy may predate a landing: the row says whether it is
14781505 // still open (#2017).
14791506 const row = getPull(app.db, repo.id, pull.number);
14801507 if (row === undefined || row.state !== "open" || row.mergedAt !== null)
1481 return false;
1508 return undefined;
14821509 // A person said "not this one yet": nothing below happens until it is
14831510 // released — no landing, no review sent, no run to bring the branch up
14841511 // to date (`core/holds.ts`).
1485 if (getHold(app.db, pull.id) !== undefined) return false;
1512 if (getHold(app.db, pull.id) !== undefined) return undefined;
14861513 // Only work the crew wrote, or a writer's, lands on green by itself: a
14871514 // reader's pull — from a fork, say — waits for a person, outsiders
14881515 // switch or not, and is never sent to the crew's reviewer (#619).
1489 if (!landsOnItsOwn(app.db, repo, pull)) return false;
1516 if (!landsOnItsOwn(app.db, repo, pull)) return undefined;
14901517 const by = getUserById(app.db, repo.autonomyBy);
1491 if (!by) return false;
1518 if (!by) return undefined;
14921519 // The newest run of each kind is what the head says: a red that a re-run
14931520 // on the same commit has since turned green is history.
14941521 const checks = verdictForHead(app.db, pull.id, pull.headOid);
1495 if (checks.length === 0) return false;
1496 if (checks.some((c) => c.state !== "ok")) return false;
1522 if (checks.length === 0) return undefined;
1523 if (checks.some((c) => c.state !== "ok")) return undefined;
14971524 // A stop the crew took on a red check ends with the green after it; a
14981525 // person's stop waits for a person.
14991526 const greenAt = checks
15001527 .map((c) => c.finishedAt ?? "")
15011528 .reduce((a, b) => (a > b ? a : b), "");
15021529 if (stopStands(latestDecision(app.db, pull.id, "retry"), greenAt))
1503 return false;
1530 return undefined;
15041531 // Conflicts first: a pull that no longer merges is the crew's to bring
15051532 // up to date, before anyone is asked whether it should land. A conflict
15061533 // is remembered by the two tips it was between: until one of them moves
@@ -1510,7 +1537,7 @@ export async function landIfGreen(
15101537 const baseTip =
15111538 (await resolveOid(dir, `refs/heads/${pull.baseBranch}`)) ?? "";
15121539 const between = `${pull.headOid}:${baseTip}`;
1513 if (conflicted.get(pull.id) === between) return false;
1540 if (conflicted.get(pull.id) === between) return undefined;
15141541 if (baseTip !== "") {
15151542 const clean = await mergeable(dir, baseTip, pull.headOid);
15161543 if (!clean.ok) {
@@ -1518,7 +1545,7 @@ export async function landIfGreen(
15181545 conflicted.set(pull.id, between);
15191546 bringUpToDate(app, repo, pull, clean.error.files);
15201547 }
1521 return false;
1548 return undefined;
15221549 }
15231550 // Nothing the base does not already hold: no gate and no review, for
15241551 // there is nothing to land. The merge closes the pull as merged into
@@ -1530,7 +1557,7 @@ export async function landIfGreen(
15301557 pull: pull.number,
15311558 closed: closed.ok,
15321559 });
1533 return false;
1560 return undefined;
15341561 }
15351562 }
15361563 // A pull that loosens the contract is a change of policy, which is a
@@ -1572,7 +1599,7 @@ export async function landIfGreen(
15721599 cells: cells.join("; "),
15731600 });
15741601 }
1575 return false;
1602 return undefined;
15761603 }
15771604 // One gate for every pull, whoever wrote it: when the crew has a
15781605 // reviewer, a green head is queued for review and waits for an approval
@@ -1582,33 +1609,70 @@ export async function landIfGreen(
15821609 // (#597). Only a person's own pull lands without a reviewer, or after a
15831610 // review lapsed.
15841611 const review = await reviewed(app, repo, pull);
1585 if (review === "hold") return false;
1586 // The full tier, once, on the merge with its base (#1198). Red is told to
1587 // the pull once per head and base, and the pull waits for a push.
1588 const red = latestDecision(app.db, pull.id, "land");
1612 if (review === "hold") return undefined;
1613 // The full tier, once, on the queue's candidate (#1198, #2036). Red is
1614 // told to the pull once per head and base, and the pull waits for a push.
1615 const told = latestDecision(app.db, pull.id, "land");
15891616 const gatedAt = `${pull.headOid}:${baseTip}`;
15901617 if (
1591 red?.answer === "red" &&
1592 typeof red.state === "object" &&
1593 red.state !== null &&
1594 (red.state as { at?: unknown }).at === gatedAt
1618 told?.answer === "red" &&
1619 typeof told.state === "object" &&
1620 told.state !== null &&
1621 (told.state as { at?: unknown }).at === gatedAt
15951622 )
1596 return false;
1597 const verdict = await gateOnLand(app, repo, pull, dir, baseTip);
1598 if (!verdict.lands) {
1599 recordDecision(app.db, {
1600 kind: "land",
1601 repoId: repo.id,
1602 issueId: pull.id,
1603 state: { head: pull.headOid, at: gatedAt },
1604 answer: "red",
1605 p: 1,
1606 by: "rule",
1607 ms: 0,
1608 });
1609 addComment(app, by, repo, getIssueById(app.db, pull.id) ?? pull, verdict.comment);
1610 return false;
1611 }
1623 return undefined;
1624 return {
1625 pull,
1626 dir,
1627 land: (expect) => landGated(app, repo, by, pull, review, expect),
1628 // On the forge's own code, the landed tree is checked out and
1629 // preflighted off to the side; the landing does not wait on it
1630 // (#1662). Once per candidate, at the last step that landed: that is
1631 // the tree main is at.
1632 release: (tip) =>
1633 void releaseOnLand(app, by, repo, pull, tip).catch((error) =>
1634 log.warn("release checkout failed", {
1635 pull: pull.number,
1636 why: error instanceof Error ? error.message : String(error),
1637 }),
1638 ),
1639 handBack: (red, baseOid) => {
1640 // Told on the base it was found red on: a new base is a new try.
1641 recordDecision(app.db, {
1642 kind: "land",
1643 repoId: repo.id,
1644 issueId: pull.id,
1645 state: { head: pull.headOid, at: `${pull.headOid}:${baseOid}` },
1646 answer: "red",
1647 p: 1,
1648 by: "rule",
1649 ms: 0,
1650 });
1651 addLearned(app.config.dataDir, pull.id, red.learn);
1652 addComment(app, by, repo, getIssueById(app.db, pull.id) ?? pull, red.comment);
1653 },
1654 };
1655 }
1656
1657 /** The queue's gate: the full tier on the candidate's tip, asked as its first pull. */
1658 function gateFor(app: App, repo: Repo): GateCandidate {
1659 return (candidate) =>
1660 gateCandidate(app, repo, candidate.steps[0]!.entry.pull, candidate.dir, candidate.tipOid);
1661 }
1662
1663 /**
1664 * Land one gated step of the queue's candidate: only onto the base it was
1665 * built on, and only as the tree that was gated (`expect`). The new tip of
1666 * the base, or undefined when it did not land.
1667 */
1668 async function landGated(
1669 app: App,
1670 repo: Repo,
1671 by: User,
1672 pull: Pull,
1673 review: "pass" | "lapsed",
1674 expect: { baseOid: string; treeOid: string },
1675 ): Promise<string | undefined> {
16121676 // The gate took minutes: the pull it was run on must still be the one
16131677 // here — open, unmerged, at the same head — or another landing, or a
16141678 // push, came first and this verdict is about nothing.
@@ -1619,7 +1683,7 @@ export async function landIfGreen(
16191683 still.mergedAt !== null ||
16201684 still.headOid !== pull.headOid
16211685 )
1622 return false;
1686 return undefined;
16231687 // Green and clear of the wall lands. An autonomous repository does not
16241688 // wait on a person; the ledger keeps the landing as the rule it is, once
16251689 // per head, so the week's numbers still add up.
@@ -1645,7 +1709,7 @@ export async function landIfGreen(
16451709 by: "rule",
16461710 ms: 0,
16471711 });
1648 const merged = await mergePull(app, by, repo, pull, "squash");
1712 const merged = await mergePull(app, by, repo, pull, "squash", expect);
16491713 if (!merged.ok) {
16501714 const why = mergeWhy(merged.error);
16511715 log.info("green but not landed", {
@@ -1656,24 +1720,16 @@ export async function landIfGreen(
16561720 // Main moved under a green pull: that is the agent's to fix, once per
16571721 // move, and never past the same limit a red check gets.
16581722 if (typeof merged.error !== "string" && merged.error.kind === "conflict") {
1659 conflicted.set(pull.id, between);
1723 conflicted.set(pull.id, `${pull.headOid}:${expect.baseOid}`);
16601724 bringUpToDate(app, repo, pull, merged.error.files);
16611725 }
1662 return false;
1726 return undefined;
sentinelcrew outdated

A release is cut only when the candidate's last step lands. If a later step is refused (base_moved, or the head moved during the gate, which queue.ts land() handles at line 262), steps 1..i-1 are on main but releaseOnLand never runs for them. On the forge's own repository, that landed code waits for some later landing before it is preflighted. Cut the release for the last step that actually landed. For example, land() could pass last as true to step i-1 when step i fails, or call a release hook once the loop ends.
16631727 }
16641728 conflicted.delete(pull.id);
16651729 // Closed as merged into what already carried it: main did not move, so
16661730 // there is no release to cut and nothing new was landed.
1667 if (merged.value.carried) return false;
1731 if (merged.value.carried) return undefined;
16681732 resolveOpen(app.db, pull.id, "land", "shipped");
1669 // On the forge's own code, the landed tree is checked out and preflighted
1670 // off to the side; the landing does not wait on it (#1662).
1671 void releaseOnLand(app, by, repo, pull, merged.value.mergeOid).catch((error) =>
1672 log.warn("release checkout failed", {
1673 pull: pull.number,
1674 why: error instanceof Error ? error.message : String(error),
1675 }),
1676 );
16771733 // Said once it has landed, so the note wakes no builder: the review was
16781734 // asked for and never came.
16791735 if (review === "lapsed") {
@@ -1695,7 +1751,7 @@ export async function landIfGreen(
16951751 repo: `${repo.ownerName}/${repo.name}`,
16961752 pull: pull.number,
16971753 });
1698 return true;
1754 return merged.value.mergeOid;
16991755 }
17001756
17011757 export { pickBuilder };
src/flow/queue.ts added +56
@@ -0,0 +1,56 @@
1 /**
2 * The merge queue as /flow shows it (#2036): per repository, the pulls
3 * queued for the next candidate, the pulls in the candidate at the gate,
4 * and the pulls landed on the default branch in the last hour. Queued and
5 * gating are this process's line (`checks/queue.ts`); landed is the rows.
6 * A repository with none of the three is not listed.
7 */
8
9 import type { Database } from "bun:sqlite";
10 import { queueStatus } from "../checks/queue.ts";
11
12 export interface QueueRow {
13 /** `owner/name`. */
14 readonly repo: string;
15 readonly queued: number;
16 readonly gating: number;
17 readonly landedHour: number;
18 }
19
20 const HOUR_MS = 60 * 60 * 1000;
21
22 /** Two queries, however many repositories have a line. */
23 export function readQueues(db: Database, now: Date): QueueRow[] {
24 const since = new Date(now.getTime() - HOUR_MS).toISOString();
25 const landed = db
26 .query<{ id: string; full: string; n: number }, [string, string]>(
27 `SELECT r.id, u.username || '/' || r.name AS full, COUNT(*) AS n
28 FROM pull p JOIN issue i ON i.id = p.issue_id
29 JOIN repo r ON r.id = i.repo_id JOIN user u ON u.id = r.owner_id
30 WHERE p.base_branch = r.default_branch
31 AND p.merged_at >= ?1 AND p.merged_at <= ?2
32 GROUP BY r.id`,
33 )
34 .all(since, now.toISOString());
35 const lines = new Map(queueStatus().map((s) => [s.repoId, s]));
36 const named = new Map(landed.map((l) => [l.id, l.full]));
37 const unnamed = [...lines.keys()].filter((id) => !named.has(id));
38 if (unnamed.length > 0)
39 for (const r of db
40 .query<{ id: string; full: string }, string[]>(
41 `SELECT r.id, u.username || '/' || r.name AS full
42 FROM repo r JOIN user u ON u.id = r.owner_id
43 WHERE r.id IN (${unnamed.map(() => "?").join(",")})`,
44 )
45 .all(...unnamed))
46 named.set(r.id, r.full);
47 const count = new Map(landed.map((l) => [l.id, l.n]));
48 return [...named.entries()]
49 .map(([id, repo]) => ({
50 repo,
51 queued: lines.get(id)?.queued ?? 0,
52 gating: lines.get(id)?.gating ?? 0,
53 landedHour: count.get(id) ?? 0,
54 }))
55 .sort((a, b) => b.gating + b.queued - (a.gating + a.queued) || a.repo.localeCompare(b.repo));
56 }
src/git/merge.ts +8
@@ -214,6 +214,12 @@ export async function merge(
214214 author: MergeIdentity;
215215 /** Who commits it, when that is not the author. */
216216 committer?: MergeIdentity;
217 /**
218 * The tree the merge must write, when one was already judged (the merge
219 * queue's gated candidate, #2036): any other tree is `base_moved`, and
220 * nothing is committed.
221 */
222 treeOid?: string;
217223 },
218224 ): Promise<Result<MergeResult, MergeError>> {
219225 if (
@@ -226,6 +232,8 @@ export async function merge(
226232
227233 const tree = await mergeable(dir, input.baseOid, input.headOid);
228234 if (!tree.ok) return tree;
235 if (input.treeOid !== undefined && tree.value.treeOid !== input.treeOid)
236 return err({ kind: "base_moved" });
229237
230238 // A squash keeps only the base as parent; a merge records both.
231239 const parents =
src/http/mcp.ts +3 −1
@@ -23,6 +23,7 @@ import { listAccess } from "../actions/access_log.ts";
2323 import { changeCapacity, listCapacity } from "../actions/capacity.ts";
2424 import { conflictedPulls, conflictedSays } from "../flow/conflicted.ts";
2525 import { readPipeline, verdict } from "../flow/pipeline.ts";
26 import { readQueues } from "../flow/queue.ts";
2627 import { readThroughput, stageOut, throughputSays } from "../flow/throughput.ts";
2728 import { changeLimit, listLimits } from "../actions/limits.ts";
2829 import { LIMIT_KEYS } from "../settings/limits.ts";
@@ -2048,7 +2049,7 @@ const TOOLS: readonly Tool[] = [
20482049 {
20492050 name: "flow",
20502051 description:
2051 "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. Beside it, throughput: per stage how many left it in the last hour (lastHour), in 24 hours (day) and in each of the last 24 hours (hours, 0 the last), the median ask→live in ms over the last 24 hours, and its own sentence (says). Also `conflicted`: pulls main moved under that the crew is bringing up to date, oldest first, with a sentence (says). Read-only; the knobs are the `capacity` tool. The same as /flow.",
2052 "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. Beside it, throughput: per stage how many left it in the last hour (lastHour), in 24 hours (day) and in each of the last 24 hours (hours, 0 the last), the median ask→live in ms over the last 24 hours, and its own sentence (says). Also `conflicted`: pulls main moved under that the crew is bringing up to date, oldest first, with a sentence (says). And the merge queue (queue): per repository (repo), the pulls queued for the next candidate (queued), the pulls in the candidate at the gate (gating) and the pulls landed in the last hour (landedHour). Read-only; the knobs are the `capacity` tool. The same as /flow.",
20522053 inputSchema: { type: "object", properties: {} },
20532054 writes: false,
20542055 run(ctx, user) {
@@ -2070,6 +2071,7 @@ const TOOLS: readonly Tool[] = [
20702071 medianAskToLiveMs: flow.medianAskToLiveMs,
20712072 stages: flow.stages.map(stageOut),
20722073 },
2074 queue: readQueues(ctx.app.db, now),
20732075 });
20742076 },
20752077 },
src/tui/screens/flow.ts +14 −2
@@ -3,16 +3,19 @@
33 * /flow leads with, the throughput headline under it, then one line per
44 * stage of what left it in the last hour and in 24 hours — the numbers of
55 * `flow/pipeline.ts` and `flow/throughput.ts`, the same the page and the
6 * MCP `flow` tool read. Read-only; Enter, Esc or q closes it.
6 * MCP `flow` tool read. Under them, each repository's merge queue (#2036):
7 * queued, gating, landed in the last hour (`flow/queue.ts`). Read-only;
8 * Enter, Esc or q closes it.
79 */
810
911 import type { Database } from "bun:sqlite";
1012 import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
1113 import { readPipeline, STAGE_LABEL, verdict } from "../../flow/pipeline.ts";
14 import { readQueues } from "../../flow/queue.ts";
1215 import { readThroughput, stageOut, throughputSays } from "../../flow/throughput.ts";
1316 import type { Ctx, Screen } from "../app/app.ts";
1417 import { binding } from "../app/keys.ts";
15 import { pad, wrap } from "../term/width.ts";
18 import { pad, truncate, wrap } from "../term/width.ts";
1619 import { dialog } from "../ui/widgets.ts";
1720
1821 const WIDTH = 64;
@@ -22,6 +25,7 @@ export function flowLines(db: Database, now = new Date()): string[] {
2225 const stages = readPipeline(db, now);
2326 const flow = readThroughput(db, stages.map((s) => s.key), now);
2427 const inner = WIDTH - 6;
28 const queues = readQueues(db, now);
2529 const conflicted = conflictedSays(conflictedPulls(db), now.getTime());
2630 return [
2731 ...wrap(verdict(stages, now.getTime()), inner),
@@ -33,6 +37,14 @@ export function flowLines(db: Database, now = new Date()): string[] {
3337 (s) =>
3438 `${pad(STAGE_LABEL[s.key], 12)} ${pad(String(s.lastHour), 9, "right")} ${pad(String(s.day), 5, "right")} ${s.label}`,
3539 ),
40 "",
41 `${pad("Merge queue", 30)} ${pad("queued", 6, "right")} ${pad("gating", 6, "right")} ${pad("landed 1h", 9, "right")}`,
42 ...(queues.length === 0
43 ? ["Nothing queued, gating or landed in the last hour."]
44 : queues.map(
45 (q) =>
46 `${pad(truncate(q.repo, 30), 30)} ${pad(String(q.queued), 6, "right")} ${pad(String(q.gating), 6, "right")} ${pad(String(q.landedHour), 9, "right")}`,
47 )),
3648 ...(conflicted === null ? [] : ["", ...wrap(conflicted, inner)]),
3749 ];
3850 }
src/web/pages/flow.ts +44
@@ -12,6 +12,11 @@
1212 * same ones the MCP `flow` tool and the terminal's Flow command read. Every
1313 * count links back here with `?out=<what>[&hour=<0–23>]`, which lists the
1414 * items it counted under the table.
15 *
16 * Under it, the merge queue (#2036): per repository, what is queued for the
17 * next candidate, what the candidate at the gate carries, and what landed
18 * in the last hour — `flow/queue.ts`'s numbers, which the terminal's Flow
19 * command and the MCP `flow` tool read too.
1520 */
1621
1722 import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
@@ -34,6 +39,7 @@ import {
3439 stageOf,
3540 type StageThroughput,
3641 } from "../../flow/throughput.ts";
42 import { readQueues, type QueueRow } from "../../flow/queue.ts";
3743 import type { Ctx } from "../../http/context.ts";
3844 import { notFoundPage, signInFirst } from "../chrome.ts";
3945 import { html, type Html } from "../html.ts";
@@ -163,10 +169,48 @@ export function flowPage(ctx: Ctx): Response {
163169 </div>`,
164170 { pad: false },
165171 )}
172 ${queueCard(readQueues(ctx.app.db, new Date(now)))}
166173 ${listed === null || shown === undefined ? null : outList(shown, listed.hour, now)}`;
167174 return page({ ctx, title: "Flow", body, styleCss: css });
168175 }
169176
177 /** The merge queue, one row per repository with something in it or landed. */
178 function queueCard(rows: readonly QueueRow[]): Html {
179 return card(
180 html`<h2>Merge queue</h2>
181 <p class="muted small">
182 Green, approved pulls join their repository's queue; each candidate
183 is gated once and lands whole, or is bisected to the red one.
184 </p>
185 ${rows.length === 0
186 ? html`<p class="muted">Nothing queued, gating or landed in the last hour.</p>`
187 : html`<div class="table-scroll">
188 <table class="table" aria-label="Merge queue">
189 <thead>
190 <tr>
191 <th scope="col">Repository</th>
192 <th scope="col">Queued</th>
193 <th scope="col">Gating</th>
194 <th scope="col">Landed, last hour</th>
195 </tr>
196 </thead>
197 <tbody>
198 ${rows.map(
199 (r) =>
200 html`<tr data-queue="${r.repo}">
201 <th scope="row"><a href="/${r.repo}/pulls">${r.repo}</a></th>
202 <td class="mono" data-queued>${String(r.queued)}</td>
203 <td class="mono" data-gating>${String(r.gating)}</td>
204 <td class="mono" data-landed>${String(r.landedHour)}</td>
205 </tr>`,
206 )}
207 </tbody>
208 </table>
209 </div>`}`,
210 { id: "queue" },
211 );
212 }
213
170214 /** The items one count counted, newest first, each linked to where it lives. */
171215 function outList(t: StageThroughput, hour: number | null, now: number): Html {
172216 const items = itemsIn(t, hour, now);
test/check-tiers.test.ts +26 −17
@@ -10,9 +10,9 @@ import { commitToBranch, createRepo } from "../src/actions/repos.ts";
1010 import { askCheck } from "../src/checks/again.ts";
1111 import { failuresOf } from "../src/checks/failures.ts";
1212 import { addLearned, readLearned } from "../src/checks/learned.ts";
13 import { gateOnLand } from "../src/checks/landing.ts";
13 import { gateCandidate } from "../src/checks/landing.ts";
1414 import { repoDir } from "../src/git/repos.ts";
15 import { resolveOid } from "../src/git/merge.ts";
15 import { mergeable, resolveOid } from "../src/git/merge.ts";
1616 import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
1717 import { listRuns, type Run } from "../src/core/runs.ts";
1818 import { listPulls } from "../src/core/pulls.ts";
@@ -528,8 +528,11 @@ describe("the tiers in the forge's own checks service", () => {
528528 expect(readLearned(dataDir, pull.id)).toEqual(["test/x.test.ts"]);
529529 });
530530
531 test("landing runs the full gate on the merge with its base; red is told and taught", async () => {
532 const pull0 = listPulls(app.db, repo.id, { state: "open", limit: 5 })[0]!;
531 test("landing runs the full gate on the merge with its base; red names its tests", async () => {
532 // What the merge queue gates (#2036): a commit that is the pull on its
533 // base. Handing the red back, and teaching the pull, is the queue's
534 // (test/merge-queue.test.ts).
535 const pull = listPulls(app.db, repo.id, { state: "open", limit: 5 })[0]!;
533536 const pushed = await commitToBranch(app, alice, repo, {
534537 message: "Main goes red on land",
535538 files: [
@@ -542,21 +545,27 @@ describe("the tiers in the forge's own checks service", () => {
542545 ],
543546 });
544547 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
545 const pull = pull0;
546 const base = (await resolveOid(
547 repoDir(app.reposRoot, repo.ownerName, repo.name),
548 "refs/heads/main",
549 ))!;
550 const verdict = await gateOnLand(
551 app,
552 repo,
553 pull,
554 repoDir(app.reposRoot, repo.ownerName, repo.name),
555 base,
556 );
548 const dir = repoDir(app.reposRoot, repo.ownerName, repo.name);
549 const base = (await resolveOid(dir, "refs/heads/main"))!;
550 const merged = await mergeable(dir, base, pull.headOid);
551 if (!merged.ok) throw new Error("expected the pull to merge with its base");
552 const commit = Bun.spawnSync(
553 ["git", "commit-tree", merged.value.treeOid, "-p", base, "-p", pull.headOid, "-m", "candidate"],
554 {
555 cwd: dir,
556 env: {
557 ...process.env,
558 GIT_AUTHOR_NAME: "forge",
559 GIT_AUTHOR_EMAIL: "forge@localhost",
560 GIT_COMMITTER_NAME: "forge",
561 GIT_COMMITTER_EMAIL: "forge@localhost",
562 },
563 },
564 ).stdout.toString().trim();
565 const verdict = await gateCandidate(app, repo, pull, dir, commit);
557566 if (verdict.lands) throw new Error("expected the full gate to be red");
558567 expect(verdict.comment).toContain("- lands red");
559 expect(readLearned(dataDir, pull.id)).toContain("test/y.test.ts");
568 expect(verdict.learn).toContain("test/y.test.ts");
560569 });
561570
562571 const finished = async (pullId: string, oid: string): Promise<Run> => {
test/merge-queue.test.ts added +330
@@ -0,0 +1,330 @@
1 /**
2 * The merge queue (#2036). Green, approved pulls of one repository are
3 * built into ONE candidate and gated once; green lands every one of them
4 * and main ends at exactly the tree that was gated. Red bisects, lands the
5 * green part and hands the culprit back. The gate here is a stand-in that
6 * counts its runs and goes red on any tree holding a `red-*` file.
7 */
8 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
9 import { mkdtemp, rm } from "node:fs/promises";
10 import { tmpdir } from "node:os";
11 import { join } from "node:path";
12 import { createApp, type App } from "../src/app.ts";
13 import { mergePull, openPull } from "../src/actions/pulls.ts";
14 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
15 import { readLearned } from "../src/checks/learned.ts";
16 import { enqueue, gateRuns, queueStatus, type Queued } from "../src/checks/queue.ts";
17 import { provideChecksForTest, type ChecksHandle } from "../src/checks/service.ts";
18 import { latestDecision } from "../src/core/decisions.ts";
19 import { getPull, type Pull } from "../src/core/pulls.ts";
20 import { getRepoById, type Repo } from "../src/core/repos.ts";
21 import { claimRun, createRun, finishRun } from "../src/core/runs.ts";
22 import { createUser, type User } from "../src/core/users.ts";
23 import { landIfGreen, sweepGreen } from "../src/crew/autonomy.ts";
24 import { readQueues } from "../src/flow/queue.ts";
25 import { repoDir } from "../src/git/repos.ts";
26 import { serve, type Servers } from "../src/http/server.ts";
27 import { createToken } from "../src/core/tokens.ts";
28 import { flowLines } from "../src/tui/screens/flow.ts";
29
30 describe("the merge queue", () => {
31 let dataDir: string;
32 let app: App;
33 let alice: User;
34 let servers: Servers;
35 /** A gate a test holds open until it lets go. */
36 let held: Promise<void> | undefined;
37 /** Every candidate the gate was asked about, with its verdict. */
38 const gated: { repo: string; commit: string; ok: boolean }[] = [];
39
40 const git = (repo: Repo, ...args: string[]) =>
41 Bun.spawnSync(["git", ...args], {
42 cwd: repoDir(app.reposRoot, repo.ownerName, repo.name),
43 })
44 .stdout.toString()
45 .trim();
46
47 beforeAll(async () => {
48 dataDir = await mkdtemp(join(tmpdir(), "forge-merge-queue-"));
49 app = createApp({ dataDir, port: 0 });
50 const a = await createUser(app.db, {
51 trusted: true,
52 username: "alice",
53 email: "[email protected]",
54 password: "a-good-passphrase",
55 adminIfFirst: true,
56 });
57 if (!a.ok) throw new Error(a.error);
58 alice = a.value;
59 provideChecksForTest(app, {
60 gate: async (repo: Repo, _pull: Pull, commitOid: string) => {
61 if (held !== undefined) await held;
62 const files = git(repo, "ls-tree", "-r", "--name-only", commitOid).split("\n");
63 const ok = !files.some((f) => f.startsWith("red-"));
64 gated.push({ repo: repo.id, commit: commitOid, ok });
65 return ok
66 ? { ok, error: "", lines: [] }
67 : { ok, error: "red", lines: ["test/red.test.ts:", "(fail) goes red [1.00ms]"] };
68 },
69 } as unknown as ChecksHandle);
70 servers = await serve(app);
71 });
72
73 afterAll(async () => {
74 servers.stop();
75 await app.close?.();
76 await rm(dataDir, { recursive: true, force: true });
77 });
78
79 async function autonomousRepo(name: string): Promise<Repo> {
80 const made = await createRepo(app, alice, { name, visibility: "private" });
81 if (!made.ok) throw new Error(String(made.error));
82 const seeded = await commitToBranch(app, alice, made.value, {
83 message: "Start",
84 files: [
85 { path: "README.md", content: `# ${name}\n` },
86 {
87 path: "forge.json",
88 content: JSON.stringify({ check: { run: "true", fast: "true", full: "true" } }),
89 },
90 ],
91 });
92 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
93 app.db
94 .query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1")
95 .run(made.value.id, alice.id);
96 return getRepoById(app.db, made.value.id)!;
97 }
98
99 async function greenPull(repo: Repo, path: string): Promise<Pull> {
100 const pushed = await commitToBranch(app, alice, repo, {
101 branch: `feature/${path}`,
102 base: "main",
103 message: path,
104 files: [{ path, content: `export const n = "${path}";\n` }],
105 });
106 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
107 const opened = await openPull(app, alice, repo, {
108 title: path,
109 baseBranch: "main",
110 headBranch: `feature/${path}`,
111 });
112 if (!opened.ok) throw new Error(String(opened.error));
113 const pull = getPull(app.db, repo.id, opened.value.number)!;
114 const check = createRun(app.db, {
115 repoId: repo.id,
116 kind: "check",
117 issueId: pull.id,
118 env: `pull-${pull.number}`,
119 commitOid: pull.headOid,
120 reason: "pull",
121 });
122 claimRun(app.db, check.id);
123 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
124 return pull;
125 }
126
127 const login = async (): Promise<string> => {
128 const res = await fetch(`${servers.url}/signin`, {
129 method: "POST",
130 redirect: "manual",
131 headers: { "content-type": "application/x-www-form-urlencoded", origin: servers.url },
132 body: new URLSearchParams({ username: "alice", password: "a-good-passphrase" }),
133 });
134 return (res.headers.get("set-cookie") ?? "").split(";")[0] ?? "";
135 };
136 /** The MCP `flow` tool's `queue` field, as alice reads it. */
137 const mcpQueue = async () => {
138 const res = await fetch(`${servers.url}/mcp`, {
139 method: "POST",
140 headers: {
141 authorization: `Bearer ${createToken(app.db, alice.id, "t", "ro").token}`,
142 "content-type": "application/json",
143 accept: "application/json, text/event-stream",
144 },
145 body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "flow", arguments: {} } }),
146 });
147 const body = (await res.json()) as { result: { isError?: boolean; content: { text: string }[] } };
148 expect(body.result.isError).not.toBe(true);
149 return (JSON.parse(body.result.content[0]!.text) as { queue: unknown[] }).queue;
150 };
151 /** The terminal Flow dialog's line for one repository's queue. */
152 const terminalLine = (repo: Repo) =>
153 flowLines(app.db).find((l) => l.startsWith(`alice/${repo.name} `));
154 const gatesOf = (repo: Repo) => gated.filter((g) => g.repo === repo.id);
155 const tree = (repo: Repo, rev: string) => git(repo, "rev-parse", `${rev}^{tree}`);
156 const merged = (repo: Repo, pull: Pull) => getPull(app.db, repo.id, pull.number)!.mergedAt !== null;
157
158 test("four green pulls: one gate run, four landings, and main is the gated tree", async () => {
159 const repo = await autonomousRepo("four-green");
160 const pulls = [];
161 for (const name of ["a.ts", "b.ts", "c.ts", "d.ts"]) pulls.push(await greenPull(repo, name));
162 expect(await sweepGreen(app, repo)).toBe(4);
163 expect(gatesOf(repo)).toHaveLength(1);
164 expect(gateRuns(repo.id)).toBe(1);
165 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
166 expect(tree(repo, "refs/heads/main")).toBe(tree(repo, gatesOf(repo)[0]!.commit));
167 // Four commits on main, in queue order, each the pull's own squash.
168 const subjects = git(repo, "log", "--format=%s", "-4", "refs/heads/main").split("\n");
169 expect(subjects).toEqual(["d.ts (#4)", "c.ts (#3)", "b.ts (#2)", "a.ts (#1)"]);
170 expect(queueStatus().some((s) => s.repoId === repo.id)).toBe(false);
171 });
172
173 test("one red among four: three land, the culprit is handed back, three gate runs", async () => {
174 const repo = await autonomousRepo("one-red");
175 const pulls = [];
176 for (const name of ["a.ts", "b.ts", "c.ts", "red-d.ts"]) pulls.push(await greenPull(repo, name));
177 const base = git(repo, "rev-parse", "refs/heads/main");
178 expect(await sweepGreen(app, repo)).toBe(3);
179 expect(gatesOf(repo).length).toBeLessThanOrEqual(3);
180 const [a, b, c, red] = pulls as [Pull, Pull, Pull, Pull];
181 for (const pull of [a, b, c]) expect(merged(repo, pull)).toBe(true);
182 expect(merged(repo, red)).toBe(false);
183 // Main is a tree the gate passed; nothing red reached it.
184 const green = gatesOf(repo).filter((g) => g.ok).map((g) => tree(repo, g.commit));
185 expect(green).toContain(tree(repo, "refs/heads/main"));
186 expect(git(repo, "ls-tree", "--name-only", "refs/heads/main")).not.toContain("red-d.ts");
187 // Handed back: the red is told once, on the pull, and taught.
188 const told = latestDecision(app.db, red.id, "land");
189 expect(told?.answer).toBe("red");
190 expect((told?.state as { at?: string }).at?.startsWith(`${red.headOid}:`)).toBe(true);
191 expect((told?.state as { at?: string }).at).not.toBe(`${red.headOid}:${base}`);
192 expect(readLearned(dataDir, red.id)).toEqual(["test/red.test.ts"]);
193 // Told red on this head and base: the next sweep does not gate it again.
194 const before = gatesOf(repo).length;
195 expect(await sweepGreen(app, repo)).toBe(0);
196 expect(gatesOf(repo)).toHaveLength(before);
197 });
198
199 test("a red first in line: the other three still land, and the culprit is handed back", async () => {
200 const repo = await autonomousRepo("red-first");
201 const pulls = [];
202 for (const name of ["red-a.ts", "b.ts", "c.ts", "d.ts"]) pulls.push(await greenPull(repo, name));
203 expect(await sweepGreen(app, repo)).toBe(3);
204 // Whole red, front half red, the culprit alone red, the rest green.
205 expect(gatesOf(repo).length).toBeLessThanOrEqual(4);
206 const [red, ...rest] = pulls as [Pull, Pull, Pull, Pull];
207 for (const pull of rest) expect(merged(repo, pull)).toBe(true);
208 expect(merged(repo, red)).toBe(false);
209 expect(latestDecision(app.db, red.id, "land")?.answer).toBe("red");
210 expect(tree(repo, "refs/heads/main")).toBe(tree(repo, gatesOf(repo).at(-1)!.commit));
211 });
212
213 test("a pull that conflicts with one ahead of it is left out; the rest land as gated", async () => {
214 const repo = await autonomousRepo("conflict");
215 const first = await greenPull(repo, "same.ts");
216 // Another branch writes the same path differently.
217 const pushed = await commitToBranch(app, alice, repo, {
218 branch: "feature/other",
219 base: "main",
220 message: "other",
221 files: [{ path: "same.ts", content: "export const n = 2;\n" }],
222 });
223 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
224 const opened = await openPull(app, alice, repo, {
225 title: "other",
226 baseBranch: "main",
227 headBranch: "feature/other",
228 });
229 if (!opened.ok) throw new Error(String(opened.error));
230 const other = getPull(app.db, repo.id, opened.value.number)!;
231 const check = createRun(app.db, {
232 repoId: repo.id,
233 kind: "check",
234 issueId: other.id,
235 env: `pull-${other.number}`,
236 commitOid: other.headOid,
237 reason: "pull",
238 });
239 claimRun(app.db, check.id);
240 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
241 const last = await greenPull(repo, "last.ts");
242 expect(await sweepGreen(app, repo)).toBe(2);
243 expect(gatesOf(repo)).toHaveLength(1);
244 expect(merged(repo, first)).toBe(true);
245 expect(merged(repo, last)).toBe(true);
246 expect(merged(repo, other)).toBe(false);
247 expect(tree(repo, "refs/heads/main")).toBe(tree(repo, gatesOf(repo)[0]!.commit));
248 });
249
250 test("a gated step lands only on the base and as the tree it was gated with", async () => {
251 const repo = await autonomousRepo("guarded");
252 const pull = await greenPull(repo, "guard.ts");
253 const tip = git(repo, "rev-parse", "refs/heads/main");
254 const wrongBase = await mergePull(app, alice, repo, pull, "squash", {
255 baseOid: git(repo, "rev-parse", "refs/heads/main~1"),
256 treeOid: tree(repo, "refs/heads/main"),
257 });
258 expect(wrongBase.ok).toBe(false);
259 if (!wrongBase.ok) expect(wrongBase.error).toEqual({ kind: "base_moved" });
260 const wrongTree = await mergePull(app, alice, repo, pull, "squash", {
261 baseOid: tip,
262 treeOid: tree(repo, tip),
263 });
264 expect(wrongTree.ok).toBe(false);
265 if (!wrongTree.ok) expect(wrongTree.error).toEqual({ kind: "base_moved" });
266 expect(git(repo, "rev-parse", "refs/heads/main")).toBe(tip);
267 expect(merged(repo, pull)).toBe(false);
268 });
269
270 test("a later step refused: the steps that landed are still released, once, at the last of them", async () => {
271 const repo = await autonomousRepo("prefix");
272 const pulls = [];
273 for (const name of ["p.ts", "q.ts", "r.ts"]) pulls.push(await greenPull(repo, name));
274 const dir = repoDir(app.reposRoot, repo.ownerName, repo.name);
275 const released: { pull: number; tip: string }[] = [];
276 // A stand-in landing: the first two land at a made-up tip, the third is
277 // refused as if its head moved during the gate.
278 const entries: Queued[] = pulls.map((pull, i) => ({
279 pull,
280 dir,
281 land: async () => (i < 2 ? `tip-${pull.number}` : undefined),
282 release: (tip) => released.push({ pull: pull.number, tip }),
283 handBack: () => {},
284 }));
285 const answers = await enqueue(repo.id, entries, async () => ({ lands: true }));
286 expect(answers).toEqual([true, true, false]);
287 expect(released).toEqual([{ pull: pulls[1]!.number, tip: `tip-${pulls[1]!.number}` }]);
288 });
289
290 test("the queue shows through every door: /flow, the terminal and MCP, while gating and after", async () => {
291 const repo = await autonomousRepo("on-flow");
292 const first = await greenPull(repo, "first.ts");
293 const second = await greenPull(repo, "second.ts");
294 const third = await greenPull(repo, "third.ts");
295 let release = (): void => {};
296 held = new Promise<void>((r) => (release = r));
297 // The first joins alone and is gated; the two after wait in line.
298 const landing = landIfGreen(app, repo, first);
299 while (!queueStatus().some((q) => q.repoId === repo.id && q.gating > 0)) await Bun.sleep(5);
300 const waiting = Promise.all([landIfGreen(app, repo, second), landIfGreen(app, repo, third)]);
301 while (!queueStatus().some((q) => q.repoId === repo.id && q.queued === 2)) await Bun.sleep(5);
302 const row = readQueues(app.db, new Date()).find((r) => r.repo === `alice/${repo.name}`);
303 expect(row).toMatchObject({ queued: 2, gating: 1 });
304 const res = await fetch(`${servers.url}/flow`, {
305 headers: { cookie: await login() },
306 redirect: "manual",
307 });
308 expect(res.status).toBe(200);
309 const page = await res.text();
310 expect(page).toMatch(
311 /data-queue="alice\/on-flow">[\s\S]*?data-queued>2<[\s\S]*?data-gating>1<[\s\S]*?data-landed>0</,
312 );
313 // The terminal's Flow dialog and the MCP tool say the same.
314 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "2", "1", "0"]);
315 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 2, gating: 1, landedHour: 0 });
316 held = undefined;
317 release();
318 expect(await landing).toBe(true);
319 // The two that waited are one candidate and one more gate.
320 expect(await waiting).toEqual([true, true]);
321 expect(gatesOf(repo)).toHaveLength(2);
322 expect(readQueues(app.db, new Date()).find((r) => r.repo === `alice/${repo.name}`)).toMatchObject({
323 queued: 0,
324 gating: 0,
325 landedHour: 3,
326 });
327 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "0", "0", "3"]);
328 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 0, gating: 0, landedHour: 3 });
329 });
330 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs