9f6b032 · add513c · d481be0 · 0fd51c8 · 0a7c3d5 · 190b1e5
@@ -13,3 +13,5 @@
A new terminal screen offered only to site admins exports its palette `entry` fr
A new read of a table that grows (`event`, `issue`, `run`, `comment`) names the index it walks, in the migration that adds it, and a test shows it from the query plan (`test/hot-reads-indexed.test.ts` is the pattern): no `SCAN`, no `TEMP B-TREE`. Judge by plan or row count, never the clock. A read that stays slow with the best index is a summary table, kept as the rows change, before it is a worker.
A new `/settings…` page route goes in `src/http/settings_routes.ts` (`settingsRoutes`), not `src/http/server.ts`.
State shared between modules through process memory, not the database, is named here. `src/runs/held.ts`: each dispatcher pass (`src/crew/dispatch.ts`, `pump`) records which queued agent runs it held and why (valve, pace, limit), and vitals (`src/kernel/vitals.ts`) leaves those out of `queue:agent`. A record older than `HELD_FRESH_MS` is ignored, so a stopped dispatcher cannot hide its queue.
@@ -1381,6 +1381,37 @@
export function queueDepth(
    .all(new Date(now).toISOString());
}
/**
 * The queued runs of one kind that may start now, weighed against the ids the
 * dispatcher holds on purpose (#9403): which held ids are still queued, and the
 * count and oldest of the rest. Two queries, however deep the queue, both on
 * `run_queue`: the `state IN` repeats the partial index's own condition so
 * SQLite may use it, and `+kind` keeps it off the per-repository agent index.
 */
export function queuedApart(
  db: Database,
      
        junipercrew
        4d ago
        outdated
      
      A limit of 500 makes the caller's row.queued - waiting wrong when more than 500 are queued. Exclude the held ids in SQL, or count held from the held map.
    
  kind: RunKind,
  heldIds: readonly string[],
  now = Date.now(),
): { held: string[]; waiting: number; oldest: string | null } {
  const ids = JSON.stringify(heldIds);
  const at = new Date(now).toISOString();
  const eligible = "state IN ('queued','running') AND state = 'queued' AND +kind = ?1 AND (not_before IS NULL OR not_before <= ?2)";
  const held = db
    .query<{ id: string }, [RunKind, string, string]>(
      `SELECT id FROM run WHERE ${eligible} AND id IN (SELECT value FROM json_each(?3))`,
    )
    .all(kind, at, ids)
    .map((r) => r.id);
  const rest = db
    .query<{ waiting: number; oldest: string | null }, [RunKind, string, string]>(
      `SELECT COUNT(*) AS waiting, MIN(created_at) AS oldest FROM run
       WHERE ${eligible} AND id NOT IN (SELECT value FROM json_each(?3))`,
    )
    .get(kind, at, ids)!;
  return { held, waiting: rest.waiting, oldest: rest.oldest };
}
/** Every running run with the repository and issue number it is about, in one query. */
export function listRunning(
  db: Database,
@@ -43,6 +43,7 @@
import { resolveOpen } from "../core/decisions.ts";
import { accountActive } from "../core/accounts.ts";
import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
import { appendRunLog } from "../runs/log.ts";
import { noteHeld, type HoldCause } from "../runs/held.ts";
import { decideAndRecord } from "./ledger.ts";
import { bugsFirst } from "../flow/bugs.ts";
import { byDials } from "./team_share.ts";
@@ -1511,7 +1512,10 @@
function pump(
  // Builder starts go to teams in proportion to their dials (`team_share.ts`),
  // then through the valves a person turned (`flow/valve_gate.ts`).
  const gated = startOrder(app);
  // What this pass holds by policy, for vitals (`runs/held.ts`, #9403).
  const held = new Map<string, HoldCause>();
  for (const shut of gated.held) {
    held.set(shut.id, "valve");
    if (noted.has(shut.id)) continue;
    noted.add(shut.id);
    appendRunLog(app.config.dataDir, shut.id, "queued, not started: its repository's valve is shut (/flow)");
@@ -1540,11 +1544,12 @@
function pump(
      const priorityGoing =
        (priorityIn.get(home.id) ?? 0) +
        active.filter((r) => r.state === "running" && r.repoId === home.id && isPriorityRun(app, r, priorityMemo)).length;
      const held = weekHold(weekFor(app, home, new Date()), home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, paceFloor(app.db));
      if (held !== null) {
      const paced = weekHold(weekFor(app, home, new Date()), home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, paceFloor(app.db));
      if (paced !== null) {
        held.set(fresh.id, "pace");
        if (!noted.has(fresh.id)) {
          noted.add(fresh.id);
          appendRunLog(app.config.dataDir, fresh.id, held);
          appendRunLog(app.config.dataDir, fresh.id, paced);
        }
        continue;
      }
@@ -1559,6 +1564,7 @@
function pump(
            runningRuns(app.db, "agent", owner, [...inFlight.keys(), ...started]),
          );
    if (full !== undefined) {
      held.set(fresh.id, "limit");
      if (!noted.has(fresh.id)) {
        noted.add(fresh.id);
        appendRunLog(app.config.dataDir, fresh.id, `queued, not started: ${full}`);
@@ -1584,4 +1590,5 @@
function pump(
      .finally(() => inFlight.delete(queued.id));
    inFlight.set(queued.id, running);
  }
  noteHeld(app.db, held);
}
@@ -19,7 +19,9 @@
import {
  leaseLapsed,
  listRunning,
  queueDepth,
  queuedApart,
} from "../core/runs.ts";
import { HOLD_WORDS, heldRuns, type HoldCause } from "../runs/held.ts";
import { STALE_AFTER_MS } from "../runs/lease.ts";
import { ownRepoPath } from "./own_repo.ts";
import { BUDGETS } from "./budgets.ts";
@@ -268,9 +270,17 @@
function holders(issues: (string | null)[], perIssue: boolean): number {
}
/** What a long queue is waiting behind: how many wait, how many run, and the slots in force. */
function queueSentence(kind: string, queued: number, running: number, slots: number | null): string {
function queueSentence(
  kind: string,
  queued: number,
  running: number,
  slots: number | null,
  held: ReadonlyMap<HoldCause, number> = new Map(),
): string {
  const behind = slots === null ? `${running} running` : `${running} of ${slots} slots running`;
  return `The ${kind} queue has waited over ${QUEUE_WAIT_MAX_MS / 60_000} minutes: ${queued} queued, ${behind}.`;
  const parts = [...held].map(([cause, n]) => `${n} by ${HOLD_WORDS[cause]}`);
  const more = parts.length > 0 ? `; held on purpose: ${parts.join(", ")}` : "";
  return `The ${kind} queue has waited over ${QUEUE_WAIT_MAX_MS / 60_000} minutes: ${queued} queued, ${behind}${more}.`;
}
/**
@@ -383,17 +393,35 @@
export function vitals(input: VitalsInput): Vitals {
        );
      continue;
    }
    const waited = now - new Date(row.oldest).getTime();
    // Agent runs the dispatcher holds on purpose (a shut valve, the week's
    // pace, an account limit) are not waiting for a slot (#9403): age and
      
        junipercrew
        4d ago
        outdated
      
      waiting is capped by the queuedOf limit, so row.queued - waiting overcounts held runs on a deep queue. Derive the held count from the holds that are still queued.
    
    // "since" come from the rest, and the sentence names how many are held.
    let waiting = row.queued;
    let oldest: string | null = row.oldest;
    const byCause = new Map<HoldCause, number>();
    const holds = row.kind === "agent" ? heldRuns(input.db, now) : new Map<string, HoldCause>();
    if (holds.size > 0) {
      const apart = queuedApart(input.db, "agent", [...holds.keys()], now);
      waiting = apart.waiting;
      oldest = apart.oldest;
      for (const id of apart.held) {
        const cause = holds.get(id)!;
        byCause.set(cause, (byCause.get(cause) ?? 0) + 1);
      }
    }
    if (oldest === null) continue;
    const waited = now - new Date(oldest).getTime();
    if (waited > QUEUE_WAIT_MAX_MS)
      need(
        `queue:${row.kind}`,
        queueSentence(
          row.kind,
          row.queued,
          waiting,
          holders(running.filter((r) => r.run.kind === row.kind).map((r) => r.run.issueId), row.kind === "agent"),
          row.kind === "agent" ? Number(capacityValue(input.db, "crew.concurrency")) : null,
          byCause,
        ),
        new Date(row.oldest).toISOString(),
        new Date(oldest).toISOString(),
        "/settings/capacity",
      );
  }
@@ -435,8 +435,7 @@
export function startLooks(app: App, options: LooksOptions): LooksHandle {
        if (shot.ok) {
          const size = shot.value.size === undefined ? {} : { size: shot.value.size };
          vitals.push({ look, entries: shot.value.entries, ...size });
        }
        else if (isNoBrowser(shot.error))
        } else if (isNoBrowser(shot.error))
          vitals.push({ look, skipped: "no browser" });
      }
      const taken =
@@ -0,0 +1,46 @@
/**
 * The queued agent runs the dispatcher chose not to start on its last pass,
 * and which policy held each: a shut valve, the week's pace, an account at its
 * limit. Policy, not a fault: vitals leaves them out of "the agent queue has
 * waited" and names how many each cause holds (#9403), as it leaves out the
 * checks a red main holds.
 *
 * Process memory, per database, replaced whole by each dispatcher pass: the
 * one place the dispatcher and vitals share state. A pass older than
 * `HELD_FRESH_MS` is ignored, so a dispatcher that stopped cannot hide its
 * queue behind the last thing it said.
 */
import type { Database } from "bun:sqlite";
/** A dispatcher pass runs every 5 s; one not heard from in two minutes has stopped. */
export const HELD_FRESH_MS = 2 * 60_000;
/** Why the dispatcher held a run. */
      
        junipercrew
        4d ago
        outdated
      
      The reason strings are stored but only the test reads them. Use them in the vitals sentence, grouped by cause, or store ids only.
    
export type HoldCause = "valve" | "pace" | "limit";
/** How vitals names each cause: "2 by a shut valve". */
export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
  valve: "a shut valve",
  pace: "the week's pace",
  limit: "an account limit",
};
interface Pass {
  readonly at: number;
  readonly held: ReadonlyMap<string, HoldCause>;
}
const passes = new WeakMap<Database, Pass>();
/** One pass's holds, replacing the last: run id → the policy that held it. */
export function noteHeld(db: Database, held: ReadonlyMap<string, HoldCause>, at = Date.now()): void {
  passes.set(db, { at, held: new Map(held) });
}
/** The holds of the latest pass, when it is fresh; empty otherwise. */
export function heldRuns(db: Database, now = Date.now()): ReadonlyMap<string, HoldCause> {
  const pass = passes.get(db);
  if (pass === undefined || now - pass.at > HELD_FRESH_MS || pass.at - now > HELD_FRESH_MS) return new Map();
  return pass.held;
}
@@ -119,7 +119,7 @@
function groupDetail(g: Group, net: NetworkTable): Html {
      <p class="small">${g.size} requests from ${g.addresses.length} addresses, confidence ${g.confidence}.</p>
      <ul class="small">${g.reasons.map((r) => html`<li>${r}</li>`)}</ul>
      <p class="small muted">${LIMITS}</p>
      ${btn("Export for an abuse report", { href: `${BASE}/export?group=${g.id}`, size: "sm" })}
      ${btn("Export for an abuse report", { href: `/settings/traffic/scrapes/export?group=${g.id}`, size: "sm" })}
      <div class="table-scroll">
        <table class="table" data-dense>
          <caption class="sr-only">Requests behind group ${g.id}</caption>
@@ -5,6 +5,7 @@
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { listEvents } from "../src/core/events.ts";
import { queuedApart } from "../src/core/runs.ts";
import { closedByReview, reviewedClosed } from "../src/core/gate_hold.ts";
import { askToLiveByPillar } from "../src/core/pillars.ts";
@@ -75,3 +76,9 @@
test("whether the Later gate review closed an issue reads the event index, not t
  expect(many.join("\n")).toContain("SEARCH e USING");
  noScan(many);
});
test("the queued agent runs weighed against the dispatcher's holds walk the queue index", () => {
  const lines = plans(() => void queuedApart(app.db, "agent", ["a", "b"]));
  expect(lines.filter((l) => l.includes("USING INDEX run_queue")).length).toBe(2);
  noScan(lines.filter((l) => !l.startsWith("SCAN json_each")));
});
@@ -27,6 +27,7 @@
import { createUser, type User } from "../src/core/users.ts";
import { startCrew } from "../src/crew/dispatch.ts";
import { serve } from "../src/http/server.ts";
import { readRunLog } from "../src/runs/log.ts";
import { heldRuns } from "../src/runs/held.ts";
import { QUOTA_KEYS, getQuota, runningLimitSentence, setQuota } from "../src/settings/quotas.ts";
import { promote } from "./helpers/members.ts";
@@ -443,6 +444,8 @@
echo '{"type":"result","subtype":"success","is_error":false,"result":"ok","total
        readRunLog(dataDir, waiting.id).lines.join("\n").includes("1 of 1 concurrent runs") ? true : undefined,
      );
      expect(count(mine, "running")).toBe(1);
      // The pass that held it says so, for vitals (#9403).
      expect(heldRuns(app.db).get(waiting.id)).toBe("limit");
      const theirs = await assignTwo(alice, "orchard");
      await until(() => (count(theirs, "running") === 2 ? true : undefined));
@@ -0,0 +1,140 @@
/** Agent runs the dispatcher holds on purpose are not a stalled queue (#9403). */
import { afterAll, beforeAll, expect, test } from "bun:test";
import { chmodSync, writeFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createRun } from "../src/core/runs.ts";
import { createUser } from "../src/core/users.ts";
import { startCrew } from "../src/crew/dispatch.ts";
import { turnValve } from "../src/flow/valve_store.ts";
import { serve } from "../src/http/server.ts";
import { HELD_FRESH_MS, heldRuns, noteHeld } from "../src/runs/held.ts";
import { QUEUE_WAIT_MAX_MS, vitals } from "../src/kernel/vitals.ts";
let dataDir: string;
let app: App;
let repoId = "";
let ownerId = "";
const roomy = () => ({ freeBytes: 100, totalBytes: 100 });
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-vitals-held-"));
  app = createApp({ dataDir, port: 0, disk: roomy });
  const owner = await createUser(app.db, {
    origin: "seed",
    trusted: true,
    username: "holder",
    email: "[email protected]",
    password: "a-good-passphrase",
  });
  if (!owner.ok) throw new Error(String(owner.error));
  ownerId = owner.value.id;
  repoId = crypto.randomUUID();
  app.db
    .query("INSERT INTO repo (id, owner_id, name, visibility, created_at) VALUES (?1, ?2, 'held', 'private', ?3)")
    .run(repoId, owner.value.id, new Date().toISOString());
});
afterAll(async () => {
  app.close();
  await rm(dataDir, { recursive: true, force: true });
});
const agentQueue = (now: number) =>
  vitals({ db: app.db, signals: app.signals, dataDir, disk: roomy, now }).degraded.find((d) => d.cause === "queue:agent");
test("a pass's holds are read while fresh, and forgotten once the dispatcher goes quiet", () => {
  const at = Date.now();
  noteHeld(app.db, new Map([["r1", "pace"]]), at);
  expect(heldRuns(app.db, at + 1000).get("r1")).toBe("pace");
  expect(heldRuns(app.db, at + HELD_FRESH_MS + 1).size).toBe(0);
  noteHeld(app.db, new Map(), at);
  expect(heldRuns(app.db, at).size).toBe(0);
});
test("queue:agent leaves out the runs the dispatcher holds, and names them", () => {
  app.db.query("DELETE FROM run").run();
  const a = createRun(app.db, { repoId, kind: "agent" });
  const b = createRun(app.db, { repoId, kind: "agent" });
  const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
  // Nothing said: both wait, as before.
  noteHeld(app.db, new Map(), later);
  expect(agentQueue(later)?.sentence).toContain("2 queued, 0 of");
  // Every queued run held by policy: no stalled queue to report.
  noteHeld(app.db, new Map([[a.id, "pace"], [b.id, "limit"]]), later);
  expect(agentQueue(later)).toBeUndefined();
  // One held, one free: the free one is the queue, the held one is named.
  noteHeld(app.db, new Map([[a.id, "pace"]]), later);
  const one = agentQueue(later);
  expect(one?.sentence).toContain("1 queued, 0 of");
  expect(one?.sentence).toContain("held on purpose: 1 by the week's pace.");
  expect(one?.since).toBe(new Date(b.createdAt).toISOString());
  // A dispatcher that stopped cannot hide its queue behind its last pass.
  noteHeld(app.db, new Map([[a.id, "pace"], [b.id, "pace"]]), later - HELD_FRESH_MS - 1);
  expect(agentQueue(later)?.sentence).toContain("2 queued, 0 of");
});
test("the held count is grouped by cause, and counts only the held runs still queued", () => {
  app.db.query("DELETE FROM run").run();
  const runs = Array.from({ length: 5 }, () => createRun(app.db, { repoId, kind: "agent" }));
  const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
  const [a, b, c] = runs.map((r) => r.id) as [string, string, string];
  // `gone` is held but no longer queued: it must not be counted.
  noteHeld(app.db, new Map([[a, "valve"], [b, "valve"], [c, "limit"], ["gone", "pace"]]), later);
  const sentence = agentQueue(later)?.sentence ?? "";
  expect(sentence).toContain("2 queued, 0 of");
  expect(sentence).toContain("held on purpose: 2 by a shut valve, 1 by an account limit.");
  expect(sentence).not.toContain("pace");
});
test("past 500 queued, the held count and the waiting count are exact", () => {
  app.db.query("DELETE FROM run").run();
  const ids = Array.from({ length: 620 }, () => createRun(app.db, { repoId, kind: "agent" }).id);
  const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
  noteHeld(app.db, new Map(ids.slice(0, 600).map((id) => [id, "pace" as const])), later);
  const sentence = agentQueue(later)?.sentence ?? "";
  expect(sentence).toContain("20 queued, 0 of");
  expect(sentence).toContain("held on purpose: 600 by the week's pace.");
});
test("with the real dispatcher, every queued run held is quiet while the pass is fresh, and reports once it goes stale", async () => {
  app.db.query("DELETE FROM run").run();
  const servers = await serve(app);
  const fake = join(dataDir, "fake-claude");
  writeFileSync(fake, `#!/bin/sh\nif [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi\nexit 1\n`);
  chmodSync(fake, 0o755);
  turnValve(app.db, repoId, 0, ownerId);
  const a = createRun(app.db, { repoId, kind: "agent" });
  const b = createRun(app.db, { repoId, kind: "agent" });
  const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString();
  app.db.query("UPDATE run SET created_at = ?1 WHERE id IN (?2, ?3)").run(hourAgo, a.id, b.id);
  const crew = await startCrew(app, {
    forgeUrl: servers.url,
    localUrl: servers.url,
    binary: fake,
    concurrency: 2,
    budgetUsd: 1,
    runTimeoutMs: 60_000,
    tickMs: 20,
  });
  try {
    const deadline = Date.now() + 10_000;
    while (heldRuns(app.db).get(b.id) !== "valve" && Date.now() < deadline) await Bun.sleep(20);
    expect(heldRuns(app.db).get(a.id)).toBe("valve");
    expect(heldRuns(app.db).get(b.id)).toBe("valve");
    await crew.stop();
    const now = Date.now();
    expect(agentQueue(now)).toBeUndefined();
    expect(agentQueue(now + HELD_FRESH_MS + 1_000)?.sentence).toContain("2 queued, 0 of");
  } finally {
    await crew.stop();
    servers.stop();
  }
});
		@@ -13,3 +13,5 @@ A new terminal screen offered only to site admins exports its palette `entry` fr
13	13	A new read of a table that grows (`event`, `issue`, `run`, `comment`) names the index it walks, in the migration that adds it, and a test shows it from the query plan (`test/hot-reads-indexed.test.ts` is the pattern): no `SCAN`, no `TEMP B-TREE`. Judge by plan or row count, never the clock. A read that stays slow with the best index is a summary table, kept as the rows change, before it is a worker.
14	14
15	15	A new `/settings…` page route goes in `src/http/settings_routes.ts` (`settingsRoutes`), not `src/http/server.ts`.
	16
	17	State shared between modules through process memory, not the database, is named here. `src/runs/held.ts`: each dispatcher pass (`src/crew/dispatch.ts`, `pump`) records which queued agent runs it held and why (valve, pace, limit), and vitals (`src/kernel/vitals.ts`) leaves those out of `queue:agent`. A record older than `HELD_FRESH_MS` is ignored, so a stopped dispatcher cannot hide its queue.
		@@ -1381,6 +1381,37 @@ export function queueDepth(
1381	1381	.all(new Date(now).toISOString());
1382	1382	}
1383	1383
	1384	/**
	1385	* The queued runs of one kind that may start now, weighed against the ids the
	1386	* dispatcher holds on purpose (#9403): which held ids are still queued, and the
	1387	* count and oldest of the rest. Two queries, however deep the queue, both on
	1388	* `run_queue`: the `state IN` repeats the partial index's own condition so
	1389	* SQLite may use it, and `+kind` keeps it off the per-repository agent index.
	1390	*/
	1391	export function queuedApart(
	1392	db: Database,
		junipercrew 4d ago outdated A limit of 500 makes the caller's `row.queued - waiting` wrong when more than 500 are queued. Exclude the held ids in SQL, or count held from the held map.
	1393	kind: RunKind,
	1394	heldIds: readonly string[],
	1395	now = Date.now(),
	1396	): { held: string[]; waiting: number; oldest: string \| null } {
	1397	const ids = JSON.stringify(heldIds);
	1398	const at = new Date(now).toISOString();
	1399	const eligible = "state IN ('queued','running') AND state = 'queued' AND +kind = ?1 AND (not_before IS NULL OR not_before <= ?2)";
	1400	const held = db
	1401	.query<{ id: string }, [RunKind, string, string]>(
	1402	`SELECT id FROM run WHERE ${eligible} AND id IN (SELECT value FROM json_each(?3))`,
	1403	)
	1404	.all(kind, at, ids)
	1405	.map((r) => r.id);
	1406	const rest = db
	1407	.query<{ waiting: number; oldest: string \| null }, [RunKind, string, string]>(
	1408	`SELECT COUNT(*) AS waiting, MIN(created_at) AS oldest FROM run
	1409	WHERE ${eligible} AND id NOT IN (SELECT value FROM json_each(?3))`,
	1410	)
	1411	.get(kind, at, ids)!;
	1412	return { held, waiting: rest.waiting, oldest: rest.oldest };
	1413	}
	1414
1384	1415	/** Every running run with the repository and issue number it is about, in one query. */
1385	1416	export function listRunning(
1386	1417	db: Database,
		@@ -43,6 +43,7 @@ import { resolveOpen } from "../core/decisions.ts";
43	43	import { accountActive } from "../core/accounts.ts";
44	44	import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
45	45	import { appendRunLog } from "../runs/log.ts";
	46	import { noteHeld, type HoldCause } from "../runs/held.ts";
46	47	import { decideAndRecord } from "./ledger.ts";
47	48	import { bugsFirst } from "../flow/bugs.ts";
48	49	import { byDials } from "./team_share.ts";
		@@ -1511,7 +1512,10 @@ function pump(
1511	1512	// Builder starts go to teams in proportion to their dials (`team_share.ts`),
1512	1513	// then through the valves a person turned (`flow/valve_gate.ts`).
1513	1514	const gated = startOrder(app);
	1515	// What this pass holds by policy, for vitals (`runs/held.ts`, #9403).
	1516	const held = new Map<string, HoldCause>();
1514	1517	for (const shut of gated.held) {
	1518	held.set(shut.id, "valve");
1515	1519	if (noted.has(shut.id)) continue;
1516	1520	noted.add(shut.id);
1517	1521	appendRunLog(app.config.dataDir, shut.id, "queued, not started: its repository's valve is shut (/flow)");
		@@ -1540,11 +1544,12 @@ function pump(
1540	1544	const priorityGoing =
1541	1545	(priorityIn.get(home.id) ?? 0) +
1542	1546	active.filter((r) => r.state === "running" && r.repoId === home.id && isPriorityRun(app, r, priorityMemo)).length;
1543		const held = weekHold(weekFor(app, home, new Date()), home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, paceFloor(app.db));
1544		if (held !== null) {
	1547	const paced = weekHold(weekFor(app, home, new Date()), home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, paceFloor(app.db));
	1548	if (paced !== null) {
	1549	held.set(fresh.id, "pace");
1545	1550	if (!noted.has(fresh.id)) {
1546	1551	noted.add(fresh.id);
1547		appendRunLog(app.config.dataDir, fresh.id, held);
	1552	appendRunLog(app.config.dataDir, fresh.id, paced);
1548	1553	}
1549	1554	continue;
1550	1555	}
		@@ -1559,6 +1564,7 @@ function pump(
1559	1564	runningRuns(app.db, "agent", owner, [...inFlight.keys(), ...started]),
1560	1565	);
1561	1566	if (full !== undefined) {
	1567	held.set(fresh.id, "limit");
1562	1568	if (!noted.has(fresh.id)) {
1563	1569	noted.add(fresh.id);
1564	1570	appendRunLog(app.config.dataDir, fresh.id, `queued, not started: ${full}`);
		@@ -1584,4 +1590,5 @@ function pump(
1584	1590	.finally(() => inFlight.delete(queued.id));
1585	1591	inFlight.set(queued.id, running);
1586	1592	}
	1593	noteHeld(app.db, held);
1587	1594	}
		@@ -19,7 +19,9 @@ import {
19	19	leaseLapsed,
20	20	listRunning,
21	21	queueDepth,
	22	queuedApart,
22	23	} from "../core/runs.ts";
	24	import { HOLD_WORDS, heldRuns, type HoldCause } from "../runs/held.ts";
23	25	import { STALE_AFTER_MS } from "../runs/lease.ts";
24	26	import { ownRepoPath } from "./own_repo.ts";
25	27	import { BUDGETS } from "./budgets.ts";
		@@ -268,9 +270,17 @@ function holders(issues: (string \| null)[], perIssue: boolean): number {
268	270	}
269	271
270	272	/** What a long queue is waiting behind: how many wait, how many run, and the slots in force. */
271		function queueSentence(kind: string, queued: number, running: number, slots: number \| null): string {
	273	function queueSentence(
	274	kind: string,
	275	queued: number,
	276	running: number,
	277	slots: number \| null,
	278	held: ReadonlyMap<HoldCause, number> = new Map(),
	279	): string {
272	280	const behind = slots === null ? `${running} running` : `${running} of ${slots} slots running`;
273		return `The ${kind} queue has waited over ${QUEUE_WAIT_MAX_MS / 60_000} minutes: ${queued} queued, ${behind}.`;
	281	const parts = [...held].map(([cause, n]) => `${n} by ${HOLD_WORDS[cause]}`);
	282	const more = parts.length > 0 ? `; held on purpose: ${parts.join(", ")}` : "";
	283	return `The ${kind} queue has waited over ${QUEUE_WAIT_MAX_MS / 60_000} minutes: ${queued} queued, ${behind}${more}.`;
274	284	}
275	285
276	286	/**
		@@ -383,17 +393,35 @@ export function vitals(input: VitalsInput): Vitals {
383	393	);
384	394	continue;
385	395	}
386		const waited = now - new Date(row.oldest).getTime();
	396	// Agent runs the dispatcher holds on purpose (a shut valve, the week's
	397	// pace, an account limit) are not waiting for a slot (#9403): age and
		junipercrew 4d ago outdated `waiting` is capped by the queuedOf limit, so `row.queued - waiting` overcounts held runs on a deep queue. Derive the held count from the holds that are still queued.
	398	// "since" come from the rest, and the sentence names how many are held.
	399	let waiting = row.queued;
	400	let oldest: string \| null = row.oldest;
	401	const byCause = new Map<HoldCause, number>();
	402	const holds = row.kind === "agent" ? heldRuns(input.db, now) : new Map<string, HoldCause>();
	403	if (holds.size > 0) {
	404	const apart = queuedApart(input.db, "agent", [...holds.keys()], now);
	405	waiting = apart.waiting;
	406	oldest = apart.oldest;
	407	for (const id of apart.held) {
	408	const cause = holds.get(id)!;
	409	byCause.set(cause, (byCause.get(cause) ?? 0) + 1);
	410	}
	411	}
	412	if (oldest === null) continue;
	413	const waited = now - new Date(oldest).getTime();
387	414	if (waited > QUEUE_WAIT_MAX_MS)
388	415	need(
389	416	`queue:${row.kind}`,
390	417	queueSentence(
391	418	row.kind,
392		row.queued,
	419	waiting,
393	420	holders(running.filter((r) => r.run.kind === row.kind).map((r) => r.run.issueId), row.kind === "agent"),
394	421	row.kind === "agent" ? Number(capacityValue(input.db, "crew.concurrency")) : null,
	422	byCause,
395	423	),
396		new Date(row.oldest).toISOString(),
	424	new Date(oldest).toISOString(),
397	425	"/settings/capacity",
398	426	);
399	427	}
		@@ -435,8 +435,7 @@ export function startLooks(app: App, options: LooksOptions): LooksHandle {
435	435	if (shot.ok) {
436	436	const size = shot.value.size === undefined ? {} : { size: shot.value.size };
437	437	vitals.push({ look, entries: shot.value.entries, ...size });
438		}
439		else if (isNoBrowser(shot.error))
	438	} else if (isNoBrowser(shot.error))
440	439	vitals.push({ look, skipped: "no browser" });
441	440	}
442	441	const taken =
		@@ -0,0 +1,46 @@
	1	/**
	2	* The queued agent runs the dispatcher chose not to start on its last pass,
	3	* and which policy held each: a shut valve, the week's pace, an account at its
	4	* limit. Policy, not a fault: vitals leaves them out of "the agent queue has
	5	* waited" and names how many each cause holds (#9403), as it leaves out the
	6	* checks a red main holds.
	7	*
	8	* Process memory, per database, replaced whole by each dispatcher pass: the
	9	* one place the dispatcher and vitals share state. A pass older than
	10	* `HELD_FRESH_MS` is ignored, so a dispatcher that stopped cannot hide its
	11	* queue behind the last thing it said.
	12	*/
	13
	14	import type { Database } from "bun:sqlite";
	15
	16	/** A dispatcher pass runs every 5 s; one not heard from in two minutes has stopped. */
	17	export const HELD_FRESH_MS = 2 * 60_000;
	18
	19	/** Why the dispatcher held a run. */
		junipercrew 4d ago outdated The reason strings are stored but only the test reads them. Use them in the vitals sentence, grouped by cause, or store ids only.
	20	export type HoldCause = "valve" \| "pace" \| "limit";
	21
	22	/** How vitals names each cause: "2 by a shut valve". */
	23	export const HOLD_WORDS: Readonly<Record<HoldCause, string>> = {
	24	valve: "a shut valve",
	25	pace: "the week's pace",
	26	limit: "an account limit",
	27	};
	28
	29	interface Pass {
	30	readonly at: number;
	31	readonly held: ReadonlyMap<string, HoldCause>;
	32	}
	33
	34	const passes = new WeakMap<Database, Pass>();
	35
	36	/** One pass's holds, replacing the last: run id → the policy that held it. */
	37	export function noteHeld(db: Database, held: ReadonlyMap<string, HoldCause>, at = Date.now()): void {
	38	passes.set(db, { at, held: new Map(held) });
	39	}
	40
	41	/** The holds of the latest pass, when it is fresh; empty otherwise. */
	42	export function heldRuns(db: Database, now = Date.now()): ReadonlyMap<string, HoldCause> {
	43	const pass = passes.get(db);
	44	if (pass === undefined \|\| now - pass.at > HELD_FRESH_MS \|\| pass.at - now > HELD_FRESH_MS) return new Map();
	45	return pass.held;
	46	}
		@@ -119,7 +119,7 @@ function groupDetail(g: Group, net: NetworkTable): Html {
119	119	<p class="small">${g.size} requests from ${g.addresses.length} addresses, confidence ${g.confidence}.</p>
120	120	<ul class="small">${g.reasons.map((r) => html`<li>${r}</li>`)}</ul>
121	121	<p class="small muted">${LIMITS}</p>
122		${btn("Export for an abuse report", { href: `${BASE}/export?group=${g.id}`, size: "sm" })}
	122	${btn("Export for an abuse report", { href: `/settings/traffic/scrapes/export?group=${g.id}`, size: "sm" })}
123	123	<div class="table-scroll">
124	124	<table class="table" data-dense>
125	125	<caption class="sr-only">Requests behind group ${g.id}</caption>
		@@ -5,6 +5,7 @@ import { tmpdir } from "node:os";
5	5	import { join } from "node:path";
6	6	import { createApp, type App } from "../src/app.ts";
7	7	import { listEvents } from "../src/core/events.ts";
	8	import { queuedApart } from "../src/core/runs.ts";
8	9	import { closedByReview, reviewedClosed } from "../src/core/gate_hold.ts";
9	10	import { askToLiveByPillar } from "../src/core/pillars.ts";
10	11
		@@ -75,3 +76,9 @@ test("whether the Later gate review closed an issue reads the event index, not t
75	76	expect(many.join("\n")).toContain("SEARCH e USING");
76	77	noScan(many);
77	78	});
	79
	80	test("the queued agent runs weighed against the dispatcher's holds walk the queue index", () => {
	81	const lines = plans(() => void queuedApart(app.db, "agent", ["a", "b"]));
	82	expect(lines.filter((l) => l.includes("USING INDEX run_queue")).length).toBe(2);
	83	noScan(lines.filter((l) => !l.startsWith("SCAN json_each")));
	84	});
		@@ -27,6 +27,7 @@ import { createUser, type User } from "../src/core/users.ts";
27	27	import { startCrew } from "../src/crew/dispatch.ts";
28	28	import { serve } from "../src/http/server.ts";
29	29	import { readRunLog } from "../src/runs/log.ts";
	30	import { heldRuns } from "../src/runs/held.ts";
30	31	import { QUOTA_KEYS, getQuota, runningLimitSentence, setQuota } from "../src/settings/quotas.ts";
31	32	import { promote } from "./helpers/members.ts";
32	33
		@@ -443,6 +444,8 @@ echo '{"type":"result","subtype":"success","is_error":false,"result":"ok","total
443	444	readRunLog(dataDir, waiting.id).lines.join("\n").includes("1 of 1 concurrent runs") ? true : undefined,
444	445	);
445	446	expect(count(mine, "running")).toBe(1);
	447	// The pass that held it says so, for vitals (#9403).
	448	expect(heldRuns(app.db).get(waiting.id)).toBe("limit");
446	449
447	450	const theirs = await assignTwo(alice, "orchard");
448	451	await until(() => (count(theirs, "running") === 2 ? true : undefined));
		@@ -0,0 +1,140 @@
	1	/** Agent runs the dispatcher holds on purpose are not a stalled queue (#9403). */
	2
	3	import { afterAll, beforeAll, expect, test } from "bun:test";
	4	import { chmodSync, writeFileSync } from "node:fs";
	5	import { mkdtemp, rm } from "node:fs/promises";
	6	import { tmpdir } from "node:os";
	7	import { join } from "node:path";
	8	import { createApp, type App } from "../src/app.ts";
	9	import { createRun } from "../src/core/runs.ts";
	10	import { createUser } from "../src/core/users.ts";
	11	import { startCrew } from "../src/crew/dispatch.ts";
	12	import { turnValve } from "../src/flow/valve_store.ts";
	13	import { serve } from "../src/http/server.ts";
	14	import { HELD_FRESH_MS, heldRuns, noteHeld } from "../src/runs/held.ts";
	15	import { QUEUE_WAIT_MAX_MS, vitals } from "../src/kernel/vitals.ts";
	16
	17	let dataDir: string;
	18	let app: App;
	19	let repoId = "";
	20	let ownerId = "";
	21	const roomy = () => ({ freeBytes: 100, totalBytes: 100 });
	22
	23	beforeAll(async () => {
	24	dataDir = await mkdtemp(join(tmpdir(), "forge-vitals-held-"));
	25	app = createApp({ dataDir, port: 0, disk: roomy });
	26	const owner = await createUser(app.db, {
	27	origin: "seed",
	28	trusted: true,
	29	username: "holder",
	30	email: "[email protected]",
	31	password: "a-good-passphrase",
	32	});
	33	if (!owner.ok) throw new Error(String(owner.error));
	34	ownerId = owner.value.id;
	35	repoId = crypto.randomUUID();
	36	app.db
	37	.query("INSERT INTO repo (id, owner_id, name, visibility, created_at) VALUES (?1, ?2, 'held', 'private', ?3)")
	38	.run(repoId, owner.value.id, new Date().toISOString());
	39	});
	40
	41	afterAll(async () => {
	42	app.close();
	43	await rm(dataDir, { recursive: true, force: true });
	44	});
	45
	46	const agentQueue = (now: number) =>
	47	vitals({ db: app.db, signals: app.signals, dataDir, disk: roomy, now }).degraded.find((d) => d.cause === "queue:agent");
	48
	49	test("a pass's holds are read while fresh, and forgotten once the dispatcher goes quiet", () => {
	50	const at = Date.now();
	51	noteHeld(app.db, new Map([["r1", "pace"]]), at);
	52	expect(heldRuns(app.db, at + 1000).get("r1")).toBe("pace");
	53	expect(heldRuns(app.db, at + HELD_FRESH_MS + 1).size).toBe(0);
	54	noteHeld(app.db, new Map(), at);
	55	expect(heldRuns(app.db, at).size).toBe(0);
	56	});
	57
	58	test("queue:agent leaves out the runs the dispatcher holds, and names them", () => {
	59	app.db.query("DELETE FROM run").run();
	60	const a = createRun(app.db, { repoId, kind: "agent" });
	61	const b = createRun(app.db, { repoId, kind: "agent" });
	62	const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
	63
	64	// Nothing said: both wait, as before.
	65	noteHeld(app.db, new Map(), later);
	66	expect(agentQueue(later)?.sentence).toContain("2 queued, 0 of");
	67
	68	// Every queued run held by policy: no stalled queue to report.
	69	noteHeld(app.db, new Map([[a.id, "pace"], [b.id, "limit"]]), later);
	70	expect(agentQueue(later)).toBeUndefined();
	71
	72	// One held, one free: the free one is the queue, the held one is named.
	73	noteHeld(app.db, new Map([[a.id, "pace"]]), later);
	74	const one = agentQueue(later);
	75	expect(one?.sentence).toContain("1 queued, 0 of");
	76	expect(one?.sentence).toContain("held on purpose: 1 by the week's pace.");
	77	expect(one?.since).toBe(new Date(b.createdAt).toISOString());
	78
	79	// A dispatcher that stopped cannot hide its queue behind its last pass.
	80	noteHeld(app.db, new Map([[a.id, "pace"], [b.id, "pace"]]), later - HELD_FRESH_MS - 1);
	81	expect(agentQueue(later)?.sentence).toContain("2 queued, 0 of");
	82	});
	83
	84	test("the held count is grouped by cause, and counts only the held runs still queued", () => {
	85	app.db.query("DELETE FROM run").run();
	86	const runs = Array.from({ length: 5 }, () => createRun(app.db, { repoId, kind: "agent" }));
	87	const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
	88	const [a, b, c] = runs.map((r) => r.id) as [string, string, string];
	89	// `gone` is held but no longer queued: it must not be counted.
	90	noteHeld(app.db, new Map([[a, "valve"], [b, "valve"], [c, "limit"], ["gone", "pace"]]), later);
	91	const sentence = agentQueue(later)?.sentence ?? "";
	92	expect(sentence).toContain("2 queued, 0 of");
	93	expect(sentence).toContain("held on purpose: 2 by a shut valve, 1 by an account limit.");
	94	expect(sentence).not.toContain("pace");
	95	});
	96
	97	test("past 500 queued, the held count and the waiting count are exact", () => {
	98	app.db.query("DELETE FROM run").run();
	99	const ids = Array.from({ length: 620 }, () => createRun(app.db, { repoId, kind: "agent" }).id);
	100	const later = Date.now() + 2 * QUEUE_WAIT_MAX_MS;
	101	noteHeld(app.db, new Map(ids.slice(0, 600).map((id) => [id, "pace" as const])), later);
	102	const sentence = agentQueue(later)?.sentence ?? "";
	103	expect(sentence).toContain("20 queued, 0 of");
	104	expect(sentence).toContain("held on purpose: 600 by the week's pace.");
	105	});
	106
	107	test("with the real dispatcher, every queued run held is quiet while the pass is fresh, and reports once it goes stale", async () => {
	108	app.db.query("DELETE FROM run").run();
	109	const servers = await serve(app);
	110	const fake = join(dataDir, "fake-claude");
	111	writeFileSync(fake, `#!/bin/sh\nif [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi\nexit 1\n`);
	112	chmodSync(fake, 0o755);
	113	turnValve(app.db, repoId, 0, ownerId);
	114	const a = createRun(app.db, { repoId, kind: "agent" });
	115	const b = createRun(app.db, { repoId, kind: "agent" });
	116	const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString();
	117	app.db.query("UPDATE run SET created_at = ?1 WHERE id IN (?2, ?3)").run(hourAgo, a.id, b.id);
	118	const crew = await startCrew(app, {
	119	forgeUrl: servers.url,
	120	localUrl: servers.url,
	121	binary: fake,
	122	concurrency: 2,
	123	budgetUsd: 1,
	124	runTimeoutMs: 60_000,
	125	tickMs: 20,
	126	});
	127	try {
	128	const deadline = Date.now() + 10_000;
	129	while (heldRuns(app.db).get(b.id) !== "valve" && Date.now() < deadline) await Bun.sleep(20);
	130	expect(heldRuns(app.db).get(a.id)).toBe("valve");
	131	expect(heldRuns(app.db).get(b.id)).toBe("valve");
	132	await crew.stop();
	133	const now = Date.now();
	134	expect(agentQueue(now)).toBeUndefined();
	135	expect(agentQueue(now + HELD_FRESH_MS + 1_000)?.sentence).toContain("2 queued, 0 of");
	136	} finally {
	137	await crew.stop();
	138	servers.stop();
	139	}
	140	});