gemini_adk_rs/live/
events.rs

1//! Semantic events emitted by the L1 processor.
2//!
3//! Subscribe via `LiveHandle::events()` (broadcast receiver) or
4//! `LiveHandle::stream()` (a [`futures_util::Stream`]). Zero-cost when no
5//! subscribers.
6
7use std::pin::Pin;
8use std::task::{Context, Poll};
9use std::time::Duration;
10
11use bytes::Bytes;
12use futures_util::Stream;
13use tokio::sync::broadcast;
14
15/// Semantic events emitted by the Live session processor.
16///
17/// The L1 equivalent of L0's [`SessionEvent`](gemini_genai_rs::prelude::SessionEvent).
18/// L0 events are wire-level; LiveEvents are semantic (extractions completed,
19/// phases transitioned, tools executed).
20///
21/// Subscribe via [`LiveHandle::events()`](super::handle::LiveHandle::events).
22/// Multiple independent subscribers supported. Zero-cost when no subscribers
23/// exist (`broadcast::send` with 0 receivers is a no-op).
24#[derive(Debug, Clone)]
25#[non_exhaustive]
26pub enum LiveEvent {
27    // -- Fast-lane events (high frequency, sync emission) --
28    /// Raw PCM audio from model. Uses `Bytes` (refcounted) — clone is
29    /// a pointer increment (~2ns), not a deep copy.
30    Audio(Bytes),
31    /// Non-audio media from the model: Gemini 3.8 Live Avatar video
32    /// (`video/mp4`), synchronized with [`Audio`](Self::Audio).
33    Media(gemini_genai_rs::session::InlineMedia),
34    /// Incremental text token from model.
35    TextDelta(String),
36    /// Complete text response (all deltas concatenated).
37    TextComplete(String),
38    /// User speech transcription.
39    InputTranscript {
40        /// The transcribed text content.
41        text: String,
42        /// Whether this is the final transcription for the utterance.
43        is_final: bool,
44    },
45    /// Model speech transcription.
46    OutputTranscript {
47        /// The transcribed text content.
48        text: String,
49        /// Whether this is the final transcription for the utterance.
50        is_final: bool,
51    },
52    /// Model reasoning/thinking content.
53    Thought(String),
54    /// Voice activity detected — user started speaking.
55    VadStart,
56    /// Voice activity ended — user stopped speaking.
57    VadEnd,
58
59    // -- Control-lane events (lower frequency, async emission) --
60    /// Extraction completed. Emitted for both the top-level result
61    /// AND each flattened key (e.g., "order.items", "order.phase").
62    Extraction {
63        /// Extractor name, or `"extractor.field"` for flattened keys.
64        name: String,
65        /// The extracted JSON value.
66        value: serde_json::Value,
67    },
68    /// Extraction failed.
69    ExtractionError {
70        /// Name of the extractor that failed.
71        name: String,
72        /// Human-readable error description.
73        error: String,
74    },
75    /// A raw extraction field was considered for promotion into authoritative state.
76    StatePromotion {
77        /// Extractor name that produced the field.
78        extractor: String,
79        /// Field name inside the extractor result.
80        field: String,
81        /// State key targeted by the promotion rule.
82        state_key: String,
83        /// Whether the promotion was accepted and written.
84        accepted: bool,
85        /// Human-readable reason for the decision.
86        reason: String,
87        /// Extracted value that was considered.
88        value: serde_json::Value,
89    },
90    /// Phase machine transitioned.
91    PhaseTransition {
92        /// Phase the machine transitioned from.
93        from: String,
94        /// Phase the machine transitioned to.
95        to: String,
96        /// Human-readable reason for the transition.
97        reason: String,
98    },
99    /// Tool dispatched and result obtained.
100    ToolExecution {
101        /// Name of the tool that was called.
102        name: String,
103        /// Arguments passed to the tool.
104        args: serde_json::Value,
105        /// Result returned by the tool.
106        result: serde_json::Value,
107    },
108    /// Tool calls cancelled — either by the server (a `ToolCallCancelled`
109    /// wire event) or locally when a user barge-in interrupted an in-flight
110    /// inline tool. No response is sent for a cancelled call, and a cancelled
111    /// call never advances the governed flow.
112    ToolCancelled {
113        /// IDs of the cancelled tool calls.
114        ids: Vec<String>,
115    },
116    /// Model completed a conversational turn.
117    TurnComplete,
118    /// Model output interrupted by user speech.
119    Interrupted,
120    /// A tool call in a stage with a filler timing has run longer than the
121    /// stage allows: play an earcon or a holding line now.
122    FillerCue {
123        /// The tool still running.
124        tool: String,
125        /// How long it has run, in milliseconds.
126        elapsed_ms: u64,
127    },
128    /// A verbatim stage's required text was compared with what the model
129    /// said this turn.
130    VerbatimChecked {
131        /// The stage.
132        step: String,
133        /// Word-level similarity, 0–1.
134        similarity: f64,
135        /// Whether it was close enough to count as verbatim.
136        passed: bool,
137    },
138    /// The user stayed silent past the active stage's reprompt timing, and
139    /// the model was asked to repeat its question.
140    Reprompted {
141        /// How long the user had been silent, in milliseconds.
142        silence_ms: u64,
143    },
144    /// Session connected to Gemini.
145    Connected,
146    /// Session disconnected.
147    Disconnected {
148        /// Optional reason for disconnection (server-provided or error message).
149        reason: Option<String>,
150    },
151    /// Unrecoverable error.
152    Error(String),
153    /// Server requesting session wind-down.
154    GoAway {
155        /// Time remaining before the server closes the connection.
156        time_left: Duration,
157    },
158
159    // -- Periodic events --
160    /// Aggregated session telemetry snapshot.
161    Telemetry(serde_json::Value),
162    /// Per-turn latency and token metrics.
163    TurnMetrics {
164        /// Turn number (1-indexed).
165        turn: u32,
166        /// End-to-end latency for this turn in milliseconds.
167        latency_ms: u32,
168        /// Number of prompt tokens consumed.
169        prompt_tokens: u32,
170        /// Number of response tokens generated.
171        response_tokens: u32,
172    },
173}
174
175/// A [`futures_util::Stream`] of [`LiveEvent`]s from a Live session.
176///
177/// Created by [`LiveHandle::stream()`](super::handle::LiveHandle::stream).
178/// Wraps the underlying [`broadcast::Receiver`] with stream semantics:
179///
180/// - **Lagged**: if this subscriber falls behind the broadcast buffer, the
181///   missed events are skipped and the stream continues with the next
182///   available event (no error item is yielded).
183/// - **Closed**: when the session's event channel closes, the stream ends
184///   (`next()` returns `None`).
185///
186/// Composes with all `futures`/`tokio-stream` combinators:
187///
188/// ```rust,ignore
189/// use futures_util::StreamExt;
190///
191/// let mut stream = handle.stream();
192/// while let Some(ev) = stream.next().await {
193///     match ev {
194///         LiveEvent::TextDelta(t) => print!("{t}"),
195///         LiveEvent::TurnComplete => println!(),
196///         _ => {}
197///     }
198/// }
199/// ```
200pub struct LiveEventStream {
201    inner: Pin<Box<dyn Stream<Item = LiveEvent> + Send>>,
202}
203
204impl LiveEventStream {
205    /// Wrap a broadcast receiver of [`LiveEvent`]s as a stream.
206    pub(crate) fn new(rx: broadcast::Receiver<LiveEvent>) -> Self {
207        let inner = futures_util::stream::unfold(rx, |mut rx| async move {
208            loop {
209                match rx.recv().await {
210                    Ok(ev) => return Some((ev, rx)),
211                    // Skip lagged (missed) events and keep going.
212                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
213                    // Channel closed: end the stream.
214                    Err(broadcast::error::RecvError::Closed) => return None,
215                }
216            }
217        });
218        Self {
219            inner: Box::pin(inner),
220        }
221    }
222}
223
224impl Stream for LiveEventStream {
225    type Item = LiveEvent;
226
227    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
228        self.inner.as_mut().poll_next(cx)
229    }
230}
231
232impl std::fmt::Debug for LiveEventStream {
233    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
234        f.debug_struct("LiveEventStream").finish_non_exhaustive()
235    }
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241    use futures_util::StreamExt;
242
243    #[tokio::test]
244    async fn stream_yields_events_in_order_and_ends_on_close() {
245        let (tx, rx) = broadcast::channel::<LiveEvent>(16);
246        let mut stream = LiveEventStream::new(rx);
247
248        tx.send(LiveEvent::VadStart).unwrap();
249        tx.send(LiveEvent::TextDelta("hi".into())).unwrap();
250        tx.send(LiveEvent::TurnComplete).unwrap();
251
252        assert!(matches!(stream.next().await, Some(LiveEvent::VadStart)));
253        match stream.next().await {
254            Some(LiveEvent::TextDelta(t)) => assert_eq!(t, "hi"),
255            other => panic!("expected TextDelta, got {other:?}"),
256        }
257        assert!(matches!(stream.next().await, Some(LiveEvent::TurnComplete)));
258
259        // Closing the channel ends the stream.
260        drop(tx);
261        assert!(stream.next().await.is_none(), "stream ends on Closed");
262    }
263
264    #[tokio::test]
265    async fn stream_skips_lagged_events_and_continues() {
266        // Capacity-2 channel: sending 5 events before polling forces a lag.
267        let (tx, rx) = broadcast::channel::<LiveEvent>(2);
268        let mut stream = LiveEventStream::new(rx);
269
270        for i in 0..5u32 {
271            tx.send(LiveEvent::TextDelta(format!("e{i}"))).unwrap();
272        }
273
274        // The first poll observes the lag, skips it, and yields the oldest
275        // event still buffered (e3), then e4 — no error, no end-of-stream.
276        match stream.next().await {
277            Some(LiveEvent::TextDelta(t)) => assert_eq!(t, "e3"),
278            other => panic!("expected e3 after lag skip, got {other:?}"),
279        }
280        match stream.next().await {
281            Some(LiveEvent::TextDelta(t)) => assert_eq!(t, "e4"),
282            other => panic!("expected e4, got {other:?}"),
283        }
284
285        // The stream is still alive after the lag.
286        tx.send(LiveEvent::TurnComplete).unwrap();
287        assert!(matches!(stream.next().await, Some(LiveEvent::TurnComplete)));
288
289        drop(tx);
290        assert!(stream.next().await.is_none());
291    }
292}
gemini_adk_rs/live/events.rs

gemini_adk_rs/live/
events.rs
