gemini_adk_rs/live/
watcher.rs

1//! State change watchers with predicate-based triggering.
2//!
3//! A [`WatcherRegistry`] holds named [`Watcher`]s that observe specific state
4//! keys and fire async actions when a [`WatchPredicate`] matches a state diff.
5//!
6//! The registry is evaluated by the control-lane processor after each mutation
7//! cycle. It returns two sets of futures: blocking (awaited sequentially on the
8//! control lane) and concurrent (spawned via `tokio::spawn`).
9
10use std::collections::{HashMap, HashSet};
11use std::sync::Arc;
12
13use gemini_genai_rs::session::SessionWriter;
14use serde_json::Value;
15
16use super::BoxFuture;
17use crate::state::{State, StateMutation};
18
19use super::contract::WatcherContract;
20
21// ── Predicate ────────────────────────────────────────────────────────────────
22
23/// Custom predicate function type for state change watchers.
24pub type PredicateFn = Arc<dyn Fn(&Value, &Value) -> bool + Send + Sync>;
25
26/// Action fired by a watcher: `(old, new, state, writer)` → future.
27pub type WatcherActionFn =
28    Arc<dyn Fn(Value, Value, State, Arc<dyn SessionWriter>) -> BoxFuture<()> + Send + Sync>;
29
30/// Condition under which a watcher fires, evaluated against (old, new) values.
31pub enum WatchPredicate {
32    /// Fires whenever the watched key's value changed (any diff entry).
33    Changed,
34    /// Fires when the new value equals the given value.
35    ChangedTo(Value),
36    /// Fires when the old value equals the given value.
37    ChangedFrom(Value),
38    /// Fires when old < threshold AND new >= threshold (both must be numeric).
39    CrossedAbove(f64),
40    /// Fires when old >= threshold AND new < threshold (both must be numeric).
41    CrossedBelow(f64),
42    /// Fires when old != true AND new == true (JSON bool).
43    BecameTrue,
44    /// Fires when old == true AND new != true (JSON bool).
45    BecameFalse,
46    /// Fires when the custom function returns true for (old, new).
47    Custom(PredicateFn),
48}
49
50impl std::fmt::Debug for WatchPredicate {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        match self {
53            Self::Changed => write!(f, "Changed"),
54            Self::ChangedTo(v) => write!(f, "ChangedTo({v})"),
55            Self::ChangedFrom(v) => write!(f, "ChangedFrom({v})"),
56            Self::CrossedAbove(t) => write!(f, "CrossedAbove({t})"),
57            Self::CrossedBelow(t) => write!(f, "CrossedBelow({t})"),
58            Self::BecameTrue => write!(f, "BecameTrue"),
59            Self::BecameFalse => write!(f, "BecameFalse"),
60            Self::Custom(_) => write!(f, "Custom(<fn>)"),
61        }
62    }
63}
64
65impl WatchPredicate {
66    /// Evaluate whether this predicate matches the given old/new value pair.
67    fn matches(&self, old: &Value, new: &Value) -> bool {
68        match self {
69            WatchPredicate::Changed => true,
70            WatchPredicate::ChangedTo(val) => new == val,
71            WatchPredicate::ChangedFrom(val) => old == val,
72            WatchPredicate::CrossedAbove(threshold) => match (as_f64(old), as_f64(new)) {
73                (Some(o), Some(n)) => o < *threshold && n >= *threshold,
74                _ => false,
75            },
76            WatchPredicate::CrossedBelow(threshold) => match (as_f64(old), as_f64(new)) {
77                (Some(o), Some(n)) => o >= *threshold && n < *threshold,
78                _ => false,
79            },
80            WatchPredicate::BecameTrue => old != &Value::Bool(true) && new == &Value::Bool(true),
81            WatchPredicate::BecameFalse => old == &Value::Bool(true) && new != &Value::Bool(true),
82            WatchPredicate::Custom(f) => f(old, new),
83        }
84    }
85}
86
87// ── Watcher ──────────────────────────────────────────────────────────────────
88
89/// A single state watcher: observes one key, fires an async action when the
90/// predicate matches.
91pub struct Watcher {
92    /// The state key to observe.
93    pub key: String,
94    /// The condition under which this watcher fires.
95    pub predicate: WatchPredicate,
96    /// Async action receiving (old_value, new_value, state, writer). The
97    /// writer is the live session's — a watcher can steer or prompt the
98    /// model, not only mutate state.
99    pub action: WatcherActionFn,
100    /// If `true`, the processor awaits this action sequentially on the control
101    /// lane. If `false`, the processor spawns it concurrently.
102    pub blocking: bool,
103}
104
105impl std::fmt::Debug for Watcher {
106    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        f.debug_struct("Watcher")
108            .field("key", &self.key)
109            .field("predicate", &self.predicate)
110            .field("blocking", &self.blocking)
111            .finish_non_exhaustive()
112    }
113}
114
115// ── WatcherRegistry ──────────────────────────────────────────────────────────
116
117/// Registry of state watchers, evaluated after each mutation cycle.
118pub struct WatcherRegistry {
119    watchers: Vec<Watcher>,
120    /// Keys that any watcher observes -- used to scope snapshot/diff.
121    observed_keys: HashSet<String>,
122}
123
124impl Default for WatcherRegistry {
125    fn default() -> Self {
126        Self::new()
127    }
128}
129
130impl WatcherRegistry {
131    /// Create an empty registry.
132    pub fn new() -> Self {
133        Self {
134            watchers: Vec::new(),
135            observed_keys: HashSet::new(),
136        }
137    }
138
139    /// Register a watcher.
140    pub fn register(&mut self, watcher: Watcher) {
141        self.observed_keys.insert(watcher.key.clone());
142        self.watchers.push(watcher);
143    }
144
145    /// The set of state keys observed by at least one watcher.
146    ///
147    /// Used by the processor to scope `State::snapshot_values()` so only
148    /// relevant keys are captured before mutations.
149    pub fn observed_keys(&self) -> &HashSet<String> {
150        &self.observed_keys
151    }
152
153    /// Return serializable contract metadata for all watchers.
154    pub fn describe(&self) -> Vec<WatcherContract> {
155        self.watchers
156            .iter()
157            .map(|watcher| WatcherContract {
158                key: watcher.key.clone(),
159                predicate: format!("{:?}", watcher.predicate),
160                blocking: watcher.blocking,
161            })
162            .collect()
163    }
164
165    /// Evaluate all watchers against the given state diffs.
166    ///
167    /// `diffs` contains `(key, old_value, new_value)` tuples produced by
168    /// `State::diff_values()`. For each diff entry, every watcher whose key
169    /// matches and whose predicate fires will have its action invoked.
170    ///
171    /// Returns `(blocking_futures, concurrent_futures)`.
172    pub fn evaluate(
173        &self,
174        diffs: &[(String, Value, Value)],
175        state: &State,
176        writer: &Arc<dyn SessionWriter>,
177    ) -> (Vec<BoxFuture<()>>, Vec<BoxFuture<()>>) {
178        let mut blocking = Vec::new();
179        let mut concurrent = Vec::new();
180
181        for (key, old, new) in diffs {
182            for watcher in &self.watchers {
183                if watcher.key == *key && watcher.predicate.matches(old, new) {
184                    let fut =
185                        (watcher.action)(old.clone(), new.clone(), state.clone(), writer.clone());
186                    if watcher.blocking {
187                        blocking.push(fut);
188                    } else {
189                        concurrent.push(fut);
190                    }
191                }
192            }
193        }
194
195        (blocking, concurrent)
196    }
197
198    /// Evaluate watchers from a batch of recorded state mutations.
199    ///
200    /// Multiple mutations for the same key are collapsed into the same net
201    /// `(old, new)` diff shape used by [`Self::evaluate`], preserving watcher
202    /// behavior while avoiding a pre-turn state snapshot.
203    pub fn evaluate_mutations(
204        &self,
205        mutations: &[StateMutation],
206        state: &State,
207        writer: &Arc<dyn SessionWriter>,
208    ) -> (Vec<BoxFuture<()>>, Vec<BoxFuture<()>>) {
209        let mut net: HashMap<String, (Option<Value>, Option<Value>)> = HashMap::new();
210
211        for mutation in mutations {
212            if !self.observed_keys.contains(&mutation.key) {
213                continue;
214            }
215
216            net.entry(mutation.key.clone())
217                .and_modify(|(_, new)| {
218                    *new = mutation.new.clone();
219                })
220                .or_insert_with(|| (mutation.old.clone(), mutation.new.clone()));
221        }
222
223        let diffs: Vec<(String, Value, Value)> = net
224            .into_iter()
225            .filter_map(|(key, (old, new))| {
226                if old == new {
227                    None
228                } else {
229                    Some((key, old.unwrap_or(Value::Null), new.unwrap_or(Value::Null)))
230                }
231            })
232            .collect();
233
234        self.evaluate(&diffs, state, writer)
235    }
236}
237
238// ── Helpers ──────────────────────────────────────────────────────────────────
239
240/// Extract an `f64` from a JSON value (only works for `Value::Number`).
241fn as_f64(v: &Value) -> Option<f64> {
242    match v {
243        Value::Number(n) => n.as_f64(),
244        _ => None,
245    }
246}
247
248// ── Tests ────────────────────────────────────────────────────────────────────
249
250#[cfg(test)]
251mod tests {
252    use super::*;
253    use serde_json::json;
254    use std::sync::atomic::{AtomicU32, Ordering};
255
256    /// Helper: the writer handed to fired actions (accepts everything).
257    fn writer() -> Arc<dyn SessionWriter> {
258        Arc::new(crate::test_helpers::MockWriter)
259    }
260
261    /// Helper: create a watcher that increments a shared counter when fired.
262    fn counting_watcher(
263        key: &str,
264        predicate: WatchPredicate,
265        counter: Arc<AtomicU32>,
266        blocking: bool,
267    ) -> Watcher {
268        Watcher {
269            key: key.to_string(),
270            predicate,
271            action: Arc::new(move |_old, _new, _state, _writer| {
272                let c = counter.clone();
273                Box::pin(async move {
274                    c.fetch_add(1, Ordering::SeqCst);
275                })
276            }),
277            blocking,
278        }
279    }
280
281    /// Helper: create a watcher that stores old+new values into state.
282    fn recording_watcher(key: &str, predicate: WatchPredicate, blocking: bool) -> Watcher {
283        Watcher {
284            key: key.to_string(),
285            predicate,
286            action: Arc::new(|old, new, state, _writer| {
287                Box::pin(async move {
288                    let _ = state.set("recorded_old", old);
289                    let _ = state.set("recorded_new", new);
290                })
291            }),
292            blocking,
293        }
294    }
295
296    // ── 1. Changed predicate fires on any diff ──────────────────────────
297
298    #[tokio::test]
299    async fn changed_fires_on_any_diff() {
300        let counter = Arc::new(AtomicU32::new(0));
301        let mut registry = WatcherRegistry::new();
302        registry.register(counting_watcher(
303            "x",
304            WatchPredicate::Changed,
305            counter.clone(),
306            false,
307        ));
308
309        let state = State::new();
310        let diffs = vec![("x".to_string(), json!(1), json!(2))];
311
312        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
313        assert!(blocking.is_empty());
314        assert_eq!(concurrent.len(), 1);
315
316        for fut in concurrent {
317            fut.await;
318        }
319        assert_eq!(counter.load(Ordering::SeqCst), 1);
320    }
321
322    // ── 2. ChangedTo fires only when new value matches ──────────────────
323
324    #[tokio::test]
325    async fn changed_to_fires_when_new_value_matches() {
326        let counter = Arc::new(AtomicU32::new(0));
327        let mut registry = WatcherRegistry::new();
328        registry.register(counting_watcher(
329            "status",
330            WatchPredicate::ChangedTo(json!("active")),
331            counter.clone(),
332            false,
333        ));
334
335        let state = State::new();
336        let diffs = vec![("status".to_string(), json!("inactive"), json!("active"))];
337
338        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
339        assert_eq!(concurrent.len(), 1);
340
341        for fut in concurrent {
342            fut.await;
343        }
344        assert_eq!(counter.load(Ordering::SeqCst), 1);
345    }
346
347    // ── 3. ChangedTo does not fire when new value doesn't match ─────────
348
349    #[tokio::test]
350    async fn changed_to_does_not_fire_when_new_value_differs() {
351        let counter = Arc::new(AtomicU32::new(0));
352        let mut registry = WatcherRegistry::new();
353        registry.register(counting_watcher(
354            "status",
355            WatchPredicate::ChangedTo(json!("active")),
356            counter.clone(),
357            false,
358        ));
359
360        let state = State::new();
361        let diffs = vec![("status".to_string(), json!("inactive"), json!("pending"))];
362
363        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
364        assert!(blocking.is_empty());
365        assert!(concurrent.is_empty());
366        assert_eq!(counter.load(Ordering::SeqCst), 0);
367    }
368
369    // ── 4. ChangedFrom fires only when old value matches ────────────────
370
371    #[tokio::test]
372    async fn changed_from_fires_when_old_value_matches() {
373        let counter = Arc::new(AtomicU32::new(0));
374        let mut registry = WatcherRegistry::new();
375        registry.register(counting_watcher(
376            "mode",
377            WatchPredicate::ChangedFrom(json!("draft")),
378            counter.clone(),
379            false,
380        ));
381
382        let state = State::new();
383        // Old is "draft" — should fire.
384        let diffs = vec![("mode".to_string(), json!("draft"), json!("published"))];
385
386        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
387        assert_eq!(concurrent.len(), 1);
388
389        for fut in concurrent {
390            fut.await;
391        }
392        assert_eq!(counter.load(Ordering::SeqCst), 1);
393
394        // Old is NOT "draft" — should not fire.
395        let diffs2 = vec![("mode".to_string(), json!("published"), json!("archived"))];
396        let (b, c) = registry.evaluate(&diffs2, &state, &writer());
397        assert!(b.is_empty());
398        assert!(c.is_empty());
399        assert_eq!(counter.load(Ordering::SeqCst), 1);
400    }
401
402    // ── 5. CrossedAbove fires when crossing threshold upward ────────────
403
404    #[tokio::test]
405    async fn crossed_above_fires_on_upward_crossing() {
406        let counter = Arc::new(AtomicU32::new(0));
407        let mut registry = WatcherRegistry::new();
408        registry.register(counting_watcher(
409            "temp",
410            WatchPredicate::CrossedAbove(100.0),
411            counter.clone(),
412            false,
413        ));
414
415        let state = State::new();
416        // 95 -> 105: crosses above 100
417        let diffs = vec![("temp".to_string(), json!(95.0), json!(105.0))];
418
419        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
420        assert_eq!(concurrent.len(), 1);
421
422        for fut in concurrent {
423            fut.await;
424        }
425        assert_eq!(counter.load(Ordering::SeqCst), 1);
426    }
427
428    // ── 6. CrossedAbove does not fire when both above threshold ─────────
429
430    #[tokio::test]
431    async fn crossed_above_does_not_fire_when_both_above() {
432        let counter = Arc::new(AtomicU32::new(0));
433        let mut registry = WatcherRegistry::new();
434        registry.register(counting_watcher(
435            "temp",
436            WatchPredicate::CrossedAbove(100.0),
437            counter.clone(),
438            false,
439        ));
440
441        let state = State::new();
442        // 110 -> 120: both above 100, no crossing
443        let diffs = vec![("temp".to_string(), json!(110.0), json!(120.0))];
444
445        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
446        assert!(blocking.is_empty());
447        assert!(concurrent.is_empty());
448        assert_eq!(counter.load(Ordering::SeqCst), 0);
449    }
450
451    // ── 7. CrossedBelow fires when crossing threshold downward ──────────
452
453    #[tokio::test]
454    async fn crossed_below_fires_on_downward_crossing() {
455        let counter = Arc::new(AtomicU32::new(0));
456        let mut registry = WatcherRegistry::new();
457        registry.register(counting_watcher(
458            "battery",
459            WatchPredicate::CrossedBelow(20.0),
460            counter.clone(),
461            false,
462        ));
463
464        let state = State::new();
465        // 25 -> 15: crosses below 20
466        let diffs = vec![("battery".to_string(), json!(25.0), json!(15.0))];
467
468        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
469        assert_eq!(concurrent.len(), 1);
470
471        for fut in concurrent {
472            fut.await;
473        }
474        assert_eq!(counter.load(Ordering::SeqCst), 1);
475    }
476
477    // ── 8. BecameTrue fires when value changes to true ──────────────────
478
479    #[tokio::test]
480    async fn became_true_fires_on_false_to_true() {
481        let counter = Arc::new(AtomicU32::new(0));
482        let mut registry = WatcherRegistry::new();
483        registry.register(counting_watcher(
484            "flag",
485            WatchPredicate::BecameTrue,
486            counter.clone(),
487            false,
488        ));
489
490        let state = State::new();
491        let diffs = vec![("flag".to_string(), json!(false), json!(true))];
492
493        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
494        assert_eq!(concurrent.len(), 1);
495
496        for fut in concurrent {
497            fut.await;
498        }
499        assert_eq!(counter.load(Ordering::SeqCst), 1);
500    }
501
502    // ── 9. BecameFalse fires when value changes from true to false ──────
503
504    #[tokio::test]
505    async fn became_false_fires_on_true_to_false() {
506        let counter = Arc::new(AtomicU32::new(0));
507        let mut registry = WatcherRegistry::new();
508        registry.register(counting_watcher(
509            "flag",
510            WatchPredicate::BecameFalse,
511            counter.clone(),
512            false,
513        ));
514
515        let state = State::new();
516        let diffs = vec![("flag".to_string(), json!(true), json!(false))];
517
518        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
519        assert_eq!(concurrent.len(), 1);
520
521        for fut in concurrent {
522            fut.await;
523        }
524        assert_eq!(counter.load(Ordering::SeqCst), 1);
525    }
526
527    // ── 10. Custom predicate ────────────────────────────────────────────
528
529    #[tokio::test]
530    async fn custom_predicate_fires_when_fn_returns_true() {
531        let counter = Arc::new(AtomicU32::new(0));
532        let mut registry = WatcherRegistry::new();
533        registry.register(counting_watcher(
534            "score",
535            WatchPredicate::Custom(Arc::new(|old, new| {
536                // Fire only when value doubled
537                match (as_f64(old), as_f64(new)) {
538                    (Some(o), Some(n)) => (n - o * 2.0).abs() < f64::EPSILON,
539                    _ => false,
540                }
541            })),
542            counter.clone(),
543            false,
544        ));
545
546        let state = State::new();
547        // 5 -> 10: exactly doubled
548        let diffs = vec![("score".to_string(), json!(5.0), json!(10.0))];
549
550        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
551        assert_eq!(concurrent.len(), 1);
552
553        for fut in concurrent {
554            fut.await;
555        }
556        assert_eq!(counter.load(Ordering::SeqCst), 1);
557
558        // 5 -> 11: not doubled
559        let diffs2 = vec![("score".to_string(), json!(5.0), json!(11.0))];
560        let (b, c) = registry.evaluate(&diffs2, &state, &writer());
561        assert!(b.is_empty());
562        assert!(c.is_empty());
563    }
564
565    // ── 11. evaluate separates blocking vs concurrent futures ───────────
566
567    #[tokio::test]
568    async fn evaluate_separates_blocking_and_concurrent() {
569        let blocking_counter = Arc::new(AtomicU32::new(0));
570        let concurrent_counter = Arc::new(AtomicU32::new(0));
571        let mut registry = WatcherRegistry::new();
572
573        // Blocking watcher
574        registry.register(counting_watcher(
575            "x",
576            WatchPredicate::Changed,
577            blocking_counter.clone(),
578            true,
579        ));
580
581        // Concurrent watcher
582        registry.register(counting_watcher(
583            "x",
584            WatchPredicate::Changed,
585            concurrent_counter.clone(),
586            false,
587        ));
588
589        let state = State::new();
590        let diffs = vec![("x".to_string(), json!(1), json!(2))];
591
592        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
593        assert_eq!(blocking.len(), 1);
594        assert_eq!(concurrent.len(), 1);
595
596        // Execute both sets
597        for fut in blocking {
598            fut.await;
599        }
600        for fut in concurrent {
601            fut.await;
602        }
603
604        assert_eq!(blocking_counter.load(Ordering::SeqCst), 1);
605        assert_eq!(concurrent_counter.load(Ordering::SeqCst), 1);
606    }
607
608    // ── 12. evaluate with no matching diffs returns empty vecs ──────────
609
610    #[test]
611    fn evaluate_with_no_matching_diffs_returns_empty() {
612        let counter = Arc::new(AtomicU32::new(0));
613        let mut registry = WatcherRegistry::new();
614        registry.register(counting_watcher(
615            "x",
616            WatchPredicate::Changed,
617            counter.clone(),
618            false,
619        ));
620
621        let state = State::new();
622        // Diff is for key "y", but watcher observes "x"
623        let diffs = vec![("y".to_string(), json!(1), json!(2))];
624
625        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
626        assert!(blocking.is_empty());
627        assert!(concurrent.is_empty());
628    }
629
630    #[tokio::test]
631    async fn evaluate_mutations_collapses_to_net_diff() {
632        let counter = Arc::new(AtomicU32::new(0));
633        let mut registry = WatcherRegistry::new();
634        registry.register(counting_watcher(
635            "x",
636            WatchPredicate::ChangedTo(json!(3)),
637            counter.clone(),
638            false,
639        ));
640
641        let state = State::new();
642        let cursor = state.mutation_cursor();
643        let _ = state.set("x", 1);
644        let _ = state.set("x", 2);
645        let _ = state.set("x", 3);
646        let _ = state.set("ignored", 10);
647
648        let (_, concurrent) =
649            registry.evaluate_mutations(&state.mutations_since(cursor), &state, &writer());
650        assert_eq!(concurrent.len(), 1);
651        for fut in concurrent {
652            fut.await;
653        }
654        assert_eq!(counter.load(Ordering::SeqCst), 1);
655    }
656
657    #[test]
658    fn evaluate_mutations_ignores_net_noop() {
659        let counter = Arc::new(AtomicU32::new(0));
660        let mut registry = WatcherRegistry::new();
661        registry.register(counting_watcher(
662            "x",
663            WatchPredicate::Changed,
664            counter,
665            false,
666        ));
667
668        let state = State::new();
669        let _ = state.set("x", 1);
670        let cursor = state.mutation_cursor();
671        let _ = state.set("x", 2);
672        let _ = state.set("x", 1);
673
674        let (blocking, concurrent) =
675            registry.evaluate_mutations(&state.mutations_since(cursor), &state, &writer());
676        assert!(blocking.is_empty());
677        assert!(concurrent.is_empty());
678    }
679
680    // ── 13. observed_keys tracks added watcher keys ─────────────────────
681
682    #[test]
683    fn observed_keys_tracks_added_watcher_keys() {
684        let counter = Arc::new(AtomicU32::new(0));
685        let mut registry = WatcherRegistry::new();
686
687        assert!(registry.observed_keys().is_empty());
688
689        registry.register(counting_watcher(
690            "alpha",
691            WatchPredicate::Changed,
692            counter.clone(),
693            false,
694        ));
695        registry.register(counting_watcher(
696            "beta",
697            WatchPredicate::Changed,
698            counter.clone(),
699            false,
700        ));
701        registry.register(counting_watcher(
702            "alpha",
703            WatchPredicate::BecameTrue,
704            counter.clone(),
705            true,
706        ));
707
708        let keys = registry.observed_keys();
709        assert_eq!(keys.len(), 2);
710        assert!(keys.contains("alpha"));
711        assert!(keys.contains("beta"));
712    }
713
714    // ── 14. multiple watchers on same key ───────────────────────────────
715
716    #[tokio::test]
717    async fn multiple_watchers_on_same_key() {
718        let counter_a = Arc::new(AtomicU32::new(0));
719        let counter_b = Arc::new(AtomicU32::new(0));
720        let mut registry = WatcherRegistry::new();
721
722        // Watcher A: fires on any change
723        registry.register(counting_watcher(
724            "x",
725            WatchPredicate::Changed,
726            counter_a.clone(),
727            false,
728        ));
729
730        // Watcher B: fires only when new == 42
731        registry.register(counting_watcher(
732            "x",
733            WatchPredicate::ChangedTo(json!(42)),
734            counter_b.clone(),
735            false,
736        ));
737
738        let state = State::new();
739        let diffs = vec![("x".to_string(), json!(1), json!(42))];
740
741        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
742        // Both watchers should fire
743        assert_eq!(concurrent.len(), 2);
744
745        for fut in concurrent {
746            fut.await;
747        }
748        assert_eq!(counter_a.load(Ordering::SeqCst), 1);
749        assert_eq!(counter_b.load(Ordering::SeqCst), 1);
750
751        // Now only watcher A should fire (new != 42)
752        let diffs2 = vec![("x".to_string(), json!(42), json!(99))];
753        let (_, concurrent2) = registry.evaluate(&diffs2, &state, &writer());
754        assert_eq!(concurrent2.len(), 1);
755
756        for fut in concurrent2 {
757            fut.await;
758        }
759        assert_eq!(counter_a.load(Ordering::SeqCst), 2);
760        assert_eq!(counter_b.load(Ordering::SeqCst), 1); // unchanged
761    }
762
763    // ── Additional edge-case tests ──────────────────────────────────────
764
765    #[tokio::test]
766    async fn action_receives_old_new_and_state() {
767        let mut registry = WatcherRegistry::new();
768        registry.register(recording_watcher("val", WatchPredicate::Changed, false));
769
770        let state = State::new();
771        let diffs = vec![("val".to_string(), json!("before"), json!("after"))];
772
773        let (_, concurrent) = registry.evaluate(&diffs, &state, &writer());
774        assert_eq!(concurrent.len(), 1);
775
776        for fut in concurrent {
777            fut.await;
778        }
779
780        assert_eq!(state.get_raw("recorded_old"), Some(json!("before")));
781        assert_eq!(state.get_raw("recorded_new"), Some(json!("after")));
782    }
783
784    #[test]
785    fn crossed_above_with_non_numeric_values_does_not_fire() {
786        let counter = Arc::new(AtomicU32::new(0));
787        let mut registry = WatcherRegistry::new();
788        registry.register(counting_watcher(
789            "x",
790            WatchPredicate::CrossedAbove(10.0),
791            counter.clone(),
792            false,
793        ));
794
795        let state = State::new();
796        let diffs = vec![("x".to_string(), json!("low"), json!("high"))];
797
798        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
799        assert!(blocking.is_empty());
800        assert!(concurrent.is_empty());
801    }
802
803    #[test]
804    fn became_true_does_not_fire_on_non_bool() {
805        let counter = Arc::new(AtomicU32::new(0));
806        let mut registry = WatcherRegistry::new();
807        registry.register(counting_watcher(
808            "x",
809            WatchPredicate::BecameTrue,
810            counter.clone(),
811            false,
812        ));
813
814        let state = State::new();
815        // "truthy" string is not json bool true
816        let diffs = vec![("x".to_string(), json!(0), json!("true"))];
817
818        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
819        assert!(blocking.is_empty());
820        assert!(concurrent.is_empty());
821    }
822
823    #[test]
824    fn empty_diffs_produce_no_futures() {
825        let counter = Arc::new(AtomicU32::new(0));
826        let mut registry = WatcherRegistry::new();
827        registry.register(counting_watcher(
828            "x",
829            WatchPredicate::Changed,
830            counter.clone(),
831            false,
832        ));
833
834        let state = State::new();
835        let diffs: Vec<(String, Value, Value)> = vec![];
836
837        let (blocking, concurrent) = registry.evaluate(&diffs, &state, &writer());
838        assert!(blocking.is_empty());
839        assert!(concurrent.is_empty());
840    }
841
842    #[test]
843    fn default_creates_empty_registry() {
844        let registry = WatcherRegistry::default();
845        assert!(registry.observed_keys().is_empty());
846    }
847}
gemini_adk_rs/live/watcher.rs

gemini_adk_rs/live/
watcher.rs
