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::atomic::{AtomicU32, Ordering};
14use std::sync::Arc;
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::phase::PhaseHook,
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::phase::PhaseHook,
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            if now.duration_since(prev) < cooldown {
99                return None;
100            }
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    /// Add a pattern to the registry.
133    pub fn add(&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 = Instant::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            if (self.filter)(evt) {
262                ts.push_back(now);
263            }
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            _: Vec<u8>,
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            _: Vec<u8>,
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::phase::PhaseHook {
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!(pattern
719            .try_fire(&state, None, &writer, t0 + Duration::from_millis(2))
720            .is_none());
721
722        // After cooldown: fires again.
723        let fut = pattern.try_fire(&state, None, &writer, t0 + Duration::from_secs(11));
724        assert!(fut.is_some());
725        fut.unwrap().await;
726        assert_eq!(counter.load(Ordering::SeqCst), 2);
727    }
728
729    // ── 12. TemporalRegistry check_all returns actions ────────────────────
730
731    #[tokio::test]
732    async fn registry_check_all_returns_actions() {
733        let counter = Arc::new(AtomicU32::new(0));
734        let state = State::new();
735        let _ = state.set("confused", true);
736        let writer = mock_writer();
737
738        let mut registry = TemporalRegistry::new();
739
740        // TurnCountDetector with required=1 — fires on first true check.
741        registry.add(TemporalPattern::new(
742            "confusion",
743            Box::new(TurnCountDetector::new(
744                Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
745                1,
746            )),
747            counting_action(counter.clone()),
748            None,
749        ));
750
751        let actions = registry.check_all(&state, None, &writer);
752        assert_eq!(actions.len(), 1);
753
754        for fut in actions {
755            fut.await;
756        }
757        assert_eq!(counter.load(Ordering::SeqCst), 1);
758    }
759
760    // ── 13. needs_timer returns true when SustainedDetector is registered ─
761
762    #[test]
763    fn needs_timer_true_with_sustained_detector() {
764        let counter = Arc::new(AtomicU32::new(0));
765        let mut registry = TemporalRegistry::new();
766
767        registry.add(TemporalPattern::new(
768            "sustained",
769            Box::new(SustainedDetector::new(
770                Arc::new(|_: &State| true),
771                Duration::from_secs(5),
772            )),
773            counting_action(counter),
774            None,
775        ));
776
777        assert!(registry.needs_timer());
778    }
779
780    // ── 14. needs_timer returns false when no SustainedDetector ───────────
781
782    #[test]
783    fn needs_timer_false_without_sustained_detector() {
784        let counter = Arc::new(AtomicU32::new(0));
785        let mut registry = TemporalRegistry::new();
786
787        registry.add(TemporalPattern::new(
788            "turn-count",
789            Box::new(TurnCountDetector::new(Arc::new(|_: &State| true), 3)),
790            counting_action(counter.clone()),
791            None,
792        ));
793
794        registry.add(TemporalPattern::new(
795            "rate",
796            Box::new(RateDetector::new(
797                Arc::new(|_: &SessionEvent| true),
798                5,
799                Duration::from_secs(10),
800            )),
801            counting_action(counter),
802            None,
803        ));
804
805        assert!(!registry.needs_timer());
806    }
807
808    // ── Additional: Default creates empty registry ────────────────────────
809
810    #[test]
811    fn default_creates_empty_registry() {
812        let registry = TemporalRegistry::default();
813        assert!(!registry.needs_timer());
814    }
815
816    // ── Additional: RateDetector reset clears timestamps ──────────────────
817
818    #[test]
819    fn rate_reset_clears_timestamps() {
820        let state = State::new();
821        let detector = RateDetector::new(
822            Arc::new(|evt: &SessionEvent| matches!(evt, SessionEvent::TurnComplete)),
823            2,
824            Duration::from_secs(10),
825        );
826
827        let t0 = Instant::now();
828        let event = SessionEvent::TurnComplete;
829
830        assert!(!detector.check(&state, Some(&event), t0));
831        detector.reset();
832        // After reset, first event should not be enough.
833        assert!(!detector.check(&state, Some(&event), t0 + Duration::from_secs(1)));
834        // Second event after reset fires.
835        assert!(detector.check(&state, Some(&event), t0 + Duration::from_secs(2)));
836    }
837
838    // ── Additional: TurnCountDetector reset clears counter ────────────────
839
840    #[test]
841    fn turn_count_reset_clears_counter() {
842        let state = State::new();
843        let _ = state.set("confused", true);
844
845        let detector = TurnCountDetector::new(
846            Arc::new(|s: &State| s.get::<bool>("confused").unwrap_or(false)),
847            3,
848        );
849
850        let t0 = Instant::now();
851
852        assert!(!detector.check(&state, None, t0));
853        assert!(!detector.check(&state, None, t0));
854        // Two consecutive trues accumulated.
855
856        detector.reset();
857
858        // After reset, need 3 more.
859        assert!(!detector.check(&state, None, t0));
860        assert!(!detector.check(&state, None, t0));
861        assert!(detector.check(&state, None, t0));
862    }
863
864    // ── Additional: ConsecutiveFailureDetector reset clears counter ───────
865
866    #[test]
867    fn consecutive_failure_reset_clears_counter() {
868        let state = State::new();
869        let _ = state.set("bg:search_failed", true);
870
871        let detector = ConsecutiveFailureDetector::new("search", 3);
872        let t0 = Instant::now();
873
874        assert!(!detector.check(&state, None, t0));
875        assert!(!detector.check(&state, None, t0));
876
877        detector.reset();
878
879        assert!(!detector.check(&state, None, t0));
880        assert!(!detector.check(&state, None, t0));
881        assert!(detector.check(&state, None, t0));
882    }
883
884    // ── Additional: SustainedDetector needs_timer is true ─────────────────
885
886    #[test]
887    fn sustained_detector_needs_timer() {
888        let detector = SustainedDetector::new(Arc::new(|_: &State| true), Duration::from_secs(5));
889        assert!(detector.needs_timer());
890    }
891
892    // ── Additional: RateDetector needs_timer is false ─────────────────────
893
894    #[test]
895    fn rate_detector_does_not_need_timer() {
896        let detector = RateDetector::new(
897            Arc::new(|_: &SessionEvent| true),
898            5,
899            Duration::from_secs(10),
900        );
901        assert!(!detector.needs_timer());
902    }
903
904    // ── Additional: TurnCountDetector needs_timer is false ────────────────
905
906    #[test]
907    fn turn_count_detector_does_not_need_timer() {
908        let detector = TurnCountDetector::new(Arc::new(|_: &State| true), 3);
909        assert!(!detector.needs_timer());
910    }
911
912    // ── Additional: Pattern without cooldown fires every time ─────────────
913
914    #[tokio::test]
915    async fn pattern_without_cooldown_fires_every_time() {
916        let counter = Arc::new(AtomicU32::new(0));
917        let state = State::new();
918        let _ = state.set("active", true);
919        let writer = mock_writer();
920
921        let pattern = TemporalPattern::new(
922            "no-cooldown",
923            Box::new(TurnCountDetector::new(
924                Arc::new(|s: &State| s.get::<bool>("active").unwrap_or(false)),
925                1,
926            )),
927            counting_action(counter.clone()),
928            None, // no cooldown
929        );
930
931        let t0 = Instant::now();
932
933        for i in 0..5u32 {
934            let fut = pattern.try_fire(&state, None, &writer, t0 + Duration::from_millis(i as u64));
935            assert!(fut.is_some(), "should fire on iteration {i}");
936            fut.unwrap().await;
937        }
938
939        assert_eq!(counter.load(Ordering::SeqCst), 5);
940    }
941}
gemini_adk_rs/live/temporal.rs

gemini_adk_rs/live/
temporal.rs
