Skip to content

Commit 55efd31

Browse files
fix(run): stop per-pass advisory churn on large catalogs (#334)
* fix(run): stop per-pass advisory churn on large catalogs Two reconcile-pass costs repeat every pass even when nothing changed: - git-exclude spawned `git rev-parse` per op per pass and failed for workspaces that are plain directories (28 agents on dev3 → ~28 spawns plus captured git stderr every pass). Probe for a `.git` marker up the ancestor chain first — the same probe is_git_tracked already uses — and answer "not a Git worktree" without a process spawn. - print_report re-printed identical warnings every pass (~7k lines in hours on dev3). The supervisor loop now deduplicates warnings that persist across passes while still re-surfacing one that clears and returns; applied to both the catalog and spec loops. Part of #314. Co-authored-by: schickling-assistant <schickling-assistant@users.noreply.github.com> agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty * fix(run): restore crash-loop surfacing in the spec supervisor Rewiring the spec loop's warning filter dropped the one-shot crash-loop surface: the GAVE UP diagnostic and the bus notification to the agent's supervisor never fired for spec-based teams even though the report still listed the task as flapping. Restore the deduplicated surfacing block alongside the warning filter. Co-authored-by: schickling-assistant <schickling-assistant@users.noreply.github.com> agent-identity: unknown agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.0.3 agent-runtime: OMP 18.0.3 tooling-profile: dotfiles@f33cd9c-dirty --------- Co-authored-by: schickling-assistant <schickling-assistant@users.noreply.github.com>
1 parent 6d58bc2 commit 55efd31

2 files changed

Lines changed: 108 additions & 2 deletions

File tree

src/materialize.rs

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -688,7 +688,23 @@ fn contains_owned_string(value: &serde_json::Value) -> bool {
688688
}
689689
}
690690

691+
fn has_git_marker(workspace: &Path) -> bool {
692+
workspace
693+
.ancestors()
694+
.any(|ancestor| ancestor.join(".git").exists())
695+
}
696+
691697
fn git_exclude(workspace: &Path, line: &str) -> Result<bool> {
698+
// A failed `git rev-parse` costs a process spawn per op per pass. On catalogs whose
699+
// workspaces are plain directories this repeats every reconcile (measured live: ~28 spawns
700+
// plus stderr captures per pass on dev3). The marker probe answers "not a repo" for free;
701+
// `.git` may be a directory or a file (linked worktrees), which `exists` covers either way.
702+
if !has_git_marker(workspace) {
703+
anyhow::bail!(
704+
"{} is not a Git worktree (no .git marker)",
705+
workspace.display()
706+
);
707+
}
692708
let output = Command::new("git")
693709
.args(["-C"])
694710
.arg(workspace)
@@ -1326,6 +1342,25 @@ mod tests {
13261342
}
13271343
}
13281344

1345+
#[test]
1346+
fn git_exclude_reports_missing_repo_without_spawning_git() {
1347+
let dir = tempfile::tempdir().unwrap();
1348+
let workspace = dir.path().join("ws");
1349+
std::fs::create_dir_all(&workspace).unwrap();
1350+
1351+
assert!(!has_git_marker(&workspace));
1352+
let error = git_exclude(&workspace, ".st2/").unwrap_err();
1353+
assert!(
1354+
error.to_string().contains("no .git marker"),
1355+
"the no-repo case must be answered by the marker probe: {error:#}"
1356+
);
1357+
1358+
// A `.git` entry anywhere above the workspace (directory or file form, as linked
1359+
// worktrees use) re-enables the real git path.
1360+
std::fs::create_dir_all(dir.path().join(".git")).unwrap();
1361+
assert!(has_git_marker(&workspace));
1362+
}
1363+
13291364
#[test]
13301365
fn deep_merge_preserves_unrelated_keys_and_replaces_arrays() {
13311366
let mut target = serde_json::json!({

src/run.rs

Lines changed: 73 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1856,6 +1856,7 @@ pub fn up_loop_specs(
18561856
let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE);
18571857
let mut presentation_cursor = PresentationPatchCursor::default();
18581858
let mut reported_flapping: HashSet<String> = HashSet::new();
1859+
let mut recurring_warnings = RecurringWarnings::default();
18591860
let park_channel = ParkChannel::for_supervisor(root, this_host);
18601861
loop {
18611862
let mut pre = UpReport::default();
@@ -1873,7 +1874,6 @@ pub fn up_loop_specs(
18731874
for id in &report.unparked {
18741875
reported_flapping.remove(id);
18751876
}
1876-
park_channel.publish(&cap, &mut report);
18771877
for cl in &report.crash_loops {
18781878
if reported_flapping.insert(cl.pty_id.clone()) {
18791879
eprintln!(
@@ -1883,6 +1883,7 @@ pub fn up_loop_specs(
18831883
surface_crash_loop(root, this_host, cl);
18841884
}
18851885
}
1886+
recurring_warnings.filter(&mut report);
18861887
on_report(&report);
18871888
if STOP.load(Ordering::SeqCst) {
18881889
break;
@@ -2048,6 +2049,24 @@ fn best_effort_catalog_watcher(
20482049
}
20492050
}
20502051

2052+
/// Suppresses warnings that persist across passes while still re-surfacing one that clears and
2053+
/// returns. An unchanged advisory failure (a non-Git workspace failing its git-exclude, say) must
2054+
/// be diagnosed once, not once per reconcile pass.
2055+
#[derive(Default)]
2056+
struct RecurringWarnings {
2057+
emitted: HashSet<String>,
2058+
}
2059+
2060+
impl RecurringWarnings {
2061+
fn filter(&mut self, report: &mut UpReport) {
2062+
let current: HashSet<_> = report.warnings.iter().cloned().collect();
2063+
self.emitted.retain(|warning| current.contains(warning));
2064+
report
2065+
.warnings
2066+
.retain(|warning| self.emitted.insert(warning.clone()));
2067+
}
2068+
}
2069+
20512070
/// The supervisor loop: reconcile on a timer AND on folder changes until interrupted. The fs-watch is
20522071
/// best-effort; the `interval` timer is the always-on fallback. `on_report` is called once per pass
20532072
pub fn up_loop(
@@ -2093,6 +2112,7 @@ fn up_loop_until(
20932112
// agent's supervisor over the native bus, so a crash-loop isn't only visible to whoever is
20942113
// watching the log.
20952114
let mut reported_flapping: HashSet<String> = HashSet::new();
2115+
let mut recurring_warnings = RecurringWarnings::default();
20962116
let park_channel = ParkChannel::for_supervisor(root, this_host);
20972117

20982118
loop {
@@ -2114,9 +2134,10 @@ fn up_loop_until(
21142134
}
21152135
// A recovered task that crash-loops again is a new crash-loop, so it must be able to surface
21162136
// again. Leaving the id in the dedup set would make every park after the first one silent.
2117-
for id in &report.unparked {
2137+
for id in report.unparked.iter() {
21182138
reported_flapping.remove(id);
21192139
}
2140+
recurring_warnings.filter(&mut report);
21202141
park_channel.publish(&cap, &mut report);
21212142
for cl in &report.crash_loops {
21222143
if reported_flapping.insert(cl.pty_id.clone()) {
@@ -2707,6 +2728,56 @@ mod tests {
27072728
);
27082729
}
27092730

2731+
#[cfg(target_os = "linux")]
2732+
#[test]
2733+
fn persistent_advisory_warnings_surface_once_not_per_pass() {
2734+
let catalog = tempfile::tempdir().unwrap();
2735+
let agent = catalog.path().join("agents/test-host/live");
2736+
std::fs::create_dir_all(&agent).unwrap();
2737+
std::fs::create_dir_all(catalog.path().join("workspace")).unwrap();
2738+
std::fs::write(
2739+
agent.join("agent.kdl"),
2740+
r#"agent "live" {
2741+
host "test-host"
2742+
command "true"
2743+
workspace "$CATALOG/workspace"
2744+
render { git-exclude "scratch.txt" }
2745+
}"#,
2746+
)
2747+
.unwrap();
2748+
let stop = AtomicBool::new(false);
2749+
let mut passes = 0usize;
2750+
let mut warnings_seen = 0usize;
2751+
2752+
std::thread::scope(|scope| {
2753+
scope.spawn(|| {
2754+
std::thread::sleep(Duration::from_millis(300));
2755+
stop.store(true, Ordering::SeqCst);
2756+
});
2757+
up_loop_until(
2758+
catalog.path(),
2759+
"test-host",
2760+
&SpawnCountingRunner::default(),
2761+
Duration::from_millis(50),
2762+
&stop,
2763+
|_, _| None,
2764+
|report| {
2765+
passes += 1;
2766+
warnings_seen += report.warnings.len();
2767+
},
2768+
)
2769+
.unwrap();
2770+
});
2771+
2772+
assert!(
2773+
passes >= 3,
2774+
"the loop must have run several passes for this to say anything: {passes}"
2775+
);
2776+
assert_eq!(
2777+
warnings_seen, 1,
2778+
"an unchanged advisory failure must be diagnosed once across {passes} passes"
2779+
);
2780+
}
27102781
/// A pass can execute a plan the task was never in: `up_once` drops an owner whose
27112782
/// materialization failed, `gate_harness_launches_on_hooks` strips gated launches, and
27122783
/// `defer_flickers` removes debounced ones — each after the pass is already committed to

0 commit comments

Comments
 (0)