gemini_adk_rs/live/
temporal.rs

1//! Temporal pattern detection for live sessions.
2//!
3//! A [`TemporalRegistry`] holds named [`TemporalPattern`]s that combine a
4//! [`PatternDetector`] (the condition) with an async action (the response).
5//! Detectors track time-based and count-based conditions such as sustained
6//! state, event rates, consecutive turns, and consecutive tool failures.
7//!
8//! The registry is evaluated by the control-lane processor on each event and
9//! optionally on a periodic timer (when [`TemporalRegistry::needs_timer`]
10//! returns `true`).
11
12use std::collections::VecDeque;
13use std::sync::Arc;
14use std::sync::atomic::{AtomicU32, Ordering};
15use std::time::{Duration, Instant};
16
17use gemini_genai_rs::session::{SessionEvent, SessionWriter};
18
19use super::BoxFuture;
20use crate::state::State;
21
22// ── PatternDetector trait ─────────────────────────────────────────────────────
23
24/// A detector that evaluates whether a temporal pattern has been triggered.
25///
26/// Implementations track internal state (timestamps, counters) to decide
27/// when a pattern fires. All interior mutability is handled via atomic
28/// operations or `parking_lot` locks so that `&self` suffices.
29pub trait PatternDetector: Send + Sync {
30    /// Evaluate whether the pattern is currently triggered.
31    ///
32    /// - `state`: the current agent state snapshot.
33    /// - `event`: the session event that prompted this check (if any).
34    /// - `now`: the current instant (passed in for testability).
35    fn check(&self, state: &State, event: Option<&SessionEvent>, now: Instant) -> bool;
36
37    /// Reset the detector's internal state (counters, timestamps, etc.).
38    fn reset(&self);
39
40    /// Whether this detector requires periodic timer checks.
41    ///
42    /// Detectors that depend on elapsed time (e.g. [`SustainedDetector`])
43    /// should return `true` so the runtime can schedule a timer.
44    fn needs_timer(&self) -> bool {
45        false
46    }
47}
48
49// ── TemporalPattern ───────────────────────────────────────────────────────────
50
51/// A named temporal pattern: detector + action + cooldown.
52pub struct TemporalPattern {
53    /// Human-readable name for logging/debugging.
54    pub name: String,
55    /// The detector that decides when to fire.
56    pub detector: Box<dyn PatternDetector>,
57    /// The async action to execute when the pattern triggers.
58    /// Receives a cloned `State` and the session writer.
59    pub action: super::SessionHook,
60    /// Optional minimum interval between successive firings.
61    pub cooldown: Option<Duration>,
62    /// Tracks when this pattern last fired (for cooldown enforcement).
63    last_triggered: parking_lot::Mutex<Option<Instant>>,
64}
65
66impl TemporalPattern {
67    /// Create a new temporal pattern.
68    pub fn new(
69        name: impl Into<String>,
70        detector: Box<dyn PatternDetector>,
71        action: super::SessionHook,
72        cooldown: Option<Duration>,
73    ) -> Self {
74        Self {
75            name: name.into(),
76            detector,
77            action,
78            cooldown,
79            last_triggered: parking_lot::Mutex::new(None),
80        }
81    }
82
83    /// Check whether the pattern fires and cooldown allows it.
84    fn try_fire(
85        &self,
86        state: &State,
87        event: Option<&SessionEvent>,
88        writer: &Arc<dyn SessionWriter>,
89        now: Instant,
90    ) -> Option<BoxFuture<()>> {
91        if !self.detector.check(state, event, now) {
92            return None;
93        }
94
95        // Enforce cooldown.
96        let mut last = self.last_triggered.lock();
97        if let (Some(cooldown), Some(prev)) = (self.cooldown, *last)
98            && now.duration_since(prev) < cooldown
99        {
100            return None;
101        }
102
103        *last = Some(now);
104
105        let s = state.clone();
106        let w = writer.clone();
107        Some((self.action)(s, w))
108    }
109}
110
111// ── TemporalRegistry ─────────────────────────────────────────────────────────
112
113/// Registry of temporal patterns evaluated on events and/or timer ticks.
114pub struct TemporalRegistry {
115    patterns: Vec<TemporalPattern>,
116}
117
118impl Default for TemporalRegistry {
119    fn default() -> Self {
120        Self::new()
121    }
122}
123
124impl TemporalRegistry {
125    /// Create an empty registry.
126    pub fn new() -> Self {
127        Self {
128            patterns: Vec::new(),
129        }
130    }
131
132    /// Register a pattern.
133    pub fn register(&mut self, pattern: TemporalPattern) {
134        self.patterns.push(pattern);
135    }
136
137    /// Check all patterns and return the futures for those that fired.
138    ///
139    /// Called by the control-lane processor on each event and optionally on
140    /// a periodic timer tick.
141    pub fn check_all(
142        &self,
143        state: &State,
144        event: Option<&SessionEvent>,
145        writer: &Arc<dyn SessionWriter>,
146    ) -> Vec<BoxFuture<()>> {
147        let now = state.clock().now();
148        self.patterns
149            .iter()
150            .filter_map(|p| p.try_fire(state, event, writer, now))
151            .collect()
152    }
153
154    /// Returns `true` if any registered pattern's detector needs periodic
155    /// timer checks (i.e. its [`PatternDetector::needs_timer`] returns `true`).
156    pub fn needs_timer(&self) -> bool {
157        self.patterns.iter().any(|p| p.detector.needs_timer())
158    }
159}
160
161// ── SustainedDetector ─────────────────────────────────────────────────────────
162
163/// Fires when a state-based condition remains true for at least `duration`.
164///
165/// On each `check()`:
166/// - If the condition is true and `became_true_at` is `None`, record `now`.
167/// - If the condition is true and `became_true_at` is `Some(t)`, return
168///   `true` when `now - t >= duration`.
169/// - If the condition is false, reset `became_true_at` to `None`.
170///
171/// This detector **needs periodic timer checks** because it depends on
172/// elapsed wall-clock time.
173pub struct SustainedDetector {
174    condition: Arc<dyn Fn(&State) -> bool + Send + Sync>,
175    duration: Duration,
176    became_true_at: parking_lot::Mutex<Option<Instant>>,
177}
178
179impl SustainedDetector {
180    /// Create a new sustained detector.
181    ///
182    /// - `condition`: evaluated against the current state.
183    /// - `duration`: how long the condition must remain true before firing.
184    pub fn new(condition: Arc<dyn Fn(&State) -> bool + Send + Sync>, duration: Duration) -> Self {
185        Self {
186            condition,
187            duration,
188            became_true_at: parking_lot::Mutex::new(None),
189        }
190    }
191}
192
193impl PatternDetector for SustainedDetector {
194    fn check(&self, state: &State, _event: Option<&SessionEvent>, now: Instant) -> bool {
195        if (self.condition)(state) {
196            let mut guard = self.became_true_at.lock();
197            match *guard {
198                None => {
199                    *guard = Some(now);
200                    false
201                }
202                Some(t) => now.duration_since(t) >= self.duration,
203            }
204        } else {
205            *self.became_true_at.lock() = None;
206            false
207        }
208    }
209
210    fn reset(&self) {
211        *self.became_true_at.lock() = None;
212    }
213
214    fn needs_timer(&self) -> bool {
215        true
216    }
217}
218
219// ── RateDetector ──────────────────────────────────────────────────────────────
220
221/// Fires when at least `count` matching events occur within `window`.
222///
223/// On each `check()`:
224/// - If `event` is `Some` and the filter accepts it, push the current
225///   timestamp.
226/// - Expire timestamps older than `window`.
227/// - Return `true` if the remaining count >= threshold.
228pub struct RateDetector {
229    filter: Arc<dyn Fn(&SessionEvent) -> bool + Send + Sync>,
230    count: u32,
231    window: Duration,
232    timestamps: parking_lot::Mutex<VecDeque<Instant>>,
233}
234
235impl RateDetector {
236    /// Create a new rate detector.
237    ///
238    /// - `filter`: predicate to select which events count.
239    /// - `count`: number of matching events required.
240    /// - `window`: sliding time window.
241    pub fn new(
242        filter: Arc<dyn Fn(&SessionEvent) -> bool + Send + Sync>,
243        count: u32,
244        window: Duration,
245    ) -> Self {
246        Self {
247            filter,
248            count,
249            window,
250            timestamps: parking_lot::Mutex::new(VecDeque::new()),
251        }
252    }
253}
254
255impl PatternDetector for RateDetector {
256    fn check(&self, _state: &State, event: Option<&SessionEvent>, now: Instant) -> bool {
257        let mut ts = self.timestamps.lock();
258
259        // Record matching event.
260        if let Some(evt) = event
261            && (self.filter)(evt)
262        {
263            ts.push_back(now);
264        }
265
266        // Expire old timestamps.
267        while let Some(&front) = ts.front() {
268            if now.duration_since(front) > self.window {
269                ts.pop_front();
270            } else {
271                break;
272            }
273        }
274
275        ts.len() as u32 >= self.count
276    }
277
278    fn reset(&self) {
279        self.timestamps.lock().clear();
280    }
281
282    // RateDetector does not need timer — it is event-driven.
283}
284
285// ── TurnCountDetector ─────────────────────────────────────────────────────────
286
287/// Fires when a state-based condition is true for `required` consecutive
288/// evaluations (typically one evaluation per turn).
289///
290/// The caller decides when to invoke `check()` — usually on `TurnComplete`
291/// events.
292pub struct TurnCountDetector {
293    condition: Arc<dyn Fn(&State) -> bool + Send + Sync>,
294    required: u32,
295    consecutive: AtomicU32,
296}
297
298impl TurnCountDetector {
299    /// Create a new turn-count detector.
300    ///
301    /// - `condition`: evaluated against the current state each turn.
302    /// - `required`: number of consecutive true results before firing.
303    pub fn new(condition: Arc<dyn Fn(&State) -> bool + Send + Sync>, required: u32) -> Self {
304        Self {
305            condition,
306            required,
307            consecutive: AtomicU32::new(0),
308        }
309    }
310}
311
312impl PatternDetector for TurnCountDetector {
313    fn check(&self, state: &State, _event: Option<&SessionEvent>, _now: Instant) -> bool {
314        if (self.condition)(state) {
315            let prev = self.consecutive.fetch_add(1, Ordering::SeqCst);
316            prev + 1 >= self.required
317        } else {
318            self.consecutive.store(0, Ordering::SeqCst);
319            false
320        }
321    }
322
323    fn reset(&self) {
324        self.consecutive.store(0, Ordering::SeqCst);
325    }
326}
327
328// ── ConsecutiveFailureDetector ────────────────────────────────────────────────
329
330/// Fires when a named tool has failed `threshold` consecutive times.
331///
332/// Uses a state-key convention: if `bg:{tool_name}_failed` is `true` the
333/// tool is considered to have failed; if `false` (or absent) the streak
334/// resets.
335pub struct ConsecutiveFailureDetector {
336    tool_name: String,
337    threshold: u32,
338    consecutive: AtomicU32,
339}
340
341impl ConsecutiveFailureDetector {
342    /// Create a new consecutive-failure detector.
343    ///
344    /// - `tool_name`: the tool whose failures are tracked.
345    /// - `threshold`: how many consecutive failures before firing.
346    pub fn new(tool_name: impl Into<String>, threshold: u32) -> Self {
347        Self {
348            tool_name: tool_name.into(),
349            threshold,
350            consecutive: AtomicU32::new(0),
351        }
352    }
353}
354
355impl PatternDetector for ConsecutiveFailureDetector {
356    fn check(&self, state: &State, _event: Option<&SessionEvent>, _now: Instant) -> bool {
357        let key = format!("bg:{}_failed", self.tool_name);
358        let failed: bool = state.get(&key).unwrap_or(false);
359
360        if failed {
361            let prev = self.consecutive.fetch_add(1, Ordering::SeqCst);
362            prev + 1 >= self.threshold
363        } else {
364            self.consecutive.store(0, Ordering::SeqCst);
365            false
366        }
367    }
368
369    fn reset(&self) {
370        self.consecutive.store(0, Ordering::SeqCst);
371    }
372}
373
374// ── Tests ─────────────────────────────────────────────────────────────────────
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379    use std::sync::atomic::{AtomicU32, Ordering};
380
381    /// Dummy SessionWriter for tests — all methods return Ok.
382    struct MockWriter;
383
384    #[async_trait::async_trait]
385    impl SessionWriter for MockWriter {
386        async fn send_audio(
387            &self,
388            _: bytes::Bytes,
389        ) -> Result<(), gemini_genai_rs::session::SessionError> {
390            Ok(())
391        }
392        async fn send_text(&self, _: String) -> Result<(), gemini_genai_rs::session::SessionError> {
393            Ok(())
394        }
395        async fn send_tool_response(
396            &self,
397            _: Vec<gemini_genai_rs::protocol::FunctionResponse>,
398        ) -> Result<(), gemini_genai_rs::session::SessionError> {
399            Ok(())
400        }
401        async fn send_client_content(
402            &self,
403            _: Vec<gemini_genai_rs::protocol::Content>,
404            _: bool,
405        ) -> Result<(), gemini_genai_rs::session::SessionError> {
406            Ok(())
407        }
408        async fn send_video(
409            &self,
410            _: bytes::Bytes,
411        ) -> Result<(), gemini_genai_rs::session::SessionError> {
412            Ok(())
413        }
414        async fn update_instruction(
415            &self,
416            _: String,
417        ) -> Result<(), gemini_genai_rs::session::SessionError> {
418            Ok(())
419        }
420        async fn signal_activity_start(
421            &self,
422        ) -> Result<(), gemini_genai_rs::session::SessionError> {
423            Ok(())
424        }
425        async fn signal_activity_end(&self) -> Result<(), gemini_genai_rs::session::SessionError> {
426            Ok(())
427        }
428        async fn disconnect(&self) -> Result<(), gemini_genai_rs::session::SessionError> {
429            Ok(())
430        }
431    }
432
433    fn mock_writer() -> Arc<dyn SessionWriter> {
434        Arc::new(MockWriter)
435    }
436
437    /// Helper: action that increments a shared counter.
438    fn counting_action(counter: Arc<AtomicU32>) -> crate::live::SessionHook {
439        Arc::new(move |_state, _writer| {
440            let c = counter.clone();
441            Box::pin(async move {
442                c.fetch_add(1, Ordering::SeqCst);
443            })
444        })
445    }
446
447    // ── 1. SustainedDetector fires after duration elapses ─────────────────
448
449    #[test]
450    fn sustained_fires_after_duration() {
451        let state = State::new();
452        let _ = state.set("hot", true);
453
454        let detector = SustainedDetector::new(
455            Arc::new(|s: &State| s.get::<bool>("hot").unwrap_or(false)),
456            Duration::from_secs(5),
457        );
458
459        let t0 = Instant::now();
460
461        // First check: records the start time, does not fire yet.
462        assert!(!detector.check(&state, None, t0));
463
464        // 3 seconds later: not yet.
465        assert!(!detector.check(&state, None, t0 + Duration::from_secs(3)));
466
467        // 5 seconds later: fires.
468        assert!(detector.check(&state, None, t0 + Duration::from_secs(5)));
469
470        // Still fires on subsequent checks while condition holds.
471        assert!(detector.check(&state, None, t0 + Duration::from_secs(6)));
472    }
473
474    // ── 2. SustainedDetector resets when condition becomes false ───────────
475
476    #[test]
477    fn sustained_resets_on_false() {
478        let state = State::new();
479        let _ = state.set("hot", true);
480
481        let detector = SustainedDetector::new(
482            Arc::new(|s: &State| s.get::<bool>("hot").unwrap_or(false)),
483            Duration::from_secs(5),
484        );
485
486        let t0 = Instant::now();
487
488        // Start tracking.
489        assert!(!detector.check(&state, None, t0));
490
491        // Condition becomes false at t0+2s — resets internal timer.
492        let _ = state.set("hot", false);
493        assert!(!detector.check(&state, None, t0 + Duration::from_secs(2)));
494
495        // Condition becomes true again at t0+3s — starts fresh.
496        let _ = state.set("hot", true);
497        assert!(!detector.check(&state, None, t0 + Duration::from_secs(3)));
498
499        // t0+7s: only 4s since re-start at t0+3s — not enough.
500        assert!(!detector.check(&state, None, t0 + Duration::from_secs(7)));
501
502        // t0+8s: 5s since t0+3s — fires.
503        assert!(detector.check(&state, None, t0 + Duration::from_secs(8)));
504    }
505
506    // ── 3. SustainedDetector reset() clears state ─────────────────────────
507
508    #[test]
509    fn sustained_reset_clears_state() {
510        let state = State::new();
511        let _ = state.set("hot", true);
512
513        let detector = SustainedDetector::new(
514            Arc::new(|s: &State| s.get::<bool>("hot").unwrap_or(false)),
515            Duration::from_secs(5),
516        );
517
518        let t0 = Instant::now();
519
520        // Start tracking.
521        assert!(!detector.check(&state, None, t0));
522
523        // Explicit reset.
524        detector.reset();
525
526        // Must start tracking from scratch — 5s from the new check.
527        assert!(!detector.check(&state, None, t0 + Duration::from_secs(4)));
528        assert!(detector.check(&state, None, t0 + Duration::from_secs(9)));
529    }
530
531    // ── 4. RateDetector fires when count reached in window ────────────────
532
533    #[test]
534    fn rate_fires_when_count_reached() {
535        let state = State::new();
536        let detector = RateDetector::new(
537            Arc::new(|evt: &SessionEvent| matches!(evt, SessionEvent::TurnComplete)),
538            3,
539            Duration::from_secs(10),
540        );
541
542        let t0 = Instant::now();
543        let event = SessionEvent::TurnComplete;
544
545        assert!(!detector.check(&state, Some(&event), t0));
546        assert!(!detector.check(&state, Some(&event), t0 + Duration::from_secs(1)));
547        // Third event: fires.
548        assert!(detector.check(&state, Some(&event), t0 + Duration::from_secs(2)));
549    }
550
551    // ── 5. RateDetector does not fire when events outside window ──────────
552
553    #[test]
554    fn rate_does_not_fire_when_events_outside_window() {
555        let state = State::new();
556        let detector = RateDetector::new(
557            Arc::new(|evt: &SessionEvent| matches!(evt, SessionEvent::TurnComplete)),
558            3,
559            Duration::from_secs(5),
560        );
561
562        let t0 = Instant::now();
563        let event = SessionEvent::TurnComplete;
564
565        // Two events at t0.
566        assert!(!detector.check(&state, Some(&event), t0));
567        assert!(!detector.check(&state, Some(&event), t0 + Duration::from_secs(1)));
568
569        // Third event at t0+10s: first two have expired.
570        assert!(!detector.check(&state, Some(&event), t0 + Duration::from_secs(10)));
571    }
572
573    // ── 6. RateDetector with filter that rejects events ───────────────────
574
575    #[test]
576    fn rate_filter_rejects_events() {
577        let state = State::new();
578        let detector = RateDetector::new(
579            Arc::new(|evt: &SessionEvent| matches!(evt, SessionEvent::TurnComplete)),
580            2,
581            Duration::from_secs(10),
582        );
583
584        let t0 = Instant::now();
585
586        // These events don't match the filter.
587        let text_event = SessionEvent::TextDelta("hello".to_string());
588        assert!(!detector.check(&state, Some(&text_event), t0));
589        assert!(!detector.check(&state, Some(&text_event), t0 + Duration::from_secs(1)));
590        assert!(!detector.check(&state, Some(&text_event), t0 + Duration::from_secs(2)));
591
592        // Still at 0 matching events — no fire.
593        assert!(!detector.check(&state, None, t0 + Duration::from_secs(3)));
594    }
595
596    // ── 7. TurnCountDetector fires after N consecutive true ───────────────
597
598    #[test]
599    fn turn_count_fires_after_n_consecutive() {
600        let state = State::new();
601        let _ = state.set("confused", true);
602
603        let detector = TurnCountDetector::new(
604            Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
605            3,
606        );
607
608        let t0 = Instant::now();
609
610        assert!(!detector.check(&state, None, t0));
611        assert!(!detector.check(&state, None, t0));
612        // Third consecutive true: fires.
613        assert!(detector.check(&state, None, t0));
614    }
615
616    // ── 8. TurnCountDetector resets on false ──────────────────────────────
617
618    #[test]
619    fn turn_count_resets_on_false() {
620        let state = State::new();
621        let _ = state.set("confused", true);
622
623        let detector = TurnCountDetector::new(
624            Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
625            3,
626        );
627
628        let t0 = Instant::now();
629
630        assert!(!detector.check(&state, None, t0));
631        assert!(!detector.check(&state, None, t0));
632
633        // Condition becomes false — resets counter.
634        let _ = state.set("confused", false);
635        assert!(!detector.check(&state, None, t0));
636
637        // Start again — need 3 more consecutive trues.
638        let _ = state.set("confused", true);
639        assert!(!detector.check(&state, None, t0));
640        assert!(!detector.check(&state, None, t0));
641        assert!(detector.check(&state, None, t0));
642    }
643
644    // ── 9. ConsecutiveFailureDetector fires after threshold ───────────────
645
646    #[test]
647    fn consecutive_failure_fires_after_threshold() {
648        let state = State::new();
649        let _ = state.set("bg:search_failed", true);
650
651        let detector = ConsecutiveFailureDetector::new("search", 3);
652
653        let t0 = Instant::now();
654
655        assert!(!detector.check(&state, None, t0));
656        assert!(!detector.check(&state, None, t0));
657        // Third consecutive failure: fires.
658        assert!(detector.check(&state, None, t0));
659    }
660
661    // ── 10. ConsecutiveFailureDetector resets on success ──────────────────
662
663    #[test]
664    fn consecutive_failure_resets_on_success() {
665        let state = State::new();
666        let _ = state.set("bg:search_failed", true);
667
668        let detector = ConsecutiveFailureDetector::new("search", 3);
669
670        let t0 = Instant::now();
671
672        assert!(!detector.check(&state, None, t0));
673        assert!(!detector.check(&state, None, t0));
674
675        // Tool succeeds — reset.
676        let _ = state.set("bg:search_failed", false);
677        assert!(!detector.check(&state, None, t0));
678
679        // Must accumulate again from 0.
680        let _ = state.set("bg:search_failed", true);
681        assert!(!detector.check(&state, None, t0));
682        assert!(!detector.check(&state, None, t0));
683        assert!(detector.check(&state, None, t0));
684    }
685
686    // ── 11. TemporalPattern cooldown prevents rapid re-firing ─────────────
687
688    #[tokio::test]
689    async fn pattern_cooldown_prevents_rapid_refiring() {
690        let counter = Arc::new(AtomicU32::new(0));
691        let state = State::new();
692        let _ = state.set("active", true);
693        let writer = mock_writer();
694
695        let pattern = TemporalPattern::new(
696            "test-cooldown",
697            Box::new(SustainedDetector::new(
698                Arc::new(|s: &State| s.get::<bool>("active").unwrap_or(false)),
699                Duration::from_secs(0), // fires immediately once became_true_at is set
700            )),
701            counting_action(counter.clone()),
702            Some(Duration::from_secs(10)), // 10s cooldown
703        );
704
705        let t0 = Instant::now();
706
707        // First check: sets became_true_at but doesn't fire (duration=0, but
708        // the first check just records the start time).
709        assert!(pattern.try_fire(&state, None, &writer, t0).is_none());
710
711        // Second check: fires (condition true + duration=0 elapsed).
712        let fut = pattern.try_fire(&state, None, &writer, t0 + Duration::from_millis(1));
713        assert!(fut.is_some());
714        fut.unwrap().await;
715        assert_eq!(counter.load(Ordering::SeqCst), 1);
716
717        // Immediate re-check: cooldown blocks.
718        assert!(
719            pattern
720                .try_fire(&state, None, &writer, t0 + Duration::from_millis(2))
721                .is_none()
722        );
723
724        // After cooldown: fires again.
725        let fut = pattern.try_fire(&state, None, &writer, t0 + Duration::from_secs(11));
726        assert!(fut.is_some());
727        fut.unwrap().await;
728        assert_eq!(counter.load(Ordering::SeqCst), 2);
729    }
730
731    // ── 12. TemporalRegistry check_all returns actions ────────────────────
732
733    #[tokio::test]
734    async fn registry_check_all_returns_actions() {
735        let counter = Arc::new(AtomicU32::new(0));
736        let state = State::new();
737        let _ = state.set("confused", true);
738        let writer = mock_writer();
739
740        let mut registry = TemporalRegistry::new();
741
742        // TurnCountDetector with required=1 — fires on first true check.
743        registry.register(TemporalPattern::new(
744            "confusion",
745            Box::new(TurnCountDetector::new(
746                Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
747                1,
748            )),
749            counting_action(counter.clone()),
750            None,
751        ));
752
753        let actions = registry.check_all(&state, None, &writer);
754        assert_eq!(actions.len(), 1);
755
756        for fut in actions {
757            fut.await;
758        }
759        assert_eq!(counter.load(Ordering::SeqCst), 1);
760    }
761
762    /// A sustained pattern measures its duration on the state's clock, so a
763    /// manual clock decides when it fires, not the wall clock.
764    #[tokio::test]
765    async fn registry_reads_time_from_the_state_clock() {
766        let counter = Arc::new(AtomicU32::new(0));
767        let clock = Arc::new(crate::clock::ManualClock::new());
768        let state = State::new().with_clock(clock.clone());
769        let writer = mock_writer();
770
771        let mut registry = TemporalRegistry::new();
772        registry.register(TemporalPattern::new(
773            "sustained",
774            Box::new(SustainedDetector::new(
775                Arc::new(|_: &State| true),
776                Duration::from_secs(5),
777            )),
778            counting_action(counter.clone()),
779            None,
780        ));
781
782        assert!(registry.check_all(&state, None, &writer).is_empty());
783        clock.advance(Duration::from_secs(4));
784        assert!(registry.check_all(&state, None, &writer).is_empty());
785        clock.advance(Duration::from_secs(1));
786        let fired = registry.check_all(&state, None, &writer);
787        assert_eq!(fired.len(), 1, "fires once five clock seconds have passed");
788        for fut in fired {
789            fut.await;
790        }
791        assert_eq!(counter.load(Ordering::SeqCst), 1);
792    }
793
794    // ── 13. needs_timer returns true when SustainedDetector is registered ─
795
796    #[test]
797    fn needs_timer_true_with_sustained_detector() {
798        let counter = Arc::new(AtomicU32::new(0));
799        let mut registry = TemporalRegistry::new();
800
801        registry.register(TemporalPattern::new(
802            "sustained",
803            Box::new(SustainedDetector::new(
804                Arc::new(|_: &State| true),
805                Duration::from_secs(5),
806            )),
807            counting_action(counter),
808            None,
809        ));
810
811        assert!(registry.needs_timer());
812    }
813
814    // ── 14. needs_timer returns false when no SustainedDetector ───────────
815
816    #[test]
817    fn needs_timer_false_without_sustained_detector() {
818        let counter = Arc::new(AtomicU32::new(0));
819        let mut registry = TemporalRegistry::new();
820
821        registry.register(TemporalPattern::new(
822            "turn-count",
823            Box::new(TurnCountDetector::new(Arc::new(|_: &State| true), 3)),
824            counting_action(counter.clone()),
825            None,
826        ));
827
828        registry.register(TemporalPattern::new(
829            "rate",
830            Box::new(RateDetector::new(
831                Arc::new(|_: &SessionEvent| true),
832                5,
833                Duration::from_secs(10),
834            )),
835            counting_action(counter),
836            None,
837        ));
838
839        assert!(!registry.needs_timer());
840    }
841
842    // ── Additional: Default creates empty registry ────────────────────────
843
844    #[test]
845    fn default_creates_empty_registry() {
846        let registry = TemporalRegistry::default();
847        assert!(!registry.needs_timer());
848    }
849
850    // ── Additional: RateDetector reset clears timestamps ──────────────────
851
852    #[test]
853    fn rate_reset_clears_timestamps() {
854        let state = State::new();
855        let detector = RateDetector::new(
856            Arc::new(|evt: &SessionEvent| matches!(evt, SessionEvent::TurnComplete)),
857            2,
858            Duration::from_secs(10),
859        );
860
861        let t0 = Instant::now();
862        let event = SessionEvent::TurnComplete;
863
864        assert!(!detector.check(&state, Some(&event), t0));
865        detector.reset();
866        // After reset, first event should not be enough.
867        assert!(!detector.check(&state, Some(&event), t0 + Duration::from_secs(1)));
868        // Second event after reset fires.
869        assert!(detector.check(&state, Some(&event), t0 + Duration::from_secs(2)));
870    }
871
872    // ── Additional: TurnCountDetector reset clears counter ────────────────
873
874    #[test]
875    fn turn_count_reset_clears_counter() {
876        let state = State::new();
877        let _ = state.set("confused", true);
878
879        let detector = TurnCountDetector::new(
880            Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
881            3,
882        );
883
884        let t0 = Instant::now();
885
886        assert!(!detector.check(&state, None, t0));
887        assert!(!detector.check(&state, None, t0));
888        // Two consecutive trues accumulated.
889
890        detector.reset();
891
892        // After reset, need 3 more.
893        assert!(!detector.check(&state, None, t0));
894        assert!(!detector.check(&state, None, t0));
895        assert!(detector.check(&state, None, t0));
896    }
897
898    // ── Additional: ConsecutiveFailureDetector reset clears counter ───────
899
900    #[test]
901    fn consecutive_failure_reset_clears_counter() {
902        let state = State::new();
903        let _ = state.set("bg:search_failed", true);
904
905        let detector = ConsecutiveFailureDetector::new("search", 3);
906        let t0 = Instant::now();
907
908        assert!(!detector.check(&state, None, t0));
909        assert!(!detector.check(&state, None, t0));
910
911        detector.reset();
912
913        assert!(!detector.check(&state, None, t0));
914        assert!(!detector.check(&state, None, t0));
915        assert!(detector.check(&state, None, t0));
916    }
917
918    // ── Additional: SustainedDetector needs_timer is true ─────────────────
919
920    #[test]
921    fn sustained_detector_needs_timer() {
922        let detector = SustainedDetector::new(Arc::new(|_: &State| true), Duration::from_secs(5));
923        assert!(detector.needs_timer());
924    }
925
926    // ── Additional: RateDetector needs_timer is false ─────────────────────
927
928    #[test]
929    fn rate_detector_does_not_need_timer() {
930        let detector = RateDetector::new(
931            Arc::new(|_: &SessionEvent| true),
932            5,
933            Duration::from_secs(10),
934        );
935        assert!(!detector.needs_timer());
936    }
937
938    // ── Additional: TurnCountDetector needs_timer is false ────────────────
939
940    #[test]
941    fn turn_count_detector_does_not_need_timer() {
942        let detector = TurnCountDetector::new(Arc::new(|_: &State| true), 3);
943        assert!(!detector.needs_timer());
944    }
945
946    // ── Additional: Pattern without cooldown fires every time ─────────────
947
948    #[tokio::test]
949    async fn pattern_without_cooldown_fires_every_time() {
950        let counter = Arc::new(AtomicU32::new(0));
951        let state = State::new();
952        let _ = state.set("active", true);
953        let writer = mock_writer();
954
955        let pattern = TemporalPattern::new(
956            "no-cooldown",
957            Box::new(TurnCountDetector::new(
958                Arc::new(|s: &State| s.get::<bool>("active").unwrap_or(false)),
959                1,
960            )),
961            counting_action(counter.clone()),
962            None, // no cooldown
963        );
964
965        let t0 = Instant::now();
966
967        for i in 0..5u32 {
968            let fut = pattern.try_fire(&state, None, &writer, t0 + Duration::from_millis(i as u64));
969            assert!(fut.is_some(), "should fire on iteration {i}");
970            fut.unwrap().await;
971        }
972
973        assert_eq!(counter.load(Ordering::SeqCst), 5);
974    }
975}
gemini_adk_rs/live/temporal.rs

gemini_adk_rs/live/
temporal.rs
