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