33a3ef4 · 920f045 · 96269b8
@@ -0,0 +1,56 @@
/**
 * Pulls closed without merging after a check stopped for CPU (#11904): the
 * machine's failure, not the pull's, so a person can queue each again. A pull
 * counts when its last check — the queued retry the close itself cancelled
 * aside — was a CPU stop (`isCpuStop`) that ended in the window, and the pull
 * is closed and unmerged now. One query, walked from `run_kind_finished`
 * (kind, finished_at): the window's finished checks, newest close first.
 */
import type { Database } from "bun:sqlite";
import { CPU_STOP_SUMMARY } from "../core/runs.ts";
import { isCpuStop } from "./cpu_stops.ts";
/** How far back a stop is looked for: a week of closes to re-queue. */
export const CPU_CLOSED_WINDOW_MS = 7 * 24 * 3_600_000;
/** At most this many pulls are listed. */
export const CPU_CLOSED_CAP = 20;
export interface CpuClosedPull {
  /** `owner/name`. */
  readonly repo: string;
  readonly number: number;
  readonly closedAt: string;
}
/** What the close itself writes on the checks it cancels (`checks/service.ts`): not the pull's last word. */
const CLOSE_SUMMARIES = ["pull closed", "pull landed"] as const;
export function cpuClosedPulls(
  db: Database,
  now: Date,
  opts: { readonly cap?: number; readonly publicOnly?: boolean } = {},
): CpuClosedPull[] {
  const since = new Date(now.getTime() - CPU_CLOSED_WINDOW_MS).toISOString();
  const rows = db
    .query<{ repo: string; number: number; closed_at: string; state: string; summary: string }, [string, string, string, string, string]>(
      `SELECT o.username || '/' || r.name AS repo, i.number AS number, i.closed_at AS closed_at, u.state AS state, u.summary AS summary
         FROM run u
         JOIN issue i ON i.id = u.issue_id AND i.kind = 'pull' AND i.state = 'closed' AND i.closed_at <= ?5
         JOIN pull p ON p.issue_id = i.id AND p.merged_at IS NULL
         JOIN repo r ON r.id = i.repo_id JOIN user o ON o.id = r.owner_id
        WHERE u.kind = 'check' AND u.finished_at IS NOT NULL AND u.finished_at >= ?1
          AND (u.state = 'failed' OR (u.state = 'cancelled' AND u.summary = ?2))
          ${opts.publicOnly === true ? "AND r.visibility = 'public'" : ""}
          AND u.id = (SELECT n.id FROM run n WHERE n.issue_id = i.id AND n.kind = 'check'
                         AND NOT (n.state = 'cancelled' AND n.summary IN (?3, ?4))
                       ORDER BY n.id DESC LIMIT 1)
        ORDER BY i.closed_at DESC, i.number DESC`,
    )
    .all(since, CPU_STOP_SUMMARY, CLOSE_SUMMARIES[0], CLOSE_SUMMARIES[1], now.toISOString());
  return rows
    .filter((r) => isCpuStop(r.state, r.summary))
    .slice(0, opts.cap ?? CPU_CLOSED_CAP)
    .map((r) => ({ repo: r.repo, number: r.number, closedAt: r.closed_at }));
}
@@ -21,6 +21,11 @@
export interface CpuStops {
  readonly failed: number;
}
/** Whether a finished check was stopped for CPU: queued again (cancelled, `CPU_STOP_SUMMARY`) or reported red. */
export function isCpuStop(state: string, summary: string): boolean {
  return state === "cancelled" ? summary === CPU_STOP_SUMMARY : state === "failed" && CPU_RED.test(summary);
}
export function cpuStops(db: Database, now = new Date()): CpuStops {
  const since = new Date(now.getTime() - HOUR_MS).toISOString();
  const rows = db
@@ -33,7 +38,7 @@
export function cpuStops(db: Database, now = new Date()): CpuStops {
  let stopped = 0;
  let failed = 0;
  for (const r of rows) {
    const cpu = r.state === "cancelled" ? r.summary === CPU_STOP_SUMMARY : CPU_RED.test(r.summary);
    const cpu = isCpuStop(r.state, r.summary);
    if (r.state === "failed" || cpu) failed++;
    if (cpu) stopped++;
  }
@@ -0,0 +1,14 @@
/**
 * /flow's CPU-stop block (#11904): the hour's checks stopped for CPU, and the
 * pulls closed unmerged after such a stop as links, so a person can queue them
 * again. Read by `flow/cpu_stops.ts` and `flow/cpu_closed.ts`. Quiet draws nothing.
 */
import type { CpuClosedPull } from "../../flow/cpu_closed.ts";
import { html, type Html } from "../html.ts";
export function cpuStopsSection(stopped: number, closed: readonly CpuClosedPull[]): Html {
  if (stopped === 0 && closed.length === 0) return html``;
  return html`<p class="mb" id="flow-cpu-stops">${stopped === 0 ? "" : html`${String(stopped)} ${stopped === 1 ? "check" : "checks"} stopped for CPU in the last hour.`}
    ${closed.length === 0 ? "" : html`Closed unmerged after a CPU stop, to queue again: ${closed.map((p, i) => html`${i === 0 ? "" : ", "}<a href="/${p.repo}/pulls/${String(p.number)}">${p.repo} #${String(p.number)}</a>`)}.`}</p>`;
}
@@ -25,6 +25,9 @@
import { readDrifts } from "../../flow/alarm_drift.ts";
import { readHandoverLists } from "../../flow/handover_read.ts";
import { readWaiting, waitingSection } from "./flow-waiting.ts";
import { clogSection, readClogView } from "./flow-clog.ts";
import { cpuStopsSection } from "./flow-cpu.ts";
import { cpuClosedPulls } from "../../flow/cpu_closed.ts";
import { cpuStops } from "../../flow/cpu_stops.ts";
const money = (usd: number): string => `$${usd.toFixed(2)}`;
const plural = (n: number, one: string, many = `${one}s`): string => `${String(n)} ${n === 1 ? one : many}`;
@@ -103,6 +106,7 @@
export async function flowPage(ctx: Ctx): Promise<Response> {
    <p class="mb">Money flows from the left to shipped on the right, as wide as this week's real spend. Every repository has a valve: drag it to give the repository a share of the week, all the way down to shut. The crew does exactly what the valves say, from the next tick, and nothing but you turns one.</p>
    ${backupRow(alarmBackup(ctx.app.config.dataDir, new Date()))}
    ${clogSection(readClogView(ctx.app.db))}
    ${cpuStopsSection(cpuStops(ctx.app.db).stopped, cpuClosedPulls(ctx.app.db, new Date()))}
    ${waitingSection(await readWaiting(ctx.app.db, ctx.app.reposRoot, new Date()))}
    ${pipesSvg(p)}
    <p class="muted small">${money(p.usd)} spent in the last 7 days; ${plural(p.shipped, "pull")} shipped.</p>
@@ -0,0 +1,123 @@
/** Pulls closed unmerged after a check stopped for CPU (#11904): listed on /flow and by `flow { kind: cpu }`, so a person can queue them again. */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createRepo } from "../src/core/repos.ts";
import { CPU_STOP_SUMMARY, createRun } from "../src/core/runs.ts";
import { createToken } from "../src/core/tokens.ts";
import { createUser } from "../src/core/users.ts";
import { cpuClosedPulls } from "../src/flow/cpu_closed.ts";
import { serve, type Servers } from "../src/http/server.ts";
import { id } from "../src/kernel/id.ts";
import { cpuStopsSection } from "../src/web/pages/flow-cpu.ts";
const NOW = new Date();
const ago = (m: number) => new Date(NOW.getTime() - m * 60_000).toISOString();
const RED_STOP = "check failed: stopped: used 16.3 cores for 31s, over twice its limit of 16; the busiest process is sh";
let dir: string;
let app: App;
let servers: Servers;
let adminTok: string;
let guestTok: string;
beforeAll(async () => {
  dir = await mkdtemp(join(tmpdir(), "forge-cpu-closed-"));
  app = createApp({ dataDir: dir, port: 0 });
  servers = await serve(app);
  const mk = async (username: string, first = false) => {
    const u = await createUser(app.db, { origin: "seed", trusted: true, username, email: `${username}@forge.local`, password: "a-good-passphrase", adminIfFirst: first });
    if (!u.ok) throw new Error(u.error);
    return u.value;
  };
  const alice = await mk("alice", true);
  const carol = await mk("carol");
  adminTok = createToken(app.db, alice.id, "rw", "rw").token;
  guestTok = createToken(app.db, carol.id, "rw", "rw").token;
  const repo = async (name: string, visibility: "public" | "private") => {
    const x = await createRepo(app.db, app.reposRoot, { ownerId: alice.id, name, visibility, description: name });
    if (!x.ok) throw new Error(x.error);
    return x.value;
  };
  const pub = await repo("engine", "public");
  const priv = await repo("vault", "private");
  let n = 0;
  /** A pull closed `closedMins` ago (merged or not) whose checks ended as `checks` say, oldest first. */
  const pull = (r: typeof pub, closedMins: number, merged: boolean, checks: [state: string, summary: string][]) => {
    const issueId = id();
    n++;
    app.db
      .query(`INSERT INTO issue (id, repo_id, number, kind, title, author_id, state, created_at, closed_at) VALUES (?1, ?2, ?3, 'pull', 'P', ?4, 'closed', ?5, ?6)`)
      .run(issueId, r.id, n, alice.id, ago(closedMins + 60), ago(closedMins));
    app.db
      .query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id, merged_at) VALUES (?1, 'main', ?2, '', '', ?3, ?4)`)
      .run(issueId, `b/${n}`, r.id, merged ? ago(closedMins) : null);
    checks.forEach(([state, summary], k) => {
      const run = createRun(app.db, { repoId: r.id, kind: "check", issueId, attempt: k });
      app.db.query(`UPDATE run SET state = ?2, summary = ?3, finished_at = ?4 WHERE id = ?1`).run(run.id, state, summary, ago(closedMins + checks.length - k));
    });
    return n;
  };
  // Stopped for CPU, queued again, then closed: the close cancelled the retry.
  pull(pub, 30, false, [["cancelled", CPU_STOP_SUMMARY], ["cancelled", "pull closed"]]);
  // Stopped for CPU too, but it landed.
  pull(pub, 20, true, [["cancelled", CPU_STOP_SUMMARY], ["cancelled", "pull landed"]]);
  // Red for its own reason and closed.
  pull(pub, 10, false, [["failed", "check failed: test/a.test.ts failed"]]);
  // Stopped once, then red for its own reason: its last check is not a stop.
  pull(pub, 15, false, [["cancelled", CPU_STOP_SUMMARY], ["failed", "check failed: test/b.test.ts failed"]]);
  // A second stop reported red, closed later, in a private repository.
  pull(priv, 5, false, [["cancelled", CPU_STOP_SUMMARY], ["failed", RED_STOP]]);
});
afterAll(async () => {
  servers.stop();
  app.close();
  await rm(dir, { recursive: true, force: true });
});
describe("cpuClosedPulls", () => {
  test("lists only the pulls closed unmerged whose last check was a CPU stop, newest close first", () => {
    expect(cpuClosedPulls(app.db, NOW).map((p) => `${p.repo}#${p.number}`)).toEqual(["alice/vault#5", "alice/engine#1"]);
    expect(cpuClosedPulls(app.db, NOW, { publicOnly: true }).map((p) => p.number)).toEqual([1]);
    expect(cpuClosedPulls(app.db, NOW, { cap: 1 }).map((p) => p.number)).toEqual([5]);
  });
  test("the clock is passed in: a week later the stops have left the window, and before the close nothing is closed", () => {
    expect(cpuClosedPulls(app.db, new Date(NOW.getTime() + 8 * 24 * 3_600_000))).toEqual([]);
    expect(cpuClosedPulls(app.db, new Date(NOW.getTime() - 7 * 60_000)).map((p) => p.number)).toEqual([1]);
  });
});
describe("/flow's block", () => {
  test("says the hour's count and links each pull; draws nothing when both are empty", () => {
    expect(String(cpuStopsSection(0, []))).toBe("");
    const s = String(cpuStopsSection(2, cpuClosedPulls(app.db, NOW)));
    expect(s).toContain("2 checks stopped for CPU in the last hour.");
    expect(s).toContain(`<a href="/alice/engine/pulls/1">alice/engine #1</a>`);
    expect(s).toContain(`<a href="/alice/vault/pulls/5">alice/vault #5</a>`);
    expect(String(cpuStopsSection(0, cpuClosedPulls(app.db, NOW)))).not.toContain("stopped for CPU in the last hour");
  });
});
describe("flow { kind: cpu }", () => {
  const call = async (token: string) => {
    const res = await fetch(`${servers.url}/mcp`, {
      method: "POST",
      headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "flow", arguments: { kind: "cpu" } } }),
    });
    const body: any = await res.json();
    return JSON.parse(body.result.content[0].text);
  };
  test("returns the same rows: an admin every repository, anyone else public ones", async () => {
    const admin = await call(adminTok);
    expect(admin.closed.map((p: { number: number }) => p.number)).toEqual([5, 1]);
    expect(typeof admin.stopped).toBe("number");
    const guest = await call(guestTok);
    expect(guest).toEqual({ stopped: 0, closed: [expect.objectContaining({ repo: "alice/engine", number: 1 })] });
  });
});
		@@ -0,0 +1,56 @@
	1	/**
	2	* Pulls closed without merging after a check stopped for CPU (#11904): the
	3	* machine's failure, not the pull's, so a person can queue each again. A pull
	4	* counts when its last check — the queued retry the close itself cancelled
	5	* aside — was a CPU stop (`isCpuStop`) that ended in the window, and the pull
	6	* is closed and unmerged now. One query, walked from `run_kind_finished`
	7	* (kind, finished_at): the window's finished checks, newest close first.
	8	*/
	9
	10	import type { Database } from "bun:sqlite";
	11	import { CPU_STOP_SUMMARY } from "../core/runs.ts";
	12	import { isCpuStop } from "./cpu_stops.ts";
	13
	14	/** How far back a stop is looked for: a week of closes to re-queue. */
	15	export const CPU_CLOSED_WINDOW_MS = 7 * 24 * 3_600_000;
	16
	17	/** At most this many pulls are listed. */
	18	export const CPU_CLOSED_CAP = 20;
	19
	20	export interface CpuClosedPull {
	21	/** `owner/name`. */
	22	readonly repo: string;
	23	readonly number: number;
	24	readonly closedAt: string;
	25	}
	26
	27	/** What the close itself writes on the checks it cancels (`checks/service.ts`): not the pull's last word. */
	28	const CLOSE_SUMMARIES = ["pull closed", "pull landed"] as const;
	29
	30	export function cpuClosedPulls(
	31	db: Database,
	32	now: Date,
	33	opts: { readonly cap?: number; readonly publicOnly?: boolean } = {},
	34	): CpuClosedPull[] {
	35	const since = new Date(now.getTime() - CPU_CLOSED_WINDOW_MS).toISOString();
	36	const rows = db
	37	.query<{ repo: string; number: number; closed_at: string; state: string; summary: string }, [string, string, string, string, string]>(
	38	`SELECT o.username \|\| '/' \|\| r.name AS repo, i.number AS number, i.closed_at AS closed_at, u.state AS state, u.summary AS summary
	39	FROM run u
	40	JOIN issue i ON i.id = u.issue_id AND i.kind = 'pull' AND i.state = 'closed' AND i.closed_at <= ?5
	41	JOIN pull p ON p.issue_id = i.id AND p.merged_at IS NULL
	42	JOIN repo r ON r.id = i.repo_id JOIN user o ON o.id = r.owner_id
	43	WHERE u.kind = 'check' AND u.finished_at IS NOT NULL AND u.finished_at >= ?1
	44	AND (u.state = 'failed' OR (u.state = 'cancelled' AND u.summary = ?2))
	45	${opts.publicOnly === true ? "AND r.visibility = 'public'" : ""}
	46	AND u.id = (SELECT n.id FROM run n WHERE n.issue_id = i.id AND n.kind = 'check'
	47	AND NOT (n.state = 'cancelled' AND n.summary IN (?3, ?4))
	48	ORDER BY n.id DESC LIMIT 1)
	49	ORDER BY i.closed_at DESC, i.number DESC`,
	50	)
	51	.all(since, CPU_STOP_SUMMARY, CLOSE_SUMMARIES[0], CLOSE_SUMMARIES[1], now.toISOString());
	52	return rows
	53	.filter((r) => isCpuStop(r.state, r.summary))
	54	.slice(0, opts.cap ?? CPU_CLOSED_CAP)
	55	.map((r) => ({ repo: r.repo, number: r.number, closedAt: r.closed_at }));
	56	}
		@@ -21,6 +21,11 @@ export interface CpuStops {
21	21	readonly failed: number;
22	22	}
23	23
	24	/** Whether a finished check was stopped for CPU: queued again (cancelled, `CPU_STOP_SUMMARY`) or reported red. */
	25	export function isCpuStop(state: string, summary: string): boolean {
	26	return state === "cancelled" ? summary === CPU_STOP_SUMMARY : state === "failed" && CPU_RED.test(summary);
	27	}
	28
24	29	export function cpuStops(db: Database, now = new Date()): CpuStops {
25	30	const since = new Date(now.getTime() - HOUR_MS).toISOString();
26	31	const rows = db
		@@ -33,7 +38,7 @@ export function cpuStops(db: Database, now = new Date()): CpuStops {
33	38	let stopped = 0;
34	39	let failed = 0;
35	40	for (const r of rows) {
36		const cpu = r.state === "cancelled" ? r.summary === CPU_STOP_SUMMARY : CPU_RED.test(r.summary);
	41	const cpu = isCpuStop(r.state, r.summary);
37	42	if (r.state === "failed" \|\| cpu) failed++;
38	43	if (cpu) stopped++;
39	44	}
		@@ -0,0 +1,14 @@
	1	/**
	2	* /flow's CPU-stop block (#11904): the hour's checks stopped for CPU, and the
	3	* pulls closed unmerged after such a stop as links, so a person can queue them
	4	* again. Read by `flow/cpu_stops.ts` and `flow/cpu_closed.ts`. Quiet draws nothing.
	5	*/
	6
	7	import type { CpuClosedPull } from "../../flow/cpu_closed.ts";
	8	import { html, type Html } from "../html.ts";
	9
	10	export function cpuStopsSection(stopped: number, closed: readonly CpuClosedPull[]): Html {
	11	if (stopped === 0 && closed.length === 0) return html``;
	12	return html`<p class="mb" id="flow-cpu-stops">${stopped === 0 ? "" : html`${String(stopped)} ${stopped === 1 ? "check" : "checks"} stopped for CPU in the last hour.`}
	13	${closed.length === 0 ? "" : html`Closed unmerged after a CPU stop, to queue again: ${closed.map((p, i) => html`${i === 0 ? "" : ", "}<a href="/${p.repo}/pulls/${String(p.number)}">${p.repo} #${String(p.number)}</a>`)}.`}</p>`;
	14	}
		@@ -25,6 +25,9 @@ import { readDrifts } from "../../flow/alarm_drift.ts";
25	25	import { readHandoverLists } from "../../flow/handover_read.ts";
26	26	import { readWaiting, waitingSection } from "./flow-waiting.ts";
27	27	import { clogSection, readClogView } from "./flow-clog.ts";
	28	import { cpuStopsSection } from "./flow-cpu.ts";
	29	import { cpuClosedPulls } from "../../flow/cpu_closed.ts";
	30	import { cpuStops } from "../../flow/cpu_stops.ts";
28	31
29	32	const money = (usd: number): string => `$${usd.toFixed(2)}`;
30	33	const plural = (n: number, one: string, many = `${one}s`): string => `${String(n)} ${n === 1 ? one : many}`;
		@@ -103,6 +106,7 @@ export async function flowPage(ctx: Ctx): Promise<Response> {
103	106	<p class="mb">Money flows from the left to shipped on the right, as wide as this week's real spend. Every repository has a valve: drag it to give the repository a share of the week, all the way down to shut. The crew does exactly what the valves say, from the next tick, and nothing but you turns one.</p>
104	107	${backupRow(alarmBackup(ctx.app.config.dataDir, new Date()))}
105	108	${clogSection(readClogView(ctx.app.db))}
	109	${cpuStopsSection(cpuStops(ctx.app.db).stopped, cpuClosedPulls(ctx.app.db, new Date()))}
106	110	${waitingSection(await readWaiting(ctx.app.db, ctx.app.reposRoot, new Date()))}
107	111	${pipesSvg(p)}
108	112	<p class="muted small">${money(p.usd)} spent in the last 7 days; ${plural(p.shipped, "pull")} shipped.</p>
		@@ -0,0 +1,123 @@
	1	/** Pulls closed unmerged after a check stopped for CPU (#11904): listed on /flow and by `flow { kind: cpu }`, so a person can queue them again. */
	2	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	3	import { mkdtemp, rm } from "node:fs/promises";
	4	import { tmpdir } from "node:os";
	5	import { join } from "node:path";
	6	import { createApp, type App } from "../src/app.ts";
	7	import { createRepo } from "../src/core/repos.ts";
	8	import { CPU_STOP_SUMMARY, createRun } from "../src/core/runs.ts";
	9	import { createToken } from "../src/core/tokens.ts";
	10	import { createUser } from "../src/core/users.ts";
	11	import { cpuClosedPulls } from "../src/flow/cpu_closed.ts";
	12	import { serve, type Servers } from "../src/http/server.ts";
	13	import { id } from "../src/kernel/id.ts";
	14	import { cpuStopsSection } from "../src/web/pages/flow-cpu.ts";
	15
	16	const NOW = new Date();
	17	const ago = (m: number) => new Date(NOW.getTime() - m * 60_000).toISOString();
	18	const RED_STOP = "check failed: stopped: used 16.3 cores for 31s, over twice its limit of 16; the busiest process is sh";
	19
	20	let dir: string;
	21	let app: App;
	22	let servers: Servers;
	23	let adminTok: string;
	24	let guestTok: string;
	25
	26	beforeAll(async () => {
	27	dir = await mkdtemp(join(tmpdir(), "forge-cpu-closed-"));
	28	app = createApp({ dataDir: dir, port: 0 });
	29	servers = await serve(app);
	30	const mk = async (username: string, first = false) => {
	31	const u = await createUser(app.db, { origin: "seed", trusted: true, username, email: `${username}@forge.local`, password: "a-good-passphrase", adminIfFirst: first });
	32	if (!u.ok) throw new Error(u.error);
	33	return u.value;
	34	};
	35	const alice = await mk("alice", true);
	36	const carol = await mk("carol");
	37	adminTok = createToken(app.db, alice.id, "rw", "rw").token;
	38	guestTok = createToken(app.db, carol.id, "rw", "rw").token;
	39	const repo = async (name: string, visibility: "public" \| "private") => {
	40	const x = await createRepo(app.db, app.reposRoot, { ownerId: alice.id, name, visibility, description: name });
	41	if (!x.ok) throw new Error(x.error);
	42	return x.value;
	43	};
	44	const pub = await repo("engine", "public");
	45	const priv = await repo("vault", "private");
	46	let n = 0;
	47	/** A pull closed `closedMins` ago (merged or not) whose checks ended as `checks` say, oldest first. */
	48	const pull = (r: typeof pub, closedMins: number, merged: boolean, checks: [state: string, summary: string][]) => {
	49	const issueId = id();
	50	n++;
	51	app.db
	52	.query(`INSERT INTO issue (id, repo_id, number, kind, title, author_id, state, created_at, closed_at) VALUES (?1, ?2, ?3, 'pull', 'P', ?4, 'closed', ?5, ?6)`)
	53	.run(issueId, r.id, n, alice.id, ago(closedMins + 60), ago(closedMins));
	54	app.db
	55	.query(`INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid, head_repo_id, merged_at) VALUES (?1, 'main', ?2, '', '', ?3, ?4)`)
	56	.run(issueId, `b/${n}`, r.id, merged ? ago(closedMins) : null);
	57	checks.forEach(([state, summary], k) => {
	58	const run = createRun(app.db, { repoId: r.id, kind: "check", issueId, attempt: k });
	59	app.db.query(`UPDATE run SET state = ?2, summary = ?3, finished_at = ?4 WHERE id = ?1`).run(run.id, state, summary, ago(closedMins + checks.length - k));
	60	});
	61	return n;
	62	};
	63	// Stopped for CPU, queued again, then closed: the close cancelled the retry.
	64	pull(pub, 30, false, [["cancelled", CPU_STOP_SUMMARY], ["cancelled", "pull closed"]]);
	65	// Stopped for CPU too, but it landed.
	66	pull(pub, 20, true, [["cancelled", CPU_STOP_SUMMARY], ["cancelled", "pull landed"]]);
	67	// Red for its own reason and closed.
	68	pull(pub, 10, false, [["failed", "check failed: test/a.test.ts failed"]]);
	69	// Stopped once, then red for its own reason: its last check is not a stop.
	70	pull(pub, 15, false, [["cancelled", CPU_STOP_SUMMARY], ["failed", "check failed: test/b.test.ts failed"]]);
	71	// A second stop reported red, closed later, in a private repository.
	72	pull(priv, 5, false, [["cancelled", CPU_STOP_SUMMARY], ["failed", RED_STOP]]);
	73	});
	74
	75	afterAll(async () => {
	76	servers.stop();
	77	app.close();
	78	await rm(dir, { recursive: true, force: true });
	79	});
	80
	81	describe("cpuClosedPulls", () => {
	82	test("lists only the pulls closed unmerged whose last check was a CPU stop, newest close first", () => {
	83	expect(cpuClosedPulls(app.db, NOW).map((p) => `${p.repo}#${p.number}`)).toEqual(["alice/vault#5", "alice/engine#1"]);
	84	expect(cpuClosedPulls(app.db, NOW, { publicOnly: true }).map((p) => p.number)).toEqual([1]);
	85	expect(cpuClosedPulls(app.db, NOW, { cap: 1 }).map((p) => p.number)).toEqual([5]);
	86	});
	87
	88	test("the clock is passed in: a week later the stops have left the window, and before the close nothing is closed", () => {
	89	expect(cpuClosedPulls(app.db, new Date(NOW.getTime() + 8 * 24 * 3_600_000))).toEqual([]);
	90	expect(cpuClosedPulls(app.db, new Date(NOW.getTime() - 7 * 60_000)).map((p) => p.number)).toEqual([1]);
	91	});
	92	});
	93
	94	describe("/flow's block", () => {
	95	test("says the hour's count and links each pull; draws nothing when both are empty", () => {
	96	expect(String(cpuStopsSection(0, []))).toBe("");
	97	const s = String(cpuStopsSection(2, cpuClosedPulls(app.db, NOW)));
	98	expect(s).toContain("2 checks stopped for CPU in the last hour.");
	99	expect(s).toContain(`<a href="/alice/engine/pulls/1">alice/engine #1</a>`);
	100	expect(s).toContain(`<a href="/alice/vault/pulls/5">alice/vault #5</a>`);
	101	expect(String(cpuStopsSection(0, cpuClosedPulls(app.db, NOW)))).not.toContain("stopped for CPU in the last hour");
	102	});
	103	});
	104
	105	describe("flow { kind: cpu }", () => {
	106	const call = async (token: string) => {
	107	const res = await fetch(`${servers.url}/mcp`, {
	108	method: "POST",
	109	headers: { "content-type": "application/json", accept: "application/json, text/event-stream", authorization: `Bearer ${token}` },
	110	body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "flow", arguments: { kind: "cpu" } } }),
	111	});
	112	const body: any = await res.json();
	113	return JSON.parse(body.result.content[0].text);
	114	};
	115
	116	test("returns the same rows: an admin every repository, anyone else public ones", async () => {
	117	const admin = await call(adminTok);
	118	expect(admin.closed.map((p: { number: number }) => p.number)).toEqual([5, 1]);
	119	expect(typeof admin.stopped).toBe("number");
	120	const guest = await call(guestTok);
	121	expect(guest).toEqual({ stopped: 0, closed: [expect.objectContaining({ repo: "alice/engine", number: 1 })] });
	122	});
	123	});