gemini_adk_rs/live/
callbacks.rs

1//! Typed callback registry for Live session events.
2//!
3//! Fast lane callbacks (sync, < 1ms): audio, text, transcripts, VAD.
4//! Control lane callbacks (async, can block): tool calls, lifecycle, interruptions.
5//! Outbound interceptors: transform tool responses, inject context at turn boundaries.
6//!
7//! # Callback Modes
8//!
9//! Each control-lane callback has an associated [`ExecutionMode`]:
10//!
11//! - [`Blocking`](ExecutionMode::Blocking) — awaited inline. The event loop
12//!   waits for completion before processing the next event. Guarantees
13//!   ordering and state consistency.
14//! - [`Concurrent`](ExecutionMode::Concurrent) — spawned as a detached tokio
15//!   task. The event loop continues immediately. Use for fire-and-forget
16//!   work (logging, background agent dispatch, analytics).
17//!
18//! Fast-lane callbacks (audio, text, VAD) are always sync and inline.
19//! Interceptors (`before_tool_response`, `on_turn_boundary`) are always blocking.
20//!
21//! `on_interrupted`, `on_turn_boundary`, and the `on_teardown` hooks default to
22//! blocking for a reason — audio forwarding resumes only after `on_interrupted`
23//! returns, the next turn proceeds only after `on_turn_boundary`, and disconnect
24//! completes only after teardown — but each can be made concurrent when the
25//! body is pure bookkeeping (see the `_mode` fields and `on_teardown_concurrent`).
26//! `on_tool_call` and `before_tool_response` are always blocking: their return
27//! value is the tool response.
28
29use std::sync::Arc;
30use std::time::Duration;
31
32use bytes::Bytes;
33use gemini_genai_rs::prelude::{FunctionCall, FunctionResponse, SessionPhase, UsageMetadata};
34use gemini_genai_rs::session::SessionWriter;
35
36use super::{BoxFuture, ExecutionMode};
37use crate::state::State;
38
39// ── Named callback types ──────────────────────────────────────────────────
40// These aliases are the vocabulary of the callback registry: every field
41// below (and the corresponding L2 setters) is one of these shapes.
42
43/// Fast-lane sync callback over a raw audio chunk.
44pub type AudioCallback = Box<dyn Fn(&Bytes) + Send + Sync>;
45/// Fast-lane sync callback over a chunk of non-audio model media (Live Avatar
46/// video).
47pub type MediaCallback = Box<dyn Fn(&gemini_genai_rs::session::InlineMedia) + Send + Sync>;
48/// Fast-lane sync callback over a text payload (delta, accumulated text, thought).
49pub type TextCallback = Box<dyn Fn(&str) + Send + Sync>;
50/// Fast-lane sync callback over a transcript chunk with its `is_final` flag.
51pub type TranscriptCallback = Box<dyn Fn(&str, bool) + Send + Sync>;
52/// Fast-lane sync callback with no payload (VAD start/end).
53pub type SignalCallback = Box<dyn Fn() + Send + Sync>;
54/// Fast-lane sync callback over a wire-level [`SessionPhase`] change
55/// (connecting → active → disconnecting …). Not the `PhaseMachine`.
56pub type SessionPhaseCallback = Box<dyn Fn(SessionPhase) + Send + Sync>;
57/// Fast-lane sync callback over usage metadata.
58pub type UsageCallback = Box<dyn Fn(&UsageMetadata) + Send + Sync>;
59
60/// Control-lane async callback with no payload.
61pub type AsyncCallback = Arc<dyn Fn() -> BoxFuture<()> + Send + Sync>;
62/// Control-lane async callback over one payload value.
63pub type AsyncCallbackWith<T> = Arc<dyn Fn(T) -> BoxFuture<()> + Send + Sync>;
64/// Control-lane async callback over two payload values.
65pub type AsyncCallbackWith2<A, B> = Arc<dyn Fn(A, B) -> BoxFuture<()> + Send + Sync>;
66/// Tool-call override: return `Some(responses)` to reply, `None` to defer to
67/// auto-dispatch via the registered `ToolDispatcher`.
68pub type ToolCallCallback =
69    Arc<dyn Fn(Vec<FunctionCall>, State) -> BoxFuture<Option<Vec<FunctionResponse>>> + Send + Sync>;
70/// Middleware over outgoing tool responses (inspect/rewrite before send).
71pub type BeforeToolResponseCallback =
72    Arc<dyn Fn(Vec<FunctionResponse>, State) -> BoxFuture<Vec<FunctionResponse>> + Send + Sync>;
73/// Sync state-reactive instruction generator (`None` = leave unchanged).
74pub type InstructionFn = Arc<dyn Fn(&State) -> Option<String> + Send + Sync>;
75
76/// Typed callback registry for Live session events.
77///
78/// Callbacks are divided into two lanes:
79/// - **Fast lane** (sync): Called inline, must be < 1ms. For audio, text, transcripts, VAD.
80/// - **Control lane** (async): Awaited on a dedicated task. For tool calls, lifecycle, interruptions.
81pub struct EventCallbacks {
82    // -- Fast lane (sync callbacks) --
83    /// Called for each audio chunk from the model (PCM16 24kHz).
84    pub on_audio: Option<AudioCallback>,
85    /// Called for each non-audio media chunk from the model — Gemini 3.8
86    /// Live Avatar video (`video/mp4`). Suppressed during barge-in, like
87    /// audio.
88    pub on_media: Option<MediaCallback>,
89    /// Called for each incremental text delta from the model.
90    pub on_text: Option<TextCallback>,
91    /// Called when the model completes a text response.
92    pub on_text_complete: Option<TextCallback>,
93    /// Called for input (user speech) transcription updates.
94    pub on_input_transcript: Option<TranscriptCallback>,
95    /// Called for output (model speech) transcription updates.
96    pub on_output_transcript: Option<TranscriptCallback>,
97    /// Called when the model emits a thought/reasoning summary (when includeThoughts is enabled).
98    pub on_thought: Option<TextCallback>,
99    /// Called when server-side VAD detects voice activity start.
100    pub on_vad_start: Option<SignalCallback>,
101    /// Called when server-side VAD detects voice activity end.
102    pub on_vad_end: Option<SignalCallback>,
103    /// Called on session phase transitions.
104    pub on_session_phase: Option<SessionPhaseCallback>,
105    /// Called when server sends token usage metadata.
106    pub on_usage: Option<UsageCallback>,
107
108    // -- Control lane (async callbacks) --
109    /// Called when the model is interrupted by barge-in.
110    pub on_interrupted: Option<AsyncCallback>,
111    /// Called when model requests tool execution.
112    /// Return `None` to use auto-dispatch (ToolDispatcher), `Some` to override.
113    /// Receives State for natural state promotion from tool results.
114    pub on_tool_call: Option<ToolCallCallback>,
115    /// Called when server cancels pending tool calls.
116    pub on_tool_cancelled: Option<AsyncCallbackWith<Vec<String>>>,
117    /// Called when the model completes its turn.
118    pub on_turn_complete: Option<AsyncCallback>,
119    /// Called when the model finishes generating a response (the wire
120    /// `GenerationComplete`). Current Live models do not send it for an
121    /// interrupted turn.
122    pub on_generation_complete: Option<AsyncCallback>,
123    /// Called when server sends GoAway (session ending soon).
124    pub on_go_away: Option<AsyncCallbackWith<Duration>>,
125    /// Called when session setup completes (connected).
126    ///
127    /// Receives a `SessionWriter` for sending messages on connect (e.g. greeting prompts).
128    pub on_connected: Option<AsyncCallbackWith<Arc<dyn SessionWriter>>>,
129    /// Called when session disconnects.
130    pub on_disconnected: Option<AsyncCallbackWith<Option<String>>>,
131    /// Teardown hooks run on disconnect, **before** `on_disconnected`.
132    ///
133    /// Additive, unlike every other callback here. Each of those is a single
134    /// `Option` that the last registration silently replaces, which is fine for
135    /// an application — it has one place to write its handler — and unusable for
136    /// an extension: `.with_memory(s).on_disconnected(f)` would drop the
137    /// extension's hook, and the reverse order would drop the application's,
138    /// with nothing reporting either.
139    ///
140    /// Extensions that must flush durable state at end of session register here
141    /// (`gemini-memory-rs` reconciles its session ledger this way). Hooks run in
142    /// registration order and are awaited before `on_disconnected` fires, so the
143    /// application's own handler observes a settled world. A hook that panics or
144    /// hangs delays disconnect — keep them bounded.
145    pub on_teardown: Vec<AsyncCallback>,
146    /// Teardown hooks that are spawned detached on disconnect rather than
147    /// awaited — for bookkeeping that must not delay the disconnect (metrics,
148    /// a final log line). Anything that flushes durable state belongs in
149    /// [`on_teardown`](Self::on_teardown).
150    pub on_teardown_concurrent: Vec<AsyncCallback>,
151    /// Called after session resumes from GoAway.
152    pub on_resumed: Option<AsyncCallback>,
153    /// Called on non-fatal errors.
154    pub on_error: Option<AsyncCallbackWith<String>>,
155    /// Called when agent transfer occurs (from, to).
156    pub on_transfer: Option<AsyncCallbackWith2<String, String>>,
157    /// Called when a TurnExtractor produces a result (extractor_name, value).
158    pub on_extracted: Option<AsyncCallbackWith2<String, serde_json::Value>>,
159    /// Called when a TurnExtractor fails (extractor_name, error_message).
160    ///
161    /// By default, extraction failures are logged via `tracing::warn!`.
162    /// Register this callback to implement custom error handling (retry, alert, etc.).
163    pub on_extraction_error: Option<AsyncCallbackWith2<String, String>>,
164
165    // -- Callback modes (control-lane only) --
166    /// Execution mode for [`on_interrupted`](Self::on_interrupted). Blocking
167    /// by default: audio forwarding resumes only after the callback returns,
168    /// which is what a playback flush needs. Concurrent is for bookkeeping only.
169    pub on_interrupted_mode: ExecutionMode,
170    /// Execution mode for [`on_turn_boundary`](Self::on_turn_boundary).
171    /// Blocking by default so injected context lands before the next turn;
172    /// concurrent is for observation only.
173    pub on_turn_boundary_mode: ExecutionMode,
174    /// Execution mode for [`on_turn_complete`](Self::on_turn_complete).
175    pub on_turn_complete_mode: ExecutionMode,
176    /// Execution mode for [`on_generation_complete`](Self::on_generation_complete).
177    pub on_generation_complete_mode: ExecutionMode,
178    /// Execution mode for [`on_connected`](Self::on_connected).
179    pub on_connected_mode: ExecutionMode,
180    /// Execution mode for [`on_disconnected`](Self::on_disconnected).
181    pub on_disconnected_mode: ExecutionMode,
182    /// Execution mode for [`on_error`](Self::on_error).
183    pub on_error_mode: ExecutionMode,
184    /// Execution mode for [`on_go_away`](Self::on_go_away).
185    pub on_go_away_mode: ExecutionMode,
186    /// Execution mode for [`on_extracted`](Self::on_extracted).
187    pub on_extracted_mode: ExecutionMode,
188    /// Execution mode for [`on_extraction_error`](Self::on_extraction_error).
189    pub on_extraction_error_mode: ExecutionMode,
190    /// Execution mode for [`on_tool_cancelled`](Self::on_tool_cancelled).
191    pub on_tool_cancelled_mode: ExecutionMode,
192    /// Execution mode for [`on_transfer`](Self::on_transfer).
193    pub on_transfer_mode: ExecutionMode,
194    /// Execution mode for [`on_resumed`](Self::on_resumed).
195    pub on_resumed_mode: ExecutionMode,
196
197    // -- Outbound interceptors (transform data going to Gemini) --
198    /// Intercept tool responses before sending to Gemini.
199    ///
200    /// Receives the tool responses and shared State. Returns (potentially modified)
201    /// responses. Use this to rewrite, augment, or filter tool results based on
202    /// conversation state.
203    pub before_tool_response: Option<BeforeToolResponseCallback>,
204
205    /// Called at turn boundaries (after extractors, before `on_turn_complete`).
206    ///
207    /// Receives shared State and a SessionWriter for injecting content into
208    /// the conversation. Use this for context stuffing, K/V injection, condensed
209    /// state summaries, or any outbound content interleaving.
210    pub on_turn_boundary: Option<AsyncCallbackWith2<State, Arc<dyn SessionWriter>>>,
211
212    /// State-reactive system instruction template (full replacement).
213    ///
214    /// Called after extractors run on each TurnComplete. If it returns
215    /// `Some(instruction)`, the system instruction is updated mid-session.
216    /// Returns `None` to leave the instruction unchanged.
217    ///
218    /// This is sync (no async) because instruction generation should be fast.
219    pub instruction_template: Option<InstructionFn>,
220
221    /// State-reactive instruction amendment (additive, not replacement).
222    ///
223    /// Called after extractors and phase transitions on each TurnComplete.
224    /// If it returns `Some(text)`, the text is appended to the current phase
225    /// instruction (separated by `\n\n`). Returns `None` to skip amendment.
226    ///
227    /// Unlike `instruction_template` (which replaces the entire instruction),
228    /// this only adds to the phase instruction — the developer never needs to
229    /// know or repeat the base instruction.
230    pub instruction_amendment: Option<InstructionFn>,
231}
232
233impl Default for EventCallbacks {
234    fn default() -> Self {
235        Self {
236            on_audio: None,
237            on_media: None,
238            on_text: None,
239            on_text_complete: None,
240            on_input_transcript: None,
241            on_output_transcript: None,
242            on_thought: None,
243            on_vad_start: None,
244            on_vad_end: None,
245            on_session_phase: None,
246            on_usage: None,
247            on_interrupted: None,
248            on_tool_call: None,
249            on_tool_cancelled: None,
250            on_turn_complete: None,
251            on_generation_complete: None,
252            on_go_away: None,
253            on_connected: None,
254            on_disconnected: None,
255            on_teardown: Vec::new(),
256            on_teardown_concurrent: Vec::new(),
257            on_resumed: None,
258            on_error: None,
259            on_transfer: None,
260            on_extracted: None,
261            on_extraction_error: None,
262            on_interrupted_mode: ExecutionMode::Blocking,
263            on_turn_boundary_mode: ExecutionMode::Blocking,
264            on_turn_complete_mode: ExecutionMode::Blocking,
265            on_generation_complete_mode: ExecutionMode::Blocking,
266            on_connected_mode: ExecutionMode::Blocking,
267            on_disconnected_mode: ExecutionMode::Blocking,
268            on_error_mode: ExecutionMode::Blocking,
269            on_go_away_mode: ExecutionMode::Blocking,
270            on_extracted_mode: ExecutionMode::Blocking,
271            on_extraction_error_mode: ExecutionMode::Blocking,
272            on_tool_cancelled_mode: ExecutionMode::Blocking,
273            on_transfer_mode: ExecutionMode::Blocking,
274            on_resumed_mode: ExecutionMode::Blocking,
275            before_tool_response: None,
276            on_turn_boundary: None,
277            instruction_template: None,
278            instruction_amendment: None,
279        }
280    }
281}
282
283impl std::fmt::Debug for EventCallbacks {
284    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
285        f.debug_struct("EventCallbacks")
286            .field("on_audio", &self.on_audio.is_some())
287            .field("on_media", &self.on_media.is_some())
288            .field("on_text", &self.on_text.is_some())
289            .field("on_text_complete", &self.on_text_complete.is_some())
290            .field("on_input_transcript", &self.on_input_transcript.is_some())
291            .field("on_output_transcript", &self.on_output_transcript.is_some())
292            .field("on_thought", &self.on_thought.is_some())
293            .field("on_vad_start", &self.on_vad_start.is_some())
294            .field("on_vad_end", &self.on_vad_end.is_some())
295            .field("on_session_phase", &self.on_session_phase.is_some())
296            .field("on_usage", &self.on_usage.is_some())
297            .field("on_interrupted", &self.on_interrupted.is_some())
298            .field("on_tool_call", &self.on_tool_call.is_some())
299            .field("on_tool_cancelled", &self.on_tool_cancelled.is_some())
300            .field("on_turn_complete", &self.on_turn_complete.is_some())
301            .field("on_go_away", &self.on_go_away.is_some())
302            .field("on_connected", &self.on_connected.is_some())
303            .field("on_disconnected", &self.on_disconnected.is_some())
304            .field("on_resumed", &self.on_resumed.is_some())
305            .field("on_error", &self.on_error.is_some())
306            .field("on_transfer", &self.on_transfer.is_some())
307            .field("on_extracted", &self.on_extracted.is_some())
308            .field("on_extraction_error", &self.on_extraction_error.is_some())
309            .field("on_teardown", &self.on_teardown.len())
310            .field("on_teardown_concurrent", &self.on_teardown_concurrent.len())
311            .field("on_interrupted_mode", &self.on_interrupted_mode)
312            .field("on_turn_boundary_mode", &self.on_turn_boundary_mode)
313            .field("on_turn_complete_mode", &self.on_turn_complete_mode)
314            .field("on_connected_mode", &self.on_connected_mode)
315            .field("on_disconnected_mode", &self.on_disconnected_mode)
316            .field("on_error_mode", &self.on_error_mode)
317            .field("on_go_away_mode", &self.on_go_away_mode)
318            .field("on_extracted_mode", &self.on_extracted_mode)
319            .field("on_extraction_error_mode", &self.on_extraction_error_mode)
320            .field("on_tool_cancelled_mode", &self.on_tool_cancelled_mode)
321            .field("on_transfer_mode", &self.on_transfer_mode)
322            .field("on_resumed_mode", &self.on_resumed_mode)
323            .field("before_tool_response", &self.before_tool_response.is_some())
324            .field("on_turn_boundary", &self.on_turn_boundary.is_some())
325            .field("instruction_template", &self.instruction_template.is_some())
326            .field(
327                "instruction_amendment",
328                &self.instruction_amendment.is_some(),
329            )
330            .finish()
331    }
332}
333
334#[cfg(test)]
335mod tests {
336    use super::*;
337
338    #[test]
339    fn default_callbacks_all_none() {
340        let cb = EventCallbacks::default();
341        assert!(cb.on_audio.is_none());
342        assert!(cb.on_text.is_none());
343        assert!(cb.on_interrupted.is_none());
344        assert!(cb.on_tool_call.is_none());
345    }
346
347    #[test]
348    fn sync_callback_callable() {
349        let mut cb = EventCallbacks::default();
350        let called = Arc::new(std::sync::atomic::AtomicBool::new(false));
351        let called_clone = called.clone();
352        cb.on_text = Some(Box::new(move |_text| {
353            called_clone.store(true, std::sync::atomic::Ordering::SeqCst);
354        }));
355        if let Some(f) = &cb.on_text {
356            f("hello");
357        }
358        assert!(called.load(std::sync::atomic::Ordering::SeqCst));
359    }
360
361    #[test]
362    fn callback_mode_defaults_to_blocking() {
363        let cb = EventCallbacks::default();
364        assert_eq!(cb.on_turn_complete_mode, ExecutionMode::Blocking);
365        assert_eq!(cb.on_connected_mode, ExecutionMode::Blocking);
366        assert_eq!(cb.on_disconnected_mode, ExecutionMode::Blocking);
367        assert_eq!(cb.on_error_mode, ExecutionMode::Blocking);
368        assert_eq!(cb.on_go_away_mode, ExecutionMode::Blocking);
369        assert_eq!(cb.on_extracted_mode, ExecutionMode::Blocking);
370        assert_eq!(cb.on_extraction_error_mode, ExecutionMode::Blocking);
371        assert_eq!(cb.on_tool_cancelled_mode, ExecutionMode::Blocking);
372        assert_eq!(cb.on_transfer_mode, ExecutionMode::Blocking);
373        assert_eq!(cb.on_resumed_mode, ExecutionMode::Blocking);
374    }
375
376    #[test]
377    fn debug_shows_registered() {
378        let cb = EventCallbacks {
379            on_audio: Some(Box::new(|_| {})),
380            ..Default::default()
381        };
382        let debug = format!("{cb:?}");
383        assert!(debug.contains("on_audio: true"));
384        assert!(debug.contains("on_text: false"));
385    }
386}
gemini_adk_rs/live/callbacks.rs

gemini_adk_rs/live/
callbacks.rs
