From 83879e7b399cd81569533fb23668ab513598f9e0 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:26:34 +0200 Subject: [PATCH 1/2] fix(resync): re-diff the whole watch set when registration changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `refresh_watches` treated a registration change as affecting only the directories it registered, and polled just those. That holds for inotify, which adds and removes descriptors on a shared fd and keeps its queue, but not for notify's macOS FSEvents backend: `watch_inner`/`unwatch_inner` stop the single shared stream, the runloop teardown calls `FSEventsPurgeEventsForDeviceUpToEventId`, and `run()` restarts at `kFSEventStreamEventIdSinceNow`. Registering one new directory therefore destroys mutations already queued for directories that never left the set. A reconcile pass that admits a newly-valid seat registers its directories, so a carrier written just before that pass could lose its only notification with nothing left to re-read it: `poll_unwatched` skips it (its parent is still watched) and it is not under a newly registered directory. The transition was then never observed at all, which is what `compile_invalid_seat_does_not_block_existing_live_resync_watch` times out on for aarch64-darwin (#368) while passing on every Linux run. Measured on macOS 26.5 with notify 8.2.0: a directory watched throughout delivers its mutation 20/20 times on its own, and 18/20 when an unrelated second directory is registered immediately after the write. The same probe loses nothing on Linux. Delivery wins the race on an idle machine within 1ms, which is why a loaded 3-core CI runner sees this and a fast Mac does not. A changed registration is the same loss a backend error means, so it takes the same containment: re-read every subscribed carrier through the ordinary classified path. Digest equality keeps it silent when nothing moved, so only the recovered mutation emits. `poll_registered` had no other caller and is gone. The regression test is platform-neutral. It records the live parents as covered without handing them to the backend — the exact state a purge leaves, a registration that will never report what already happened — so it fails on Linux too without this change. Refs #368 agent-identity: dev3.direct.claude.paqjmjfq agent-persona: generalist agent-supervisor: unavailable agent-tool: Claude Code agent-tool-version: 2.1.250 agent-runtime: Claude Code 2.1.250 tooling-profile: dotfiles@a1a5f89 --- docs/vrs/06-resync/spec.md | 11 +++ src/resync.rs | 133 ++++++++++++++++++++++++++++++------- 2 files changed, 120 insertions(+), 24 deletions(-) diff --git a/docs/vrs/06-resync/spec.md b/docs/vrs/06-resync/spec.md index 2d48f209..06af4b19 100644 --- a/docs/vrs/06-resync/spec.md +++ b/docs/vrs/06-resync/spec.md @@ -147,6 +147,17 @@ meaningful. - A runtime watcher-backend error may mean mutation events were dropped, so it schedules every changed carrier through the same pending-aware classified path. Equal states remain silent. +- A change to the registration set is read as the same kind of drop, for the + whole watch set and not only the directories that changed. A backend is not + obliged to leave its other subscriptions undisturbed while it registers or + drops one: `notify`'s macOS FSEvents backend stops the single shared stream + on every watch and unwatch, purges the device's pending events, and restarts + at `kFSEventStreamEventIdSinceNow`, so a mutation already queued for a + directory that never left the set is destroyed. Linux inotify keeps its queue + across descriptor changes. Every subscribed carrier is therefore re-diffed + whenever a registration actually changed, through the same classified path; + equal states remain silent, so the recovered mutation is the only thing that + emits. - Reads open carriers nonblocking and without following the final symlink (every component for confined carriers). A proven regular file becomes `present()`; `ENOENT` or a stable non-regular replacement becomes diff --git a/src/resync.rs b/src/resync.rs index c1d071c2..90a79924 100644 --- a/src/resync.rs +++ b/src/resync.rs @@ -971,10 +971,15 @@ impl Worker { // Diff paths that were blind before registering newly recovered parents; otherwise the // new watch suppresses polling of mutations that happened during the blind interval. self.poll_unwatched(); - let registered = self.refresh_watches(); - // Registration closes the event gap first; this second digest pass covers writes between - // the pre-registration poll and watch installation. - self.poll_registered(®istered); + // Registration closes the event gap first; a second digest pass then covers writes + // between the pre-registration poll and watch installation. That pass spans the whole + // watch set, not only the directories this refresh touched: changing the registration + // can cost the backend its already-queued events for subscriptions that never moved + // (see `refresh_watches`), which is the same loss a backend error means and takes the + // same containment. Equal digests keep it silent. + if self.refresh_watches() { + self.rescan_all(); + } } fn apply_watch_sets(&mut self, refresh: WatchRefresh) { @@ -1098,19 +1103,6 @@ impl Worker { self.poll_paths(unwatched); } - fn poll_registered(&mut self, directories: &[PathBuf]) { - let paths = self - .carriers - .keys() - .filter(|path| { - path.parent() - .is_some_and(|parent| directories.iter().any(|dir| dir == parent)) - }) - .cloned() - .collect(); - self.poll_paths(paths); - } - fn rescan_all(&mut self) { self.poll_paths(self.carriers.keys().cloned().collect()); } @@ -1119,8 +1111,16 @@ impl Worker { /// dropping directories that left the set or were replaced (identity change). A replaced watch /// stays blind until the next pass rebuilds it — bounded by the reconcile interval, the same /// tradeoff `CatalogDeclarationWatcher` accepts for declarations. - fn refresh_watches(&mut self) -> Vec { - let mut registered = Vec::new(); + /// + /// Reports whether the registration set actually changed, because a backend is not obliged to + /// leave its other subscriptions undisturbed while it does. `notify`'s macOS FSEvents backend + /// stops the single shared stream on every `watch`/`unwatch`, purges the device's pending + /// events, and restarts at `kFSEventStreamEventIdSinceNow` — so registering one new directory + /// destroys mutations already queued for directories that were watched the whole time. Linux + /// inotify adds and removes descriptors on a shared fd and keeps its queue. Callers must treat + /// a changed registration as a possible drop across the entire watch set. + fn refresh_watches(&mut self) -> bool { + let mut changed = false; let mut desired: Vec = Vec::new(); for path in self.carriers.keys() { if let Some(parent) = path.parent() { @@ -1136,6 +1136,7 @@ impl Worker { if stale { if let Some(watcher) = self.watcher.as_mut() { let _ = watcher.unwatch(dir); + changed = true; } } !stale @@ -1153,11 +1154,11 @@ impl Worker { .watch(&dir, notify::RecursiveMode::NonRecursive) .is_ok() { - registered.push(dir.clone()); + changed = true; self.watched.insert(dir, identity); } } - registered + changed } fn mark_mutated(&mut self, paths: Vec) { @@ -1195,9 +1196,8 @@ impl Worker { } drop(dirty_here); } - if extend { - let registered = self.refresh_watches(); - self.poll_registered(®istered); + if extend && self.refresh_watches() { + self.rescan_all(); } } @@ -2312,6 +2312,91 @@ mod tests { assert!(events[0].contains("binding: goal")); } + #[test] + fn registering_another_directory_rediffs_the_untouched_watch_set() { + // A registration change is not free of the subscriptions it does not name: notify's + // macOS FSEvents backend stops the one shared stream on every `watch`, purges the + // device's pending events, and restarts at "since now", so mutations already queued for + // a directory that stayed in the set are destroyed. The state that leaves behind is a + // registered watch that will never report a change that already happened, and the model + // here is exact — the live parents are recorded as covered but never handed to the + // backend, so no event about them can exist. Only re-diffing the whole set recovers it. + let root = tempfile::tempdir().unwrap(); + let write_agent = |identity: &str| { + let agent_dir = root.path().join("agents/alias").join(identity); + let resources = agent_dir.join("resources"); + std::fs::create_dir_all(&resources).unwrap(); + std::fs::write( + agent_dir.join("agent.kdl"), + format!( + r#"agent "{identity}" {{ + host "alias" + command "agent" + resource "goal" uri="resources/goal.md" reason="Mission." +}}"# + ), + ) + .unwrap(); + let goal = resources.join("goal.md"); + std::fs::write(&goal, "before\n").unwrap(); + (agent_dir, goal) + }; + let (live_dir, live_goal) = write_agent("live"); + write_agent("joining"); + crate::event::publish_owner_binding_for_test(root.path(), "alias").unwrap(); + + let specs = crate::discover_strict(root.path()).specs; + let set_for = |identity: &str| { + let spec = specs + .iter() + .find(|spec| spec.path.starts_with(root.path().join("agents/alias").join(identity))) + .expect("both declarations are valid"); + watch_set_for(spec, "alias", &ResourceProfileRegistry::empty()) + }; + let live_set = set_for("live"); + let joining_set = set_for("joining"); + + let (tx, _rx) = channel::(); + let mut worker = Worker { + root: root.path().to_path_buf(), + this_host: "alias".to_owned(), + carriers: rebuild_carriers( + BTreeMap::new(), + refresh_for(vec![live_set.clone()]), + &BTreeMap::new(), + ), + subscription_sequences: BTreeMap::new(), + deadlines: BTreeMap::new(), + watched: BTreeMap::new(), + watcher: make_watcher(tx), + }; + worker.watched = worker + .carriers + .keys() + .filter_map(|path| path.parent()) + .map(|dir| (dir.to_path_buf(), dir_identity(dir))) + .collect(); + + std::fs::write(&live_goal, "changed with no watch able to report it\n").unwrap(); + + // The joining seat contributes directories the backend has not seen, so this refresh + // changes the registration set without touching the live subscription's own paths. + worker.apply_watch_sets(refresh_for(vec![live_set, joining_set])); + worker.flush_due(Instant::now() + IMMEDIATE_WINDOW + Duration::from_secs(1)); + + let event = resync_inbox_event(&live_dir); + assert_eq!(event_field(&event, "binding"), "goal"); + // The joining seat has no inbox at all: its baseline seeded silently, as a new + // subscription must, so the rescan is not simply emitting for everything it re-reads. + assert!( + !root + .path() + .join("agents/alias/joining/resources/inbox") + .exists(), + "the joining seat seeds its baseline silently" + ); + } + #[cfg(unix)] #[test] fn digesting_a_fifo_fails_without_blocking_the_worker() { From e2d4fc98ab0cedef975a1ab820462eed8fc980f6 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:36:26 +0200 Subject: [PATCH 2/2] fix(resync): count the watch attempt, not the successful watch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FSEvents purge is in the attempt. `watch_inner` calls `stop()` — which tears down the runloop and purges the device's pending events — before `append_path` runs, so a `watch` that fails because the directory went missing since `dir_identity` looked at it costs exactly as many queued events as one that succeeds. `unwatch_inner` already had this shape and was already counted. Keying the rescan on `watch(..).is_ok()` therefore left one carrier with a missing parent directory purging every other agent's queued mutations on every pass, with nothing left to re-read them: a standing version of the same defect rather than the transient one. Refs #368 agent-identity: dev3.direct.claude.paqjmjfq agent-persona: generalist agent-supervisor: unavailable agent-tool: Claude Code agent-tool-version: 2.1.250 agent-runtime: Claude Code 2.1.250 tooling-profile: dotfiles@a1a5f89 --- src/resync.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/resync.rs b/src/resync.rs index 90a79924..4e01854e 100644 --- a/src/resync.rs +++ b/src/resync.rs @@ -1150,11 +1150,14 @@ impl Worker { // Degraded mode: apply_watch_sets diffs digests at refresh cadence instead. break; }; + // The purge is in the attempt, not the outcome: FSEvents' `watch_inner` stops and + // restarts the stream before `append_path` can reject a directory that has since + // gone missing. Count the attempt, exactly as the unwatch above does. + changed = true; if watcher .watch(&dir, notify::RecursiveMode::NonRecursive) .is_ok() { - changed = true; self.watched.insert(dir, identity); } }