fix(run): stop per-pass advisory churn on large catalogs

src/materialize.rs

@@ -531,7 +531,8 @@ fn deep_merge(
             // longer states are removed, so a matcher group holding a user hook beside an st2
             // one keeps the user's; containers left with nothing but empty arrays are dropped.
             target.retain_mut(|element| {
-                if patch.contains(element) || !supersede_managed_hooks
+                if patch.contains(element)
+                    || !supersede_managed_hooks
                     || !contains_owned_string(element)
                 {
                     return true;
@@ -602,7 +603,23 @@ fn contains_owned_string(value: &serde_json::Value) -> bool {
     }
 }
 
+fn has_git_marker(workspace: &Path) -> bool {
+    workspace
+        .ancestors()
+        .any(|ancestor| ancestor.join(".git").exists())
+}
+
 fn git_exclude(workspace: &Path, line: &str) -> Result<bool> {
+    // A failed `git rev-parse` costs a process spawn per op per pass. On catalogs whose
+    // workspaces are plain directories this repeats every reconcile (measured live: ~28 spawns
+    // plus stderr captures per pass on dev3). The marker probe answers "not a repo" for free;
+    // `.git` may be a directory or a file (linked worktrees), which `exists` covers either way.
+    if !has_git_marker(workspace) {
+        anyhow::bail!(
+            "{} is not a Git worktree (no .git marker)",
+            workspace.display()
+        );
+    }
     let output = Command::new("git")
         .args(["-C"])
         .arg(workspace)
@@ -1169,6 +1186,25 @@ mod tests {
         }
     }
 
+    #[test]
+    fn git_exclude_reports_missing_repo_without_spawning_git() {
+        let dir = tempfile::tempdir().unwrap();
+        let workspace = dir.path().join("ws");
+        std::fs::create_dir_all(&workspace).unwrap();
+
+        assert!(!has_git_marker(&workspace));
+        let error = git_exclude(&workspace, ".st2/").unwrap_err();
+        assert!(
+            error.to_string().contains("no .git marker"),
+            "the no-repo case must be answered by the marker probe: {error:#}"
+        );
+
+        // A `.git` entry anywhere above the workspace (directory or file form, as linked
+        // worktrees use) re-enables the real git path.
+        std::fs::create_dir_all(dir.path().join(".git")).unwrap();
+        assert!(has_git_marker(&workspace));
+    }
+
     #[test]
     fn deep_merge_preserves_unrelated_keys_and_replaces_arrays() {
         let mut target = serde_json::json!({

src/run.rs

@@ -1856,6 +1856,7 @@ pub fn up_loop_specs(
     let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE);
     let mut presentation_cursor = PresentationPatchCursor::default();
     let mut reported_flapping: HashSet<String> = HashSet::new();
+    let mut recurring_warnings = RecurringWarnings::default();
     let park_channel = ParkChannel::for_supervisor(root, this_host);
     loop {
         let mut pre = UpReport::default();
@@ -1873,7 +1874,6 @@ pub fn up_loop_specs(
         for id in &report.unparked {
             reported_flapping.remove(id);
         }
-        park_channel.publish(&cap, &mut report);
         for cl in &report.crash_loops {
             if reported_flapping.insert(cl.pty_id.clone()) {
                 eprintln!(
@@ -1883,6 +1883,7 @@ pub fn up_loop_specs(
                 surface_crash_loop(root, this_host, cl);
             }
         }
+        recurring_warnings.filter(&mut report);
         on_report(&report);
         if STOP.load(Ordering::SeqCst) {
             break;
@@ -2048,6 +2049,24 @@ fn best_effort_catalog_watcher(
     }
 }
 
+/// Suppresses warnings that persist across passes while still re-surfacing one that clears and
+/// returns. An unchanged advisory failure (a non-Git workspace failing its git-exclude, say) must
+/// be diagnosed once, not once per reconcile pass.
+#[derive(Default)]
+struct RecurringWarnings {
+    emitted: HashSet<String>,
+}
+
+impl RecurringWarnings {
+    fn filter(&mut self, report: &mut UpReport) {
+        let current: HashSet<_> = report.warnings.iter().cloned().collect();
+        self.emitted.retain(|warning| current.contains(warning));
+        report
+            .warnings
+            .retain(|warning| self.emitted.insert(warning.clone()));
+    }
+}
+
 /// The supervisor loop: reconcile on a timer AND on folder changes until interrupted. The fs-watch is
 /// best-effort; the `interval` timer is the always-on fallback. `on_report` is called once per pass
 pub fn up_loop(
@@ -2093,6 +2112,7 @@ fn up_loop_until(
     // agent's supervisor over the native bus, so a crash-loop isn't only visible to whoever is
     // watching the log.
     let mut reported_flapping: HashSet<String> = HashSet::new();
+    let mut recurring_warnings = RecurringWarnings::default();
     let park_channel = ParkChannel::for_supervisor(root, this_host);
 
     loop {
@@ -2114,9 +2134,10 @@ fn up_loop_until(
         }
         // A recovered task that crash-loops again is a new crash-loop, so it must be able to surface
         // again. Leaving the id in the dedup set would make every park after the first one silent.
-        for id in &report.unparked {
+        for id in report.unparked.iter() {
             reported_flapping.remove(id);
         }
+        recurring_warnings.filter(&mut report);
         park_channel.publish(&cap, &mut report);
         for cl in &report.crash_loops {
             if reported_flapping.insert(cl.pty_id.clone()) {
@@ -2707,6 +2728,56 @@ mod tests {
         );
     }
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn persistent_advisory_warnings_surface_once_not_per_pass() {
+        let catalog = tempfile::tempdir().unwrap();
+        let agent = catalog.path().join("agents/test-host/live");
+        std::fs::create_dir_all(&agent).unwrap();
+        std::fs::create_dir_all(catalog.path().join("workspace")).unwrap();
+        std::fs::write(
+            agent.join("agent.kdl"),
+            r#"agent "live" {
+  host "test-host"
+  command "true"
+  workspace "$CATALOG/workspace"
+  render { git-exclude "scratch.txt" }
+}"#,
+        )
+        .unwrap();
+        let stop = AtomicBool::new(false);
+        let mut passes = 0usize;
+        let mut warnings_seen = 0usize;
+
+        std::thread::scope(|scope| {
+            scope.spawn(|| {
+                std::thread::sleep(Duration::from_millis(300));
+                stop.store(true, Ordering::SeqCst);
+            });
+            up_loop_until(
+                catalog.path(),
+                "test-host",
+                &SpawnCountingRunner::default(),
+                Duration::from_millis(50),
+                &stop,
+                |_, _| None,
+                |report| {
+                    passes += 1;
+                    warnings_seen += report.warnings.len();
+                },
+            )
+            .unwrap();
+        });
+
+        assert!(
+            passes >= 3,
+            "the loop must have run several passes for this to say anything: {passes}"
+        );
+        assert_eq!(
+            warnings_seen, 1,
+            "an unchanged advisory failure must be diagnosed once across {passes} passes"
+        );
+    }
     /// A pass can execute a plan the task was never in: `up_once` drops an owner whose
     /// materialization failed, `gate_harness_launches_on_hooks` strips gated launches, and
     /// `defer_flickers` removes debounced ones — each after the pass is already committed to