From 24a75bfc7a42d96ab8b0ff32d8f27260fee6234b Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sat, 1 Aug 2026 11:24:50 +0200 Subject: [PATCH] fix(run): key presentation repair fairness by stable ID agent-tool: Codex CLI agent-tool-version: 0.145.0 agent-runtime: Codex CLI 0.145.0 agent-session-lookup: sha256:ad381a588faa911d70bf08ee2ae4305e96d980701e6405897e32fe38c3163d98 tooling-profile: dotfiles@de765ec --- src/eval_run.rs | 5 +- src/flapping.rs | 10 --- src/run.rs | 180 +++++++++++++++++++++++++++++++++++++++++++----- tests/run.rs | 16 ++--- 4 files changed, 176 insertions(+), 35 deletions(-) diff --git a/src/eval_run.rs b/src/eval_run.rs index 8c95794a..9d7a3399 100644 --- a/src/eval_run.rs +++ b/src/eval_run.rs @@ -1151,6 +1151,7 @@ fn run_eval_inner(spec: &Spec, eval: &Eval, spec_dir: &Path, catalog: &Path, hos let supervise_runner = SystemRunner::new(catalog.to_path_buf(), catalog.join("exec")); let mut sup_cap = crate::flapping::FlappingCap::default(); let mut sup_debounce = crate::run::LivenessDebounce::new(Duration::from_secs(2)); + let mut sup_presentation_cursor = crate::run::PresentationPatchCursor::default(); // Crash-ding state: the boot gate immediately above proved every declared task alive, so // carry that proof into supervision. PTY may self-reap a fast exit before the first // post-kickoff list snapshot; starting empty would then misclassify the proven-live task @@ -1208,14 +1209,16 @@ fn run_eval_inner(spec: &Spec, eval: &Eval, spec_dir: &Path, catalog: &Path, hos &supervise_runner, &mut sup_cap, &mut sup_debounce, + &mut sup_presentation_cursor, ) } - Err(_) => crate::run::reconcile_pass_specs( + Err(_) => crate::run::reconcile_pass_specs_with_cursor( &specs, host, &supervise_runner, &mut sup_cap, &mut sup_debounce, + &mut sup_presentation_cursor, ), }; if !report.launched.is_empty() { diff --git a/src/flapping.rs b/src/flapping.rs index 92fd231d..fa873c6f 100644 --- a/src/flapping.rs +++ b/src/flapping.rs @@ -32,7 +32,6 @@ pub struct FlappingCap { launches: HashMap>, last_launch: HashMap, parked: HashSet, - presentation_batch_cursor: usize, } impl FlappingCap { @@ -50,15 +49,6 @@ impl FlappingCap { self.parked.iter() } - pub(crate) fn presentation_batch_start(&mut self, total: usize, batch: usize) -> usize { - if total == 0 { - return 0; - } - let start = self.presentation_batch_cursor % total; - self.presentation_batch_cursor = (start + batch.min(total)) % total; - start - } - /// Decide whether `id` may be (re)launched at `now` under `policy`. On `Allow` the caller should /// spawn and then call [`record`](Self::record). pub fn decide(&mut self, id: &str, now: Instant, policy: &Restart) -> RestartDecision { diff --git a/src/run.rs b/src/run.rs index 860d4cab..6bfab6db 100644 --- a/src/run.rs +++ b/src/run.rs @@ -42,6 +42,30 @@ const PTY_LIST_TIMEOUT: Duration = Duration::from_secs(2); const PTY_DAEMON_SHUTDOWN_WAIT: Duration = Duration::from_secs(6); const MAX_PRESENTATION_PATCHES_PER_PASS: usize = 8; +#[derive(Debug, Default)] +pub(crate) struct PresentationPatchCursor { + after_id: Option, +} + +impl PresentationPatchCursor { + fn batch<'a>(&mut self, presentation: &'a [PtyPresentation]) -> Vec<&'a PtyPresentation> { + let mut ordered = presentation.iter().collect::>(); + ordered.sort_by(|left, right| left.pty_id.cmp(&right.pty_id)); + if ordered.is_empty() { + return Vec::new(); + } + let start = self.after_id.as_ref().map_or(0, |after_id| { + let next = ordered.partition_point(|item| item.pty_id <= *after_id); + if next == ordered.len() { 0 } else { next } + }); + let batch = (0..ordered.len().min(MAX_PRESENTATION_PATCHES_PER_PASS)) + .map(|offset| ordered[(start + offset) % ordered.len()]) + .collect::>(); + self.after_id = batch.last().map(|item| item.pty_id.clone()); + batch + } +} + /// Run a non-interactive child with bounded output capture. Regular temporary files keep an escaped /// descendant that inherited stdout/stderr from blocking cleanup after the direct child times out. /// The child still gets a fresh process group so the common wrapper-and-descendants case is reaped. @@ -1035,6 +1059,22 @@ pub fn execute( runner: &dyn Runner, cap: &mut FlappingCap, report: &mut UpReport, +) { + execute_with_presentation_cursor( + plan, + runner, + cap, + &mut PresentationPatchCursor::default(), + report, + ); +} + +fn execute_with_presentation_cursor( + plan: &ReconcilePlan, + runner: &dyn Runner, + cap: &mut FlappingCap, + presentation_cursor: &mut PresentationPatchCursor, + report: &mut UpReport, ) { // The corpses tied to a launch target (dead, non-keep, active ptys) are reaped inside the launch // loop so a parked flapper keeps its evidence. Everything else in `gc` (e.g. a retired agent's @@ -1119,16 +1159,8 @@ pub fn execute( // Presentation never delays lifecycle convergence. Drift repair is bounded to eight sequential // children, keeping its worst-case 2s-per-child containment below the 30s supervisor cadence; - // the persistent cursor rotates remaining drift through later passes without starvation. - let presentation_count = plan - .presentation - .len() - .min(MAX_PRESENTATION_PATCHES_PER_PASS); - let presentation_start = - cap.presentation_batch_start(plan.presentation.len(), presentation_count); - for offset in 0..presentation_count { - let presentation = - &plan.presentation[(presentation_start + offset) % plan.presentation.len()]; + // remaining drift is observed and retried on later passes. + for presentation in presentation_cursor.batch(&plan.presentation) { if let Err(error) = runner.patch_presentation(presentation) { report .errors @@ -1239,6 +1271,7 @@ fn reconcile_pass( runner: &dyn Runner, cap: &mut FlappingCap, debounce: &mut LivenessDebounce, + presentation_cursor: &mut PresentationPatchCursor, ) -> UpReport { let found = crate::discover(root); let mut report = UpReport { @@ -1310,7 +1343,7 @@ fn reconcile_pass( Some(error) => anyhow::bail!("{error}"), None => Ok(()), }); - execute(&plan, runner, cap, &mut report); + execute_with_presentation_cursor(&plan, runner, cap, presentation_cursor, &mut report); report } @@ -1364,6 +1397,7 @@ pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Res runner, &mut FlappingCap::default(), &mut debounce, + &mut PresentationPatchCursor::default(), )) } @@ -1381,6 +1415,24 @@ pub fn reconcile_pass_specs( runner: &dyn Runner, cap: &mut FlappingCap, debounce: &mut LivenessDebounce, +) -> UpReport { + reconcile_pass_specs_with_cursor( + specs, + this_host, + runner, + cap, + debounce, + &mut PresentationPatchCursor::default(), + ) +} + +pub(crate) fn reconcile_pass_specs_with_cursor( + specs: &[agent_spec::spec::AgentSpec], + this_host: &str, + runner: &dyn Runner, + cap: &mut FlappingCap, + debounce: &mut LivenessDebounce, + presentation_cursor: &mut PresentationPatchCursor, ) -> UpReport { let mut report = UpReport::default(); let sessions = match runner.list_sessions() { @@ -1393,7 +1445,15 @@ pub fn reconcile_pass_specs( return report; } }; - reconcile_pass_specs_with_sessions(specs, &sessions, this_host, runner, cap, debounce) + reconcile_pass_specs_with_sessions( + specs, + &sessions, + this_host, + runner, + cap, + debounce, + presentation_cursor, + ) } /// Reconcile an in-memory team against an already captured session snapshot. Eval supervision uses @@ -1407,13 +1467,14 @@ pub(crate) fn reconcile_pass_specs_with_sessions( runner: &dyn Runner, cap: &mut FlappingCap, debounce: &mut LivenessDebounce, + presentation_cursor: &mut PresentationPatchCursor, ) -> UpReport { let mut report = UpReport::default(); let now = Instant::now(); debounce.observe(sessions, now); let mut plan = crate::reconcile(specs, sessions, this_host); report.deferred = debounce.defer_flickers(&mut plan, now); - execute(&plan, runner, cap, &mut report); + execute_with_presentation_cursor(&plan, runner, cap, presentation_cursor, &mut report); report } @@ -1536,9 +1597,17 @@ pub fn up_loop_specs( install_signal_handler(); let mut cap = FlappingCap::default(); let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE); + let mut presentation_cursor = PresentationPatchCursor::default(); let mut reported_flapping: HashSet = HashSet::new(); loop { - let report = reconcile_pass_specs(specs, this_host, runner, &mut cap, &mut debounce); + let report = reconcile_pass_specs_with_cursor( + specs, + this_host, + runner, + &mut cap, + &mut debounce, + &mut presentation_cursor, + ); for cl in &report.crash_loops { if reported_flapping.insert(cl.pty_id.clone()) { eprintln!( @@ -1690,6 +1759,7 @@ fn up_loop_until( // Carries per-id liveness across passes so a transient `pty list` flicker under load isn't // destructively GC'd (R21c). Fresh throwaway in `up_once` — a single pass has no flicker to absorb. let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE); + let mut presentation_cursor = PresentationPatchCursor::default(); // Surface each parked crash-loop once (not every pass): an stderr line AND a message to the // agent's supervisor over the native bus, so a crash-loop isn't only visible to whoever is @@ -1697,7 +1767,14 @@ fn up_loop_until( let mut reported_flapping: HashSet = HashSet::new(); loop { - let report = reconcile_pass(root, this_host, runner, &mut cap, &mut debounce); + let report = reconcile_pass( + root, + this_host, + runner, + &mut cap, + &mut debounce, + &mut presentation_cursor, + ); for cl in &report.crash_loops { if reported_flapping.insert(cl.pty_id.clone()) { eprintln!( @@ -1805,7 +1882,7 @@ pub fn detect_host() -> String { mod tests { use super::*; use agent_spec::spec::{AgentSpec, JobType, Task, TaskKind, TaskLifecycle}; - use std::cell::Cell; + use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, BTreeSet}; use std::ffi::OsStr; @@ -1848,6 +1925,77 @@ mod tests { } } + #[derive(Default)] + struct PersistentPatchRunner { + patched: RefCell>, + } + + impl Runner for PersistentPatchRunner { + fn list_sessions(&self) -> anyhow::Result> { + unreachable!("presentation execution does not list sessions") + } + + fn spawn(&self, _target: &TaskTarget, _spec_dir: &Path) -> anyhow::Result<()> { + unreachable!("presentation-only plan must not spawn") + } + + fn kill(&self, _pty_id: &str) -> anyhow::Result<()> { + unreachable!("presentation-only plan must not kill") + } + + fn remove(&self, _pty_id: &str) -> anyhow::Result<()> { + unreachable!("presentation-only plan must not remove") + } + + fn patch_presentation(&self, presentation: &PtyPresentation) -> anyhow::Result<()> { + self.patched.borrow_mut().push(presentation.pty_id.clone()); + if presentation.pty_id.as_str() < "host.presented.08" { + anyhow::bail!("simulated persistent metadata failure"); + } + Ok(()) + } + } + + #[test] + fn bounded_presentation_batches_are_deterministic_and_do_not_starve() { + let plan = ReconcilePlan { + presentation: (0..12) + .rev() + .map(|index| PtyPresentation { + pty_id: format!("host.presented.{index:02}"), + display_name: None, + tags: BTreeMap::new(), + }) + .collect(), + ..ReconcilePlan::default() + }; + let runner = PersistentPatchRunner::default(); + let mut cap = FlappingCap::default(); + let mut cursor = PresentationPatchCursor::default(); + + for _ in 0..2 { + execute_with_presentation_cursor( + &plan, + &runner, + &mut cap, + &mut cursor, + &mut UpReport::default(), + ); + } + + let attempted = runner.patched.borrow(); + assert_eq!( + &attempted[..8], + &(0..8) + .map(|index| format!("host.presented.{index:02}")) + .collect::>() + ); + assert_eq!(attempted.len(), 16); + for index in 8..12 { + assert!(attempted.contains(&format!("host.presented.{index:02}"))); + } + } + #[test] fn selected_codex_gate_suppresses_launch_on_stale_hooks() { let spec = AgentSpec { diff --git a/tests/run.rs b/tests/run.rs index 650f03c6..82ed49a8 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -546,6 +546,7 @@ fn lifecycle_work_precedes_a_bounded_presentation_batch() { presentation: None, }; let presentation = (0..10) + .rev() .map(|index| PtyPresentation { pty_id: format!("host.presented.{index}"), display_name: Some(Some(format!("Presented {index}"))), @@ -566,15 +567,18 @@ fn lifecycle_work_precedes_a_bounded_presentation_batch() { }; let runner = FakeRunner::default(); let mut report = UpReport::default(); - let mut cap = FlappingCap::default(); - - execute(&plan, &runner, &mut cap, &mut report); + execute(&plan, &runner, &mut FlappingCap::default(), &mut report); assert_eq!( &runner.ops.borrow()[..2], ["spawn:host.owner.work", "kill:host.retired.work"] ); - assert_eq!(runner.patched.borrow().len(), 8); + assert_eq!( + *runner.patched.borrow(), + (0..8) + .map(|index| format!("host.presented.{index}")) + .collect::>() + ); assert!( report .warnings @@ -582,10 +586,6 @@ fn lifecycle_work_precedes_a_bounded_presentation_batch() { .any(|warning| warning.contains("deferred 2 presentation patches")) ); assert!(report.is_noteworthy()); - - execute(&plan, &runner, &mut cap, &mut UpReport::default()); - let patched = runner.patched.borrow(); - assert!((0..10).all(|index| patched.contains(&format!("host.presented.{index}")))); } /// A v2 service job: a pty agent + an exec ding.