forge_workspace/
idle_sweep.rs

1//! The idle lead sweep: forge injects the board sweep when a project's
2//! lead has nothing in flight.
3//!
4//! The trigger is this module's whole subject, not the procedure: the
5//! lead charter already carries how a sweep is walked, so the injected
6//! prompt only names the moment and the ending it owes. The pass never
7//! wakes a sleeping session - a lead with no session is skipped, and its
8//! next session start sweeps, as the charter says.
9
10use std::sync::Arc;
11use std::time::SystemTime;
12
13use forge_primitives::tasks::TaskStatus;
14use tracing::Instrument;
15
16use crate::SessionSlot;
17use crate::chase::{Deliver, Delivery, PromptDeliver};
18use crate::mcp::workers::types::WorkerEntry;
19use crate::workspace::Workspace;
20
21/// How often the loop looks. A lead that goes idle is swept within this.
22pub const IDLE_SWEEP_INTERVAL_SECS: u64 = 60;
23
24/// At most one idle-triggered sweep per lead inside this window.
25pub const IDLE_SWEEP_MIN_GAP_SECS: u64 = 2 * 60 * 60;
26
27/// A lead that has stayed busy this long is swept anyway: the prompt
28/// queues behind the running turn, so it is a nudge and not an
29/// interruption.
30pub const IDLE_SWEEP_FALLBACK_SECS: u64 = 6 * 60 * 60;
31
32/// The idle path's prompt.
33pub const IDLE_SWEEP_PROMPT: &str = "Nothing is in flight on this board. Sweep it now, and end \
34     with the next three picks from the backlog.";
35
36/// The fallback's own lead-in. The fallback fires precisely when the lead
37/// is NOT idle, so it must not claim the board is quiet.
38pub const IDLE_SWEEP_FALLBACK_PROMPT: &str = "Six hours have passed since the last sweep. Sweep \
39     the board when you can, and end with the next three picks from the backlog.";
40
41/// One lead's injected-sweep clock, held in memory so a restart re-arms
42/// every lead.
43#[derive(Debug, Clone, Copy)]
44pub(crate) struct SweepClock {
45    /// When this workspace first saw the lead live, re-seeded on every
46    /// tick that finds the seat asleep. The fallback measures from here
47    /// only until a sweep has landed this process.
48    pub(crate) first_seen: SystemTime,
49    /// When the last injected sweep was delivered. Kept across a session
50    /// restart: the two-hour gap belongs to the lead, and only a process
51    /// restart re-arms it.
52    pub(crate) last_delivered: Option<SystemTime>,
53    /// Whether the last delivery attempt failed, so a failure episode is
54    /// logged once rather than on every retry tick.
55    pub(crate) failing: bool,
56}
57
58/// How long `now` is after `since`, and zero when the clock went
59/// backwards. An age that cannot be measured is unknown, never
60/// infinitely old: one NTP step must not sweep every lead.
61fn age_secs(since: SystemTime, now: SystemTime) -> u64 {
62    now.duration_since(since).map_or(0, |d| d.as_secs())
63}
64
65impl Workspace {
66    /// One pass over every project: deliver the injected sweep to each
67    /// lead that is idle, or overdue for one. Returns how many prompts
68    /// reached a session.
69    pub fn idle_sweep(&self, deliver: &dyn Deliver, now: SystemTime) -> usize {
70        let views = self.list_projects();
71        // A project removed from the config takes its lead's clock with
72        // it.
73        self.idle_sweeps.lock().retain(|slot, _| {
74            views.iter().any(|view| view.org == slot.org() && view.name == slot.project())
75        });
76        let mut delivered = 0;
77        for view in views {
78            let lead = SessionSlot::lead(&view.org, &view.name);
79            // Never wake a sleeping session: its next start sweeps, as
80            // the charter says. Its clock keeps `last_delivered` - a
81            // session restart is not a process restart - while
82            // `first_seen` re-arms from the wake.
83            if !self.seat_is_live(&lead) {
84                if let Some(clock) = self.idle_sweeps.lock().get_mut(&lead) {
85                    clock.first_seen = now;
86                }
87                continue;
88            }
89            let idle = lead_is_idle(self, &lead);
90            let due = {
91                let mut clocks = self.idle_sweeps.lock();
92                let clock = clocks.entry(lead.clone()).or_insert(SweepClock {
93                    first_seen: now,
94                    last_delivered: None,
95                    failing: false,
96                });
97                let age = clock
98                    .last_delivered
99                    .map_or(age_secs(clock.first_seen, now), |last| age_secs(last, now));
100                match clock.last_delivered {
101                    None => idle || age >= IDLE_SWEEP_FALLBACK_SECS,
102                    Some(_) => {
103                        (idle && age >= IDLE_SWEEP_MIN_GAP_SECS) || age >= IDLE_SWEEP_FALLBACK_SECS
104                    }
105                }
106            };
107            if !due {
108                continue;
109            }
110            let text = if idle { IDLE_SWEEP_PROMPT } else { IDLE_SWEEP_FALLBACK_PROMPT };
111            match deliver.to_seat(&lead, text) {
112                Delivery::Delivered => {
113                    let recovered = {
114                        let mut clocks = self.idle_sweeps.lock();
115                        clocks.get_mut(&lead).is_some_and(|clock| {
116                            clock.last_delivered = Some(now);
117                            std::mem::replace(&mut clock.failing, false)
118                        })
119                    };
120                    delivered += 1;
121                    tracing::debug!(
122                        target: "forge_workspace::idle_sweep",
123                        event_name = "idle_sweep_injected",
124                        slot = %lead.display(),
125                        fallback = !idle,
126                        "injected the board sweep",
127                    );
128                    if recovered {
129                        tracing::info!(
130                            target: "forge_workspace::idle_sweep",
131                            event_name = "idle_sweep_recovered",
132                            slot = %lead.display(),
133                            "the board sweep reached the lead again",
134                        );
135                    }
136                }
137                Delivery::NoSession(reason) => {
138                    let first_failure = {
139                        let mut clocks = self.idle_sweeps.lock();
140                        match clocks.get_mut(&lead) {
141                            Some(clock) => !std::mem::replace(&mut clock.failing, true),
142                            None => true,
143                        }
144                    };
145                    if first_failure {
146                        tracing::warn!(
147                            target: "forge_workspace::idle_sweep",
148                            event_name = "idle_sweep_not_delivered",
149                            slot = %lead.display(),
150                            reason = %reason,
151                            "the board sweep could not be delivered; retrying each tick",
152                        );
153                    }
154                }
155            }
156        }
157        delivered
158    }
159
160    /// Spawn the idle sweep: a background pass every
161    /// [`IDLE_SWEEP_INTERVAL_SECS`] that injects the board sweep into a
162    /// lead with nothing in flight. The registry makes a second call a
163    /// no-op; started once at boot. Mirrors
164    /// [`Workspace::start_chase_sweep`].
165    pub fn start_idle_sweep(self: &Arc<Self>) {
166        let weak = Arc::downgrade(self);
167        let span = tracing::info_span!("idle_sweep");
168        self.supervisor.spawn("idle_sweep", forge_supervise::Policy::backoff(), move || {
169            let weak = weak.clone();
170            let span = span.clone();
171            async move {
172                let period = std::time::Duration::from_secs(IDLE_SWEEP_INTERVAL_SECS);
173                let mut interval =
174                    tokio::time::interval_at(tokio::time::Instant::now() + period, period);
175                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
176                loop {
177                    interval.tick().await;
178                    let Some(workspace) = weak.upgrade() else {
179                        return;
180                    };
181                    workspace.idle_sweep(
182                        &PromptDeliver { workspace: Arc::clone(&workspace) },
183                        SystemTime::now(),
184                    );
185                }
186            }
187            .instrument(span)
188        });
189    }
190}
191
192/// Whether `lead` is idle: it is the project's own agent, its session is
193/// live, no turn is running, it owns no row in progress, and the project
194/// has no live worker.
195fn lead_is_idle(ws: &Workspace, lead: &SessionSlot) -> bool {
196    if !lead.is_lead() || !ws.seat_is_live(lead) {
197        return false;
198    }
199    if ws.domain_session_for(lead).is_some_and(|domain| domain.lock().turn_in_flight()) {
200        return false;
201    }
202    let busy_row = ws
203        .tasks_for_project(lead.project())
204        .iter()
205        .any(|task| task.status == TaskStatus::InProgress && task.owner.as_ref() == Some(lead));
206    if busy_row {
207        return false;
208    }
209    let Some(view) = ws.list_projects().into_iter().find(|view| view.name == lead.project()) else {
210        return false;
211    };
212    // `is_live` is "in flight": a `Restarting` worker is not running and
213    // holds no work for the lead to wait on, though it still spends a cap
214    // slot (`holds_slot`). The sweep may inject beside one.
215    !ws.list_live_workers(&view.key).iter().any(WorkerEntry::is_live)
216}
217
218#[cfg(test)]
219mod tests {
220    use std::sync::Arc;
221    use std::time::Duration;
222
223    use parking_lot::Mutex;
224    use tempfile::tempdir;
225
226    use super::*;
227    use crate::SessionSlot;
228    use crate::chase::{Deliver, Delivery};
229    use forge_primitives::tasks::{Task, TaskId, TaskStatus};
230
231    const PROJECT: &str = "proj";
232
233    fn epoch(secs: u64) -> SystemTime {
234        SystemTime::UNIX_EPOCH + Duration::from_secs(secs)
235    }
236
237    fn fixture(dir: &std::path::Path) -> Arc<Workspace> {
238        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.to_owned());
239        ws.seed_test_project(PROJECT, "/tmp/tp-idle-sweep");
240        let db = crate::store::Db::open(&dir.join("db.redb")).expect("open db");
241        ws.install_db_for_test(db);
242        ws
243    }
244
245    fn lead() -> SessionSlot {
246        SessionSlot::lead("TestOrg", PROJECT)
247    }
248
249    /// `project`'s lead seat behind a live session. The receiver must be
250    /// held for the stub to stay.
251    fn live_lead_for(
252        ws: &Workspace,
253        project: &str,
254    ) -> tokio::sync::mpsc::UnboundedReceiver<forge_primitives::AgentCommand> {
255        ws.install_testing_stub(&SessionSlot::lead("TestOrg", project))
256    }
257
258    fn live_lead(
259        ws: &Workspace,
260    ) -> tokio::sync::mpsc::UnboundedReceiver<forge_primitives::AgentCommand> {
261        live_lead_for(ws, PROJECT)
262    }
263
264    /// Drop the seat's session, leaving the slot as a sleeping one.
265    fn sleep(ws: &Workspace) {
266        ws.domain_handles.lock().remove(&lead());
267    }
268
269    fn task(id: &str, status: TaskStatus, owner: Option<SessionSlot>) -> Task {
270        Task {
271            id: TaskId::from(id),
272            project_name: PROJECT.to_owned(),
273            subject: format!("subject {id}"),
274            active_form: None,
275            detail: None,
276            status,
277            owner,
278            parent: None,
279            waiting_on: None,
280            estimate: None,
281            rank: None,
282            verify: None,
283            attested: None,
284            reserved: None,
285            links: Vec::new(),
286            attempt: 0,
287            archived_at: None,
288            created_at: epoch(0),
289            updated_at: epoch(0),
290        }
291    }
292
293    fn live_worker(ws: &Workspace, label: &str) {
294        let key = ws
295            .list_projects()
296            .into_iter()
297            .find(|view| view.name == PROJECT)
298            .expect("view")
299            .key
300            .clone();
301        ws.insert_live_worker(
302            &key,
303            crate::WorkerEntry {
304                label: label.to_owned(),
305                charter: "charter".to_owned(),
306                slot: SessionSlot::worker("TestOrg", PROJECT, label),
307                session_id: None,
308                status: forge_primitives::WorkerLiveness::Running,
309                spawned_at: epoch(0),
310                spawned_by: lead(),
311                needs_tag: false,
312                is_git_repo_at_spawn: false,
313                cwd: None,
314                diagnostic: None,
315                kick: None,
316            },
317        );
318    }
319
320    #[derive(Default)]
321    struct Recording {
322        seen: Mutex<Vec<(SessionSlot, String)>>,
323    }
324
325    impl Deliver for Recording {
326        fn to_seat(&self, slot: &SessionSlot, text: &str) -> Delivery {
327            self.seen.lock().push((slot.clone(), text.to_owned()));
328            Delivery::Delivered
329        }
330    }
331
332    /// A delivery that fails until `open` is set, so a retry can be
333    /// watched without a live dispatch.
334    #[derive(Default)]
335    struct Flaky {
336        open: std::sync::atomic::AtomicBool,
337        seen: Mutex<usize>,
338    }
339
340    impl Deliver for Flaky {
341        fn to_seat(&self, _slot: &SessionSlot, _text: &str) -> Delivery {
342            *self.seen.lock() += 1;
343            if self.open.load(std::sync::atomic::Ordering::SeqCst) {
344                Delivery::Delivered
345            } else {
346                Delivery::NoSession("the seat's channel closed".to_owned())
347            }
348        }
349    }
350
351    fn sweep(ws: &Arc<Workspace>, deliver: &dyn Deliver, at: u64) -> usize {
352        ws.idle_sweep(deliver, epoch(at))
353    }
354
355    /// One captured tracing record: its level, its target and its fields.
356    #[derive(Clone, Default)]
357    struct Caught(std::sync::Arc<std::sync::Mutex<Vec<(tracing::Level, String, String)>>>);
358
359    #[derive(Default)]
360    struct CaughtFields(String);
361
362    impl tracing::field::Visit for CaughtFields {
363        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
364            if !self.0.is_empty() {
365                self.0.push(' ');
366            }
367            let _ = std::fmt::write(&mut self.0, format_args!("{}={value:?}", field.name()));
368        }
369    }
370
371    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Caught {
372        fn on_event(
373            &self,
374            event: &tracing::Event<'_>,
375            _ctx: tracing_subscriber::layer::Context<'_, S>,
376        ) {
377            let mut fields = CaughtFields::default();
378            event.record(&mut fields);
379            self.0.lock().expect("capture").push((
380                *event.metadata().level(),
381                event.metadata().target().to_owned(),
382                fields.0,
383            ));
384        }
385    }
386
387    fn warnings(caught: &Caught) -> Vec<(tracing::Level, String, String)> {
388        caught
389            .0
390            .lock()
391            .expect("capture")
392            .iter()
393            .filter(|(level, target, _)| {
394                *level == tracing::Level::WARN && target == "forge_workspace::idle_sweep"
395            })
396            .cloned()
397            .collect()
398    }
399
400    fn recoveries(caught: &Caught) -> Vec<(tracing::Level, String, String)> {
401        caught
402            .0
403            .lock()
404            .expect("capture")
405            .iter()
406            .filter(|(level, target, fields)| {
407                *level == tracing::Level::INFO
408                    && target == "forge_workspace::idle_sweep"
409                    && fields.contains("idle_sweep_recovered")
410            })
411            .cloned()
412            .collect()
413    }
414
415    #[test]
416    fn a_live_lead_with_nothing_in_flight_is_idle() {
417        let dir = tempdir().expect("tempdir");
418        let ws = fixture(dir.path());
419        let _rx = live_lead(&ws);
420
421        assert!(lead_is_idle(&ws, &lead()));
422    }
423
424    #[test]
425    fn a_running_turn_is_not_idle() {
426        let dir = tempdir().expect("tempdir");
427        let ws = fixture(dir.path());
428        let _rx = live_lead(&ws);
429        let domain = ws.domain_session_for(&lead()).expect("domain");
430        domain.lock().commit_turn();
431
432        assert!(!lead_is_idle(&ws, &lead()), "a lead with a turn running has something in flight");
433    }
434
435    #[test]
436    fn a_row_in_progress_under_the_lead_is_not_idle() {
437        let dir = tempdir().expect("tempdir");
438        let ws = fixture(dir.path());
439        let _rx = live_lead(&ws);
440        ws.seed_test_task(task("t-1", TaskStatus::InProgress, Some(lead())));
441
442        assert!(
443            !lead_is_idle(&ws, &lead()),
444            "a lead holding an in-progress row has work in flight",
445        );
446    }
447
448    #[test]
449    fn a_row_in_progress_under_a_worker_leaves_the_lead_idle() {
450        let dir = tempdir().expect("tempdir");
451        let ws = fixture(dir.path());
452        let _rx = live_lead(&ws);
453        ws.seed_test_task(task(
454            "t-1",
455            TaskStatus::InProgress,
456            Some(SessionSlot::worker("TestOrg", PROJECT, "w-1")),
457        ));
458
459        assert!(lead_is_idle(&ws, &lead()), "a worker's row is the worker's work, not the lead's");
460    }
461
462    #[test]
463    fn a_live_worker_in_the_project_is_not_idle() {
464        let dir = tempdir().expect("tempdir");
465        let ws = fixture(dir.path());
466        let _rx = live_lead(&ws);
467        live_worker(&ws, "w-1");
468
469        assert!(!lead_is_idle(&ws, &lead()), "a project with a live worker has work in flight");
470    }
471
472    #[test]
473    fn a_worker_seat_is_not_idle() {
474        let dir = tempdir().expect("tempdir");
475        let ws = fixture(dir.path());
476        let worker = SessionSlot::worker("TestOrg", PROJECT, "w-1");
477        let _rx = ws.install_testing_stub(&worker);
478
479        assert!(
480            !lead_is_idle(&ws, &worker),
481            "the sweep is for the project's own agent, never a worker",
482        );
483    }
484
485    #[test]
486    fn a_sleeping_lead_is_skipped_and_never_woken() {
487        let dir = tempdir().expect("tempdir");
488        let ws = fixture(dir.path());
489        let deliver = Recording::default();
490
491        assert_eq!(sweep(&ws, &deliver, 1_000), 0);
492        assert!(deliver.seen.lock().is_empty(), "a sleeping lead is never woken");
493    }
494
495    #[test]
496    fn a_lead_that_was_swept_and_then_slept_is_not_woken() {
497        let dir = tempdir().expect("tempdir");
498        let ws = fixture(dir.path());
499        let _rx = live_lead(&ws);
500        let deliver = Recording::default();
501        assert_eq!(sweep(&ws, &deliver, 1_000), 1, "the first sweep lands");
502
503        sleep(&ws);
504        assert_eq!(
505            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_FALLBACK_SECS + 60),
506            0,
507            "a sleeping lead is skipped even when its fallback is long overdue",
508        );
509        assert_eq!(deliver.seen.lock().len(), 1);
510    }
511
512    #[test]
513    fn a_live_idle_lead_is_swept_once() {
514        let dir = tempdir().expect("tempdir");
515        let ws = fixture(dir.path());
516        let _rx = live_lead(&ws);
517        let deliver = Recording::default();
518
519        assert_eq!(sweep(&ws, &deliver, 1_000), 1);
520        let seen = deliver.seen.lock();
521        assert_eq!(seen.len(), 1);
522        assert_eq!(seen[0].0, lead(), "the prompt goes to the project's lead");
523        assert_eq!(
524            seen[0].1,
525            "Nothing is in flight on this board. Sweep it now, and end with the next three picks \
526             from the backlog.",
527            "the idle path carries the plan's prompt verbatim",
528        );
529    }
530
531    #[test]
532    fn two_ticks_inside_the_gap_deliver_one_prompt() {
533        let dir = tempdir().expect("tempdir");
534        let ws = fixture(dir.path());
535        let _rx = live_lead(&ws);
536        let deliver = Recording::default();
537
538        assert_eq!(sweep(&ws, &deliver, 1_000), 1);
539        assert_eq!(sweep(&ws, &deliver, 1_060), 0, "inside the two-hour gap");
540        assert_eq!(sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_MIN_GAP_SECS), 1, "the gap is spent");
541        assert_eq!(
542            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_MIN_GAP_SECS + 60),
543            0,
544            "a delivery re-arms the gap from the delivery, not from the first sighting",
545        );
546        assert_eq!(deliver.seen.lock().len(), 2);
547    }
548
549    #[test]
550    fn a_lead_that_slept_keeps_its_gap_across_the_restart() {
551        let dir = tempdir().expect("tempdir");
552        let ws = fixture(dir.path());
553        let lead_rx = live_lead(&ws);
554        let deliver = Recording::default();
555        assert_eq!(sweep(&ws, &deliver, 1_000), 1);
556
557        drop(lead_rx);
558        sleep(&ws);
559        assert_eq!(sweep(&ws, &deliver, 1_100), 0, "asleep, nothing is delivered");
560
561        let _rx = live_lead(&ws);
562        assert_eq!(
563            sweep(&ws, &deliver, 1_200),
564            0,
565            "the wake does not reset the lead's two-hour gap",
566        );
567        assert_eq!(
568            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_MIN_GAP_SECS),
569            1,
570            "the gap still runs from the delivery before the sleep",
571        );
572    }
573
574    #[test]
575    fn a_busy_lead_is_swept_after_six_hours() {
576        let dir = tempdir().expect("tempdir");
577        let ws = fixture(dir.path());
578        let _rx = live_lead(&ws);
579        let domain = ws.domain_session_for(&lead()).expect("domain");
580        domain.lock().commit_turn();
581        let deliver = Recording::default();
582
583        assert_eq!(sweep(&ws, &deliver, 1_000), 0, "a busy lead is not idle");
584        assert_eq!(
585            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_FALLBACK_SECS - 1),
586            0,
587            "the fallback is not due yet",
588        );
589        assert_eq!(
590            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_FALLBACK_SECS),
591            1,
592            "the fallback lands even mid-turn",
593        );
594        {
595            let seen = deliver.seen.lock();
596            assert_eq!(seen.len(), 1);
597            assert_eq!(
598                seen[0].1,
599                "Six hours have passed since the last sweep. Sweep the board when you can, and \
600                 end with the next three picks from the backlog.",
601                "the fallback does not claim the board is quiet",
602            );
603        }
604        assert_eq!(
605            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_FALLBACK_SECS + 60),
606            0,
607            "one fallback, not a loop",
608        );
609        assert_eq!(
610            sweep(&ws, &deliver, 1_000 + 2 * IDLE_SWEEP_FALLBACK_SECS),
611            1,
612            "the next six-hour boundary delivers again",
613        );
614    }
615
616    #[test]
617    fn a_lead_busy_on_its_own_row_is_swept_after_six_hours() {
618        let dir = tempdir().expect("tempdir");
619        let ws = fixture(dir.path());
620        let _rx = live_lead(&ws);
621        ws.seed_test_task(task("t-1", TaskStatus::InProgress, Some(lead())));
622        let deliver = Recording::default();
623
624        assert_eq!(sweep(&ws, &deliver, 1_000), 0, "a lead on its own row is not idle");
625        assert_eq!(
626            sweep(&ws, &deliver, 1_000 + IDLE_SWEEP_FALLBACK_SECS),
627            1,
628            "the fallback covers a busy row as well as a busy turn",
629        );
630    }
631
632    #[test]
633    fn a_backwards_clock_does_not_sweep() {
634        let dir = tempdir().expect("tempdir");
635        let ws = fixture(dir.path());
636        let _rx = live_lead(&ws);
637        let deliver = Recording::default();
638
639        assert_eq!(sweep(&ws, &deliver, 2_000), 1);
640        assert_eq!(
641            sweep(&ws, &deliver, 1_900),
642            0,
643            "a clock that went backwards is unknown, not infinitely old",
644        );
645        assert_eq!(deliver.seen.lock().len(), 1);
646    }
647
648    #[test]
649    fn a_failed_delivery_is_retried_on_the_next_tick() {
650        let dir = tempdir().expect("tempdir");
651        let ws = fixture(dir.path());
652        let _rx = live_lead(&ws);
653        let deliver = Flaky::default();
654
655        assert_eq!(sweep(&ws, &deliver, 1_000), 0, "the failed attempt delivers nothing");
656        assert_eq!(*deliver.seen.lock(), 1);
657        assert_eq!(
658            sweep(&ws, &deliver, 1_060),
659            0,
660            "the failing seat is retried, not stamped as swept",
661        );
662        deliver.open.store(true, std::sync::atomic::Ordering::SeqCst);
663        assert_eq!(sweep(&ws, &deliver, 1_120), 1, "the retry succeeds");
664        assert_eq!(*deliver.seen.lock(), 3);
665    }
666
667    /// The shipped clocks, pinned where the charter's words live: a
668    /// changed interval or gap would silently falsify the paragraph.
669    #[test]
670    fn the_product_clocks_are_the_shipped_ones() {
671        assert_eq!(IDLE_SWEEP_INTERVAL_SECS, 60, "the lead is swept within a minute");
672        assert_eq!(IDLE_SWEEP_MIN_GAP_SECS, 2 * 60 * 60, "the idle gap is two hours");
673        assert_eq!(IDLE_SWEEP_FALLBACK_SECS, 6 * 60 * 60, "the fallback is six hours");
674    }
675
676    #[test]
677    fn each_project_keeps_its_own_clock_and_a_removed_project_is_pruned() {
678        let dir = tempdir().expect("tempdir");
679        let ws = fixture(dir.path());
680        let _proj_rx = live_lead(&ws);
681        let deliver = Recording::default();
682        assert_eq!(sweep(&ws, &deliver, 1_000), 1);
683
684        ws.seed_test_project("other", "/tmp/tp-idle-sweep-other");
685        let other_lead = SessionSlot::lead("TestOrg", "other");
686        let _other_rx = live_lead_for(&ws, "other");
687        assert_eq!(
688            sweep(&ws, &deliver, 1_060),
689            1,
690            "the new project's lead sweeps on its own clock",
691        );
692        {
693            let seen = deliver.seen.lock();
694            assert_eq!(seen.len(), 2);
695            assert_eq!(seen[1].0, other_lead, "the second prompt goes to the other lead");
696        }
697
698        ws.remove_test_project("other");
699        assert_eq!(sweep(&ws, &deliver, 1_120), 0);
700        let clocks = ws.idle_sweeps.lock();
701        assert_eq!(
702            clocks.len(),
703            1,
704            "a removed project's clock is pruned: {:?}",
705            clocks.keys().collect::<Vec<_>>(),
706        );
707        assert!(clocks.contains_key(&lead()));
708    }
709
710    #[test]
711    fn a_wake_re_arms_the_fallback_from_the_wake() {
712        let dir = tempdir().expect("tempdir");
713        let ws = fixture(dir.path());
714        let _rx = live_lead(&ws);
715        let domain = ws.domain_session_for(&lead()).expect("domain");
716        domain.lock().commit_turn();
717        let deliver = Recording::default();
718
719        assert_eq!(sweep(&ws, &deliver, 1_000), 0, "a busy lead is not idle");
720        sleep(&ws);
721        assert_eq!(sweep(&ws, &deliver, 30_000), 0, "asleep, nothing is delivered");
722
723        let _rx = live_lead(&ws);
724        let domain = ws.domain_session_for(&lead()).expect("domain");
725        domain.lock().commit_turn();
726        assert_eq!(
727            sweep(&ws, &deliver, 30_060),
728            0,
729            "the fallback re-arms from the wake, not from before the sleep",
730        );
731        assert_eq!(
732            sweep(&ws, &deliver, 30_000 + IDLE_SWEEP_FALLBACK_SECS),
733            1,
734            "a full fallback after the wake delivers",
735        );
736    }
737
738    /// A failure episode is one warning, its recovery one record, and the
739    /// next episode after a delivered sweep warns again.
740    #[test]
741    fn a_failure_episode_warns_once_and_its_recovery_is_recorded() {
742        use tracing_subscriber::layer::SubscriberExt;
743
744        let dir = tempdir().expect("tempdir");
745        let ws = fixture(dir.path());
746        let _rx = live_lead(&ws);
747        let deliver = Flaky::default();
748        let caught = Caught::default();
749        let subscriber = tracing_subscriber::Registry::default().with(caught.clone());
750        let _default = tracing::subscriber::set_default(subscriber);
751
752        assert_eq!(sweep(&ws, &deliver, 1_000), 0);
753        assert_eq!(sweep(&ws, &deliver, 1_060), 0);
754        let warns = warnings(&caught);
755        assert_eq!(warns.len(), 1, "two failed ticks are one episode: {warns:?}");
756        assert!(warns[0].2.contains("idle_sweep_not_delivered"));
757        assert!(warns[0].2.contains("TestOrg/proj/lead"), "the warning names the lead");
758        assert!(
759            warns[0].2.contains("the seat's channel closed"),
760            "the warning carries the dispatch's reason: {}",
761            warns[0].2,
762        );
763
764        deliver.open.store(true, std::sync::atomic::Ordering::SeqCst);
765        assert_eq!(sweep(&ws, &deliver, 1_120), 1);
766        assert_eq!(recoveries(&caught).len(), 1, "the recovery is its own record");
767        assert_eq!(warnings(&caught).len(), 1, "a delivery does not add a warning");
768
769        deliver.open.store(false, std::sync::atomic::Ordering::SeqCst);
770        assert_eq!(
771            sweep(&ws, &deliver, 1_120 + IDLE_SWEEP_MIN_GAP_SECS),
772            0,
773            "the next due attempt fails",
774        );
775        assert_eq!(
776            warnings(&caught).len(),
777            2,
778            "a failure after a delivered sweep is a new episode",
779        );
780        assert_eq!(recoveries(&caught).len(), 1);
781    }
782
783    /// A sleep between two failures does not end the episode: the seat
784    /// was down, not recovered.
785    #[test]
786    fn a_sleep_does_not_split_a_failure_episode() {
787        use tracing_subscriber::layer::SubscriberExt;
788
789        let dir = tempdir().expect("tempdir");
790        let ws = fixture(dir.path());
791        let _rx = live_lead(&ws);
792        let deliver = Flaky::default();
793        let caught = Caught::default();
794        let subscriber = tracing_subscriber::Registry::default().with(caught.clone());
795        let _default = tracing::subscriber::set_default(subscriber);
796
797        assert_eq!(sweep(&ws, &deliver, 1_000), 0);
798        assert_eq!(warnings(&caught).len(), 1);
799        sleep(&ws);
800        assert_eq!(sweep(&ws, &deliver, 1_060), 0);
801        let _rx = live_lead(&ws);
802        assert_eq!(sweep(&ws, &deliver, 1_120), 0);
803        assert_eq!(
804            warnings(&caught).len(),
805            1,
806            "a sleep is not a recovery, so the episode keeps its one warning",
807        );
808    }
809
810    /// The acceptance end to end: the boot member ticks and the prompt
811    /// reaches the lead's dispatch path.
812    #[tokio::test(start_paused = true)]
813    async fn the_boot_member_injects_the_sweep_on_its_tick() {
814        let dir = tempdir().expect("tempdir");
815        let ws = fixture(dir.path());
816        let _rx = live_lead(&ws);
817        ws.enable_test_dispatch_intercept();
818        ws.start_idle_sweep();
819
820        let names: Vec<String> =
821            ws.supervisor().snapshot().into_iter().map(|member| member.name).collect();
822        assert!(
823            names.contains(&"idle_sweep".to_owned()),
824            "the sweep is a supervised member: {names:?}"
825        );
826
827        // Let the supervisor's driver poll the member, so its interval is
828        // armed at the frozen instant zero before time moves.
829        for _ in 0..10 {
830            tokio::task::yield_now().await;
831        }
832        tokio::time::advance(Duration::from_secs(IDLE_SWEEP_INTERVAL_SECS - 1)).await;
833        for _ in 0..10 {
834            tokio::task::yield_now().await;
835        }
836        assert!(
837            ws.drain_test_dispatch_buffer().is_empty(),
838            "no eager pass lands before the interval",
839        );
840
841        tokio::time::advance(Duration::from_secs(1)).await;
842        for _ in 0..10 {
843            tokio::task::yield_now().await;
844        }
845        let prompts: Vec<(SessionSlot, String)> = ws
846            .drain_test_dispatch_buffer()
847            .iter()
848            .filter_map(|command| match command {
849                crate::protocol::Command::PromptUnder { key, text, .. } => {
850                    Some((key.clone(), text.clone()))
851                }
852                _ => None,
853            })
854            .collect();
855        assert_eq!(
856            prompts,
857            vec![(lead(), IDLE_SWEEP_PROMPT.to_owned())],
858            "one sweep, addressed to the project's lead, on the interval",
859        );
860    }
861}
forge_workspace/idle_sweep.rs

forge_workspace/
idle_sweep.rs
