gemini_genai_rs/session/
events.rs

1//! Session events, commands, and turn tracking.
2//!
3//! [`SessionEvent`] — events emitted by the server for application consumption.
4//! [`SessionCommand`] — commands sent from the application to the transport.
5//! [`Turn`] — tracking for a single model response turn.
6//! [`recv_event`] — broadcast lag-tolerant event receiver.
7
8use super::errors::SessionError;
9use super::state::SessionPhase;
10use crate::protocol::{Content, FunctionCall, FunctionResponse, UsageMetadata};
11use bytes::Bytes;
12use std::time::{Duration, Instant};
13use tokio::sync::broadcast;
14
15// ---------------------------------------------------------------------------
16// Events (server -> application)
17// ---------------------------------------------------------------------------
18
19/// Events emitted by the session, consumed by application code.
20#[derive(Debug, Clone)]
21#[non_exhaustive]
22pub enum SessionEvent {
23    /// Session connected and setup complete.
24    Connected,
25    /// Incremental text from model response.
26    TextDelta(String),
27    /// Complete text of a finished model turn.
28    TextComplete(String),
29    /// Audio data from model response (PCM16 samples, base64-decoded).
30    ///
31    /// Uses [`bytes::Bytes`] for zero-copy fan-out: cloning a `Bytes` handle
32    /// bumps an `Arc` refcount instead of copying the underlying data.
33    AudioData(bytes::Bytes),
34    /// Non-audio media from the model, such as Gemini 3.8 Live Avatar video
35    /// (`video/mp4`, 24 FPS, synchronized with the speech in
36    /// [`AudioData`](Self::AudioData)).
37    Media(InlineMedia),
38    /// Input transcription from server.
39    InputTranscription(String),
40    /// Output transcription from server.
41    OutputTranscription(String),
42    /// Thought/reasoning summary from the model (when includeThoughts is enabled).
43    Thought(String),
44    /// Model requested tool calls.
45    ToolCall(Vec<FunctionCall>),
46    /// Server cancelled pending tool calls.
47    ToolCallCancelled(Vec<String>),
48    /// Progress of a longer interaction (`serverContent.interactionStatus`),
49    /// e.g. `"IN_PROGRESS"` when Gemini 3.8 Live Extended Thinking has said a
50    /// holding line and will answer in a later turn.
51    InteractionStatus(String),
52    /// Model turn is complete (it's the user's turn now).
53    TurnComplete,
54    /// Model finished generating its full response.
55    ///
56    /// Fires even if the generation was interrupted — tells you the model's
57    /// internal generation pipeline has stopped. Distinct from `TurnComplete`
58    /// which is the turn-taking signal.
59    GenerationComplete,
60    /// Model was interrupted by barge-in.
61    Interrupted,
62    /// Session phase changed.
63    PhaseChanged(SessionPhase),
64    /// Server sent GoAway: it will close the connection after `time_left`
65    /// (when it said how long). The transport reconnects on its own; this
66    /// is the moment to drain buffered audio and expect a resumption handle.
67    GoAway(Option<Duration>),
68    /// Session disconnected (with optional reason).
69    Disconnected(Option<String>),
70    /// Non-fatal error. The session loop keeps running (reconnecting where
71    /// the transport allows); match on the [`SessionError`] variants to tell
72    /// a refused setup from a dropped socket.
73    Error(SessionError),
74    /// Session resumption update with handle, resumability, and consumed index.
75    SessionResumeUpdate(ResumeInfo),
76    /// Server-side voice activity detected (user started speaking).
77    VoiceActivityStart,
78    /// Server-side voice activity ended (user stopped speaking).
79    VoiceActivityEnd,
80    /// Token usage metadata from server (for context window tracking).
81    ///
82    /// Contains full token breakdown: prompt, response, cached, tool-use,
83    /// thinking tokens, plus per-modality details.
84    Usage(UsageMetadata),
85}
86
87/// A chunk of non-audio media the model produced inline.
88#[derive(Debug, Clone, PartialEq, Eq)]
89pub struct InlineMedia {
90    /// MIME type as the server gave it, e.g. `"video/mp4"`.
91    pub mime_type: String,
92    /// The decoded bytes. Cloning shares them (see [`bytes::Bytes`]).
93    pub data: bytes::Bytes,
94}
95
96impl InlineMedia {
97    /// Whether this is a video chunk (`video/*`).
98    pub fn is_video(&self) -> bool {
99        self.mime_type.starts_with("video/")
100    }
101}
102
103/// Session resumption information from the server.
104#[derive(Debug, Clone)]
105pub struct ResumeInfo {
106    /// Opaque handle for session resumption.
107    pub handle: String,
108    /// Whether the session is currently resumable.
109    pub resumable: bool,
110    /// Index of the last client message consumed by the server.
111    pub last_consumed_index: Option<String>,
112}
113
114// ---------------------------------------------------------------------------
115// Commands (application -> server)
116// ---------------------------------------------------------------------------
117
118/// Commands sent from application code to the session transport.
119#[derive(Debug, Clone)]
120pub enum SessionCommand {
121    /// Send audio data (raw PCM16 bytes, will be base64-encoded).
122    ///
123    /// [`Bytes`] both ways: the same handle type as [`SessionEvent::AudioData`],
124    /// so audio can be forwarded without a copy.
125    SendAudio(Bytes),
126    /// Send a text message.
127    SendText(String),
128    /// Send tool responses.
129    SendToolResponse(Vec<FunctionResponse>),
130    /// Signal activity start (client VAD detected speech).
131    ActivityStart,
132    /// Signal activity end (client VAD detected silence).
133    ActivityEnd,
134    /// Send client content (conversation history or context injection).
135    SendClientContent {
136        /// Conversation turns to include.
137        turns: Vec<Content>,
138        /// Whether this completes the client's turn.
139        turn_complete: bool,
140    },
141    /// Send video/image data (raw JPEG bytes, will be base64-encoded).
142    SendVideo(Bytes),
143    /// Update system instruction mid-session (sends client_content with role=system).
144    UpdateInstruction(String),
145    /// Gracefully disconnect.
146    Disconnect,
147}
148
149// ---------------------------------------------------------------------------
150// Turn tracking
151// ---------------------------------------------------------------------------
152
153/// Represents a single model response turn.
154#[derive(Debug, Clone)]
155pub struct Turn {
156    /// Unique turn identifier.
157    pub id: String,
158    /// Accumulated text parts.
159    pub text: String,
160    /// Whether this turn included audio.
161    pub has_audio: bool,
162    /// Tool calls requested in this turn.
163    pub tool_calls: Vec<FunctionCall>,
164    /// When the turn started.
165    pub started_at: Instant,
166    /// When the turn completed (if complete).
167    pub completed_at: Option<Instant>,
168    /// Whether the turn was interrupted.
169    pub interrupted: bool,
170}
171
172impl Turn {
173    /// Create a new turn.
174    pub fn new() -> Self {
175        Self {
176            id: uuid::Uuid::new_v4().to_string(),
177            text: String::new(),
178            has_audio: false,
179            tool_calls: Vec::new(),
180            started_at: Instant::now(),
181            completed_at: None,
182            interrupted: false,
183        }
184    }
185
186    /// Duration of the turn.
187    pub fn duration(&self) -> std::time::Duration {
188        let end = self.completed_at.unwrap_or_else(Instant::now);
189        end.duration_since(self.started_at)
190    }
191}
192
193impl Default for Turn {
194    fn default() -> Self {
195        Self::new()
196    }
197}
198
199// ---------------------------------------------------------------------------
200// Broadcast lag helper
201// ---------------------------------------------------------------------------
202
203/// Receive the next event from a broadcast receiver, handling lag gracefully.
204///
205/// If the receiver falls behind (too slow to keep up with the sender), the
206/// skipped events are logged and the next available event is returned.
207/// Returns `None` when the channel is closed.
208///
209/// # Example
210///
211/// ```no_run
212/// # use gemini_genai_rs::session::{SessionHandle, recv_event};
213/// # async fn run(handle: SessionHandle) {
214/// let mut events = handle.subscribe();
215/// while let Some(event) = recv_event(&mut events).await {
216///     // handle event
217/// }
218/// # }
219/// ```
220pub async fn recv_event(rx: &mut broadcast::Receiver<SessionEvent>) -> Option<SessionEvent> {
221    loop {
222        match rx.recv().await {
223            Ok(event) => return Some(event),
224            Err(broadcast::error::RecvError::Lagged(n)) => {
225                tracing::warn!(skipped = n, "Event subscriber lagged, skipped {n} events");
226                continue;
227            }
228            Err(broadcast::error::RecvError::Closed) => return None,
229        }
230    }
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236
237    #[tokio::test]
238    async fn recv_event_returns_events_normally() {
239        let (tx, mut rx) = broadcast::channel(16);
240
241        tx.send(SessionEvent::Connected).unwrap();
242        tx.send(SessionEvent::TurnComplete).unwrap();
243
244        let event = recv_event(&mut rx).await;
245        assert!(matches!(event, Some(SessionEvent::Connected)));
246
247        let event = recv_event(&mut rx).await;
248        assert!(matches!(event, Some(SessionEvent::TurnComplete)));
249    }
250
251    #[tokio::test]
252    async fn recv_event_returns_none_on_closed_channel() {
253        let (tx, mut rx) = broadcast::channel::<SessionEvent>(16);
254        drop(tx);
255
256        let event = recv_event(&mut rx).await;
257        assert!(event.is_none(), "should return None when channel is closed");
258    }
259
260    #[tokio::test]
261    async fn recv_event_handles_lag() {
262        // Create a tiny broadcast channel (capacity 2)
263        let (tx, mut rx) = broadcast::channel(2);
264
265        // Send 4 events — the receiver will lag behind
266        for i in 0..4 {
267            let _ = tx.send(SessionEvent::TextDelta(format!("msg{i}")));
268        }
269
270        // recv_event should skip the lagged events and return the next available
271        let event = recv_event(&mut rx).await;
272        assert!(event.is_some(), "should get an event after lag");
273    }
274}
gemini_genai_rs/session/events.rs

gemini_genai_rs/session/
events.rs
