fix(proof): never report proof sources watched after a lost, replaced or unsettled watch (#1813)

Gaps in the narrowed proof source watcher found in review of 8b236e06: a watcher's runtime error was cleared by the next good read without replacing it; a replaced or re-pointed directory kept its old watch, since a watch was reused by path alone, including when the platform reports a change without a filename; and the reconciliation could stop with its latest dependency set never subscribed while still reporting watching. A runtime watcher error now leaves the instance unavailable for its lifetime. Every watched directory, the missions root included, is also watched at each ancestor up to the project's parent, filtered to the next segment; a change there naming that child, or naming nothing, drops the watches at and below it, and they are attached again by path. Coverage is certified only by a read whose every dependency was already watched; if the dependencies keep moving past a small budget, the sources are reported unavailable instead.

packages/daemon/src/domain/proof/source-watch.ts

@@ -3,7 +3,7 @@ import * as path from "node:path";
 import { createHash } from "node:crypto";
 import type { EventBus } from "../event-bus.js";
 import type { ScopeFsDeps } from "../scope/scope-view-projection.js";
-import { createProofPolicyRead, proofFs, readProjectReadiness } from "./judgments.js";
+import { createProofPolicyRead, proofFs, readProjectReadiness, resolveProjectRoot } from "./judgments.js";
 
 export interface ProofSourceWatch {
   close(): void;
@@ -16,22 +16,32 @@ export function proofSourceObservation(c: { get: (key: never) => unknown }) {
 /** A needed watch could not be attached: the sources are not observed (distinct from an input that can't be read). */
 class WatchFailure extends Error {}
 
-/** Per directory outside the missions root, the entry names a readiness evaluation read there. */
+/** Per directory outside the missions root, the entry names that matter there. */
 type Shallow = Map<string, Set<string>>;
 
+const within = (p: string, root: string) => p === root || p.startsWith(root + path.sep);
+function add(out: Shallow, dir: string, name: string) {
+  const names = out.get(dir) ?? new Set<string>();
+  names.add(name);
+  out.set(dir, names);
+}
+
 /** Where each path the evaluation read outside the missions root is observed: its parent, shallow, filtered to its
- *  name; for a missing parent, the nearest existing ancestor, filtered to the next segment toward it. */
-function shallowWatches(missionsRoot: string, read: Iterable<string>): Shallow {
+ *  name (for a missing parent, the nearest existing ancestor, filtered to the next segment toward it). Every watched
+ *  directory, the missions root included, is also anchored at each ancestor up to `top`, filtered to the next segment,
+ *  so a replaced or re-pointed directory on the way is seen and its watch rebound. */
+function shallowWatches(missionsRoot: string, top: string, read: Iterable<string>): Shallow {
   const out: Shallow = new Map();
   for (const file of read) {
     const target = path.resolve(file);
-    if (target === missionsRoot || target.startsWith(missionsRoot + path.sep)) continue;
+    if (within(target, missionsRoot)) continue;
     let child = target, dir = path.dirname(target);
     while (dir !== child && !proofFs.exists(dir)) { child = dir; dir = path.dirname(dir); }
     if (dir === child) continue;
-    const names = out.get(dir) ?? new Set<string>();
-    names.add(path.basename(child));
-    out.set(dir, names);
+    add(out, dir, path.basename(child));
+  }
+  for (const watched of [...out.keys(), missionsRoot]) {
+    for (let p = watched; within(p, top) && p !== top; p = path.dirname(p)) add(out, path.dirname(p), path.basename(p));
   }
   return out;
 }
@@ -44,6 +54,9 @@ function shallowWatches(missionsRoot: string, read: Iterable<string>): Shallow {
  */
 export function watchProofSources(missionsRoot: string, invalidate: () => void, bus: EventBus): ProofSourceWatch {
   missionsRoot = path.resolve(missionsRoot);
+  let projectRoot = path.dirname(missionsRoot);
+  try { projectRoot = resolveProjectRoot(missionsRoot) ?? projectRoot; } catch { /* Legacy roots keep the missions root's parent. */ }
+  const top = path.dirname(path.resolve(projectRoot));
   // ponytail: bounded local workspaces use a full semantic read after a file burst.
   // Keep this workload measured; subtree indexing is warranted only when that bound is exceeded.
   const basis = () => {
@@ -57,36 +70,51 @@ export function watchProofSources(missionsRoot: string, invalidate: () => void,
       isDirectory: (p) => proofFs.isDirectory(note(p)),
     };
     const hash = createHash("sha256").update(JSON.stringify(readProjectReadiness(missionsRoot, createProofPolicyRead(), undefined, io).missions.map(m => [m.name, m.revision]))).digest("hex");
-    return { hash, watches: shallowWatches(missionsRoot, read) };
+    return { hash, watches: shallowWatches(missionsRoot, top, read) };
   };
-  let closed = false, covered = false, readable = false;
+  // failed latches for this instance: a watcher that errored is not trusted to be replaced by a later read.
+  let closed = false, failed = false, covered = false, readable = false;
   let revision = "unavailable", timer: NodeJS.Timeout | undefined;
   let wanted: Shallow = new Map();
   const shallow = new Map<string, fs.FSWatcher>();
   let recursive: fs.FSWatcher | undefined;
   const notify = (next: string) => { try { bus.emit({ type: "proof.sources_changed", scope: missionsRoot, revision: next }); } catch { /* Quiet refresh remains the repair path. */ } };
+  /** A directory at or below `changed` may have been replaced: its watches are dropped, to be reattached by path. */
+  const stale = (changed: string) => {
+    for (const [dir, watcher] of shallow) if (within(dir, changed)) { watcher.close(); shallow.delete(dir); }
+    if (recursive && within(missionsRoot, changed)) { recursive.close(); recursive = undefined; }
+  };
   const onEvent = (dir: string | null) => (_event: string, filename: string | Buffer | null) => {
     if (closed) return;
-    // No filename (null, or none given) means "something here changed": re-read rather than guess.
-    if (dir !== null && filename != null && !wanted.get(dir)?.has(filename.toString())) return;
+    if (dir !== null) {
+      // No filename (null, or none given): anything here may have been replaced, so every watch at or below this
+      // directory is rebound by path on the next read (the recursive root too, if it lies below).
+      if (filename == null) stale(dir);
+      else if (!wanted.get(dir)?.has(filename.toString())) return;
+      else stale(path.join(dir, filename.toString()));
+    }
     schedule();
   };
   const onError = () => {
     if (closed) return;
-    covered = false; revision = "unavailable";
+    failed = true; covered = false; revision = "unavailable";
     invalidate(); notify("unavailable");
   };
-  /** Attach watches for directories newly needed, then release ones no longer needed (new coverage first).
-   *  Returns whether any were added. */
+  const watchDir = (dir: string, recursiveWatch: boolean) => {
+    try {
+      const watcher = fs.watch(dir, { recursive: recursiveWatch, persistent: false }, onEvent(recursiveWatch ? null : dir));
+      watcher.on("error", onError);
+      return watcher;
+    } catch (e) { throw new WatchFailure(`${dir}: ${(e as Error).message}`); }
+  };
+  /** Attach watches for directories newly needed (the recursive root too, if it was dropped), then release ones no
+   *  longer needed (new coverage first). Returns whether anything was attached. */
   const resubscribe = (next: Shallow): boolean => {
     let added = false;
+    if (!recursive) { recursive = watchDir(missionsRoot, true); added = true; }
     for (const dir of next.keys()) {
       if (shallow.has(dir)) continue;
-      let watcher: fs.FSWatcher;
-      try { watcher = fs.watch(dir, { recursive: false, persistent: false }, onEvent(dir)); }
-      catch (e) { throw new WatchFailure(`${dir}: ${(e as Error).message}`); }
-      watcher.on("error", onError);
-      shallow.set(dir, watcher);
+      shallow.set(dir, watchDir(dir, false));
       added = true;
     }
     wanted = next;
@@ -94,13 +122,17 @@ export function watchProofSources(missionsRoot: string, invalidate: () => void,
     return added;
   };
   /** Read the sources and keep the subscriptions in step with what was read, even when the result is unchanged.
-   *  After attaching new watches it reads again, so a change between the read and the watch is not missed. */
+   *  Coverage is certified only by a read whose every dependency was already watched before it; if the dependencies
+   *  keep moving past a small budget, the sources are reported unobserved rather than read forever. */
   const evaluate = (): string => {
     let read = basis();
-    for (let i = 0; i < 2 && resubscribe(read.watches); i++) read = basis();
-    covered = true;
-    return read.hash;
+    for (let i = 0; ; i++) {
+      if (!resubscribe(read.watches)) { covered = true; return read.hash; }
+      if (i === 2) { covered = false; return read.hash; }
+      read = basis();
+    }
   };
+  const observed = () => !closed && !failed && covered && readable;
   const schedule = () => {
     if (timer || closed) return;
     timer = setTimeout(() => {
@@ -125,8 +157,7 @@ export function watchProofSources(missionsRoot: string, invalidate: () => void,
     timer.unref();
   };
   try {
-    recursive = fs.watch(missionsRoot, { recursive: true, persistent: false }, onEvent(null));
-    recursive.on("error", onError);
+    recursive = watchDir(missionsRoot, true);
     try { revision = evaluate(); readable = true; } catch (e) {
       if (e instanceof WatchFailure) throw e;
       // Direct reads name unavailable inputs; watching may recover them.
@@ -140,7 +171,7 @@ export function watchProofSources(missionsRoot: string, invalidate: () => void,
     return { close() {}, observation: () => ({ state: "unavailable", revision }) };
   }
   return {
-    observation: () => ({ state: !closed && covered && readable ? "watching" : "unavailable", revision }),
+    observation: () => ({ state: observed() ? "watching" : "unavailable", revision }),
     close: () => {
       closed = true;
       if (timer) clearTimeout(timer);

packages/daemon/test/proof-policy-reuse.test.ts

@@ -156,13 +156,15 @@ describe("request-local proof-policy reuse", () => {
 describe("watch-local proof-policy reuse", () => {
   function observe(missions: string) {
     vi.useFakeTimers();
-    let changed!: () => void;
-    vi.spyOn(fs, "watch").mockImplementation(((_root: unknown, _options: unknown, listener: () => void) => {
-      changed = listener;
+    // Each watched directory's listener, by path: "a change" is delivered by the missions root's own watch.
+    const listeners = new Map<string, () => void>();
+    vi.spyOn(fs, "watch").mockImplementation(((root: string, _options: unknown, listener: () => void) => {
+      listeners.set(root, listener);
       return { on: vi.fn(), close: vi.fn() };
-    }) as typeof fs.watch);
+    }) as unknown as typeof fs.watch);
     const invalidate = vi.fn(), emit = vi.fn();
     const watch = watchProofSources(missions, invalidate, { emit } as unknown as EventBus);
+    const changed = () => listeners.get(missions)!();
     const oracle = () => createHash("sha256").update(JSON.stringify(readProjectReadiness(missions).missions.map(m => [m.name, m.revision]))).digest("hex");
     const event = () => {
       const previous = watch.observation().revision, calls = invalidate.mock.calls.length;
@@ -269,3 +271,66 @@ describe("watch-local proof-policy reuse", () => {
     } finally { o.watch.close(); }
   });
 });
+
+describe("QA bounded adverse coverage at 8b236e06", () => {
+  function adverseFixture() {
+    const f = fixture(), dir = f.dirs[0]!, outside = path.join(f.root, "build", "out"), evidence = path.join(outside, "report.md");
+    f.write(evidence, "accepted observation");
+    const item = readSliceReadiness(dir).items[0]!;
+    recordJudgment(f.missions, { scope: "trial/slices/slice-0", item: item.id, verdict: "accept", reason: "observed", evidence: [evidence], expectedRevision: item.revision, expectedPrevious: null }, "judge@trial", "fixture");
+    vi.useFakeTimers();
+    const handles = new Map<string, { event: (event: string, name: string | null) => void; error?: () => void }>();
+    vi.spyOn(fs, "watch").mockImplementation(((root: string, _options: unknown, listener: (event: string, name: string | null) => void) => {
+      const h = { event: listener, error: undefined as (() => void) | undefined }; handles.set(root, h);
+      return { on: (_kind: string, cb: () => void) => { h.error = cb; }, close: vi.fn() };
+    }) as unknown as typeof fs.watch);
+    const invalidate = vi.fn(), emit = vi.fn(), watch = watchProofSources(f.missions, invalidate, { emit } as unknown as EventBus);
+    const oracle = () => createHash("sha256").update(JSON.stringify(readProjectReadiness(f.missions).missions.map(m => [m.name, m.revision]))).digest("hex");
+    return { f, dir, outside, evidence, handles, watch, oracle, invalidate, emit };
+  }
+  it("QA retains unavailable after an async required-handle error and another valid event unless coverage is replaced", () => {
+    const t = adverseFixture();
+    try {
+      expect(t.watch.observation()).toEqual({ state: "watching", revision: t.oracle() });
+      const failed = t.handles.get(t.outside)!; failed.error!();
+      expect(t.watch.observation().state).toBe("unavailable");
+      t.f.write(t.evidence, "changed observation");
+      t.handles.get(t.f.missions)!.event("change", "trial/slices/slice-0/SPEC.md"); vi.advanceTimersByTime(50);
+      expect(readSliceReadiness(t.dir).state).toBe("unknown");
+      expect(t.watch.observation().revision).toBe(t.oracle());
+      // A fresh semantic read is not repaired notification coverage for the errored, unreplaced handle.
+      expect(t.watch.observation().state).toBe("unavailable");
+    } finally { t.watch.close(); }
+  });
+  it("rebinds a replaced evidence parent when its structural parent reports a change without a filename", () => {
+    const t = adverseFixture();
+    try {
+      const opened = () => vi.mocked(fs.watch).mock.calls.filter(([root]) => root === t.outside).length;
+      expect(opened()).toBe(1);
+      fs.renameSync(t.outside, `${t.outside}.old`); t.f.write(t.evidence, "changed observation");
+      // The platform names nothing: only the structural parent of the replaced directory reports a change.
+      t.handles.get(path.dirname(t.outside))!.event("rename", null); vi.advanceTimersByTime(50);
+      expect(readSliceReadiness(t.dir).state).toBe("unknown");
+      expect(t.watch.observation().revision).toBe(t.oracle());
+      // The old same-path handle cannot certify coverage: it is replaced, or the sources are reported unobserved.
+      expect(opened() === 2 || t.watch.observation().state === "unavailable").toBe(true);
+    } finally { t.watch.close(); }
+  });
+  it("QA observes replacement of an external evidence parent or honestly reports unavailable", () => {
+    const t = adverseFixture();
+    try {
+      const previous = t.watch.observation().revision;
+      expect(readSliceReadiness(t.dir).state).toBe("ready");
+      fs.renameSync(t.outside, `${t.outside}.old`); t.f.write(t.evidence, "changed observation");
+      // Directory replacement is visible at structural parents, not as a leaf event on the old directory inode.
+      for (const [root, h] of t.handles) {
+        if (root === path.dirname(t.outside)) h.event("rename", path.basename(t.outside));
+        if (root === t.f.root) h.event("rename", "build");
+      }
+      vi.advanceTimersByTime(50);
+      expect(readSliceReadiness(t.dir).state).toBe("unknown");
+      expect(t.oracle()).not.toBe(previous);
+      expect(t.watch.observation().state === "unavailable" || t.watch.observation().revision === t.oracle()).toBe(true);
+    } finally { t.watch.close(); }
+  });
+});

packages/daemon/test/proof-source-boundary.test.ts

@@ -28,8 +28,10 @@ it("watches only the missions root recursively; the project and evidence outside
   reads(() => [["exists", path.join(project, "project.yaml")], ["readBytes", path.join(project, "build", "out", "report.md")]]);
   const close = handle();
   const instance = watchProofSources(missions, vi.fn(), { emit: vi.fn() } as unknown as EventBus);
+  // Inputs at their parents, plus shallow structural anchors up to the project's parent (for replaced directories).
   expect(watch.mock.calls.map(([root, options]) => [root, options.recursive])).toEqual([
     [missions, true], [project, false], [path.join(project, "build", "out"), false],
+    [path.dirname(project), false], [path.join(project, "build"), false],
   ]);
   // Neither the project (with its sibling checkouts and build trees) nor any parent is watched recursively.
   expect(watch.mock.calls.filter(([root, options]) => options.recursive && root !== missions)).toEqual([]);
@@ -48,11 +50,12 @@ it("watches only the missions root recursively; the project and evidence outside
   notifyEvidence("rename", "report.md");
   vi.advanceTimersByTime(50);
   expect(readProjectReadiness).toHaveBeenCalledTimes(4);
+  // No filename: that directory's watch is rebound (closed, reattached by path) and the read verified again.
   notifyEvidence("change", null);
   vi.advanceTimersByTime(50);
-  expect(readProjectReadiness).toHaveBeenCalledTimes(5);
+  expect(readProjectReadiness).toHaveBeenCalledTimes(6);
   instance.close();
-  expect(close).toHaveBeenCalledTimes(3);
+  expect(close).toHaveBeenCalledTimes(6);
 });
 
 it("watches the nearest existing ancestor of a missing input, filtered to the next path segment", () => {
@@ -64,7 +67,7 @@ it("watches the nearest existing ancestor of a missing input, filtered to the ne
   reads(() => [["readBytes", path.join(project, "build", "out", "report.md")]]);
   handle();
   const instance = watchProofSources(missions, vi.fn(), { emit: vi.fn() } as unknown as EventBus);
-  expect(watch.mock.calls.map(([root, options]) => [root, options.recursive])).toEqual([[missions, true], [project, false]]);
+  expect(watch.mock.calls.map(([root, options]) => [root, options.recursive])).toEqual([[missions, true], [project, false], [path.dirname(project), false]]);
   const notify = watch.mock.calls[1]![2];
   notify("rename", "unrelated");
   vi.advanceTimersByTime(50);
@@ -90,15 +93,39 @@ it("reports unavailable and closes what it opened when a required watch fails, w
   expect(close).toHaveBeenCalledOnce();
 });
 
-it("uses only the missions watcher when the evaluation reads nothing outside it", () => {
+it("watches nothing else recursively when the evaluation reads nothing outside the missions root", () => {
   const missions = path.resolve("fixture/missions");
   resolveProjectRoot.mockReturnValue(path.dirname(missions));
   reads((root) => [["exists", path.join(root, "m1", "mission.yaml")]]);
   const close = handle();
   const instance = watchProofSources(missions, vi.fn(), { emit: vi.fn() } as unknown as EventBus);
-  expect(watch.mock.calls.map(([root, options]) => [root, options.recursive])).toEqual([[missions, true]]);
+  // Only the missions root's structural anchors are added, shallow.
+  expect(watch.mock.calls.map(([root, options]) => [root, options.recursive])).toEqual([
+    [missions, true], [path.dirname(missions), false], [path.dirname(path.dirname(missions)), false],
+  ]);
+  instance.close();
+  expect(close).toHaveBeenCalledTimes(3);
+});
+
+it("does not report watching when the inputs keep changing past the reconciliation budget", () => {
+  const project = path.resolve("fixture/project");
+  const missions = path.join(project, "missions");
+  resolveProjectRoot.mockReturnValue(project);
+  // Each read finds its evidence in a different directory (A, B, C, then C again), with the same semantic result.
+  const sets = ["a", "b", "c"].map((d) => path.join(project, d, "report.md"));
+  let call = 0;
+  readProjectReadiness.mockImplementation((_m: string, _r: unknown, _root: unknown, io?: Io) => {
+    io?.readBytes(sets[Math.min(call++, sets.length - 1)]!);
+    return { missions: [] };
+  });
+  handle();
+  const instance = watchProofSources(missions, vi.fn(), { emit: vi.fn() } as unknown as EventBus);
+  const attached = new Set(watch.mock.calls.map(([root]) => root));
+  // Either the final dependency set is actually watched and verified, or the monitor says it is not watching.
+  const finalDir = path.dirname(sets[2]!);
+  expect(instance.observation().state === "unavailable" || attached.has(finalDir)).toBe(true);
+  if (instance.observation().state === "watching") expect(readProjectReadiness.mock.calls.length).toBeGreaterThan(3);
   instance.close();
-  expect(close).toHaveBeenCalledOnce();
 });
 
 it("after close, a queued callback neither re-reads, reopens nor emits", () => {
@@ -116,7 +143,7 @@ it("after close, a queued callback neither re-reads, reopens nor emits", () => {
   notifyMissions("change", "y");
   vi.advanceTimersByTime(100);
   expect(readProjectReadiness).toHaveBeenCalledTimes(2);
-  expect(watch).toHaveBeenCalledTimes(2);
+  expect(watch).toHaveBeenCalledTimes(3);
   expect(emit).not.toHaveBeenCalled();
   expect(instance.observation().state).toBe("unavailable");
 });