Bound live observation work across requests and polling

README.md

@@ -264,9 +264,32 @@ graph updates, no reload. Add `--poll <secs>` (with `--env`) to also re-query li
 drift on an interval and push updates when a node's status changes:
 
 ```sh
-behold serve ./infra --env prod --poll 30   # watch source + poll drift every 30s
+behold serve ./infra --env prod --poll 30   # wait 30s between completed drift sweeps
 ```
 
+**Read budget.** HTTP observations, background polling, and frame captures share
+one Chant subprocess budget: two reads at once by default (one on a one-CPU
+host). `BEHOLD_ESTATE_CONCURRENCY` overrides that process-wide limit. Up to 64
+distinct reads can wait; excess reads fail explicitly rather than growing an
+unbounded queue. Identical in-flight reads share work only when the project,
+resolved Chant, source stamp, argv, and effective environment match. Completed
+live results are not cached. Source watcher invalidation also separates new
+requests from work begun before an edit.
+
+A running read has a 180-second deadline, configurable with
+`BEHOLD_READ_TIMEOUT_MS`. A disconnected GET releases its subscription; when no
+callers remain, the read is canceled. On Unix, cancellation stops the whole
+npm/tsx/Node process group, escalating from TERM to KILL after one second. On
+Windows only the direct child is terminated. Delegated writes retain their
+existing lifecycle and are never deduplicated as reads.
+
+The browser runs one refresh at a time and collapses intervening notifications
+into one follow-up. It no longer starts additional reads at 3/8/15-second offsets.
+Polling uses the same estate namespace bindings as the HTTP overlay and reuses
+the primary member's observation for lanes capture. Slow sweeps extend the poll
+period; they do not overlap the next sweep. These bounds control duplicate work;
+they do not eliminate Chant's underlying live discovery cost.
+
 behold shells the **project's own** chant (resolved from the project's
 `node_modules` first), so the project decides the chant version — pin it to
 `@intentius/chant ^0.18.1` or later for the live overlay (`graph --live` observed

src/chant-read-integration.test.ts

@@ -0,0 +1,120 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { runChantRaw, isScheduledRead, resolveChant } from "./chant.ts";
+import { createApp } from "./server.ts";
+import { invalidateReadGeneration, withReadSignal } from "./read-scheduler.ts";
+
+// Real subprocesses, no cluster. The fixture writes its instrumentation outside
+// the member so observing it cannot accidentally change the source stamp.
+let scratch: string, project: string, log: string;
+beforeEach(() => {
+  scratch = mkdtempSync(join(tmpdir(), "behold-read-test-"));
+  project = join(scratch, "project"); log = join(scratch, "events");
+  const pkg = join(project, "node_modules/@intentius/chant");
+  mkdirSync(pkg, { recursive: true });
+  writeFileSync(join(project, "source.ts"), "// fixture source\n");
+  writeFileSync(join(pkg, "package.json"), JSON.stringify({ name: "@intentius/chant", version: "0.54.0", main: "chant.cjs", bin: { chant: "chant.cjs" } }));
+  writeFileSync(join(pkg, "chant.cjs"), `#!/usr/bin/env node
+const fs = require('node:fs');
+const log = ${JSON.stringify(log)};
+const event = (value) => fs.appendFileSync(log, JSON.stringify(value) + '\\n');
+event({ event: 'start', pid: process.pid, args: process.argv.slice(2) });
+if (process.argv.includes('--tree')) {
+  const child = require('node:child_process').spawn(process.execPath, ['-e',
+    "process.on('SIGTERM', () => {}); require('node:fs').appendFileSync(" + JSON.stringify(log) + ", JSON.stringify({event:'child',pid:process.pid})+'\\\\n'); setInterval(() => {}, 1000);"
+  ], { stdio: 'ignore' });
+  process.on('SIGTERM', () => process.exit(0));
+  setInterval(() => {}, 1000);
+} else {
+  setTimeout(() => {
+    event({ event: 'finish', pid: process.pid });
+    process.stdout.write(JSON.stringify({ pid: process.pid, variant: process.env.READ_TEST_VARIANT }));
+  }, process.argv.includes('--slow') ? 1000 : 150);
+}
+`, { mode: 0o755 });
+  expect(resolveChant(project).source).toBe("project");
+  vi.stubEnv("BEHOLD_ESTATE_CONCURRENCY", "2");
+});
+afterEach(() => { vi.unstubAllEnvs(); rmSync(scratch, { recursive: true, force: true }); });
+const events = (): { event: string; pid: number; args?: string[] }[] => {
+  try { return readFileSync(log, "utf8").trim().split("\n").map((s) => JSON.parse(s)); } catch { return []; }
+};
+
+describe("real Chant read scheduling", () => {
+  it("matching HTTP and background reads share a process, then the next read is fresh", async () => {
+    const args = ["graph", "source.ts", "--live", "--namespace", "app"];
+    const first = withReadSignal(new AbortController().signal, () => runChantRaw(args, project));
+    const second = runChantRaw(args, project);
+    const [a, b] = await Promise.all([first, second]);
+    expect(a.stdout).toBe(b.stdout);
+    expect(events().filter((e) => e.event === "start")).toHaveLength(1);
+    a.stdout = "caller mutation";
+    expect(b.stdout).not.toBe(a.stdout);
+    const fresh = await runChantRaw(args, project);
+    expect(fresh.stdout).not.toBe(b.stdout);
+  });
+
+  it("the HTTP middleware isolates cancellation between matching requests", async () => {
+    const app = createApp({ projectDir: project, port: 0 });
+    app.onError((error, c) => c.json({ error: error.message }, 500));
+    const caller = new AbortController();
+    const first = app.request(new Request("http://localhost/api/diff?env=home", { signal: caller.signal }));
+    const second = app.request("/api/diff?env=home");
+    await vi.waitFor(() => expect(events().filter((e) => e.event === "start")).toHaveLength(1));
+    caller.abort(new Error("client disconnected"));
+    expect((await first).status).toBe(500);
+    expect((await second).status).toBe(200);
+    expect(events().filter((e) => e.event === "start")).toHaveLength(1);
+  });
+
+  it("namespace, effective environment, source edits, and watcher invalidation isolate reads", async () => {
+    const args = ["graph", "source.ts", "--live"];
+    const reads = [runChantRaw(args, project)];
+    reads.push(runChantRaw([...args, "--namespace", "other"], project));
+    reads.push(runChantRaw(args, project, { READ_TEST_VARIANT: "other" }));
+    writeFileSync(join(project, "source.ts"), "// changed while reading\n");
+    reads.push(runChantRaw(args, project));
+    invalidateReadGeneration();
+    reads.push(runChantRaw(args, project));
+    const results = await Promise.all(reads);
+    expect(new Set(results.map((r) => JSON.parse(r.stdout).pid)).size).toBe(5);
+    let active = 0, peak = 0;
+    for (const e of events()) { active += e.event === "start" ? 1 : -1; peak = Math.max(peak, active); }
+    expect(peak).toBe(2);
+  });
+
+  it("a timed-out process releases the shared budget", async () => {
+    vi.stubEnv("BEHOLD_READ_TIMEOUT_MS", "50");
+    await expect(runChantRaw(["graph", "source.ts", "--slow"], project)).rejects.toThrow("exceeded 50ms");
+    vi.stubEnv("BEHOLD_READ_TIMEOUT_MS", "3000");
+    expect((await runChantRaw(["graph", "source.ts"], project)).code).toBe(0);
+  });
+
+  it.skipIf(process.platform === "win32")("disconnect kills the whole worker group, including a descendant ignoring TERM", async () => {
+    const caller = new AbortController();
+    const read = withReadSignal(caller.signal, () => runChantRaw(["graph", "source.ts", "--tree"], project));
+    // Handle rejection immediately; disconnect rejects the subscriber before
+    // the worker group has necessarily finished closing its pipes.
+    const rejected = read.catch((error) => error);
+    let child = 0;
+    try {
+      await vi.waitFor(() => { child = events().find((e) => e.event === "child")?.pid ?? 0; expect(child).toBeGreaterThan(0); });
+      caller.abort(new Error("client left"));
+      expect((await rejected).message).toBe("client left");
+      await vi.waitFor(() => expect(() => process.kill(child, 0)).toThrow(), { timeout: 3000 });
+    } finally {
+      caller.abort(new Error("client left"));
+      for (const e of events()) { try { process.kill(e.pid, "SIGKILL"); } catch {} }
+    }
+  });
+
+  it("delegated mutations are never scheduled or deduplicated as reads", async () => {
+    for (const args of [["approve", "op", "gate"], ["run", "apply"], ["emulator", "up"], ["carve", "emit"], ["build"]]) {
+      expect(isScheduledRead(args)).toBe(false);
+    }
+    const results = await Promise.all([runChantRaw(["approve", "op", "gate"], project), runChantRaw(["approve", "op", "gate"], project)]);
+    expect(results[0].stdout).not.toBe(results[1].stdout);
+  });
+});

src/chant.test.ts

@@ -562,11 +562,11 @@ describe("applyArgs", () => {
 describe("runChantRaw — env override reaches the spawn", () => {
   beforeEach(() => vi.mocked(spawnMock).mockReset());
 
-  it("spawns with no explicit `env` option when no override is given — inherits process.env as before", async () => {
+  it("snapshots the inherited environment before a read can queue", async () => {
     vi.mocked(spawnMock).mockReturnValue(fakeProc(0, "{}"));
     await runChantRaw(["graph", "src", "--format", "ir"], "/proj");
     const opts = vi.mocked(spawnMock).mock.calls[0]![2] as { env?: unknown } | undefined;
-    expect(opts?.env).toBeUndefined();
+    expect(opts?.env).toEqual(process.env);
   });
 
   it("merges the env override over process.env for exactly this spawn (M2 tier/target lenses)", async () => {

src/chant.ts

@@ -26,6 +26,9 @@ import { detectProject } from "./project.ts";
 // (package.json `exports["./yaml"]` has no compiled-JS condition), which a
 // plain `node dist/cli.js` cannot import unbundled (Node refuses to
 // type-strip files under node_modules).
+import { createHash } from "node:crypto";
+import { memberSourceStamp } from "./member-source.ts";
+import { ReadScheduler, currentReadSignal, readGeneration } from "./read-scheduler.ts";
 import { parseYAML } from "@intentius/chant/yaml";
 import { carveStatusArgs, type CarveStatusJson } from "./carve-manifest.ts";
 import { dropForeignDeclarations } from "./foreign.ts";
@@ -302,41 +305,94 @@ export interface ChantRun {
 }
 
 /** Run the chant bin, capturing stdout/stderr and the exit code. Never rejects on
- * a non-zero exit (only on a spawn failure) — a failing exit is data.
+ * a non-zero exit — a failing exit is data. Scheduled reads can also reject
+ * on queue saturation, cancellation, or deadline expiry.
  * `envOverride` (M2, #54: the tier/target lenses' `envOverridesFor`) merges over
  * `process.env` for this one spawn only — never a global mutation, so a picked
  * lens on one request can't bleed into a concurrent request on another. */
+// Positive read allowlist: delegated mutations (including build/carve emit) must
+// never be coalesced or canceled by the observation scheduler.
+export function isScheduledRead(args: string[]): boolean {
+  return args[0] === "graph" ||
+    (args[0] === "components" && args[1] === "status") ||
+    (args[0] === "lifecycle" && ["diff", "plan"].includes(args[1])) ||
+    (args[0] === "helm" && ["renders", "diff"].includes(args[1])) ||
+    (args[0] === "run" && args[1] === "status") ||
+    (args[0] === "operator" && ["status", "log"].includes(args[1]));
+}
+const reads = new ReadScheduler<ChantRun>();
+let unstampableRead = 0;
+
 export function runChantRaw(
   args: string[],
   projectDir?: string,
   envOverride?: Record<string, string>,
 ): Promise<ChantRun> {
+  const chant = resolveChant(projectDir);
+  if (!isScheduledRead(args)) return spawnChant(chant.bin, args, projectDir, envOverride);
+  const dir = resolve(projectDir ?? process.cwd());
+  const stamp = memberSourceStamp(dir);
+  const effectiveEnv = { ...process.env, ...envOverride };
+  const environment = Object.entries(effectiveEnv).sort(([a], [b]) => a.localeCompare(b));
+  // Include the resolved compiler, exact argv (namespace/lens/env included),
+  // source identity and effective environment. Unreadable source cannot share.
+  const key = createHash("sha256").update(JSON.stringify([
+    dir, chant.bin, chant.version, args, environment, readGeneration(), stamp ?? ++unstampableRead,
+  ])).digest("hex");
+  return reads.read(key, (signal) => spawnChant(chant.bin, args, projectDir, effectiveEnv, signal), currentReadSignal())
+    .then((result) => ({ ...result })); // callers own their result; JSON is parsed separately
+}
+
+function spawnChant(
+  bin: string,
+  args: string[],
+  projectDir?: string,
+  envOverride?: NodeJS.ProcessEnv,
+  signal?: AbortSignal,
+): Promise<ChantRun> {
+  if (signal?.aborted) return Promise.reject(signal.reason);
   return new Promise((resolvePromise, reject) => {
-    // Run in the project dir: `chant graph --live` reads the current working
-    // directory (not the path arg), so the cwd must be the project for the live
-    // and overlay paths to observe the right environment.
-    const proc = spawn(chantBin(projectDir), args, {
+    // Read workers own a process group: killing only npx leaves tsx/Node alive.
+    // Writes retain their existing lifetime and process behavior.
+    const grouped = !!signal && process.platform !== "win32";
+    const proc = spawn(bin, args, {
       ...(projectDir ? { cwd: projectDir } : {}),
-      ...(envOverride ? { env: { ...process.env, ...envOverride } } : {}),
+      ...(envOverride ? { env: signal ? envOverride : { ...process.env, ...envOverride } } : {}),
+      ...(grouped ? { detached: true } : {}),
       stdio: ["ignore", "pipe", "pipe"],
     });
-    // Accumulate raw Buffer chunks and decode once at the end. Coercing each
-    // chunk to a string as it arrives (`s += d`) corrupts a multi-byte UTF-8
-    // character that straddles a chunk boundary — which for loomster's ~200KB
-    // entity-graph IR reliably mangles the JSON near the 64KB highWaterMark and
-    // makes `JSON.parse` throw. Concatenating bytes first avoids the split.
     const outChunks: Buffer[] = [];
     const errChunks: Buffer[] = [];
+    let escalation: ReturnType<typeof setTimeout> | undefined;
+    const kill = (name: NodeJS.Signals): void => {
+      try {
+        if (grouped && proc.pid) process.kill(-proc.pid, name);
+        else proc.kill(name);
+      } catch (error) {
+        if ((error as NodeJS.ErrnoException).code !== "ESRCH") proc.kill(name);
+      }
+    };
+    const abort = (): void => {
+      kill("SIGTERM");
+      escalation = setTimeout(() => kill("SIGKILL"), 1000);
+    };
+    signal?.addEventListener("abort", abort, { once: true });
+    const cleanup = (): void => {
+      signal?.removeEventListener("abort", abort);
+      clearTimeout(escalation);
+      // A wrapper can exit before a descendant that ignores TERM. Finish the
+      // group before releasing its budget, even if its pipes already closed.
+      if (signal?.aborted && grouped) kill("SIGKILL");
+    };
+    // Decode once: a UTF-8 character can straddle stdout chunks.
     proc.stdout.on("data", (d: Buffer) => outChunks.push(d));
     proc.stderr.on("data", (d: Buffer) => errChunks.push(d));
-    proc.on("error", reject);
-    proc.on("close", (code) =>
-      resolvePromise({
-        code: code ?? 1,
-        stdout: Buffer.concat(outChunks).toString("utf8"),
-        stderr: Buffer.concat(errChunks).toString("utf8"),
-      }),
-    );
+    proc.on("error", (error) => { cleanup(); reject(error); });
+    proc.on("close", (code) => {
+      cleanup();
+      if (signal?.aborted) { reject(signal.reason); return; }
+      resolvePromise({ code: code ?? 1, stdout: Buffer.concat(outChunks).toString("utf8"), stderr: Buffer.concat(errChunks).toString("utf8") });
+    });
   });
 }
 

src/estate.ts

@@ -21,7 +21,7 @@
 import { joinCarvedSources } from "./carve-manifest.ts";
 import { carveStatesFor } from "./carve-discovery.ts";
 import { statSync } from "node:fs";
-import { availableParallelism } from "node:os";
+import { readConcurrency } from "./read-scheduler.ts";
 import { join, resolve, sep } from "node:path";
 import { composeStacks, shortStackNames, type GraphIR } from "@intentius/pinhole";
 import { graphIr, meetsFloor, resolveChant, type GraphOptions } from "./chant.ts";
@@ -41,15 +41,10 @@ import { CLUSTER_SCOPED } from "./zoom-notes.ts";
 // host busy without drowning it.
 // ---------------------------------------------------------------------------
 
-/** How many member chant processes one estate read runs at once: the host's
- * parallelism capped at 4 (each spawn wants a core-plus for its TS eval),
- * never more lanes than members. `BEHOLD_ESTATE_CONCURRENCY` overrides the
- * cap for tuning a live estate; anything unparseable or < 1 is ignored
- * rather than honoured into a stall. Exported for testing. */
+/** Per-composition pipelining; runChantRaw also enforces this budget across
+ * all simultaneous estate requests, background polls, and frame captures. */
 export function estateReadPool(members: number, env: Record<string, string | undefined> = process.env): number {
-  const override = Number.parseInt(env.BEHOLD_ESTATE_CONCURRENCY ?? "", 10);
-  const cap = Number.isInteger(override) && override >= 1 ? override : Math.min(4, availableParallelism());
-  return Math.max(1, Math.min(cap, members));
+  return Math.max(1, Math.min(readConcurrency(env), members));
 }
 
 /** `Promise.all(items.map(fn))` with at most `width` calls in flight.