test(daemon): guard health and queue reads during nodes bursts (#2469)

The daemon's burst test now also covers a burst of seat inventory reads. The existing N=24 real-HTTP cell in ps-summary-stall-red.test.ts runs a second case. A child daemon on the host-shaped fixture with today's migrations serves 24 /api/rigs/:rigId/nodes requests across its active rigs while health and an exact queue read are sampled. Every inventory reply must be non-empty and for its own rig, and the worst health and queue reads must stay under the shipped 250 ms responsiveness budget. In the run that added it, the worst reads were about 117 ms and 124 ms, against 6 ms and 9 ms without the burst. Only one paired probe completed during the burst, so this guards against a regression rather than measuring a distribution. The health route is the fixture's event-loop stub, not daemon health. The ps+summary case is unchanged. Test only; production is unchanged.

packages/daemon/test/fixtures/ps-stall-http-child.ts

@@ -19,13 +19,24 @@ import { AgentActivityStore } from "../../src/domain/agent-activity-store.js";
 import { PsProjectionService } from "../../src/domain/ps-projection.js";
 import { psRoutes } from "../../src/routes/ps.js";
 import { rigsRoutes } from "../../src/routes/rigs.js";
+import { nodesRoutes } from "../../src/routes/sessions.js";
+import { queueRoutes } from "../../src/routes/queue.js";
+import { QueueRepository } from "../../src/domain/queue-repository.js";
+import { ContextUsageStore } from "../../src/domain/context-usage-store.js";
+import { createDb } from "../../src/db/connection.js";
+import { migrate } from "../../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../../src/db/all-migrations.js";
 import { createMigratedDb, seedHostShaped } from "../helpers/seed-host-shaped.js";
 
-export function buildStallApp(db: Database.Database): Hono {
+export const NODES_QUEUE_PROBE_ID = "qitem-nodes-burst-probe";
+
+export function buildStallApp(db: Database.Database, includeNodes = false): Hono {
   const repo = new RigRepository(db);
   const eventBus = new EventBus(db);
   const agentActivity = new AgentActivityStore({ db, eventBus });
   const psService = new PsProjectionService({ db, agentActivity }); // agentActivity REQUIRED; seatActivity ABSENT
+  const queueRepo = includeNodes ? new QueueRepository(db, eventBus, { loadHumanRegistry: () => ({ ok: true, entities: [] }) }) : null;
+  const contextUsage = includeNodes ? new ContextUsageStore(db, { stateDir: "" }) : null;
 
   const app = new Hono();
   app.use("*", async (c, next) => {
@@ -33,9 +44,17 @@ export function buildStallApp(db: Database.Database): Hono {
     c.set("rigRepo" as never, repo as never);
     c.set("eventBus" as never, eventBus as never);
     c.set("agentActivityStore" as never, agentActivity as never);
+    if (includeNodes) {
+      c.set("queueRepo" as never, queueRepo as never);
+      c.set("contextUsageStore" as never, contextUsage as never);
+    }
     await next();
   });
   app.route("/api/ps", psRoutes);
+  if (includeNodes) {
+    app.route("/api/rigs/:rigId/nodes", nodesRoutes);
+    app.route("/api/queue", queueRoutes());
+  }
   app.route("/api/rigs", rigsRoutes);
   // Trivial diagnostic /healthz — NOT daemon health semantics; it measures pure
   // event-loop availability (a synchronous ps/summary handler blocks even this).
@@ -62,9 +81,17 @@ if (runAsChild) {
   process.on("SIGINT", () => closeAll(0));
   process.on("uncaughtException", (e) => { process.stderr.write(`child uncaught: ${(e as Error).message}\n`); closeAll(1); });
   try {
-    db = createMigratedDb();
+    const includeNodes = process.argv.includes("--nodes");
+    db = includeNodes ? createDb() : createMigratedDb();
+    if (includeNodes) migrate(db, ALL_MIGRATIONS);
     seedHostShaped(db);
-    const app = buildStallApp(db);
+    if (includeNodes) {
+      // Keep the frozen rig/node/event counts, with today's runtime spelling and schema.
+      db.prepare("UPDATE nodes SET runtime = 'claude-code' WHERE runtime = 'claude'").run();
+      db.prepare("INSERT INTO queue_items (qitem_id, source_session, destination_session, body, state, ts_created, ts_updated) VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)")
+        .run(NODES_QUEUE_PROBE_ID, "coordinator@fixture", "worker@fixture", "Exact queue read during nodes burst");
+    }
+    const app = buildStallApp(db, includeNodes);
     server = serve({ fetch: app.fetch, port: 0, hostname: "127.0.0.1" }, (info) => {
       process.stdout.write(`READY ${info.port}\n`);
     });

packages/daemon/test/ps-summary-stall-red.test.ts

@@ -1,7 +1,7 @@
 // slice-04 /api/ps event-loop stall — daemon regression pins + GREEN characterization.
 // The D1/D2 titles carry the word "regression": they were genuine RED at the
-// test-only gate and pin the target behavior. The production fix is NOT yet
-// committed or shipped — these pins guard the pending candidate.
+// test-only gate and pin the shipped ps/summary behavior. The default-nodes
+// mode characterizes another route; no production change is assumed necessary.
 // qitem-20260721000001-ps-stall-driver. Host-shaped SYNTHETIC only (never the
 // copied host DB). CI-rerunnable; reads no external lane marker.
 //
@@ -16,8 +16,8 @@ import net from "node:net";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { createFullTestDb } from "./helpers/test-app.js";
-import { seedHostShaped } from "./helpers/seed-host-shaped.js";
-import { buildStallApp } from "./fixtures/ps-stall-http-child.js";
+import { seedHostShaped, ACTIVE_RIGS } from "./helpers/seed-host-shaped.js";
+import { buildStallApp, NODES_QUEUE_PROBE_ID } from "./fixtures/ps-stall-http-child.js";
 import { HEALTHZ_RESPONSIVENESS_BUDGET_MS } from "../src/domain/event-loop-monitor.js";
 
 const HERE = path.dirname(fileURLToPath(import.meta.url));
@@ -149,7 +149,7 @@ describe("slice-04 ps/summary event-loop regression", () => {
   }, 60_000);
 
   // ---------- D2 (regression pin): real localhost N=24 burst health budget ----------
-  it("D2 regression: N=24 real-HTTP ps+summary burst keeps worst /healthz < budget", async () => {
+  it.each([{ name: "ps+summary", nodes: false }, { name: "default nodes", nodes: true }])("D2 regression/characterization: N=24 real-HTTP $name burst keeps reads responsive", async ({ nodes }) => {
     const env: NodeJS.ProcessEnv = { ...process.env };
     for (const k of ["OPENRIG_URL", "OPENRIG_PORT", "OPENRIG_HOME", "OPENRIG_HOST", "OPENRIG_DB",
       "RIGGED_URL", "RIGGED_PORT", "RIGGED_HOME", "RIGGED_HOST", "RIGGED_DB"]) delete env[k];
@@ -168,11 +168,15 @@ describe("slice-04 ps/summary event-loop regression", () => {
     const psStatuses: number[] = [];
     const sumStatuses: number[] = [];
     const health: { ms: number; status: number }[] = [];
+    const queue: { ms: number; status: number }[] = [];
+    const baselineHealth: { ms: number; status: number }[] = [];
+    const baselineQueue: { ms: number; status: number }[] = [];
+    const nodeCounts: number[] = [];
     let pidGone = false;
     let portClosed = false;
 
     try {
-      child = spawn(process.execPath, ["--import", "tsx", CHILD], {
+      child = spawn(process.execPath, ["--import", "tsx", CHILD, ...(nodes ? ["--nodes"] : [])], {
         cwd: REPO_ROOT, env, stdio: ["ignore", "pipe", "pipe"],
       });
       // whole-run watchdog: force-terminate the child if the run overruns 45s.
@@ -212,10 +216,36 @@ describe("slice-04 ps/summary event-loop regression", () => {
       await readyHealth.text();
       expect(readyStatus).toBe(200);
 
+      const read = async (route: string) => {
+        const t = performance.now();
+        try {
+          const r = await fetch(base + route, { signal: probeAbort.signal });
+          const body = await r.json();
+          const valid = route === "/healthz" ? body.ok === true : body.qitemId === NODES_QUEUE_PROBE_ID;
+          return { ms: +(performance.now() - t).toFixed(1), status: valid ? r.status : 0 };
+        } catch { return { ms: +(performance.now() - t).toFixed(1), status: 0 }; }
+      };
+      const sample = async (h: typeof health, q: typeof queue) => {
+        const [hs, qs] = await Promise.all([read("/healthz"), read(`/api/queue/${NODES_QUEUE_PROBE_ID}`)]);
+        h.push(hs); q.push(qs);
+      };
+      if (nodes) {
+        // Warm the same projection once before measuring without and with concurrent reads.
+        const warm = await fetch(base + "/api/rigs/rig-act-00/nodes");
+        expect(warm.status).toBe(200);
+        expect((await warm.json()).length).toBeGreaterThan(0);
+        for (let i = 0; i < 8; i++) await sample(baselineHealth, baselineQueue);
+      }
+
       // one sequential paced health-probe loop, awaited after the burst
       const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
-      probe = (async () => {
+      const startProbe = () => (async () => {
         while (!stop) {
+          if (nodes) {
+            await sample(health, queue);
+            if (!stop) await sleep(5);
+            continue;
+          }
           const t = performance.now();
           let status = 0;
           try { const r = await fetch(base + "/healthz", { signal: probeAbort.signal }); status = r.status; await r.text().catch(() => {}); }
@@ -224,12 +254,24 @@ describe("slice-04 ps/summary event-loop regression", () => {
           if (!stop) await sleep(100);
         }
       })();
+      if (!nodes) probe = startProbe();
 
       const reqs: Promise<unknown>[] = [];
       for (let i = 0; i < 24; i++) {
+        if (nodes) {
+          const rig = `rig-act-${String(i % ACTIVE_RIGS).padStart(2, "0")}`;
+          reqs.push(fetch(`${base}/api/rigs/${rig}/nodes`).then(async (r) => {
+            psStatuses.push(r.status);
+            const rows = await r.json();
+            nodeCounts.push(Array.isArray(rows) && rows.every((n: { rigId: string }) => n.rigId === rig) ? rows.length : 0);
+          }));
+          continue;
+        }
         reqs.push(fetch(base + "/api/ps").then((r) => { psStatuses.push(r.status); return r.text(); }));
         reqs.push(fetch(base + "/api/rigs/summary").then((r) => { sumStatuses.push(r.status); return r.text(); }));
       }
+      // For nodes, queue the competing reads after the burst requests, not ahead of all of them.
+      if (nodes) probe = startProbe();
       await Promise.all(reqs);
       stop = true;
       await probe;
@@ -263,15 +305,29 @@ describe("slice-04 ps/summary event-loop regression", () => {
       const measured = health.length > 0
         ? `${Math.max(...health.map((h) => h.ms))} ms over ${health.length} probes`
         : "no probes completed";
-      console.log(`[ps-summary D2] worst /healthz ${measured} (budget ${HEALTHZ_RESPONSIVENESS_BUDGET_MS} ms; `
-        + `ps 200s ${psStatuses.filter((st) => st === 200).length}/24, summary 200s ${sumStatuses.filter((st) => st === 200).length}/24)`);
+      console.log(`[${nodes ? "nodes" : "ps-summary"} D2] worst /healthz ${measured} (budget ${HEALTHZ_RESPONSIVENESS_BUDGET_MS} ms; `
+        + `${nodes ? "nodes" : "ps"} 200s ${psStatuses.filter((st) => st === 200).length}/24`
+        + (nodes ? ")" : `, summary 200s ${sumStatuses.filter((st) => st === 200).length}/24)`));
+      if (nodes) {
+        const stats = (samples: typeof health) => ({ count: samples.length, worstMs: Math.max(0, ...samples.map(s => s.ms)), statuses: samples.map(s => s.status) });
+        console.log("[nodes baseline/burst] " + JSON.stringify({ baseline: { health: stats(baselineHealth), queue: stats(baselineQueue) }, burst: { health: stats(health), queue: stats(queue) }, nodeCounts }));
+      }
     }
 
     // Teardown is guaranteed done above; now assert (the budget pin is the LAST assertion).
     expect(psStatuses).toHaveLength(24);
     expect(psStatuses.every((s) => s === 200)).toBe(true);
-    expect(sumStatuses).toHaveLength(24);
-    expect(sumStatuses.every((s) => s === 200)).toBe(true);
+    if (nodes) {
+      expect(nodeCounts).toHaveLength(24);
+      expect(nodeCounts.every(n => n > 0)).toBe(true);
+      expect(baselineHealth).toHaveLength(8);
+      expect(baselineQueue).toHaveLength(8);
+      expect([...baselineHealth, ...baselineQueue, ...queue].every(s => s.status === 200)).toBe(true);
+      expect(queue.length).toBeGreaterThan(0);
+    } else {
+      expect(sumStatuses).toHaveLength(24);
+      expect(sumStatuses.every((s) => s === 200)).toBe(true);
+    }
     expect(health.length).toBeGreaterThan(0);
     expect(health.every((h) => h.status === 200)).toBe(true);
     expect(pidGone).toBe(true);
@@ -282,5 +338,6 @@ describe("slice-04 ps/summary event-loop regression", () => {
     // starve the loop so worst /healthz exceeds the shipped 250ms responsiveness
     // budget; this pin asserts the target fix keeps it under (the acceptance test).
     expect(worst).toBeLessThan(HEALTHZ_RESPONSIVENESS_BUDGET_MS);
+    if (nodes) expect(Math.max(...queue.map(s => s.ms))).toBeLessThan(HEALTHZ_RESPONSIVENESS_BUDGET_MS);
   }, 60_000);
 });