gemini_adk_rs/live/
reactor.rs

1//! First-class reactor vocabulary for Live sessions.
2//!
3//! This module is intentionally small: it defines the normalized events,
4//! reactions, and typed effects that existing mechanisms can migrate onto
5//! incrementally without rewriting the current control plane in one step.
6
7use std::sync::Mutex;
8use std::time::{Duration, Instant};
9
10use gemini_genai_rs::prelude::Content;
11
12use crate::state::StateMutation;
13
14use super::ExecutionMode;
15use super::events::LiveEvent;
16
17/// A normalized event that can drive ADK-level reactions.
18#[derive(Debug, Clone)]
19pub enum ReactorEvent {
20    /// Existing semantic Live event.
21    Live(LiveEvent),
22    /// State changed since the last reactor cursor.
23    StateChanged(Vec<StateMutation>),
24    /// Periodic tick for timers and sustained conditions.
25    TimerTick {
26        /// Time observed by the control lane for this tick.
27        now: Instant,
28    },
29    /// Client-side playback drained after model audio was generated.
30    PlaybackDrained {
31        /// Whether the control plane has armed a deferred model prompt.
32        prompt_pending: bool,
33    },
34    /// Client detected that the user started speaking.
35    UserSpeechStarted,
36    /// Client detected that the user stopped speaking.
37    UserSpeechEnded {
38        /// Whether the control plane has armed a deferred model prompt.
39        prompt_pending: bool,
40    },
41    /// User speech ended, but the model may or may not produce a turn.
42    SoftTurnComplete,
43}
44
45/// Voice-flow state owned by the reactor.
46#[derive(Debug, Clone, Default)]
47pub struct VoiceRuntimeState {
48    /// Whether the client believes the user is currently speaking.
49    pub user_speaking: bool,
50    /// Whether browser playback is believed to be active.
51    pub playback_active: bool,
52    /// Whether a deferred model prompt is armed.
53    pub prompt_pending: bool,
54    /// Monotonic epoch bumped whenever a prompt is cancelled or armed state changes.
55    pub prompt_epoch: u64,
56    /// Last time the client reported barge-in/user speech start.
57    pub last_barge_in_at: Option<Instant>,
58    /// Last time browser playback reported drained.
59    pub last_playback_drained_at: Option<Instant>,
60}
61
62impl VoiceRuntimeState {
63    /// Apply an incoming event to the voice runtime state before rules run.
64    pub fn apply_event(&mut self, event: &ReactorEvent) {
65        match event {
66            ReactorEvent::PlaybackDrained { prompt_pending } => {
67                self.playback_active = false;
68                self.prompt_pending = *prompt_pending;
69                self.last_playback_drained_at = Some(Instant::now());
70            }
71            ReactorEvent::UserSpeechStarted => {
72                self.user_speaking = true;
73                self.playback_active = false;
74                if self.prompt_pending {
75                    self.prompt_epoch = self.prompt_epoch.saturating_add(1);
76                }
77                self.prompt_pending = false;
78                self.last_barge_in_at = Some(Instant::now());
79            }
80            ReactorEvent::UserSpeechEnded { prompt_pending } => {
81                self.user_speaking = false;
82                self.prompt_pending = *prompt_pending && !self.playback_active;
83            }
84            _ => {}
85        }
86    }
87}
88
89/// Execution policy for an effect.
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct EffectPolicy {
92    /// Whether this effect blocks later effects from running.
93    pub mode: ExecutionMode,
94    /// Optional maximum time budget for the effect.
95    pub timeout: Option<Duration>,
96}
97
98impl Default for EffectPolicy {
99    fn default() -> Self {
100        Self {
101            mode: ExecutionMode::Blocking,
102            timeout: None,
103        }
104    }
105}
106
107/// A typed runtime effect emitted by a reaction.
108#[derive(Debug, Clone)]
109pub enum LiveEffect {
110    /// No operation; useful for conditional reaction builders.
111    Noop,
112    /// Add state/context turns to the session.
113    SendContext(Vec<Content>),
114    /// Ask the model to generate from accumulated context.
115    PromptModel,
116    /// Cancel a deferred model prompt while leaving queued context intact.
117    CancelDeferredPrompt,
118    /// Tell the Live API that user speech activity started.
119    SignalUserActivityStart,
120    /// Tell the Live API that user speech activity ended.
121    SignalUserActivityEnd,
122    /// Replace or amend the active instruction.
123    UpdateInstruction(String),
124    /// Emit a semantic event for observers.
125    Emit(LiveEvent),
126}
127
128/// A policy-wrapped effect.
129#[derive(Debug, Clone)]
130pub struct Reaction {
131    /// Rule or subsystem that produced the reaction.
132    pub source: &'static str,
133    /// Runtime effect requested by the rule.
134    pub effect: LiveEffect,
135    /// Execution policy for the effect.
136    pub policy: EffectPolicy,
137}
138
139impl Reaction {
140    /// Create a blocking reaction.
141    pub fn blocking(source: &'static str, effect: LiveEffect) -> Self {
142        Self {
143            source,
144            effect,
145            policy: EffectPolicy::default(),
146        }
147    }
148
149    /// Create a concurrent reaction.
150    pub fn concurrent(source: &'static str, effect: LiveEffect) -> Self {
151        Self {
152            source,
153            effect,
154            policy: EffectPolicy {
155                mode: ExecutionMode::Concurrent,
156                ..EffectPolicy::default()
157            },
158        }
159    }
160}
161
162/// A rule that reacts to normalized events and emits typed effects.
163pub trait ReactorRule: Send + Sync {
164    /// Stable rule name for diagnostics and reaction provenance.
165    fn name(&self) -> &str;
166    /// Produce reactions for a normalized event.
167    fn react(&self, event: &ReactorEvent, voice: &VoiceRuntimeState) -> Vec<Reaction>;
168}
169
170/// Ordered collection of reactor rules.
171pub struct LiveReactor {
172    rules: Vec<Box<dyn ReactorRule>>,
173    voice: Mutex<VoiceRuntimeState>,
174}
175
176impl Default for LiveReactor {
177    fn default() -> Self {
178        Self {
179            rules: Vec::new(),
180            voice: Mutex::new(VoiceRuntimeState::default()),
181        }
182    }
183}
184
185impl LiveReactor {
186    /// Create an empty reactor.
187    pub fn new() -> Self {
188        Self::default()
189    }
190
191    /// Create a reactor with the default voice-flow rules.
192    pub fn voice_defaults() -> Self {
193        let mut reactor = Self::new();
194        reactor.add_rule(PromptOnPlaybackDrained);
195        reactor.add_rule(UserSpeechActivityRule);
196        reactor
197    }
198
199    /// Add a rule to the end of the ordered rule list.
200    pub fn add_rule(&mut self, rule: impl ReactorRule + 'static) {
201        self.rules.push(Box::new(rule));
202    }
203
204    /// Run all rules against an event and collect reactions in rule order.
205    pub fn react(&self, event: &ReactorEvent) -> Vec<Reaction> {
206        let voice = {
207            let mut voice = self.voice.lock().expect("voice reactor state poisoned");
208            voice.apply_event(event);
209            voice.clone()
210        };
211
212        self.rules
213            .iter()
214            .flat_map(|rule| rule.react(event, &voice))
215            .collect()
216    }
217
218    /// Return a snapshot of the current voice runtime state.
219    pub fn voice_state(&self) -> VoiceRuntimeState {
220        self.voice
221            .lock()
222            .expect("voice reactor state poisoned")
223            .clone()
224    }
225}
226
227/// Prompt the model when browser playback is fully drained and a prompt is armed.
228pub struct PromptOnPlaybackDrained;
229
230impl ReactorRule for PromptOnPlaybackDrained {
231    fn name(&self) -> &str {
232        "prompt_on_playback_drained"
233    }
234
235    fn react(&self, event: &ReactorEvent, voice: &VoiceRuntimeState) -> Vec<Reaction> {
236        if matches!(event, ReactorEvent::PlaybackDrained { .. })
237            && voice.prompt_pending
238            && !voice.user_speaking
239            && !voice.playback_active
240        {
241            vec![Reaction::blocking(
242                "prompt_on_playback_drained",
243                LiveEffect::PromptModel,
244            )]
245        } else {
246            Vec::new()
247        }
248    }
249}
250
251/// Cancel pending model prompts and signal activity around user speech.
252pub struct UserSpeechActivityRule;
253
254impl ReactorRule for UserSpeechActivityRule {
255    fn name(&self) -> &str {
256        "user_speech_activity"
257    }
258
259    fn react(&self, event: &ReactorEvent, voice: &VoiceRuntimeState) -> Vec<Reaction> {
260        match event {
261            ReactorEvent::UserSpeechStarted => vec![
262                Reaction::blocking("user_speech_activity", LiveEffect::CancelDeferredPrompt),
263                Reaction::blocking("user_speech_activity", LiveEffect::SignalUserActivityStart),
264            ],
265            ReactorEvent::UserSpeechEnded { .. } => {
266                let mut reactions = vec![Reaction::blocking(
267                    "user_speech_activity",
268                    LiveEffect::SignalUserActivityEnd,
269                )];
270                if voice.prompt_pending && !voice.playback_active {
271                    reactions.push(Reaction::blocking(
272                        "user_speech_activity",
273                        LiveEffect::PromptModel,
274                    ));
275                }
276                reactions
277            }
278            _ => Vec::new(),
279        }
280    }
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286
287    #[test]
288    fn reactor_collects_reactions_in_rule_order() {
289        let mut reactor = LiveReactor::new();
290        reactor.add_rule(PromptOnPlaybackDrained);
291
292        let reactions = reactor.react(&ReactorEvent::PlaybackDrained {
293            prompt_pending: true,
294        });
295        assert_eq!(reactions.len(), 1);
296        assert_eq!(reactions[0].source, "prompt_on_playback_drained");
297        assert_eq!(reactions[0].policy.mode, ExecutionMode::Blocking);
298        assert!(matches!(reactions[0].effect, LiveEffect::PromptModel));
299    }
300
301    #[test]
302    fn playback_drained_without_pending_prompt_is_noop() {
303        let reactor = LiveReactor::voice_defaults();
304
305        let reactions = reactor.react(&ReactorEvent::PlaybackDrained {
306            prompt_pending: false,
307        });
308
309        assert!(reactions.is_empty());
310    }
311
312    #[test]
313    fn user_speech_started_cancels_prompt_and_signals_activity() {
314        let reactor = LiveReactor::voice_defaults();
315
316        let prompt_reactions = reactor.react(&ReactorEvent::PlaybackDrained {
317            prompt_pending: true,
318        });
319        assert_eq!(prompt_reactions.len(), 1);
320
321        let reactions = reactor.react(&ReactorEvent::UserSpeechStarted);
322
323        assert_eq!(reactions.len(), 2);
324        assert!(matches!(
325            reactions[0].effect,
326            LiveEffect::CancelDeferredPrompt
327        ));
328        assert!(matches!(
329            reactions[1].effect,
330            LiveEffect::SignalUserActivityStart
331        ));
332        let voice = reactor.voice_state();
333        assert!(voice.user_speaking);
334        assert!(!voice.prompt_pending);
335        assert_eq!(voice.prompt_epoch, 1);
336    }
337
338    #[test]
339    fn user_speech_ended_signals_activity_end() {
340        let reactor = LiveReactor::voice_defaults();
341
342        let reactions = reactor.react(&ReactorEvent::UserSpeechEnded {
343            prompt_pending: false,
344        });
345
346        assert_eq!(reactions.len(), 1);
347        assert!(matches!(
348            reactions[0].effect,
349            LiveEffect::SignalUserActivityEnd
350        ));
351        assert!(!reactor.voice_state().user_speaking);
352    }
353
354    #[test]
355    fn speech_end_prompts_when_playback_already_drained_and_prompt_pending() {
356        let reactor = LiveReactor::voice_defaults();
357
358        reactor.react(&ReactorEvent::UserSpeechStarted);
359        let drain_reactions = reactor.react(&ReactorEvent::PlaybackDrained {
360            prompt_pending: true,
361        });
362        assert!(drain_reactions.is_empty());
363
364        let reactions = reactor.react(&ReactorEvent::UserSpeechEnded {
365            prompt_pending: true,
366        });
367
368        assert_eq!(reactions.len(), 2);
369        assert!(matches!(
370            reactions[0].effect,
371            LiveEffect::SignalUserActivityEnd
372        ));
373        assert!(matches!(reactions[1].effect, LiveEffect::PromptModel));
374    }
375
376    #[test]
377    fn playback_drained_does_not_prompt_while_user_is_speaking() {
378        let reactor = LiveReactor::voice_defaults();
379
380        reactor.react(&ReactorEvent::UserSpeechStarted);
381        let reactions = reactor.react(&ReactorEvent::PlaybackDrained {
382            prompt_pending: true,
383        });
384
385        assert!(reactions.is_empty());
386        let voice = reactor.voice_state();
387        assert!(voice.user_speaking);
388        assert!(voice.prompt_pending);
389    }
390
391    #[test]
392    fn speech_start_wins_after_pending_prompt_snapshot() {
393        let reactor = LiveReactor::voice_defaults();
394
395        reactor.react(&ReactorEvent::PlaybackDrained {
396            prompt_pending: true,
397        });
398        reactor.react(&ReactorEvent::UserSpeechStarted);
399        let reactions = reactor.react(&ReactorEvent::PlaybackDrained {
400            prompt_pending: false,
401        });
402
403        assert!(reactions.is_empty());
404        let voice = reactor.voice_state();
405        assert!(voice.user_speaking);
406        assert!(!voice.prompt_pending);
407        assert_eq!(voice.prompt_epoch, 1);
408    }
409}
gemini_adk_rs/live/reactor.rs

gemini_adk_rs/live/
reactor.rs
