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

gemini_adk_rs/live/
events.rs
