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

gemini_adk_rs/live/
callbacks.rs
