Skip to content

Commit 94e07a7

Browse files
author
Alex
committed
fix(terraphim_orchestrator): gate compound-review fires against per-occurrence cursor
Adds `last_compound_review_fired_at: Option<DateTime<Utc>>` to `AgentOrchestrator` and rewrites the compound-review branch of `check_cron_schedules` so the cursor is recorded **before** the `.await` on `handle_schedule_event`. Mirrors the per-agent `last_cron_fire` pattern at the start of the same function. Why: `reconcile_tick` is wrapped in a 90 s `tokio::time::timeout` safety net. When the future is cancelled mid-await, `last_tick_time` is never updated, so the previous `should_fire` check kept returning true on every subsequent tick, spawning a fresh review worktree every 30 s (the bigbox storm). Recording the cursor synchronously before the await makes cancellation safe: the next iteration sees the cursor and short-circuits via `already_fired = fire_time <= prev`. Verified via new regression test `test_compound_review_cursor_advances_on_cancellation`: plants a `last_tick_time` 2 h in the past, runs `check_cron_schedules` twice without advancing wall-clock, and asserts the cursor is `Some(_)` after the first call and unchanged after the second. Note on design drift: the design doc (docs/design/adf-worktree-lifecycle-design.md section 4.1) used `take_while(|t| *t <= now)` plus an explicit `already_fired` gate. I initially collapsed both into `compound_sched.after(&cursor).next()` for brevity but reverted to the documented shape after the regression test exposed the catch-up vs gating semantic difference -- the design's `last_tick_time`-anchored `next_fire` plus separate cursor gate is the correct read of "same occurrence, do not re-fire". Line numbers in the design (`:241`, `:817`, `:7137-7161`, `:7712`) all matched current source unchanged. Layer 0 of the ADF worktree lifecycle epic (Gitea #1567). Out of scope: the per-agent cron path at lib.rs:7484, the WorktreeGuard refactor (Layer 1, Gitea #1569), startup sweep (Layer 2, #1570), and the adf-cleanup.sh hardening (Layer 3, #1571). Refs #1562 (Gitea)
1 parent f1e6ece commit 94e07a7

1 file changed

Lines changed: 134 additions & 13 deletions

File tree

  • crates/terraphim_orchestrator/src

crates/terraphim_orchestrator/src/lib.rs

Lines changed: 134 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,14 @@ pub struct AgentOrchestrator {
239239
/// Per-agent last cron fire timestamp to prevent re-triggering within same schedule window.
240240
/// Key: agent name. Value: timestamp of last fire.
241241
last_cron_fire: HashMap<String, chrono::DateTime<chrono::Utc>>,
242+
/// Last compound-review fire time, used to gate the compound schedule
243+
/// independently of `last_tick_time`. Mirrors the `last_cron_fire`
244+
/// pattern for per-agent crons. Without this cursor, if the
245+
/// `reconcile_tick` future is cancelled mid-await by its 90 s
246+
/// `tokio::time::timeout` safety wrapper, `last_tick_time` is never
247+
/// advanced and the same compound-review occurrence re-fires on the
248+
/// very next tick, producing a worktree storm (#1562).
249+
last_compound_review_fired_at: Option<chrono::DateTime<chrono::Utc>>,
242250
/// Lazy-initialised Gitea tracker for gitea-issue pre-check.
243251
pre_check_tracker: Option<terraphim_tracker::GiteaTracker>,
244252
/// Active flow executions keyed by flow name.
@@ -824,6 +832,7 @@ impl AgentOrchestrator {
824832
circuit_breakers: Arc::new(Mutex::new(HashMap::new())),
825833
last_run_commits: HashMap::new(),
826834
last_cron_fire: HashMap::new(),
835+
last_compound_review_fired_at: None,
827836
pre_check_tracker: None,
828837
active_flows: HashMap::new(),
829838
mention_cursors: HashMap::new(),
@@ -7137,26 +7146,46 @@ Remove the pause flag once the underlying failure is resolved:\n\n\
71377146
if let Some(compound_sched) = self.scheduler.compound_review_schedule() {
71387147
debug!(
71397148
last_tick = %self.last_tick_time,
7149+
last_fired = ?self.last_compound_review_fired_at,
71407150
now = %now,
71417151
"checking compound review schedule"
71427152
);
71437153

7144-
// Get next fire times for debugging
7145-
let upcoming: Vec<_> = compound_sched.after(&self.last_tick_time).take(3).collect();
7146-
debug!(upcoming = ?upcoming, "compound schedule upcoming times");
7147-
7148-
let should_fire = compound_sched
7154+
// Compute the earliest occurrence strictly after
7155+
// `last_tick_time` that is also <= now. This is the same
7156+
// occurrence the buggy code would have refired forever when
7157+
// the reconcile-tick future was cancelled mid-await by the
7158+
// 90 s `tokio::time::timeout` safety wrapper (#1562).
7159+
let next_fire = compound_sched
71497160
.after(&self.last_tick_time)
71507161
.take_while(|t| *t <= now)
7151-
.next()
7152-
.is_some();
7153-
7154-
debug!(should_fire = should_fire, "compound review fire check");
7162+
.next();
7163+
debug!(next_fire = ?next_fire, "compound schedule next fire");
7164+
7165+
if let Some(fire_time) = next_fire {
7166+
// Gate against re-firing the same occurrence. The
7167+
// cursor `last_compound_review_fired_at` is the per-
7168+
// occurrence dedup key, mirroring `last_cron_fire` for
7169+
// per-agent crons. It is updated *before* the `.await`
7170+
// below so a cancelled future cannot lose the update.
7171+
let already_fired = self
7172+
.last_compound_review_fired_at
7173+
.map(|prev| fire_time <= prev)
7174+
.unwrap_or(false);
71557175

7156-
if should_fire {
7157-
info!("compound review schedule fired, starting review");
7158-
self.handle_schedule_event(ScheduleEvent::CompoundReview)
7159-
.await;
7176+
if !already_fired {
7177+
// Record fire time BEFORE awaiting
7178+
// `handle_schedule_event` so that future
7179+
// cancellation cannot lose the update and
7180+
// re-trigger the same occurrence on the next tick.
7181+
self.last_compound_review_fired_at = Some(fire_time);
7182+
info!(
7183+
fire_time = %fire_time,
7184+
"compound review schedule fired, starting review"
7185+
);
7186+
self.handle_schedule_event(ScheduleEvent::CompoundReview)
7187+
.await;
7188+
}
71607189
}
71617190
}
71627191
}
@@ -7713,6 +7742,19 @@ Remove the pause flag once the underlying failure is resolved:\n\n\
77137742
self.last_tick_time = time;
77147743
}
77157744

7745+
/// Test helper: read the compound-review fire cursor.
7746+
#[doc(hidden)]
7747+
pub fn last_compound_review_fired_at(&self) -> Option<chrono::DateTime<chrono::Utc>> {
7748+
self.last_compound_review_fired_at
7749+
}
7750+
7751+
/// Test helper: clear the compound-review fire cursor for synthetic
7752+
/// testing of the cancellation property.
7753+
#[doc(hidden)]
7754+
pub fn clear_last_compound_review_fired_at(&mut self) {
7755+
self.last_compound_review_fired_at = None;
7756+
}
7757+
77167758
/// Test helper: access the telemetry store for assertions.
77177759
#[doc(hidden)]
77187760
pub fn telemetry_store(&self) -> &control_plane::TelemetryStore {
@@ -8094,6 +8136,85 @@ mod tests {
80948136
assert_eq!(result.agents_failed, 0);
80958137
}
80968138

8139+
/// Regression test for #1562.
8140+
///
8141+
/// Property: when `check_cron_schedules` fires the compound review,
8142+
/// `last_compound_review_fired_at` advances **before** the `await`
8143+
/// on `handle_schedule_event`. Calling `check_cron_schedules` a
8144+
/// second time without advancing wall-clock time must NOT re-fire
8145+
/// the same occurrence; the cursor stays put.
8146+
///
8147+
/// This is the property that breaks if the cursor is dropped: the
8148+
/// 90 s `tokio::time::timeout` wrapping `reconcile_tick` cancels
8149+
/// the future mid-await, `last_tick_time` is never updated, and
8150+
/// the next tick re-evaluates the same cron occurrence as "should
8151+
/// fire", spawning a new worktree every tick (the bigbox storm).
8152+
#[tokio::test]
8153+
async fn test_compound_review_cursor_advances_on_cancellation() {
8154+
// Build a test config and override compound_review so that the
8155+
// workflow has no review groups -- it still creates a worktree
8156+
// on the workspace git repo, but no agent subprocesses are
8157+
// launched. This mirrors `test_orchestrator_compound_review_manual`.
8158+
let mut config = test_config();
8159+
let tmp_worktree = TempDir::new().expect("tempdir");
8160+
config.compound_review.worktree_root = tmp_worktree.path().to_path_buf();
8161+
// Schedule fires hourly so we can use a recent `last_tick_time`.
8162+
// 5-field cron: minute 0 of every hour.
8163+
config.compound_review.schedule = "0 * * * *".to_string();
8164+
8165+
let mut orch = AgentOrchestrator::new(config).expect("orchestrator");
8166+
8167+
// Replace the compound workflow with one that uses an empty
8168+
// group list so the cron-fire path is a no-op apart from the
8169+
// worktree creation/removal. The orchestrator's
8170+
// `repo_path`/`base_branch` are inherited from the test config.
8171+
let repo_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..");
8172+
let swarm_config = crate::compound::SwarmConfig {
8173+
groups: vec![],
8174+
timeout: Duration::from_secs(60),
8175+
worktree_root: tmp_worktree.path().to_path_buf(),
8176+
repo_path,
8177+
base_branch: "main".to_string(),
8178+
max_concurrent_agents: 1,
8179+
create_prs: false,
8180+
};
8181+
orch.compound_workflow = crate::compound::CompoundReviewWorkflow::new(swarm_config);
8182+
8183+
// Plant `last_tick_time` 2 hours ago so at least one occurrence
8184+
// of `0 * * * *` lies in [last_tick_time, now]. Clear the new
8185+
// cursor so the first call has nothing to compare against.
8186+
let two_hours_ago = chrono::Utc::now() - chrono::Duration::hours(2);
8187+
orch.set_last_tick_time(two_hours_ago);
8188+
orch.clear_last_compound_review_fired_at();
8189+
assert!(
8190+
orch.last_compound_review_fired_at().is_none(),
8191+
"cursor should start empty",
8192+
);
8193+
8194+
// First call: should advance the cursor to a past fire time.
8195+
orch.check_cron_schedules().await;
8196+
let cursor_after_first = orch
8197+
.last_compound_review_fired_at()
8198+
.expect("cursor should be Some after first fire");
8199+
assert!(
8200+
cursor_after_first <= chrono::Utc::now(),
8201+
"cursor should be in the past, got {}",
8202+
cursor_after_first
8203+
);
8204+
8205+
// Second call without advancing wall-clock or `last_tick_time`:
8206+
// the cursor must NOT advance (the same occurrence is gated).
8207+
orch.check_cron_schedules().await;
8208+
let cursor_after_second = orch
8209+
.last_compound_review_fired_at()
8210+
.expect("cursor should still be Some");
8211+
assert_eq!(
8212+
cursor_after_first, cursor_after_second,
8213+
"cursor must not re-advance on a re-check without new occurrences \
8214+
(#1562 storm regression)",
8215+
);
8216+
}
8217+
80978218
#[test]
80988219
fn test_orchestrator_from_toml() {
80998220
let toml_str = r#"

0 commit comments

Comments
 (0)