/*
 * `stall_check` — a predicate over the board and the bus, plus a HALT flag.
 *
 * Stall has been "notice the room" until now, which means it is noticed when
 * somebody happens to look. The measured cost elsewhere: ~42 hours of dead merge
 * path across four incidents in three days, with `main` advancing throughout so
 * every commit-based liveness check read green.
 *
 * A MISS IS SILENT TO THE DUTY OFFICER AND NEVER SILENT TO THE RECORD.
 *
 * This is the whole design constraint (3.3b). "HIT DMs, MISS silent" answers the
 * noise question and makes a DEAD CLOCK look exactly like a healthy fleet — the
 * absence-read-as-evidence rule, inside the verb built to watch for absence. So
 * every run leaves a mark, HIT or MISS, and "no alert" becomes distinguishable
 * from "nothing ran". A check that only speaks when it fires cannot be told from
 * a broken one.
 */
import { activeTransport, isLocallyProbeable } from "../transports/index.js";
import { readFleetTick, tickByAgent, tickVerdict, type FleetTick, type SeatTick } from "./tick.js";
import { existsSync, readFileSync, writeFileSync, mkdirSync, readdirSync } from "node:fs";
import { execFileSync } from "node:child_process";
import path from "node:path";
import { z } from "zod";
import { parseWorkDoc, workstreamsV1RowsOf, workstreamsExtensionsOf, isMalformedRow, queueItemsOf, workStateOf, coarseOf, type WorkState, type WorkstreamsV1Row } from "@davidbalzan/groundwork-seam";
import { ROOT, AGENTS_FILE, readJson } from "../store.js";
import { classifyBoardRef, refInCell, rowKindOf, cellKindOf } from "./board-ref.js";
import { loadLiveTransports } from "./registry.js";
import { verdictsFor, shaAgrees, verdictShasIn, verdictBoundTo } from "../gated-head.js";
import { closingsIn, ACCEPTED_FORMS, type Closing } from "../closing-line.js";

const BOARD_DOC = "docs/WORKSTREAMS.md";

/*
 * ⟨q-3d82f1a9⟩ — A BOARD THAT FAILS TO PARSE RENDERS AS A QUIET FLEET.
 *
 * Measured live 2026-09-14: ad80825 rewrote line 27 — the FIRST row of the
 * table — from 6 columns to 5. The seam ended the workstreams.v1 block at
 * that line and recorded NO malformed row (rows:1, malformed:0), so the other
 * 24 rows fell out of every count, and stall_check answered checked:0 with an
 * EMPTY unmeasurable list and an EMPTY blind list: the healthiest reading the
 * instrument can produce, while two seats were working. Every other
 * degradation here pushes an `unmeasurable` entry; an unreadable board pushes
 * nothing, because there is no row to attach the complaint to.
 *
 * So the parse state is a STATEMENT in this verb's own output, on its own key —
 * never an inference a careful reader must know to draw from the role count.
 * The discriminator is TEXT-LEVEL because the seam's is not: rows PRESENT in
 * the Active Streams table (pipe-lines after the header and separator) versus
 * rows the v1 block PARSED, with the block's own malformed rows counted so a
 * refused-but-recorded row is not double-charged. `unparsed > 0` means rows
 * exist that no record describes. A genuinely EMPTY board — header, separator,
 * nothing — is 0 present / 0 parsed and reads as readable-empty, not suspect.
 */
export type BoardParse = {
  section: "Active Streams";
  headerFound: boolean;
  rowsPresent: number;
  rowsParsed: number;
  malformed: number;
  unparsed: number;
  readable: boolean;
  why: string;
};
export function boardParseOf(text: string): BoardParse {
  const lines = String(text ?? "").split("\n");
  const header = lines.findIndex((l) => /^\|\s*Stream\s*\|/i.test(l));
  const doc = parseWorkDoc(String(text ?? ""));
  const block = doc.blocks.find((b) => b.kind === "board" && (b as { schema?: string }).schema === "workstreams.v1") as
    | { rows?: unknown[] }
    | undefined;
  const blockRows = (block?.rows ?? []) as Parameters<typeof isMalformedRow>[0][];
  const rowsParsed = workstreamsV1RowsOf(doc).length;
  const malformed = blockRows.filter((r) => isMalformedRow(r)).length;
  if (header === -1) {
    const pipeLines = lines.filter((l) => /^\s*\|/.test(l)).length;
    const readable = pipeLines === 0 && rowsParsed === 0;
    return {
      section: "Active Streams", headerFound: false, rowsPresent: pipeLines, rowsParsed, malformed, unparsed: Math.max(0, pipeLines - rowsParsed - malformed), readable,
      why: readable
        ? "no Active Streams table in the file and no table-shaped lines — an empty board, readable"
        : `no \`| Stream |\` header found but ${pipeLines} table-shaped line(s) exist — the board is not a workstreams.v1 table this verb can read; UNPARSEABLE, which is not the same as empty`,
    };
  }
  let i = header + 1;
  if (i < lines.length && /^\s*\|\s*:?-+/.test(lines[i]!)) i++; // the alignment row is scaffold, not a row
  let rowsPresent = 0;
  for (; i < lines.length && /^\s*\|/.test(lines[i]!); i++) rowsPresent++;
  const unparsed = Math.max(0, rowsPresent - rowsParsed - malformed);
  // A MALFORMED ROW IS UNREADABLE TOO. Measured while writing the control: a
  // short row that is not the table's first line is recorded by the seam as
  // malformed (refused, kept verbatim) rather than ending the block — and a
  // refused lane is exactly as invisible to this clock as a dropped one. Both
  // conjuncts are required; the ad80825 shape trips the first, the mid-table
  // shape trips the second.
  const readable = unparsed === 0 && malformed === 0;
  return {
    section: "Active Streams", headerFound: true, rowsPresent, rowsParsed, malformed, unparsed, readable,
    why: readable
      ? rowsPresent === 0
        ? "the Active Streams table is present and EMPTY — 0 rows present, 0 parsed; readable"
        : `${rowsPresent} row(s) present, ${rowsParsed} parsed — every row is accounted for`
      : `UNPARSEABLE — ${rowsPresent} row(s) present in the Active Streams table but only ${rowsParsed} parsed` +
        `${malformed ? `; ${malformed} row(s) REFUSED by the parser as malformed (wrong column count) and therefore invisible to this clock` : ""}` +
        `${unparsed ? `; ${unparsed} row(s) exist that no record describes — one bad row ends the table for the parser and nothing below it is measured` : ""}. ` +
        `"Nothing measured" is not "nothing wrong".`,
  };
}
const STALL_MS = 30 * 60 * 1000;
/**
 * How long a lane may sit in review before it is a HIT. Four times the
 * authoring window, not equal to it: a review is a queue on another seat, and
 * the number that matters is how long that queue has held this row — it is not
 * a tuned threshold on branch activity, which a review-frozen branch fails by
 * design (⟨q-507e80c4⟩).
 */
const REVIEW_STALL_MINUTES = 120;
/** ⟨q-7b2f6c04⟩ — the claim window: a lane claimed this long ago with nothing on origin is a hit. Its own window, independent of the VCS one. */
export const CLAIM_STALL_MINUTES = 120;

const runFile = () => path.join(ROOT, "stall-check.json");
const haltFile = () => path.join(ROOT, "halt.json");

/*
 * ⟨q-7e3b10c9⟩ — EVERY HIT CARRIES ITS AUDIENCE. Measured 2026-09-11: the
 * clock's first alarm of the day was a `stale-row` — a squash-landed branch
 * under a 🚧 row — and the WORKER was told about a lane it had finished. The
 * predicate had already said "not a stall, a stale row"; the delivery ignored
 * it. A stall alarm that cries on finished lanes is disabled by its third
 * firing. So the hit says who it is FOR, and every relayer — the clock script,
 * a seat forwarding `dm:true` — routes on that rather than on `agentId`:
 *   duty         the lane's owner may be stuck — no heartbeat, no commits, too
 *                long in review, an open PR nobody verdicted, a pushed branch
 *                nobody proposed. Somebody must look at the WORK.
 *   board-owner  bookkeeping: a row pointing at a landed branch, a routed item
 *                with no row, a merge with no verdict comment. Somebody must
 *                fix the RECORD; the lane's owner did nothing wrong.
 *   david        ⟨q-11257590⟩ nobody IN THE FLEET can act on this: the coordinator
 *                seat itself is the thing missing, so there is no duty owner and no
 *                board-owner left to fix the record. David's own ruling: "coord
 *                should never be absent in reality... if we have a comms issue that
 *                just needs to be surfaced to me then i tackle it" — this audience
 *                exists so a relayer routes it to him and nowhere else.
 */
export type HitAudience = "duty" | "board-owner" | "david";
export const audienceOf = (kind: StallHitBody["kind"]): HitAudience =>
  kind === "coordinator-absent"
    ? "david"
    : kind === "stale-row" || kind === "routed-without-row" || kind === "unverdicted-merge" || kind === "merge-window-write" || kind === "lane-left-population"
      ? "board-owner"
      : "duty";
const withAudience = (hits: StallHitBody[]): StallHit[] => hits.map((h) => ({ ...h, audience: audienceOf(h.kind) }) as StallHit);
export type StallHit = StallHitBody & { audience: HitAudience };
export type StallHitBody =
  | { kind: "no-heartbeat"; agentId: string; stream: string; minutes: number }
  | { kind: "no-vcs-activity"; agentId: string; branch: string; minutes: number }
  /** ⟨q-1c95f7d4⟩ Task 5 — an EXTERNAL observer says the seat is waiting on a person. Not inferred from the branch, and not a verdict about the lane. */
  | { kind: "seat-blocked"; agentId: string; stream: string; source: string; why: string }
  /**
   * ⟨q-7b2f6c04⟩ — THE CLAIM AXIS: a lane claimed (its row entered in-flight,
   * read from the board's git history) whose branch has nothing on origin —
   * not pushed, or pushed empty at claim — for longer than the claim window.
   * The window between `claim` and the first push used to be UNMEASURABLE by
   * construction; a 35-hour fleet-wide stop read as healthy inside it.
   */
  | { kind: "unpushed-claim"; agentId: string; stream: string; branch: string; minutes: number; since: string; why: string }
  /**
   * ⟨q-d0e83b41⟩ (09-14 amend) — a lane LEFT the scored population since the
   * last run with no closing state on the board: its row is gone. Without this
   * a seat leaving the denominator reads as an IMPROVED ratio.
   */
  | { kind: "lane-left-population"; agentId: string; stream: string; why: string }
  /**
   * A 🚧 row whose branch HAS ALREADY LANDED. Not a stall — the opposite: the
   * work finished and nobody closed the row.
   *
   * It exists because the VCS axis was catching this BY ACCIDENT and is about to
   * stop. A merged ref is frozen, so it reported an ever-growing "no activity"
   * age, and that false stall was the only thing surfacing a real routing
   * failure: a lane sitting In Progress with nothing routed to it. Making the
   * stall honest would have silently removed the signal, so the signal gets its
   * own name instead of inheriting a wrong one.
   */
  | { kind: "stale-row"; agentId: string; branch: string; why: string }
  /**
   * A 🔍 row that has DECLARED itself in review for longer than the review
   * window. Measured from the board's own git history — the commit that first
   * (contiguously) put this row in review — never from the branch, which is
   * frozen for the correct reason while a gate reads it.
   */
  | { kind: "in-review-too-long"; agentId: string; stream: string; minutes: number; since: string }
  /**
   * ⟨q-6f0a3d81⟩ — A REPOSITORY FACT, NOT A BUS EMISSION. An open PR whose head
   * sha has no verdict record in any room log, older than the stall limit. It
   * needs no board row, no DONE:, no heartbeat: it is read from `gh` and the
   * log, so it fires when every seat stays silent — which is the one case every
   * other signal here is blind to by construction.
   */
  /**
   * ⟨q-dcbaf544⟩ — `why` splits the one predicate (open PR, no typed verdict at
   * its head) into its two causes: NOBODY gated it, or someone whose role cannot
   * emit `verdict` reported a typed gate line for this head and no gate-runner
   * scribed it. The second fires without waiting: the claim exists, the record
   * does not, and every minute it stands the audit trail is wrong, not late.
   */
  | { kind: "unverdicted-pr"; agentId: string; pr: number; head: string; branch: string; minutes: number; why: "nobody-gated" | "reported-unscribed"; claimedBy?: string; claimedAt?: string; claimedResult?: string }
  /**
   * ⟨q-8f1e604b⟩ — CONVENTION (e) AS A CHECK: a PR merged within the window whose
   * page carries NO verdict comment bound to the head that merged. #299 is the
   * live instance (David's own merge, no verdict). A comment that merely says
   * PASS or FAIL in prose is not a verdict — the aide measured that a substring
   * scan would have counted #299's "not a verdict" review.
   */
  | { kind: "unverdicted-merge"; pr: number; head: string; mergedAt: string; minutes: number; why: string }
  /** ⟨q-fee7239f⟩ — a merge closing on the bus asserting a deletion in neither accepted form, after the floor. The send path refuses new ones; this lists what got through elsewhere. */
  | { kind: "unread-delete-claim"; agentId: string; pr: number | null; closedAt: string; minutes: number; why: string }
  /**
   * ⟨q-8f1e604b⟩ — CONVENTION (f) AS A CHECK: an item ROUTED by a GO on the bus
   * (named on the GO line, not merely mentioned) that is still open in the queue
   * and has no board row naming it as its subject. `next_unblocked` can offer
   * such an item again; the coordinator's cold free-set derivation nearly did.
   */
  | { kind: "routed-without-row"; itemId: string; routedBy: string; routedAt: string; minutes: number; why: string }
  /**
   * ⟨q-4e08b3c1⟩ — CONVENTION (g) AS A CHECK: a commit touching a record
   * document, authored INSIDE an open MERGE WINDOW. The window is read from
   * the bus (qa's typed lines), the write from git — the one convention whose
   * data lives in two places at once, which is why it had no check.
   */
  | { kind: "merge-window-write"; pr: number; sha: string; author: string; authoredAt: string; secondsIntoWindow: number; why: string }
  /**
   * A pushed branch ahead of main with no open PR, older than the limit. Same
   * class, one step earlier. ⚠ STRANDED IS NOT LOST: the first two live hits
   * were superseded drafts of a file that landed under other PRs (measured by
   * the aide, same seven tests on all three). The axis can see that nobody
   * cited the artefact; only its owner can say whether it is superseded,
   * abandoned or genuinely unlanded — so the hit SAYS so, or a true alarm
   * with a false implication teaches the reader to ignore the axis.
   */
  | { kind: "unproposed-branch"; agentId: string; branch: string; head: string; ahead: number; minutes: number; note: string }
  /**
   * ⟨q-11257590⟩ — David's ruling: coordinator absence is a FAULT to be surfaced,
   * never a state to be silently covered by a stand-in. `agentId` is read from the
   * board's own Rooms table (`boardOwnerOf`), the same name every seat is told at
   * `join` — never hardcoded, so a renamed coordinator seat is still found. Fires
   * when the seat the board itself names has no registry entry at all, or has one
   * but no currently-live transport marker (`loadLiveTransports`, which already
   * verifies pid/pane/remote-heartbeat liveness for every transport kind — this
   * reuses that verdict rather than re-deriving it). A present-and-active
   * coordinator produces NO hit: the negative control this row's acceptance requires.
   */
  | { kind: "coordinator-absent"; agentId: string; why: string };

/* ────────────────────────────────────────────────────────────────────────────
 * ⟨q-6f0a3d81⟩ — THE HANDOVER IS THE THING BEING MEASURED, AND EVERY INSTRUMENT
 * KEYED ON AN ARTEFACT PRODUCED *BY* A HANDOVER. A cited DONE: proves work
 * ARRIVED; its absence proves nothing. Measured 2026-09-11: #244 sat open,
 * green and never cited, and stall_check (lane rows), QA's queue (cited
 * arrivals) and the board (a row nobody updated) each said "nothing wrong" —
 * each correctly. Three instruments, one blind spot, because all three begin
 * "when someone reports…".
 *
 * So this axis reads the ARTEFACT: what `gh` says is open, what origin says is
 * pushed, and whether the bus holds a verdict bound to that head. No seat has
 * to speak for it to fire. Injected, like `rotate`'s facts and `merge`'s PR
 * facts, so the negative control the item demands — fires with every seat
 * silent — is provable offline.
 *
 * ⛔ AND "COULD NOT READ" IS NEVER "NOTHING STRANDED": an unreachable `gh` or
 * origin lands on `unmeasurable`, named, exactly as an unreadable branch does.
 * ──────────────────────────────────────────────────────────────────────────── */
export type OpenPrFact = { n: number; headRefOid: string; headRefName: string; updatedAt: string };
export type RepoArtefacts = {
  /** `gh pr list --state open`, or null when gh could not answer. */
  openPrs: OpenPrFact[] | null;
  /** `git ls-remote --heads origin`, or null when origin could not be read. */
  remoteHeads: { name: string; sha: string }[] | null;
  /**
   * `gh pr list --state merged` — the FORGE's record of what landed, rank 1 in
   * docs/LANDEDNESS.md. A branch whose tip is a merged PR's head is landed,
   * whatever ancestry or patch ids say. Null when gh could not answer.
   */
  mergedPrs?: { headRefName: string; headRefOid: string }[] | null;
  /**
   * ⟨q-8f1e604b⟩ — the RECENT merges with their PR comments, so convention (e)
   * can be checked: `gh pr list --state merged --limit 50 --json number,headRefOid,mergedAt,comments`.
   * Null when gh could not answer.
   */
  recentMerges?: RecentMerge[] | null;
};
export type RecentMerge = { n: number; headRefOid: string; mergedAt: string; comments: { body: string }[]; files?: string[] };

/*
 * ⛔ AN EXPRESS-BY-DESIGN MERGE WITH NO VERDICT IS NOT A DEFECT — ruled by the
 * coordinator 2026-09-15 after this predicate reported `#966` as a stall. `#966` is
 * ONE FILE, `+3/-3`, on `docs/WORKSTREAMS.md`, and David's standing rule is that ANY
 * `.md` change goes direct to main: no PR, no branch, no CI. It legitimately required
 * no verdict, so reporting it trained the room to ignore the check.
 *
 * ⭐ THE EXEMPTION IS MEASURED FROM THE MERGE ITSELF, NOT FROM A LABEL SOMEBODY HAS TO
 * REMEMBER TO SET. A docs-only file set is a fact `gh` already returns; a lane marker
 * is a convention, and this row exists because a convention with zero adoption silently
 * broke the instrument that depended on it.
 *
 * ⚠ AND IT COVERS ONLY THE DERIVABLE HALF, WHICH IS SAID RATHER THAN IMPLIED: a SMALL
 * CODE PR merged express is also exempt by policy and is NOT visible in the file set.
 * That needs a real lane input from the caller — `expressPrs` below — which nothing
 * populates yet. So: docs-only is closed here; express-code remains open, and a caller
 * that knows the lane can pass it without touching this file.
 */
export const isDocsOnlyMerge = (files: string[] | undefined): boolean =>
  Array.isArray(files) && files.length > 0 && files.every((f) => /\.mdx?$/i.test(String(f)));
const NET_TIMEOUT_MS = 15_000;
export const realArtefacts = (repo: string): RepoArtefacts => {
  let openPrs: OpenPrFact[] | null = null;
  try {
    const out = execFileSync("gh", ["pr", "list", "--state", "open", "--json", "number,headRefOid,headRefName,updatedAt"], {
      cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], timeout: NET_TIMEOUT_MS,
    });
    openPrs = (JSON.parse(out) as { number: number; headRefOid: string; headRefName: string; updatedAt: string }[]).map((p) => ({
      n: p.number, headRefOid: p.headRefOid, headRefName: p.headRefName, updatedAt: p.updatedAt,
    }));
  } catch { /* reported as unmeasurable by the caller */ }
  let mergedPrs: { headRefName: string; headRefOid: string }[] | null = null;
  try {
    const out = execFileSync("gh", ["pr", "list", "--state", "merged", "--limit", "1000", "--json", "headRefName,headRefOid"], {
      cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], timeout: NET_TIMEOUT_MS,
    });
    mergedPrs = JSON.parse(out) as { headRefName: string; headRefOid: string }[];
  } catch { /* landedness falls back to patch ids; said so by the caller */ }
  let recentMerges: RecentMerge[] | null = null;
  try {
    const out = execFileSync("gh", ["pr", "list", "--state", "merged", "--limit", "50", "--json", "number,headRefOid,mergedAt,comments,files"], {
      cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], timeout: NET_TIMEOUT_MS, maxBuffer: 64 * 1024 * 1024,
    });
    recentMerges = (JSON.parse(out) as { number: number; headRefOid: string; mergedAt: string; comments?: { body?: string }[]; files?: { path?: string }[] }[]).map((p) => ({
      n: p.number, headRefOid: p.headRefOid, mergedAt: p.mergedAt, comments: (p.comments ?? []).map((c) => ({ body: String(c.body ?? "") })),
      files: (p.files ?? []).map((f) => String(f.path ?? "")).filter(Boolean),
    }));
  } catch { /* said by the caller */ }
  let remoteHeads: { name: string; sha: string }[] | null = null;
  try {
    const out = execFileSync("git", ["ls-remote", "--heads", "origin"], {
      cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], timeout: NET_TIMEOUT_MS,
    });
    remoteHeads = out.split("\n").filter(Boolean).map((l) => {
      const [sha, ref] = l.split(/\s+/);
      return { sha: sha ?? "", name: (ref ?? "").replace(/^refs\/heads\//, "") };
    }).filter((h) => h.sha && h.name);
  } catch { /* reported as unmeasurable by the caller */ }
  return { openPrs, remoteHeads, mergedPrs, recentMerges };
};
/** Every inbox on this bus, concatenated — a GO delivered by DM is a routing record too. */
export const readAllInboxLogs = (): string => {
  const dir = path.join(ROOT, "inbox");
  try {
    return readdirSync(dir).filter((f) => f.endsWith(".jsonl")).map((f) => readFileSync(path.join(dir, f), "utf8")).join("\n");
  } catch {
    return "";
  }
};

/* ────────────────────────────────────────────────────────────────────────────
 * ⟨q-8f1e604b⟩ — TWO RECORD-KEEPING CONVENTIONS BECOME CHECKS. Both were
 * adopted on 2026-09-14 by seats that were cycled the same afternoon; a
 * convention does not survive its holder, a check does.
 *
 * (e) VERDICTS ON THE PR. A merged PR must carry a verdict COMMENT bound to the
 *     head that merged. "Verdict" is a TYPED line — `QA GATE — **PASS** @ \`sha\``
 *     or the merge-time `**GATE: PASS** … @ \`sha\`` — never a substring: #299's
 *     page says "not a verdict", "PASS" and "my verdict above" in three review
 *     comments and carries no verdict at all. Bound by sha, prefix-tolerant
 *     (`shaAgrees`): a verdict on an OLDER head is the #268 breach, not a gate.
 * (f) A GO WRITES THE BOARD ROW. An item is ROUTED when a `go` record names it
 *     ON THE GO LINE — `GO ⟨q-…⟩ —`, `GO: … ⟨q-…⟩` — read by POSITION, because a
 *     GO's prose mentions neighbouring items as context. A routed item that is
 *     still open and has no board row naming it as its SUBJECT is a hit:
 *     `next_unblocked` would offer it again. Room logs AND inboxes are read —
 *     this fleet routes by DM.
 * Both are bounded by a window: a merge or a GO older than it is listed, not
 * raised, so the day's real instances (#299) fire once and age out.
 * ──────────────────────────────────────────────────────────────────────────── */
export const CONVENTION_WINDOW_MS = 3 * 24 * 60 * 60 * 1000;
/**
 * WHEN (e) WAS ADOPTED, as a dated fact: qa posted at 2026-09-14 13:28Z that all
 * five of its verdicts were now on their PRs. Measured live before this floor
 * existed: 39 of the 50 most recent merges had no verdict comment — every one
 * merged BEFORE that instant, when the convention did not exist. Raising them
 * would make the check's first three days pure noise, which is how a check gets
 * switched off. Merges before the floor are LISTED with `beforeAdoption`, never
 * raised; the window subsumes the floor after three days.
 */
export const CONVENTION_E_ADOPTED_MS = Date.parse("2026-09-14T13:28:00Z");
// The verdict grammar lives with the gate predicate now (⟨q-5a93c2d7⟩); re-exported so nothing that reached it here breaks.
export { VERDICT_COMMENT, verdictShasIn, verdictClaimsIn, verdictBoundTo } from "../gated-head.js";
/** Items named ON A GO LINE of a `go` record — position, not mention. */
export function routedItemsIn(logText: string): { itemId: string; by: string; ts: number }[] {
  const out: { itemId: string; by: string; ts: number }[] = [];
  for (const l of logText.split("\n")) {
    if (!l.trim()) continue;
    let o: { ts?: number; from?: string; text?: string; record?: { type?: string } };
    try { o = JSON.parse(l); } catch { continue; }
    if (o.record?.type !== "go") continue;
    for (const line of String(o.text ?? "").split("\n")) {
      if (!/\bGO\b/.test(line)) continue;
      for (const m of line.matchAll(/q-[0-9a-f]{8}/g)) out.push({ itemId: m[0], by: String(o.from ?? ""), ts: Number(o.ts ?? 0) });
    }
  }
  return out;
}
/* ────────────────────────────────────────────────────────────────────────────
 * ⟨q-4e08b3c1⟩ — THE MERGE WINDOW, JOINED. qa posts `MERGE WINDOW: #N — hold
 * queue/board writes` as the trailing line of its typed PASS and `MERGE WINDOW
 * CLOSED` as the trailing line of its MERGED done; a seat's write is a commit
 * touching docs/QUEUE.md, docs/DONE.md or docs/WORKSTREAMS.md, timestamped by
 * git. The incident (#312): two coordinator board commits at 16:34 and 16:37,
 * before a window opened at ~16:40 — nobody violated anything, the protocol's
 * party list was incomplete. `217a6af` bound every record writer; that is the
 * dated floor. Measured today before this existed: five windows, each under
 * 70 seconds, 67 record commits, none inside a window.
 *
 * ⚠ AN UNCLOSED WINDOW IS CAPPED, NEVER OPEN-ENDED (the ruling): at the next
 * window's open or 30 minutes, whichever comes first, and reported `unclosed`
 * by name — a window nobody closed must not condemn every write after it.
 * ──────────────────────────────────────────────────────────────────────────── */
export const MERGE_WINDOW_ADOPTED_MS = Date.parse("2026-09-14T16:43:54Z");
export const MERGE_WINDOW_CAP_MS = 30 * 60 * 1000;
export const RECORD_DOCS = ["docs/QUEUE.md", "docs/DONE.md", "docs/WORKSTREAMS.md"];
export type MergeWindow = { pr: number; open: number; close: number | null; end: number; unclosed: boolean; openedBy: string };
/** Windows as the bus records them: an OPEN line and, for the same PR, the next CLOSED line after it. */
export function windowsIn(logText: string): MergeWindow[] {
  const opens: { pr: number; ts: number; from: string }[] = [];
  const closes: { pr: number | null; ts: number }[] = [];
  for (const l of logText.split("\n")) {
    if (!l.trim()) continue;
    let o: { ts?: number; from?: string; text?: string };
    try { o = JSON.parse(l); } catch { continue; }
    const t = String(o.text ?? "");
    const om = /MERGE WINDOW: #(\d+)/.exec(t);
    if (om) opens.push({ pr: Number(om[1]), ts: Number(o.ts ?? 0), from: String(o.from ?? "") });
    if (/MERGE WINDOW CLOSED/.test(t)) {
      // The PR is named wherever the message names it (`DONE: MERGED owner/repo#N … MERGE WINDOW CLOSED`); a close naming no PR closes the open window.
      const pm = /#(\d+)/.exec(t);
      closes.push({ pr: pm ? Number(pm[1]) : null, ts: Number(o.ts ?? 0) });
    }
  }
  opens.sort((a, b) => a.ts - b.ts);
  return opens.map((w, i) => {
    const close = closes.filter((c) => c.ts > w.ts && (c.pr === null || c.pr === w.pr)).sort((a, b) => a.ts - b.ts)[0]?.ts ?? null;
    const nextOpen = opens[i + 1]?.ts ?? Infinity;
    const end = close ?? Math.min(nextOpen, w.ts + MERGE_WINDOW_CAP_MS);
    return { pr: w.pr, open: w.ts, close, end, unclosed: close === null, openedBy: w.from };
  });
}
/** `mergeOf`: the PR whose landing commit this is, read off the FULL subject's trailing `(#N)` before any display slicing — the live #325 subject is 119 chars and the marker sat past an 80-char cut. */
export type RecordWrite = { sha: string; at: number; author: string; subject: string; mergeOf: number | null };
/** The PR a landing commit lands, from the forge's `(#N)` marker at the END of the subject; null for any other commit. */
export const mergeOf = (fullSubject: string): number | null => { const m = /\(#(\d+)\)\s*$/.exec(fullSubject); return m ? Number(m[1]) : null; };
/**
 * ⟨q-4e08b3c1⟩ follow-up, MEASURED LIVE 2026-09-14 19:38Z by this file's own
 * live-population test: #325's SQUASH (3b8702e) touched docs/QUEUE.md — the PR
 * migrated a queue receipt — and its author time is the merge time, 13s into
 * MERGE WINDOW #325. The window's own landing commit is the MERGE the window
 * exists to protect, not a seat's write into it; it is listed under the window
 * with `landing: true` and never raised.
 */
export const isLandingOf = (write: Pick<RecordWrite, "mergeOf">, pr: number): boolean => write.mergeOf !== null && write.mergeOf === pr;
/** Commits touching a record document on the shared branch, by AUTHOR time — when the seat wrote, not when it landed. */
export function recordWritesOn(repo: string, sinceMs: number): RecordWrite[] | null {
  const git = (args: string[]) => execFileSync("git", args, { cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"] }).trim();
  let base: string | null = null;
  for (const cand of ["origin/main", "main", "origin/master", "master"]) {
    try { git(["rev-parse", "--verify", "--quiet", `${cand}^{commit}`]); base = cand; break; } catch { /* next */ }
  }
  if (!base) return null;
  try {
    const out = git(["log", base, `--since=${new Date(sinceMs).toISOString()}`, "--format=%H|%aI|%an|%s", "--", ...RECORD_DOCS]);
    return out.split("\n").filter(Boolean).map((l) => {
      const [sha, aI, author, ...rest] = l.split("|");
      const full = rest.join("|");
      return { sha: sha ?? "", at: Date.parse(aI ?? ""), author: author ?? "", subject: full.slice(0, 120), mergeOf: mergeOf(full) };
    });
  } catch {
    return null;
  }
}
export function mergeWindowChecks(input: { logText: string; writes: RecordWrite[] | null; now: number; floorMs?: number; horizonMs?: number }): {
  hits: StallHit[];
  windows: (MergeWindow & { writes: (RecordWrite & { beforeAdoption: boolean })[] })[];
  unmeasurable: { agentId: string; value: string; why: string }[];
} {
  const floor = input.floorMs ?? MERGE_WINDOW_ADOPTED_MS;
  const horizon = input.horizonMs ?? CONVENTION_WINDOW_MS;
  const hits: StallHitBody[] = [];
  const unmeasurable: { agentId: string; value: string; why: string }[] = [];
  const windows = windowsIn(input.logText).filter((w) => input.now - w.open <= horizon);
  if (input.writes === null) {
    unmeasurable.push({ agentId: "-", value: "git log -- docs/{QUEUE,DONE,WORKSTREAMS}.md", why: "record-document writes could not be read from the shared branch — convention (g), the merge window, is UNMEASURED, which is not the same as kept" });
    return { hits: [], windows: windows.map((w) => ({ ...w, writes: [] })), unmeasurable };
  }
  const out = windows.map((w) => {
    const inside = input.writes!.filter((c) => Number.isFinite(c.at) && c.at >= w.open && c.at <= w.end).map((c) => ({ ...c, beforeAdoption: c.at < floor, landing: isLandingOf(c, w.pr) }));
    for (const c of inside) {
      if (c.beforeAdoption || c.landing) continue;
      const secs = Math.round((c.at - w.open) / 1000);
      hits.push({
        kind: "merge-window-write", pr: w.pr, sha: c.sha, author: c.author, authoredAt: new Date(c.at).toISOString(), secondsIntoWindow: secs,
        why: `${c.sha.slice(0, 7)} (${c.author}) wrote a record document ${secs}s into MERGE WINDOW #${w.pr}${w.unclosed ? " (window never closed; capped)" : ""}: "${c.subject}". Convention (g): queue/board/DONE writes pause between MERGE WINDOW and MERGE WINDOW CLOSED.`,
      });
    }
    return { ...w, writes: inside };
  });
  return { hits: withAudience(hits), windows: out, unmeasurable };
}

/**
 * ⟨q-caa2959e⟩ — WHETHER THE MERGE-WINDOW CONVENTION CAN BE JUDGED ON THIS BUS RIGHT NOW,
 * decided from OWN LANDINGS IN THE HORIZON and never from the window count.
 *
 * The live check used to assert `windows.length > 0`. A stand-down left the fleet quiet: 0 window
 * announcements inside the horizon while record commits remained, so a control pinned to a live
 * baseline failed for everyone — it expired when the fleet went quiet, not when anything broke.
 *
 * ⛔ AND APPLICABILITY IS NEVER READ OFF THE PREDICATE'S OWN COUNT. Deciding it from `windows.length`
 * is the by-mention trap ⟨q-552b9912⟩ taught us, inverted: a fleet that landed PRs and announced NO
 * window — precisely the violation this check exists to catch — would skip itself green. The
 * population is what the fleet DID (its landings); the windows are what it is judged ON.
 *
 * Applicable when the horizon holds at least one landing of this fleet's (a squash carrying the
 * forge's `(#N)` marker). Then every one of them MUST carry a window announcement, and a missing one
 * is a problem. With no landing there is nothing to judge: NOT APPLICABLE, named with both counts.
 */
export function mergeWindowLivePopulation(input: { logText: string; writes: RecordWrite[] | null; now: number; floorMs?: number; horizonMs?: number }): {
  applicable: boolean;
  reason: string;
  population: string;
  landings: RecordWrite[];
  missingWindow: { sha: string; pr: number }[];
  windows: ReturnType<typeof mergeWindowChecks>["windows"];
  hits: ReturnType<typeof mergeWindowChecks>["hits"];
} {
  const floor = input.floorMs ?? MERGE_WINDOW_ADOPTED_MS;
  const horizon = input.horizonMs ?? CONVENTION_WINDOW_MS;
  const r = mergeWindowChecks(input);
  const since = new Date(input.now - horizon).toISOString();
  const head = `horizon ${since} → ${new Date(input.now).toISOString()}, floor ${new Date(floor).toISOString()}`;
  if (input.writes === null) {
    const reason = `NOT APPLICABLE: record-document writes could not be read from the shared branch — ${head}. UNMEASURED is not the same as kept.`;
    return { applicable: false, reason, population: reason, landings: [], missingWindow: [], windows: r.windows, hits: r.hits };
  }
  const inHorizon = input.writes.filter((c) => Number.isFinite(c.at) && c.at >= input.now - horizon);
  const landings = inHorizon.filter((c) => c.mergeOf !== null);
  const population = `${head}: ${inHorizon.length} record commit(s), ${landings.length} own landing(s), ${r.windows.length} window(s), ${r.hits.length} hit(s)`;
  if (landings.length === 0) {
    return {
      applicable: false,
      reason: `NOT APPLICABLE: this fleet landed nothing in the horizon, so there is no window to require — ${population}. (A quiet fleet, not a broken convention; the window count is never what decides this.)`,
      population, landings, missingWindow: [], windows: r.windows, hits: r.hits,
    };
  }
  const missingWindow = landings
    .filter((c) => !r.windows.some((w) => w.pr === c.mergeOf))
    .map((c) => ({ sha: c.sha, pr: c.mergeOf as number }));
  return { applicable: true, reason: `APPLICABLE — ${population}`, population, landings, missingWindow, windows: r.windows, hits: r.hits };
}

export function conventionChecks(input: {
  recentMerges: RecentMerge[] | null | undefined;
  /** PRs the caller knows merged express. Nothing populates this yet — see isDocsOnlyMerge. */
  expressPrs?: number[];
  routingLogText: string;
  queueText: string | null;
  boardRows: WorkstreamsV1Row[];
  now: number;
  windowMs?: number;
}): {
  hits: StallHit[];
  unmeasurable: { agentId: string; value: string; why: string }[];
  merges: { pr: number; head: string; minutes: number; verdictAtHead: boolean; inWindow: boolean; beforeAdoption: boolean; exempt?: string }[] | null;
  routed: { itemId: string; by: string; minutes: number; open: boolean | null; onBoard: boolean; inWindow: boolean }[];
  closings: (Closing & { inWindow: boolean })[];
} {
  const windowMs = input.windowMs ?? CONVENTION_WINDOW_MS;
  const hits: StallHitBody[] = [];
  const unmeasurable: { agentId: string; value: string; why: string }[] = [];
  let merges: ReturnType<typeof conventionChecks>["merges"] = null;
  const expressPrs = new Set(input.expressPrs ?? []);
  if (input.recentMerges === null || input.recentMerges === undefined) {
    unmeasurable.push({ agentId: "-", value: "gh pr list --state merged (with comments)", why: "recent merges could not be read from gh — convention (e), verdicts on the PR, is UNMEASURED, which is not the same as kept" });
  } else {
    merges = [];
    for (const m of input.recentMerges) {
      const mergedMs = Date.parse(m.mergedAt);
      const minutes = Math.max(0, Math.round((input.now - mergedMs) / 60000));
      const inWindow = input.now - mergedMs <= windowMs;
      const beforeAdoption = mergedMs < CONVENTION_E_ADOPTED_MS;
      // ⛔ KEYED ON THE CLAIM, NOT ON A HEADER PHRASE. This used to test `verdictShasIn`,
      // i.e. #313's `QA GATE — **PASS** @ sha` wording, which had ZERO adoption: on the 9
      // merges of 2026-09-15 it matched 0 lines while all 9 carried a verdict, so this hit
      // fired 9 times on 9 correct merges. `verdictBoundTo` asks the binding question
      // instead — a disposition, on one line, at a sha equal to the head that merged — and
      // accepts the split halves (#964's CONTENT PASS and READINESS RELEASED were two
      // separate comments).
      const bound = verdictBoundTo(m.comments, m.headRefOid);
      const verdictAtHead = bound.bound;
      // ⛔ EXEMPT BEFORE REPORTING, NEVER AFTER. A docs-only merge (or one the caller
      // names as express) owed no verdict, so it is not a hit — it is not a defect
      // whose report we suppress. `exempt` is carried on the merge row so a reader can
      // see the population and why a member left it.
      const exemptWhy = isDocsOnlyMerge(m.files) ? "docs-only" : expressPrs.has(m.n) ? "express lane (caller-supplied)" : null;
      merges.push({ pr: m.n, head: m.headRefOid, minutes, verdictAtHead, inWindow, beforeAdoption, ...(exemptWhy ? { exempt: exemptWhy } : {}) });
      if (inWindow && !beforeAdoption && !verdictAtHead && !exemptWhy) {
        hits.push({
          kind: "unverdicted-merge", pr: m.n, head: m.headRefOid, mergedAt: m.mergedAt, minutes,
          why: `#${m.n} merged at ${m.headRefOid.slice(0, 7)} with no verdict comment bound to that head on the PR — ${bound.why} (${m.comments.length} comment(s)). Convention (e): the verdict lives on the PR, not only on the bus. ⚠ This predicate does NOT know about the express lane, so an express merge (no QA verdict by design) still reports here — see the kit row on #966.`,
        });
      }
    }
  }
  const openIds = new Set<string>();
  const closedIds = new Set<string>();
  if (input.queueText !== null) {
    for (const i of queueItemsOf(parseWorkDoc(input.queueText))) (i.done ? closedIds : openIds).add(i.id);
  }
  const subjectOf = (r: WorkstreamsV1Row) => [...String(r.stream).matchAll(/\b(q-[0-9a-f]{8})\b/g)].map((x) => x[1] as string);
  const onBoard = new Set(input.boardRows.flatMap(subjectOf));
  const routed: ReturnType<typeof conventionChecks>["routed"] = [];
  const seen = new Set<string>();
  for (const r of routedItemsIn(input.routingLogText).sort((a, b) => b.ts - a.ts)) {
    if (seen.has(r.itemId)) continue; // the latest GO for an item is the one that binds
    seen.add(r.itemId);
    const minutes = Math.max(0, Math.round((input.now - r.ts) / 60000));
    const inWindow = input.now - r.ts <= windowMs;
    const open = input.queueText === null ? null : openIds.has(r.itemId) ? true : closedIds.has(r.itemId) ? false : null;
    const has = onBoard.has(r.itemId);
    routed.push({ itemId: r.itemId, by: r.by, minutes, open, onBoard: has, inWindow });
    if (inWindow && open === true && !has) {
      hits.push({
        kind: "routed-without-row", itemId: r.itemId, routedBy: r.by, routedAt: new Date(r.ts).toISOString(), minutes,
        why: `⟨${r.itemId}⟩ was routed by a GO from ${r.by} ${minutes}m ago, is still open in the queue, and no board row names it as its subject — next_unblocked can offer it again. Convention (f): a GO writes the board row (claim does it for you).`,
      });
    }
  }
  // ⟨q-fee7239f⟩ — closing lines: every MERGED done on the bus, judged by the
  // grammar; those at or before the floor are LISTED with beforeAdoption.
  const closings = closingsIn(input.routingLogText).map((c) => ({ ...c, inWindow: input.now - c.ts <= windowMs }));
  for (const c of closings) {
    if (!c.inWindow || c.beforeAdoption || !c.claimsDelete || c.form !== null) continue;
    const minutes = Math.max(0, Math.round((input.now - c.ts) / 60000));
    hits.push({
      kind: "unread-delete-claim", agentId: c.from, pr: c.pr, closedAt: new Date(c.ts).toISOString(), minutes,
      why: `${c.from}'s closing line for ${c.pr === null ? "a merge" : `#${c.pr}`} asserts a branch deletion in neither accepted form — ${ACCEPTED_FORMS}. The remote was not read; the record may be premature or false.`,
    });
  }
  return { hits: withAudience(hits), unmeasurable, merges, routed, closings };
}
/** Every room log on this bus, concatenated — a verdict for a PR counts from any room. */
export const readAllRoomLogs = (): string => {
  const dir = path.join(ROOT, "rooms");
  try {
    return readdirSync(dir).filter((f) => f.endsWith(".jsonl")).map((f) => readFileSync(path.join(dir, f), "utf8")).join("\n");
  } catch {
    return "";
  }
};
/** The seat a branch belongs to, by the prefix `claim` writes — identity is in the artefact, not the account. */
const seatOfBranch = (name: string): string => (name.includes("/") ? name.split("/")[0]! : "unknown");
const BASE_BRANCHES = new Set(["main", "master", "HEAD"]);

/*
 * ⚠ A CLOCK VERB CANNOT TAKE A MINUTE. Measured live: 212 remote heads × ~5 git
 * processes each = 65s per run. Landedness is MONOTONE — once a sha's patches
 * are on main they stay there — so it is memoised per sha across runs, and a
 * not-landed answer is memoised against the base sha it was measured at and
 * re-measured when main moves. A per-run BUDGET bounds the uncached work; what
 * does not fit is reported as deferred, never silently skipped, and the next
 * run picks it up. Steady state is ls-remote + a handful of new branches.
 */
type LandedBy = "forge" | "patch" | "squash" | null;
type StrandedMemo = Record<string, { landed: boolean; landedBy: LandedBy; ahead: number; committedAt: string; base: string }>;
const memoFile = () => path.join(ROOT, "stall-stranded-memo.json");
const readMemo = (): StrandedMemo => {
  try {
    return JSON.parse(readFileSync(memoFile(), "utf8")) as StrandedMemo;
  } catch {
    return {};
  }
};
const writeMemo = (m: StrandedMemo) => {
  try {
    mkdirSync(ROOT, { recursive: true });
    writeFileSync(memoFile(), JSON.stringify(m));
  } catch { /* a lost memo costs time on the next run, never correctness */ }
};
export const STRANDED_BUDGET_MS = 15_000;

/**
 * Did the WHOLE RANGE land as ONE commit? `git cherry` compares commit by
 * commit, so a squash of several commits — one patch id for the range, none
 * for the parts — reads `+` on every line and the branch looks unproposed
 * forever. Measured live: 47 of 209 ahead-by-ancestry branches survived the
 * cherry test, and every young one was a 2-commit branch merged that day.
 *
 * So index main's recent commits by the patch id of EACH commit — ONE process
 * (`git log -p | git patch-id`), measured at 1.8s for 1000 commits — and ask,
 * per branch, whether the patch id of `mergeBase..sha` as ONE diff is in it.
 * A hit means main carries this branch's cumulative change as a single
 * commit: a squash. Bounded to the last 1000 first-parent commits, so a
 * branch that landed further back than that reads as not landed — the
 * direction that over-reports, never the one that hides stranded work.
 */
const SQUASH_INDEX_DEPTH = 1000;
/** `git patch-id --stable` over a diff, without a shell: the inputs are shas and refs from origin, and data is not a command line. */
const patchIds = (repo: string, diff: string): string =>
  execFileSync("git", ["patch-id", "--stable"], { cwd: repo, encoding: "utf8", input: diff, stdio: ["pipe", "pipe", "ignore"], maxBuffer: 64 * 1024 * 1024 });
function squashIndex(repo: string, base: string): Map<string, string> {
  const map = new Map<string, string>();
  try {
    const log = execFileSync("git", ["log", "-p", "--first-parent", `--max-count=${SQUASH_INDEX_DEPTH}`, "--format=%H", base, "--"], {
      cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], maxBuffer: 256 * 1024 * 1024,
    });
    const out = patchIds(repo, log);
    for (const l of out.split("\n")) {
      const [id, commit] = l.trim().split(/\s+/);
      if (id && commit) map.set(id, commit);
    }
  } catch { /* an empty index reads every branch as not landed — over-reports */ }
  return map;
}
function landedBySquash(repo: string, base: string, sha: string, index: Map<string, string>): boolean {
  if (index.size === 0) return false;
  try {
    const git = (args: string[]) =>
      execFileSync("git", args, { cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"], maxBuffer: 64 * 1024 * 1024 });
    const mb = git(["merge-base", base, sha]).trim();
    if (!mb) return false;
    const id = patchIds(repo, git(["diff", mb, sha, "--"])).trim().split(/\s+/)[0] ?? "";
    return id.length > 0 && index.has(id);
  } catch {
    return false;
  }
}

/**
 * The stranded-work axis. Pure over its inputs so it is testable without a
 * network; the caller supplies the artefacts and the log.
 */
/*
 * ⟨q-dcbaf544⟩ — A GATE RESULT REPORTED BY A SEAT THAT CANNOT RECORD IT. The
 * typed gate line (#313's grammar, `GATE: PASS @ sha`) inside a NON-verdict
 * record that cites the PR — a repo-owner's or console's `done`/`fyi` — is a
 * claim that the head was gated. If no typed `verdict` names that head, the
 * claim was never scribed, and `unverdicted-pr` says so instead of "nobody".
 * Measured live 2026-09-14 before this existed: 0 such claims in 14 rooms +
 * inboxes — the convention was applied by prose, which this cannot read, and
 * that is the point: the typed line is what makes a report scribable.
 */
export type GateClaim = { from: string; ts: number; result: string; sha: string; type: string };
export function gateClaimsIn(logText: string, pr: string, head: string): GateClaim[] {
  const out: GateClaim[] = [];
  for (const l of logText.split("\n")) {
    if (!l.trim()) continue;
    let o: { ts?: number; from?: string; text?: string; record?: { type?: string; cites?: { ref?: string }[] } };
    try { o = JSON.parse(l); } catch { continue; }
    const r = o.record;
    if (!r || r.type === "verdict") continue;
    if (!(r.cites ?? []).some((c) => (String(c?.ref ?? "").match(/#(\d+)/) ?? [])[1] === pr)) continue;
    for (const v of verdictShasIn([{ body: String(o.text ?? "") }])) {
      if (shaAgrees(v.sha, head)) out.push({ from: String(o.from ?? ""), ts: Number(o.ts ?? 0), result: v.result, sha: v.sha, type: String(r.type ?? "") });
    }
  }
  return out.sort((a, b) => a.ts - b.ts);
}

export function strandedWork(
  repo: string,
  artefacts: RepoArtefacts,
  logText: string,
  now: number,
  limitMs: number,
  /** Seats the bus knows. A branch hit needs a recipient; a branch nobody owns is reported, not DMed. */
  registeredSeats: ReadonlySet<string> = new Set(),
  /** Uncached branch classification stops here; the rest is DEFERRED and named. */
  budgetMs: number = STRANDED_BUDGET_MS,
): {
  hits: StallHit[];
  unmeasurable: { agentId: string; value: string; why: string }[];
  openPrs: { pr: number; head: string; branch: string; minutes: number; verdictAtHead: boolean; claimAtHead: GateClaim | null }[] | null;
  pushedBranches: { branch: string; head: string; ahead: number; landed: boolean; landedBy: LandedBy; minutes: number; hasPr: boolean }[] | null;
} {
  const hits: StallHitBody[] = [];
  const unmeasurable: { agentId: string; value: string; why: string }[] = [];
  const git = (args: string[]) =>
    execFileSync("git", args, { cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"] }).trim();

  let openPrs: ReturnType<typeof strandedWork>["openPrs"] = null;
  if (artefacts.openPrs === null) {
    unmeasurable.push({ agentId: "-", value: "gh pr list", why: "open PRs could not be read from gh — stranded PRs are UNMEASURED, which is not the same as none" });
  } else {
    openPrs = [];
    for (const p of artefacts.openPrs) {
      const { verdicts } = verdictsFor(logText, String(p.n));
      const verdictAtHead = verdicts.some((v) => shaAgrees(v.head, p.headRefOid));
      const minutes = Math.max(0, Math.round((now - Date.parse(p.updatedAt)) / 60000));
      const claimAtHead = verdictAtHead ? null : gateClaimsIn(logText, String(p.n), p.headRefOid)[0] ?? null;
      openPrs.push({ pr: p.n, head: p.headRefOid, branch: p.headRefName, minutes, verdictAtHead, claimAtHead });
      // A verdict bound to an OLDER head does not count: the head moved, and a
      // PASS on what used to be there is the #268 breach one step earlier.
      // ⟨q-dcbaf544⟩ — an UNSCRIBED gate report fires at once; "nobody gated" waits the limit.
      if (!verdictAtHead && (claimAtHead || now - Date.parse(p.updatedAt) > limitMs)) {
        hits.push({
          kind: "unverdicted-pr", agentId: seatOfBranch(p.headRefName), pr: p.n, head: p.headRefOid, branch: p.headRefName, minutes,
          ...(claimAtHead
            ? { why: "reported-unscribed" as const, claimedBy: claimAtHead.from, claimedAt: new Date(claimAtHead.ts).toISOString(), claimedResult: claimAtHead.result }
            : { why: "nobody-gated" as const }),
        });
      }
    }
  }

  let pushedBranches: ReturnType<typeof strandedWork>["pushedBranches"] = null;
  if (artefacts.remoteHeads === null) {
    unmeasurable.push({ agentId: "-", value: "git ls-remote --heads origin", why: "origin's branches could not be read — pushed-but-unproposed work is UNMEASURED, which is not the same as none" });
  } else {
    pushedBranches = [];
    const prHeads = new Set((artefacts.openPrs ?? []).map((p) => p.headRefOid));
    // RANK 1, docs/LANDEDNESS.md: the forge's own record. Measured live before it
    // was asked: a 15-day-old branch whose PR #170 had merged with this very tip
    // as its head read "not landed" — its squash sat beyond the patch index's
    // window. A patch-id negative is inconclusive by construction; the forge's
    // positive is not.
    const forgeLanded = new Set((artefacts.mergedPrs ?? []).map((p) => p.headRefOid));
    if (artefacts.mergedPrs === null || artefacts.mergedPrs === undefined) {
      unmeasurable.push({ agentId: "-", value: "gh pr list --state merged", why: "the forge's merged-PR record could not be read — landedness falls back to patch ids, whose NEGATIVES are inconclusive in a squash-merging repo (docs/LANDEDNESS.md); a branch reported unproposed here may have landed" });
    }
    const prBranches = new Set((artefacts.openPrs ?? []).map((p) => p.headRefName));
    let base: string | null = null;
    for (const cand of ["origin/main", "main", "origin/master", "master"]) {
      try { git(["rev-parse", "--verify", "--quiet", `${cand}^{commit}`]); base = cand; break; } catch { /* next */ }
    }
    let squashes: Map<string, string> | null = null; // built once per run, only if a branch needs it
    const memo = readMemo();
    let baseSha = "";
    try { baseSha = base ? git(["rev-parse", base]) : ""; } catch { /* handled below as no base */ }
    const started = Date.now();
    let deferred = 0;
    // The budget is spent on the branches that can RAISE something first: a
    // registered seat's branch is the only kind that becomes a hit, and on a
    // 212-head origin the alphabet puts `docs/…` and `fix/…` ahead of every
    // `groundwork-kit-worker-*/…`. Stable within each group.
    const ordered = [...artefacts.remoteHeads].sort(
      (x, y) => Number(registeredSeats.has(seatOfBranch(y.name))) - Number(registeredSeats.has(seatOfBranch(x.name))),
    );
    for (const h of ordered) {
      if (BASE_BRANCHES.has(h.name)) continue;
      const hasPr = prHeads.has(h.sha) || prBranches.has(h.name);
      if (!base) {
        unmeasurable.push({ agentId: seatOfBranch(h.name), value: h.name, why: "no main/master to measure distance from" });
        continue;
      }
      // ⛔ THE FORGE BEFORE THE MEMO — QA's row B (#307 FAIL @ 0f5f334): a branch
      // memoised as not-landed against a base that has not moved stayed a hit
      // AFTER the forge said its PR had merged, because the memo answered first.
      // Live shape: PR merged, local origin/main not yet fetched, owner DMed that
      // merged work is stranded — the instrument mirroring the aide's clearance
      // error. The forge is rank 1 and a set lookup; it is asked first, always,
      // and its answer overwrites the memo.
      const cached = memo[h.sha];
      let ahead: number;
      let landed: boolean;
      let landedBy: LandedBy;
      let committedAt: string;
      if (forgeLanded.has(h.sha) && cached && !cached.landed) {
        ({ ahead, committedAt } = cached);
        landed = true;
        landedBy = "forge";
        memo[h.sha] = { landed, landedBy, ahead, committedAt, base: baseSha };
      } else if (cached && (cached.landed || cached.base === baseSha)) {
        ({ ahead, landed, landedBy, committedAt } = cached);
      } else {
        if (Date.now() - started > budgetMs) {
          deferred++;
          continue;
        }
        try {
          git(["cat-file", "-e", `${h.sha}^{commit}`]);
        } catch {
          unmeasurable.push({ agentId: seatOfBranch(h.name), value: h.name, why: `origin/${h.name} @ ${h.sha.slice(0, 7)} is not fetched here — its age and distance from main cannot be read; not counted as clean` });
          continue;
        }
        ahead = Number(git(["rev-list", "--count", `${base}..${h.sha}`]) || 0);
        committedAt = git(["log", "-1", "--format=%cI", h.sha, "--"]);
      // ⛔⛆ ANCESTRY CANNOT SAY "LANDED" ON A SQUASH-MERGING FLEET. Measured live
      // before this line existed: 205 hits, and every young one was a branch
      // whose PR had merged that afternoon — `rev-list --count main..sha` reads
      // a squash-landed branch as ahead forever. `git cherry` compares PATCH
      // IDS, the same instrument `classifyBoardRef` and `refresh_worktrees` use:
      // every line `-` means every patch is upstream. Same caveat as theirs: a
      // squash of SEVERAL commits into one changes the combined patch id, so a
      // multi-commit landed branch can still read as unproposed — the direction
      // that over-reports, never the one that hides stranded work.
        landed = false;
        landedBy = null;
        if (ahead > 0) {
          if (forgeLanded.has(h.sha)) {
            landed = true;
            landedBy = "forge";
          }
          if (!landed) {
            try {
              const cherry = git(["cherry", base, h.sha]);
              landed = cherry.length > 0 && cherry.split("\n").every((l) => l.trim().startsWith("-"));
              if (landed) landedBy = "patch";
            } catch { /* unreadable: inconclusive, which over-reports */ }
          }
          if (!landed) {
            if (!squashes) squashes = squashIndex(repo, base);
            landed = landedBySquash(repo, base, h.sha, squashes);
            if (landed) landedBy = "squash";
          }
        }
        memo[h.sha] = { landed, landedBy, ahead, committedAt, base: baseSha };
      }
      const minutes = Math.max(0, Math.round((now - Date.parse(committedAt)) / 60000));
      pushedBranches.push({ branch: h.name, head: h.sha, ahead, landed, landedBy, minutes, hasPr });
      // ⚠ A BRANCH HIT IS RAISED ONLY FOR A SEAT THE BUS KNOWS. Measured live
      // after the two landedness tests: the survivors were `docs/…`, `fix/…`
      // branches from before the seat-prefix convention — abandoned work with no
      // owner to DM. 40+ hits every run is how a signal gets switched off. They
      // stay VISIBLE in `pushedBranches` (ahead, not landed, no PR); the DM goes
      // only where there is a lane to answer it. The PR half is NOT filtered:
      // an open PR is somebody's by construction, and it is the acceptance.
      const seat = seatOfBranch(h.name);
      if (ahead > 0 && !landed && !hasPr && registeredSeats.has(seat) && now - Date.parse(committedAt) > limitMs) {
        hits.push({
          kind: "unproposed-branch", agentId: seat, branch: h.name, head: h.sha, ahead, minutes,
          note: "STRANDED, not necessarily LOST: nobody has cited this artefact. Its owner classifies it — superseded draft (delete) · abandoned (say so) · unlanded work (open the PR).",
        });
      }
    }
    // Forget shas origin no longer has, so the memo cannot grow without bound.
    const live = new Set(artefacts.remoteHeads.map((h) => h.sha));
    for (const k of Object.keys(memo)) if (!live.has(k)) delete memo[k];
    writeMemo(memo);
    if (deferred > 0) {
      unmeasurable.push({ agentId: "-", value: `${deferred} branch(es)`, why: `DEFERRED — the ${budgetMs}ms budget for uncached branch classification ran out; these are UNMEASURED this run, not clean, and the next run continues from the memo` });
    }
  }
  return { hits: withAudience(hits), unmeasurable, openPrs, pushedBranches };
}

// ---------- halt ----------

export const setHaltSchema = {
  reason: z.string().min(1),
  by: z.string().min(1),
  clear: z.boolean().optional(),
};

/**
 * A NAMED state, never a mood.
 *
 * "Production feels down" is not a halt: a halt blocks every claim in the fleet,
 * so the thing that sets it must be nameable and therefore arguable — a board
 * cutover, a cited `BLOCKER:`, a documented red pipeline. The reason is required
 * for that reason, not for the log.
 */
export async function setHaltTool(args: { reason: string; by: string; clear?: boolean }) {
  mkdirSync(ROOT, { recursive: true });
  if (args.clear) {
    writeFileSync(haltFile(), JSON.stringify({ halted: false, clearedBy: args.by, clearedAt: Date.now(), lastReason: args.reason }, null, 2));
    return { ok: true as const, halted: false, clearedBy: args.by };
  }
  const state = { halted: true, reason: args.reason, by: args.by, at: Date.now() };
  writeFileSync(haltFile(), JSON.stringify(state, null, 2));
  return { ok: true as const, ...state };
}

export function haltState(): { halted: boolean; reason?: string; by?: string; at?: number } {
  try {
    const raw = JSON.parse(readFileSync(haltFile(), "utf8"));
    return raw?.halted ? raw : { halted: false };
  } catch {
    return { halted: false };
  }
}

// ---------- the run mark ----------

/** Every run leaves this, HIT or MISS. It is what makes a dead clock visible. */
/**
 * Record a run that FAILED.
 *
 * Acceptance from the queue item, and the clause most easily skipped: "a
 * scheduled check reports its FETCH FAILURES, or a broken check is
 * indistinguishable from a quiet registry". Without this, a clock that fires
 * every 30 minutes and throws every time leaves NO marks at all — identical on
 * disk to a clock that was never installed.
 */
export function markRunFailure(reason: string): void {
  mkdirSync(ROOT, { recursive: true });
  let history: RunMark[] = [];
  try {
    history = JSON.parse(readFileSync(runFile(), "utf8")).history ?? [];
  } catch {
    /* first run */
  }
  history.push({ at: Date.now(), hits: 0, checked: 0, failed: reason });
  writeFileSync(runFile(), JSON.stringify({ history: history.slice(-200) }, null, 2));
}

/**
 * ⟨q-d0e83b41⟩ (09-14 amend) — THE POPULATION TRAVELS WITH THE RUN. A ratio is
 * a claim about a population; a run that records only the ratio cannot tell
 * "a lane got measurable" from "a lane left the denominator". So each run
 * records WHO was scored, from which board (path@sha), and the next run
 * reports the delta as its own signal beside the ratio.
 */
export type RunPopulation = { repo: string; source: string; scored: string[]; roles: string[]; deliberate: string[]; held: string[] };
type RunMark = { at: number; hits: number; checked: number; measurable?: number; failed?: string; population?: RunPopulation };
export function lastPopulationFor(repo: string): { at: number; population: RunPopulation } | null {
  let history: RunMark[] = [];
  try { history = JSON.parse(readFileSync(runFile(), "utf8")).history ?? []; } catch { return null; }
  for (let i = history.length - 1; i >= 0; i--) {
    const h = history[i]!;
    if (!h.failed && h.population && h.population.repo === repo) return { at: h.at, population: h.population };
  }
  return null;
}
/** `docs/WORKSTREAMS.md@<sha>`, with `+uncommitted` when the working copy differs from that commit. */
export function boardSourceOf(repo: string): string {
  let sha = "unknown";
  try { sha = gitOut(repo, ["log", "-1", "--format=%h", "--", BOARD_DOC]) || "uncommitted"; } catch { /* no history */ }
  let dirty = false;
  try { dirty = gitOut(repo, ["status", "--porcelain", "--", BOARD_DOC]).length > 0; } catch { /* unknown */ }
  return `${BOARD_DOC}@${sha}${dirty ? "+uncommitted" : ""}`;
}

function markRun(result: { hits: StallHit[]; checked: number; measurable?: number; population?: RunPopulation }) {
  mkdirSync(ROOT, { recursive: true });
  let history: RunMark[] = [];
  try {
    history = JSON.parse(readFileSync(runFile(), "utf8")).history ?? [];
  } catch {
    /* first run */
  }
  // COVERAGE IS RECORDED WITH THE RUN, because a status tool that cannot
  // express its own blindness is worse than no status tool. Measured: the clock
  // reported `runs 8, failures 0` — all green — while every one of those runs
  // had covered 0 of 3 agents. Nothing in the mark could say so, and this is the
  // instrument meant to cover an absence.
  history.push({ at: Date.now(), hits: result.hits.length, checked: result.checked, measurable: result.measurable ?? 0, ...(result.population ? { population: result.population } : {}) });
  // A RUN OF MISSES MUST BE VISIBLE AS RUNS, not as absence — so the marks are a
  // list, not a single timestamp. "Ten quiet checks" and "one check ten hours ago"
  // are different states and only the first is a healthy fleet.
  writeFileSync(runFile(), JSON.stringify({ history: history.slice(-200) }, null, 2));
}

export const lastRanSchema = { maxAgeMinutes: z.number().optional() };

/**
 * Is the clock alive? Readable by a human or another check, which is the point —
 * MISS is silent to the WATCHER, not to the record.
 */
export async function stallClockStatusTool(args: { maxAgeMinutes?: number }) {
  const maxAge = (args.maxAgeMinutes ?? 60) * 60 * 1000;
  let history: RunMark[] = [];
  try {
    history = JSON.parse(readFileSync(runFile(), "utf8")).history ?? [];
  } catch {
    return {
      ok: false as const,
      error:
        "stall_check has NEVER run — no run mark exists. That is not a quiet fleet, it is an unwatched one: a check that only speaks when it fires cannot be told from a broken one.",
    };
  }
  const last = history[history.length - 1];
  // A RUN THAT FAILED IS NOT A RUN THAT PASSED. Without this the clock reads
  // "fresh" off a mark it wrote while erroring — the age is honest and the
  // health is not, which is the same shape as a fresh heartbeat from a stuck
  // agent.
  const failures = history.filter((h) => h.failed);
  const lastFailed = last?.failed;
  const age = Date.now() - (last?.at ?? 0);
  // `>=`, NOT `>`, AND THE DIFFERENCE IS A REAL RACE RATHER THAN PEDANTRY.
  //
  // With `>`, a window of 0 and a mark written in the SAME MILLISECOND gives
  // `0 > 0` = false: the clock reads FRESH at the instant it was asked to treat
  // everything as stale. It passed locally and on one CI run and failed on
  // another — green in two environments, red in one — because it depended on at
  // least a millisecond elapsing.
  //
  // The window having ELAPSED is the condition, so equality is inside it: a
  // 0-minute window means nothing is ever fresh, which is what a caller asking
  // for one means.
  const stale = age >= maxAge;
  const misses = history.filter((h) => !h.failed && h.hits === 0).length;
  const hits = history.filter((h) => !h.failed && h.hits > 0).length;
  // A RUN THAT COVERED NOTHING IS NOT A RUN THAT FOUND NOTHING. Marks written
  // before coverage was recorded carry no `measurable` field: they are reported
  // as UNKNOWN rather than assumed covered, because assuming is what produced
  // the green this exists to correct.
  const graded = history.filter((h) => !h.failed && typeof h.measurable === "number");
  const blindRuns = graded.filter((h) => h.checked > 0 && h.measurable === 0).length;
  const ungraded = history.filter((h) => !h.failed && typeof h.measurable !== "number").length;
  const lastBlind = last && !last.failed && (last.checked ?? 0) > 0 && last.measurable === 0;
  return {
    // A FAILING CLOCK IS NOT A HEALTHY ONE. It writes marks on schedule, so the
    // age looks fresh while nothing is being measured — a fresh heartbeat from
    // a stuck agent, one level up.
    ok: !stale && !lastFailed && !lastBlind,
    ...(stale
      ? {
          error: `stall_check last ran ${Math.round(age / 60000)}m ago, past the ${args.maxAgeMinutes ?? 60}m window — THE CLOCK IS STOPPED. No alerts is not the same as no stalls.`,
        }
      : lastFailed
        ? {
            error: `the clock is RUNNING but its last run FAILED: ${lastFailed}. It is firing on schedule and measuring nothing, which reads as fresh and is not.`,
          }
        : lastBlind
          ? {
              error:
                `the clock is RUNNING and BLIND: its last run checked ${last?.checked} in-flight row(s) and could measure ${last?.measurable} of them. ` +
                `It fires on schedule, reports no hits, and that "no hits" is not evidence of a healthy fleet — it is evidence of nothing. ` +
                `Read stall_check's own 'unmeasurable' list for the cause; the usual one is a board 'Branch · Worktree' cell holding a PATH rather than a branch ref.`,
            }
          : {}),
    lastRanMinutesAgo: Math.round(age / 60000),
    runs: history.length,
    misses,
    hits,
    failures: failures.length,
    // The number the old shape could not express.
    coverage: {
      lastRun: last && !last.failed ? { checked: last.checked, measurable: last.measurable ?? null } : null,
      blindRuns,
      ungraded,
      note:
        "blindRuns are runs that checked rows and measured none of them — green by every other field. " +
        "ungraded are marks written before coverage was recorded: UNKNOWN, never assumed covered.",
    },
  };
}

// ---------- the predicate ----------

export const stallCheckSchema = {
  repo: z.string().optional(),
  stallMinutes: z.number().optional(),
  /** The 🔍 window, in minutes. Default REVIEW_STALL_MINUTES. Independent of stallMinutes on purpose. */
  reviewMinutes: z.number().optional(),
  /** ⟨q-7b2f6c04⟩ — the claim-axis window, in minutes. Default CLAIM_STALL_MINUTES. */
  claimMinutes: z.number().optional(),
};

const gitOut = (repo: string, args: string[]) =>
  execFileSync("git", args, { cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"] }).trim();

/** The key a row keeps while its cells are edited: its stream text. */
export const rowKey = (r: { stream: string }) => r.stream.replace(/\s+/g, " ").trim();

export type ReviewAge = { since: string; minutes: number; commits: number };

/**
 * ⛔⛆ AN `🔍 In Review` ROW IS NOT SCORED ON BRANCH ACTIVITY AT ALL (⟨q-507e80c4⟩).
 *
 * Measured 2026-09-14: `no-vcs-activity` fired on a review lane at 41m, again at
 * 45m, and would have fired on every correctly-behaving review lane given an
 * hour — a branch under review is SUPPOSED to stop moving, so that predicate
 * reaches its guaranteed end state on every healthy lane. The stall question in
 * review is a DIFFERENT one: how long has the row been in review.
 *
 * The board carries no timestamp column, but it is a git-tracked document, so
 * the answer is in its history: walk the commits that touched the board from
 * newest to oldest while this row (by stream key) still declares in-review; the
 * oldest such commit is when the row ENTERED review. A row that declares review
 * only in the working tree entered it just now (`since: "uncommitted"`).
 *
 * Returns `null` only when the board's history cannot be read — which is
 * reported as unmeasurable, never as healthy.
 */
export function reviewEnteredAt(repo: string, key: string, now = Date.now()): ReviewAge | null {
  return stateEnteredAt(repo, key, now, (row) => workStateOf(row.status) === "in-review");
}
/**
 * ⟨q-7b2f6c04⟩ — when the row was CLAIMED: the oldest board commit in which it
 * declares an in-flight state. The claim axis reads the board's history the
 * way the review axis does, so a lane is observable from its claim, not from
 * its first push.
 */
export function claimEnteredAt(repo: string, key: string, now = Date.now()): ReviewAge | null {
  return stateEnteredAt(repo, key, now, (row) => coarseOf(workStateOf(row.status)) === "in-flight");
}
function stateEnteredAt(repo: string, key: string, now: number, matches: (row: WorkstreamsV1Row) => boolean): ReviewAge | null {
  let log: string;
  try {
    log = gitOut(repo, ["log", "-n", "300", "--format=%H%x09%cI", "--", BOARD_DOC]);
  } catch {
    return null;
  }
  let since: string | null = null;
  let commits = 0;
  for (const line of log.split("\n").filter(Boolean)) {
    const [sha, iso] = line.split("\t");
    let text: string;
    try {
      text = gitOut(repo, ["show", `${sha}:${BOARD_DOC}`]);
    } catch {
      break;
    }
    const row = workstreamsV1RowsOf(parseWorkDoc(text)).find((r) => rowKey(r) === key);
    if (!row || !matches(row)) break;
    since = iso;
    commits++;
  }
  if (!since) return { since: "uncommitted", minutes: 0, commits: 0 };
  return { since, minutes: Math.round((now - Date.parse(since)) / 60000), commits };
}

export type RowFreshness = {
  stream: string;
  owner: string;
  state: WorkState;
  /** ISO of the commit that last touched this row's line, `uncommitted` for a working-tree edit, `unknown` if blame failed. */
  lastUpdated: string;
  minutesAgo: number | null;
  commit: string | null;
};

/**
 * ⛆ THE BOARD'S OWN FRESHNESS, PER ROW — the acceptance clause the original row
 * lacked. A predicate reading stale cells is not measuring lanes, it is
 * measuring the board's maintenance; on 2026-09-14 one silent seat froze the
 * board and the clock then fired on three HEALTHY lanes, and nothing in its
 * output could tell the two apart. So every scored row now says when its line
 * was last written, read from `git blame` on the row's own bytes.
 */
export function rowFreshness(repo: string, boardText: string, rows: { stream: string; owner: string; status: string; raw?: string }[], now = Date.now()): RowFreshness[] {
  const lines = boardText.split("\n");
  return rows.map((r) => {
    const base = { stream: r.stream.slice(0, 60), owner: r.owner.replace(/[`*]/g, "").trim(), state: workStateOf(r.status) };
    const idx = r.raw ? lines.indexOf(r.raw) : -1;
    if (idx < 0) return { ...base, lastUpdated: "unknown", minutesAgo: null, commit: null };
    try {
      const out = gitOut(repo, ["blame", "-L", `${idx + 1},${idx + 1}`, "--porcelain", "--", BOARD_DOC]);
      const sha = out.split(/\s/)[0] ?? "";
      if (/^0+$/.test(sha)) return { ...base, lastUpdated: "uncommitted", minutesAgo: 0, commit: null };
      const t = Number(/^committer-time (\d+)/m.exec(out)?.[1]);
      if (!Number.isFinite(t)) return { ...base, lastUpdated: "unknown", minutesAgo: null, commit: sha.slice(0, 8) };
      return { ...base, lastUpdated: new Date(t * 1000).toISOString(), minutesAgo: Math.round((now - t * 1000) / 60000), commit: sha.slice(0, 8) };
    } catch {
      return { ...base, lastUpdated: "unknown", minutesAgo: null, commit: null };
    }
  });
}

/**
 * A row somebody is WORKING — the population the stall clock watches.
 *
 * It was `/🚧/` alone, and that read a fleet living in review as an EMPTY
 * fleet (q-507e80c4). Measured on a consumer fleet 2026-09-02: a board of 3× `🔍 In
 * Review` and 0× `🚧` returned `checked: 0, measurable: 0`, and `coord_away`
 * armed at 0/0 — correct and vacuous at once. A clock watching nothing and a
 * clock watching a healthy fleet return identical green.
 *
 * `🔍 In Review` is a lane with an owner and a branch that can stop moving
 * exactly as an authoring lane can; the review that never gets re-gated is
 * THE stall shape of a QA-gated fleet. It is read here, on the row as written
 * — relabelling `🔍`→`🚧` to satisfy the old predicate was the item's
 * forbidden move, because it satisfies a guard by editing what it reads and
 * reports authoring lanes that do not exist.
 *
 * NOT widened further, deliberately: `⏸ Parked`, `⏳ Queued`, `⛔ Blocked`,
 * `🚫 Unstaffable`, `✅ Done` have no lane to stall. Widening to everything
 * would be the 0/0 defect inverted. They are REPORTED, though — see
 * `notWatched` below — because a row the clock declines to watch and a row
 * the clock cannot see produce the same silence.
 *
 * One predicate, exported: `coord_away` measures coverage by calling
 * `stall_check` and reading `checked`/`measurable`, so it inherits this
 * population without a change of its own. Anything else that asks "which rows
 * are in flight" should ask here rather than re-derive it from a glyph.
 *
 * ⛔⛆ DERIVED FROM THE SEAM, NOT MATCHED ON A GLYPH (⟨q-a42503cb⟩). This was
 * `IN_FLIGHT_STATUS = /🚧|🔍/`, and the sentence above it — "ask here rather
 * than re-derive it from a glyph" — described a single authority that WAS a
 * glyph regex. The board's vocabulary grew to eleven documented states
 * (`workStateOf`, seam ⟨q-5b3e9a04⟩) and this predicate stayed at two, so
 * every `⏸` row was invisible here AND in `next_unblocked`'s routing
 * exclusion, which shares it: a seat that wrote `⏸ MERGE-HELD` instead of a
 * stale `🚧` was doing the right thing and punished for it by a silent drop
 * from both populations.
 *
 * The seam reads the LEADING glyph after stripping decoration, so
 * `⛔ Blocked — was 🚧 yesterday` is blocked here and was in-flight under the
 * regex. Measured on the live board at origin/main 687a74f (25 rows): the two
 * predicates agree on every row, 4 in flight under each — the swap changes no
 * verdict today; it changes what the file can SAY about the other 21.
 */
export const isInFlightStatus = (status: string): boolean => coarseOf(workStateOf(status)) === "in-flight";

/** A hold the board declares for this lane, or null. Blocker cell first (the lane's own statement), then the Cutover Gates table. */
export function holdOf(row: WorkstreamsV1Row, cutoverGates: Record<string, string>[]): { text: string; by: "blocker" | "cutover-gate" } | null {
  const blocker = String(row.blocker ?? "").replace(/[`*]/g, "").trim();
  if (blocker && !/^(—|-|–|none|n\/a)$/i.test(blocker)) return { text: blocker.slice(0, 160), by: "blocker" };
  const ids = [...String(row.stream).matchAll(/\b(q-[0-9a-f]{8})\b/g)].map((m) => m[1] as string);
  const prs = [...String(row.stream).matchAll(/#(\d{2,})\b/g)].map((m) => `#${m[1]}`);
  const ref = refInCell(row.branchWorktree);
  for (const g of cutoverGates) {
    const gate = String(g["Gate"] ?? Object.values(g)[0] ?? "");
    if (ids.some((id) => gate.includes(id)) || prs.some((p) => gate.includes(p)) || (ref && gate.includes(ref))) {
      return { text: gate.replace(/[`*]/g, "").trim().slice(0, 160), by: "cutover-gate" };
    }
  }
  return null;
}
/**
 * ⟨q-7e3b10c9⟩ — THE BOARD NAMES ITS OWNER. The Rooms table's owner cell says
 * who holds the topic — this fleet's `groundwork-kit-coordinator (topic owner)`
 * — so the clock can route bookkeeping hits without a bus-meta dependency.
 * Null when the board does not say; the clock then says so and falls back.
 */
export function boardOwnerOf(boardText: string): string | null {
  const ext = workstreamsExtensionsOf(parseWorkDoc(boardText));
  for (const r of ext.rooms) {
    for (const v of Object.values(r)) {
      const m = /([\w.-]+-coordinator)\b/.exec(String(v));
      if (m) return m[1]!;
    }
  }
  return null;
}

export async function stallCheckTool(
  args: { repo?: string; stallMinutes?: number; reviewMinutes?: number; claimMinutes?: number },
  artefacts: (repo: string) => RepoArtefacts = realArtefacts,
  readRooms: () => string = readAllRoomLogs,
  readInboxes: () => string = readAllInboxLogs,
  /** ⟨q-1c95f7d4⟩ the transport the external tick is read through; injected by tests. */
  tickTransport: Parameters<typeof readFleetTick>[1] = activeTransport(),
) {
  const repo = args.repo ?? process.cwd();
  const limit = (args.stallMinutes ?? 30) * 60 * 1000;
  const reviewLimit = args.reviewMinutes ?? REVIEW_STALL_MINUTES;
  const claimLimit = args.claimMinutes ?? CLAIM_STALL_MINUTES;
  const board = path.join(repo, BOARD_DOC);
  if (!existsSync(board)) return { ok: false as const, error: `no ${BOARD_DOC} under '${repo}'` };

  const liveTransports = await loadLiveTransports();
  // ⟨q-1c95f7d4⟩ Task 5 — the EXTERNAL tick, read once per run for the whole fleet. On a
  // fleet with no herdr seat this asks nothing and answers "measured NOTHING", which is
  // what keeps a blind axis from reading as a calm one.
  const fleetTick: FleetTick = await readFleetTick(liveTransports.values(), tickTransport);
  const tickOf = tickByAgent(fleetTick);
  const boardText = readFileSync(board, "utf8");
  const boardDoc = parseWorkDoc(boardText);
  const rows = workstreamsV1RowsOf(boardDoc);
  // ⟨q-7e3b10c9⟩ — holds DECLARED on the board: the Cutover Gates table's Gate
  // cell names what is held; a lane row's Blocker cell names what it waits on.
  const cutoverGates = workstreamsExtensionsOf(boardDoc).cutoverGates;
  // ⟨q-3d82f1a9⟩ — SAY THE PARSE STATE FIRST. An unreadable board is refused
  // as a measurement, not answered as a quiet fleet: ok:false, dm:true, and
  // the counts named, so the duty officer is told the board broke rather than
  // shown an empty list that reads as healthy.
  const boardParse = boardParseOf(boardText);
  if (!boardParse.readable) {
    markRunFailure(`unparseable board — ${boardParse.rowsPresent} row(s) present, ${boardParse.rowsParsed} parsed`);
    return {
      ok: false as const,
      error: `${BOARD_DOC} is ${boardParse.why}`,
      boardParse,
      checked: 0,
      measurable: 0,
      blind: [] as string[],
      hits: [] as StallHit[],
      unmeasurable: [] as { agentId: string; value: string; why: string }[],
      halted: haltState().halted,
      dm: true,
      note: `UNPARSEABLE BOARD — ${boardParse.rowsPresent} row(s) present, ${boardParse.rowsParsed} parsed. This is NOT a quiet fleet: nothing was measured. Fix the row the parser stopped at and re-run.`,
    };
  }
  // ⟨q-5d1c8e04⟩ — LANES ARE SCORED; ROLES ARE REPORTED. A role row (🪑, branch
  // `—`) is a standing seat with nothing to stall; it is listed as present or
  // absent from the registry and live or not on its transport, and never
  // enters `checked`. Two roles and zero lanes read 0 of 0, not 0 of 2. A 🪑
  // row that carries a ref is a lane in disguise and is scored below.
  const reg = await readJson<Record<string, { lastHeartbeat: number }>>(AGENTS_FILE, {});
  const kinds = rows.map((r) => ({ row: r, kind: rowKindOf(r) }));
  const roles = kinds
    .filter((k) => k.kind === "role")
    .map(({ row }) => {
      const owner = row.owner.replace(/[`*]/g, "").trim();
      return { stream: row.stream.slice(0, 60), owner, present: !!reg[owner], active: liveTransports.has(owner) };
    });
  const disguised = kinds.filter((k) => k.kind === "role-with-ref").map((k) => k.row);
  const inFlight = [...rows.filter((r) => isInFlightStatus(r.status)), ...disguised];
  // THE ROWS THIS CLOCK DOES NOT WATCH, NAMED AS THEIR OWN STATE (⟨q-a42503cb⟩).
  // A parked row is not a stalled one, so it is not in `checked` — but a
  // clock that cannot name what it declined to watch reads the same as one that
  // never saw it. Finished rows are left out: there is nothing to watch or
  // route behind a `✅`. Everything else that is not in flight is listed with
  // the state the seam read, so a reader can tell "held pending a merge" from
  // "blocked" from "parked awaiting a human" without opening the board.
  const byState: Partial<Record<WorkState, number>> = {};
  for (const r of rows) {
    const s = workStateOf(r.status);
    byState[s] = (byState[s] ?? 0) + 1;
  }
  const notWatched = rows
    .filter((r) => {
      const c = coarseOf(workStateOf(r.status));
      return c !== "in-flight" && c !== "finished";
    })
    .map((r) => ({
      stream: r.stream.slice(0, 60),
      owner: r.owner.replace(/[`*]/g, "").trim(),
      state: workStateOf(r.status),
      status: r.status.slice(0, 80),
    }));
  const now = Date.now();
  const hits: StallHitBody[] = [];
  // ⟨q-11257590⟩ — COORDINATOR ABSENCE IS A FAULT, SURFACED, NEVER COVERED. Read the
  // seat's own name from the board (`boardOwnerOf`, the Rooms table's "topic owner"
  // cell — the same identity every seat is told at `join`), not hardcoded, so a
  // renamed or re-elected coordinator is still found. `reg` and `liveTransports` are
  // already loaded above for the role rows; this reuses their verdicts rather than
  // re-deriving liveness. NO board owner named → nothing to check, not a hit: an
  // unnamed owner is a board-authoring gap, a different row's problem.
  const coordId = boardOwnerOf(boardText);
  if (coordId) {
    if (!reg[coordId]) {
      hits.push({
        kind: "coordinator-absent", agentId: coordId,
        why: `the board names '${coordId}' as coordinator (topic owner) but the bus has no registry entry for it — never joined, or evicted`,
      });
    } else if (!liveTransports.has(coordId)) {
      hits.push({
        kind: "coordinator-absent", agentId: coordId,
        why: `'${coordId}' is registered but has no live transport marker right now — registered once, unreachable now`,
      });
    }
    // Present AND active → no hit. This is the row's required negative control: a
    // healthy coordinator must not surface, or this becomes a standing false alarm.
  }
  /** The base's tip, for telling an empty claim-time push from a lane with commits. */
  let baseTip: string | null = null;
  for (const cand of ["origin/main", "origin/master", "main", "master"]) {
    try { baseTip = gitOut(repo, ["rev-parse", "--verify", "--quiet", `${cand}^{commit}`]); break; } catch { /* next */ }
  }
  /** Rows whose VCS activity could not be measured. NOT stall claims. */
  const unmeasurable: { agentId: string; value: string; why: string }[] = [];
  /**
   * Rows a predicate actually reached a VERDICT about — hit or clean.
   *
   * Tracked explicitly rather than inferred from "2 predicates minus the
   * unmeasurable ones", because a predicate that is SKIPPED pushes no
   * unmeasurable entry and the arithmetic then counts it as having measured.
   * That is the same defect as the `continue` above, one level up: a silent
   * skip and a clean pass produce the same number.
   */
  const measured = new Set<string>();
  /**
   * ⟨q-5d1c8e04⟩ — DELIBERATELY BRANCHLESS LANES, out of the population. The
   * cell holds a statement (`docs-direct · no per-agent branch`) rather than a
   * ref; the seat said so in words. Reported by name, subtracted from
   * `checked`, never "unmeasurable" — the coverage ratio stops carrying rows
   * that were never measurable by design. A blank or `—` on a lane is NOT
   * this: that lane should carry a branch and names none, and stays unmeasurable.
   */
  const deliberate: { agentId: string; stream: string; cell: string }[] = [];
  /**
   * ⟨q-7e3b10c9⟩ — HELD, NOT STALLED AND NOT UNATTENDED. A lane ageing behind a
   * hold the board DECLARES — its Blocker cell names a hold, or a Cutover Gates
   * row names the lane — is reported as held, with the hold, and is not scored:
   * a frozen branch behind a PASS-HOLD is the gate working, not a seat stuck.
   * Measured as a verdict (coverage counts it). An UNDECLARED wait is not a
   * hold: the same frozen branch with an empty Blocker cell still fires.
   */
  const held: { agentId: string; stream: string; hold: string; declaredBy: "blocker" | "cutover-gate" }[] = [];
  const scored: typeof inFlight = [];
  /**
   * ⟨q-7b2f6c04⟩ — WHICH AXIS EACH ROW WAS MEASURED ON. One number over three
   * axes is how all three stayed hidden; every scored row now says which
   * predicate reached a verdict about it, or that none could.
   */
  type Axis = "vcs" | "review" | "claim" | "landed" | "tick";
  const axes: { agentId: string; stream: string; axis: Axis | null; measurable: boolean; why?: string }[] = [];
  const onAxis = (agentId: string, row: { stream: string }, axis: Axis) => { measured.add(agentId); axes.push({ agentId, stream: row.stream.slice(0, 60), axis, measurable: true }); };
  const offAxis = (agentId: string, row: { stream: string }, why: string) => axes.push({ agentId, stream: row.stream.slice(0, 60), axis: null, measurable: false, why });
  /** The claim axis: scored from the board's history; fires past the claim window. */
  const scoreClaim = (agentId: string, row: WorkstreamsV1Row, branch: string, because: string) => {
    const age = claimEnteredAt(repo, rowKey(row), now);
    if (!age) {
      unmeasurable.push({ agentId, value: BOARD_DOC, why: "the board's git history could not be read, so the claim time is unknown — unknown is not healthy" });
      offAxis(agentId, row, "board history unreadable");
      return;
    }
    onAxis(agentId, row, "claim");
    if (age.minutes > claimLimit) {
      hits.push({
        kind: "unpushed-claim", agentId, stream: row.stream.slice(0, 60), branch, minutes: age.minutes, since: age.since,
        why: `claimed ${age.minutes}m ago (row in flight since ${age.since}) and ${because} — nothing from this lane has reached origin inside the ${claimLimit}m claim window. A worker thinking hard and a worker whose model has wedged look the same from here; say which.`,
      });
    }
  };

  for (const row of inFlight) {
    const agentId = row.owner.replace(/[`*]/g, "").trim();
    // Only a row the VCS half would score can declare itself branchless. A 🔍
    // row is scored on time-in-review from the board (#306), whatever its cell
    // says — so its prose is a note, not an exemption.
    if (cellKindOf(row.branchWorktree) === "prose" && workStateOf(row.status) !== "in-review") {
      deliberate.push({ agentId, stream: row.stream.slice(0, 60), cell: String(row.branchWorktree).trim().slice(0, 80) });
      continue;
    }
    scored.push(row);
    const hold = holdOf(row, cutoverGates);
    if (hold) {
      held.push({ agentId, stream: row.stream.slice(0, 60), hold: hold.text, declaredBy: hold.by });
      measured.add(agentId);
      continue;
    }
    // ⟨q-1c95f7d4⟩ 5.2 — THE TICK IS APPLIED FIRST AND SUPPRESSES NOTHING. It may add a
    // hit and it may credit coverage; every branch and heartbeat predicate below still
    // runs and still reaches its own verdict, because "the seat is moving" and "the work
    // is moving" are different questions and only one of them is answered here.
    const tickSeat: SeatTick | undefined = tickOf.get(agentId);
    const tick = tickVerdict(tickSeat);
    if (tick && tickSeat) {
      if (tick.measured) {
        onAxis(agentId, row, "tick");
        if (tick.hit && tickSeat.readable) hits.push({ kind: "seat-blocked", agentId, stream: row.stream.slice(0, 60), source: tickSeat.source, why: tick.why });
      } else {
        // 5.3 — A HERDR SEAT WHOSE SIGNAL CANNOT BE READ IS UNMEASURABLE, NEVER HEALTHY.
        unmeasurable.push({ agentId, value: tickSeat.transport, why: `external tick unreadable: ${tick.why}` });
      }
    }
    const entry = reg[agentId];
    if (!entry) {
      // NOT SKIPPED IN SILENCE. An owner with no registry entry has no
      // heartbeat to read, which is a missing signal and must be reported as
      // one — a row whose owner the bus has never heard of is exactly the row
      // you would want named before going away.
      unmeasurable.push({ agentId, value: "", why: "no registry entry for this owner — there is no heartbeat to read, and the name may not be an agent id at all" });
    }
    if (entry) {
      const age = now - entry.lastHeartbeat;
      // A HEARTBEAT IS ONLY EVIDENCE WHERE SOMETHING WRITES ONE.
      //
      // `heartbeat` is called by the PUSHER, never by the agent. Only
      // `coord-pusher.mjs` (the REMOTE pusher) calls it, and it must: a remote
      // marker cannot be pid-probed across machines, so its liveness IS the
      // heartbeat. `hooks/tmux-pusher.mjs` — what this fleet actually runs —
      // never calls it, because a LOCAL marker's liveness is `isPidAlive`.
      //
      // So for a local transport there is no heartbeat SOURCE at all. Before
      // kit#137, `list_agents` stamped these agents and that fabrication was
      // the only thing keeping the field moving; removing it left the field
      // honest and empty. Measured: three attached, working agents at an
      // IDENTICAL 44.7m — the uniform signature of one shared cause, not three
      // stalls.
      //
      // WHY NOT JUST MAKE tmux-pusher HEARTBEAT: because the signal would mean
      // "the pusher process is alive", which `isPidAlive` already answers for
      // local markers. During the 17-hour stall every transport was live the
      // whole time, so a pusher heartbeat would have read FRESH for all 17
      // hours. It would restore a field without restoring a detector — and the
      // case this verb exists for is exactly the one it would miss. The vcs
      // half is what catches "alive and not progressing"; saying so is more
      // honest than a green field.
      //
      // Unknown is not stalled (kit#138). This does NOT narrow Task 3.5: an
      // agent with no transport, or a REMOTE one where the heartbeat genuinely
      // is the liveness mechanism, still HITs on a dead heartbeat.
      const marker = liveTransports.get(agentId);
      if (age > limit && marker && isLocallyProbeable(marker.transport)) {
        unmeasurable.push({
          agentId,
          value: marker.transport,
          why: `heartbeat is ${Math.round(age / 60000)}m old, but nothing writes heartbeats for a local 'tmux-push' transport — hooks/tmux-pusher.mjs does not call heartbeat, and this marker's liveness is its pid. There is no heartbeat SOURCE here, so the age measures nothing about this agent`,
        });
        // FALL THROUGH TO THE VCS PREDICATE — do not `continue`.
        //
        // This used to skip the row entirely, and the consequence was total: on
        // this fleet, 3 of 3 in-flight rows were local tmux-push agents, so
        // EVERY row exited here and the check reported `checked 3 · measurable
        // 0`. The half that was blind (heartbeat, deliberately, for the reason
        // above) was taking the half that works (vcs activity) down with it.
        //
        // The two predicates answer different questions and only one of them is
        // unanswerable for a local transport. "Is it alive" has no source here;
        // "is its branch moving" has a perfectly good one, and it is the half
        // that catches the case this verb exists for — alive and not
        // progressing. An unmeasurable heartbeat is a missing signal, not a
        // reason to stop measuring the signal that is present.
      } else if (age > limit) {
        hits.push({ kind: "no-heartbeat", agentId, stream: row.stream.slice(0, 60), minutes: Math.round(age / 60000) });
        measured.add(agentId);
        continue;
      }
      // A FRESH HEARTBEAT IS *NOT* A VERDICT, AND THIS IS A CORRECTION TO WHAT
      // THIS CODE CLAIMED WHEN IT MERGED.
      //
      // It read "a fresh heartbeat IS a verdict — alive" and credited coverage
      // for it. Measured since, by reading agents.json directly (`list_agents`
      // refreshes the CALLER's mark, so it cannot be used to measure the
      // caller): an agent's mark aged from 9611s to 9623s across a `post_status`
      // call. BUS TOOL CALLS DO NOT WRITE HEARTBEATS.
      //
      // `heartbeat` age is TIME SINCE JOIN. Not activity, and not even
      // time-since-restart — a server can restart mid-session without the mark
      // moving, because nothing rejoined. So freshness says "recently joined",
      // and crediting it as coverage lets `/coord-away` arm on a fleet whose
      // only evidence is that somebody reconnected.
      //
      // The inverse is worse and is why this is not merely tidiness: reading
      // age as activity marks the two most CONTINUOUSLY ACTIVE agents on a bus
      // as stale at 2.6h, while a freshly-rejoined idle agent reads healthy —
      // a false-stall generator aimed at exactly the agents that must not be
      // false-stalled while nobody is watching.
      //
      // The HIT above is kept: for an agent with no transport at all, "has not
      // rejoined and has no live marker" is still the death signal Task 3.5
      // specifies. What is removed is the coverage credit for freshness, which
      // leaves coverage resting on the vcs predicate — the one that measures
      // something the agent DID.
    }
    // A FRESH HEARTBEAT IS NOT PROGRESS. An agent can be alive and stuck, which is
    // the case "notice the room" never catches: the pane is responsive, so nobody
    // looks. Ask the branch instead.
    //
    // RESOLVE A REF, AND SAY SO WHEN THE VALUE IS NOT ONE.
    //
    // `git log -1 <value>` accepts a PATHSPEC exactly as readily as a ref, and
    // the old guard only required a `/` — which every path has. So the board's
    // `Branch · Worktree` cells, which hold PATHS, were fed to git and silently
    // measured as paths: the aide was reported at 2,271 minutes, the age of
    // `docs/phases/phase5`'s last commit, not of any activity by that agent.
    // Verified: `docs/phases/phase5` does not resolve as a ref, yet
    // `git log -1 --format=%cI docs/phases/phase5` returns a date.
    //
    // Other rows read plausibly only by coincidence — a path that happens to be
    // committed often looks like an active branch.
    //
    // UNKNOWN IS NOT STALLED, which this verb already gets right for
    // heartbeats. An unresolvable value is REPORTED as unmeasurable rather than
    // skipped in silence: a silent `continue` and a healthy agent produce the
    // same output, which is the failure this whole verb exists to avoid.
    // ONE CLASSIFIER, SHARED WITH `claim` (board-ref.ts). The old block asked
    // only "does it resolve", and three of the four ways this cell has been
    // wrong resolved cleanly — a shared ref, a local-only ref, and a merged
    // ref. Resolution was never the property; the property is whether the ref's
    // movement is THIS AGENT'S WORK.
    const verdict = classifyBoardRef(repo, agentId, row.branchWorktree);
    // ⟨q-7b2f6c04⟩ — a ref sitting EXACTLY on the base's tip is the empty push
    // `ensure_worktree` makes at claim: zero commits of its own, so the
    // classifier reads it as landed and the VCS axis would read the base's
    // age. It is a CLAIM, and the claim axis scores it — checked before either.
    if ((verdict.kind === "merged" || verdict.kind === "measurable") && baseTip) {
      const cellRef = refInCell(row.branchWorktree).replace(/^origin\//, "");
      let tip: string | null = null;
      try { tip = gitOut(repo, ["rev-parse", "--verify", "--quiet", `origin/${cellRef}^{commit}`]); } catch { /* not on origin */ }
      if (tip && tip === baseTip) {
        scoreClaim(agentId, row, `origin/${cellRef}`, "its branch sits on the base with no commits of its own (the empty claim-time push)");
        continue;
      }
    }
    if (verdict.kind === "merged") {
      // A VERDICT, NOT AN ABSENCE — and it counts as coverage, because the check
      // did learn something about this lane: its work landed and its row is
      // stale. Reported as its own kind rather than as a stall, which is what
      // the frozen ref was reporting it as.
      hits.push({ kind: "stale-row", agentId, branch: refInCell(row.branchWorktree), why: verdict.why });
      onAxis(agentId, row, "landed");
      continue;
    }
    if (workStateOf(row.status) === "in-review") {
      // ⛔ NOT THE VCS PREDICATE. A review-frozen branch is frozen for the
      // correct reason; scoring it on activity fires on every healthy review
      // lane eventually. The question here is how long the row has been in
      // review, read from the board's own history — which also means a 🔍 row
      // whose cell holds a path is MEASURED here rather than unmeasurable: the
      // predicate needs the board, not the ref.
      const age = reviewEnteredAt(repo, rowKey(row), now);
      if (!age) {
        unmeasurable.push({ agentId, value: BOARD_DOC, why: "the board's git history could not be read, so time in review is unknown — unknown is not healthy" });
        offAxis(agentId, row, "board history unreadable");
        continue;
      }
      onAxis(agentId, row, "review");
      if (age.minutes > reviewLimit) {
        hits.push({ kind: "in-review-too-long", agentId, stream: row.stream.slice(0, 60), minutes: age.minutes, since: age.since });
      }
      continue;
    }
    // ⟨q-7b2f6c04⟩ — a branch with nothing on origin is not unmeasurable: it is
    // a lane between claim and first push, and the CLAIM axis measures it.
    if (verdict.kind === "unpushed" || verdict.kind === "local-only") {
      scoreClaim(agentId, row, refInCell(row.branchWorktree), `its branch is not on origin (${verdict.kind})`);
      continue;
    }
    if (verdict.kind !== "measurable") {
      unmeasurable.push({ agentId, value: refInCell(row.branchWorktree), why: verdict.why });
      offAxis(agentId, row, verdict.why);
      continue;
    }
    try {
      // The resolved SHA, and `--`: a ref can then never be re-read as a
      // pathspec, which is the ambiguity that produced the wrong number.
      const sha = execFileSync("git", ["rev-parse", "--verify", "--quiet", `${verdict.ref}^{commit}`], {
        cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"],
      }).trim();
      const iso = execFileSync("git", ["log", "-1", "--format=%cI", sha, "--"], {
        cwd: repo, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"],
      }).trim();
      const age = now - Date.parse(iso);
      onAxis(agentId, row, "vcs");
      if (age > limit) {
        hits.push({ kind: "no-vcs-activity", agentId, branch: verdict.ref, minutes: Math.round(age / 60000) });
      }
    } catch {
      unmeasurable.push({ agentId, value: verdict.ref, why: "resolved as a ref but its log could not be read" });
      offAxis(agentId, row, "resolved as a ref but its log could not be read");
    }
  }

  // COVERAGE, not "did it run". A row is COVERED when a predicate produced a
  // verdict about it; a row where every predicate came back unmeasurable was
  // looked at and not measured, and counting it as checked is how "the clock
  // ran" gets mistaken for "the fleet is observed".
  const owners = scored.map((r) => r.owner.replace(/[`*]/g, "").trim()).filter(Boolean);
  const blind = [...new Set(owners)].filter((id) => !measured.has(id));
  const measurable = Math.max(0, scored.length - blind.length);
  // ⟨q-d0e83b41⟩ (09-14 amend) — the population's source and its delta since
  // the last run of THIS repo. A lane that left with no closing state on the
  // board (its row is gone) is a board-owner hit, never an improved ratio.
  const runPopulation: RunPopulation = {
    repo, source: boardSourceOf(repo),
    scored: scored.map(rowKey), roles: roles.map((r) => r.stream), deliberate: deliberate.map((d) => d.stream), held: held.map((h) => h.stream),
  };
  const prev = lastPopulationFor(repo);
  const ownerOf = (key: string) => (rows.find((r) => rowKey(r) === key)?.owner ?? "").replace(/[`*]/g, "").trim();
  const whereNow = (key: string): string => {
    const row = rows.find((r) => rowKey(r) === key);
    if (!row) return "gone";
    if (runPopulation.held.includes(row.stream.slice(0, 60))) return "held";
    if (runPopulation.deliberate.includes(row.stream.slice(0, 60))) return "deliberate";
    return workStateOf(row.status);
  };
  const delta = prev
    ? {
        since: new Date(prev.at).toISOString(),
        left: prev.population.scored.filter((k) => !runPopulation.scored.includes(k)).map((k) => ({ stream: k.slice(0, 60), to: whereNow(k) })),
        joined: runPopulation.scored.filter((k) => !prev.population.scored.includes(k)).map((k) => ({ stream: k.slice(0, 60), owner: ownerOf(k) })),
      }
    : { since: null, left: [], joined: [] };
  for (const l of delta.left) {
    if (l.to !== "gone") continue;
    hits.push({
      kind: "lane-left-population", agentId: boardOwnerOf(boardText) ?? "-", stream: l.stream,
      why: `"${l.stream}" was in the scored population at the last run (${delta.since}) and its row is GONE from ${runPopulation.source} with no closing state — the coverage ratio would otherwise read as improved. A lane leaves the board through a state (✅, ⏸, ⛔), never by deletion.`,
    });
  }
  // THE BOARD'S OWN FRESHNESS BESIDE EVERY VERDICT. A stall report over stale
  // cells is a report about the board's maintenance, and the reader must be
  // able to see which it is without opening git.
  const freshness = rowFreshness(repo, boardText, scored, now);
  const aged = freshness.map((f) => f.minutesAgo).filter((m): m is number => typeof m === "number");

  // ⟨q-6f0a3d81⟩ — THE AXIS THAT NEEDS NOBODY TO SPEAK. Computed over the
  // repository, not the board: a stranded PR has no row, which is the point.
  // `checked`/`measurable` stay the lane population `coord_away` reads; these
  // hits join `hits` so the duty officer is DMed. Its unreadable SOURCES are
  // reported under its own key, not in the per-row `unmeasurable` list: that
  // list is a population of lane rows (a test pins "two halves of one row"),
  // and an unreachable gh is not a row. Separate populations, separate keys.
  const arte = artefacts(repo);
  const roomText = readRooms();
  const stranded = strandedWork(repo, arte, roomText, now, limit, new Set(Object.keys(reg)));
  hits.push(...stranded.hits);
  // ⟨q-8f1e604b⟩ — the two conventions, checked from the artefacts: PR pages
  // for (e), the bus's own routing records against the board and queue for (f).
  const queuePath = path.join(repo, "docs/QUEUE.md");
  const conventions = conventionChecks({
    recentMerges: arte.recentMerges,
    routingLogText: `${roomText}\n${readInboxes()}`,
    queueText: existsSync(queuePath) ? readFileSync(queuePath, "utf8") : null,
    boardRows: rows,
    now,
  });
  hits.push(...conventions.hits);
  // ⟨q-4e08b3c1⟩ — convention (g): the bus's windows joined to git's write times.
  const mergeWindows = mergeWindowChecks({ logText: roomText, writes: recordWritesOn(repo, now - CONVENTION_WINDOW_MS), now });
  hits.push(...mergeWindows.hits);

  const stamped = withAudience(hits);
  const dmBy = { duty: stamped.filter((h) => h.audience === "duty").length, "board-owner": stamped.filter((h) => h.audience === "board-owner").length };
  const result = {
    hits: stamped,
    // ⟨q-7e3b10c9⟩ — who each hit is for, counted, so a relayer can split.
    dmBy,
    held,
    // ⟨q-3d82f1a9⟩ — the parse state travels with every answer, readable or not.
    boardParse,
    // The population the clock SCORES: lanes with a ref position to read. Roles
    // and deliberately branchless lanes are reported beside it, not inside it.
    checked: scored.length,
    unmeasurable,
    measurable,
    blind,
    roles,
    deliberate,
    disguisedRoles: disguised.map((r) => r.stream.slice(0, 60)),
    predicates: {
      "in-progress": `heartbeat (where a source exists) + branch activity, window ${Math.round(limit / 60000)}m`,
      "in-review": `time in review from the board's git history, window ${reviewLimit}m — branch activity is NOT scored`,
      claimed: `time since the row entered in-flight, from the board's git history, for a lane whose branch has nothing on origin (unpushed, local-only, or pushed empty at claim); window ${claimLimit}m`,
    },
    boardFreshness: {
      rows: freshness,
      stalestMinutes: aged.length ? Math.max(...aged) : null,
      uncommitted: freshness.filter((f) => f.lastUpdated === "uncommitted").length,
      note:
        "when each scored row's line was last written (git blame). A stall over a row nobody has updated for hours may be the board's maintenance, not the lane.",
    },
    // Population beside every count: `checked` is the in-flight rows; this is
    // the rest of the board by the state the seam read, and the not-in-flight,
    // not-finished rows by name.
    population: { rows: rows.length, inFlight: inFlight.length, byState, source: runPopulation.source, delta },
    notWatched,
    // ⟨q-7b2f6c04⟩ — per-row axis, beside the one number.
    axes,
    // ⟨q-1c95f7d4⟩ Task 5 — ONE ENTRY PER SEAT WITH AN EXTERNAL OBSERVER, and nothing that
    // grows with the fleet's branches or history. Measured on this fleet the day it was
    // added: 0 seats carry a herdr marker, so this key is 121 bytes of an answer whose
    // other keys are 76 KB (`stranded.pushedBranches` alone is 35.7 KB).
    tick: fleetTick,
    stranded: { openPrs: stranded.openPrs, pushedBranches: stranded.pushedBranches, unmeasurable: stranded.unmeasurable },
    conventions: { merges: conventions.merges, routed: conventions.routed, closings: conventions.closings, unmeasurable: conventions.unmeasurable },
    mergeWindows: { windows: mergeWindows.windows, floor: new Date(MERGE_WINDOW_ADOPTED_MS).toISOString(), unmeasurable: mergeWindows.unmeasurable },
  };
  markRun({ hits: stamped, checked: scored.length, measurable, population: runPopulation });
  return {
    ok: true as const,
    ...result,
    halted: haltState().halted,
    // MISS is silent to the DUTY OFFICER — the caller decides whether to DM — and
    // never silent to the record, which markRun just wrote.
    dm: hits.length > 0,
    note:
      hits.length === 0
        ? `MISS — ${scored.length} lane(s) scored${held.length ? `, ${held.length} held behind a declared hold` : ""}${roles.length ? `, ${roles.length} role(s) present and not scored` : ""}${deliberate.length ? `, ${deliberate.length} deliberately branchless lane(s) out of population` : ""}, none stalled${unmeasurable.length ? `; ${unmeasurable.length} row(s) UNMEASURABLE for VCS activity (${unmeasurable.map((u) => u.agentId).join(", ")}) — reported, not counted as healthy` : ""}. No DM. The run IS recorded: read it with stall_clock_status, because no alert and nothing running look identical from here.`
        : undefined,
  };
}