gemini_adk_rs/live/
processor.rs

1//! Three-lane event processor for Live sessions.
2//!
3//! **Fast lane**: audio, text, VAD (sync callbacks, never blocks)
4//! **Control lane**: tool calls, interruptions, lifecycle, transcript accumulation,
5//!   extractors, phases, watchers (async callbacks, can block)
6//! **Telemetry lane**: SessionSignals + SessionTelemetry (debounced state writes,
7//!   runs on its own broadcast receiver — zero work on the router hot path)
8
9use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
10use std::sync::Arc;
11use std::time::Duration;
12
13use bytes::Bytes;
14use tokio::sync::{broadcast, mpsc};
15use tokio_util::sync::CancellationToken;
16
17use gemini_genai_rs::prelude::{SessionEvent, SessionPhase};
18use gemini_genai_rs::session::SessionWriter;
19
20use crate::state::State;
21use crate::tool::ToolDispatcher;
22
23use super::background_tool::BackgroundToolTracker;
24use super::callbacks::EventCallbacks;
25use super::computed::ComputedRegistry;
26use super::context_writer::PendingContext;
27use super::control_plane::run_control_lane;
28use super::events::LiveEvent;
29use super::extractor::TurnExtractor;
30use super::needs::NeedsFulfillment;
31use super::persistence::SessionPersistence;
32use super::phase::PhaseMachine;
33use super::session_signals::SessionSignals;
34use super::soft_turn::SoftTurnDetector;
35use super::steering::{ContextDelivery, SteeringMode};
36use super::telemetry::SessionTelemetry;
37use super::temporal::TemporalRegistry;
38use super::watcher::WatcherRegistry;
39
40/// Backpressure (delivery) policy for a single class of fast-lane events.
41///
42/// The event router forwards fast-lane frames (audio, text, transcripts,
43/// thoughts, VAD, phase) over a bounded channel to the fast-lane consumer. When
44/// that consumer falls behind and the channel fills, the policy decides what the
45/// router does — and crucially, whether the router *blocks*. Because the router
46/// is shared by both the fast lane and the control lane, a blocking fast-lane
47/// send stalls routing for *all* events, including control-lane lifecycle and
48/// tool events. The policy lets callers trade frame durability for router
49/// responsiveness on a per-class basis.
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
51pub enum Delivery {
52    /// Never drop a frame: `tx.send(ev).await` — the router awaits when the
53    /// channel is full. This is the historical (and default) behavior and is
54    /// byte-for-byte identical to the pre-policy code path. Use it when every
55    /// frame matters and a slow consumer applying backpressure to the router is
56    /// acceptable.
57    #[default]
58    Lossless,
59    /// Drop the *newest* frame on overflow: `tx.try_send(ev)` and, on
60    /// [`TrySendError::Full`](tokio::sync::mpsc::error::TrySendError::Full),
61    /// discard the just-produced frame and bump a dropped-frame counter. The
62    /// router never blocks on this class, so a slow fast-lane consumer can no
63    /// longer stall control-lane routing. Use it for high-frequency, loss-
64    /// tolerant streams (e.g. partial transcripts, thoughts) where freshness of
65    /// already-queued frames matters less than keeping the router moving.
66    ///
67    /// A drop-oldest / latest-only variant is intentionally *not* provided:
68    /// tokio's `mpsc` has no clean "evict the oldest queued item" primitive, so
69    /// implementing it correctly would require a custom ring buffer. That is
70    /// left as future work rather than shipped half-working.
71    LossyDropNewest,
72}
73
74/// Per-event-class delivery (backpressure) policy for the fast lane.
75///
76/// Each fast-lane event class carries its own [`Delivery`] policy. The
77/// [`Default`] impl sets **every** class to [`Delivery::Lossless`], which makes
78/// the whole feature behavior-preserving: with the default config the router
79/// uses the same `send().await` path it always has. Callers opt into lossy
80/// behavior per class via the builder setters.
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub struct DeliveryConfig {
83    /// Policy for raw PCM audio frames.
84    pub audio: Delivery,
85    /// Policy for incremental text deltas (and text-complete frames).
86    pub text: Delivery,
87    /// Policy for input/output transcript frames (fast-lane callback copy only;
88    /// control-lane accumulation is unaffected and always lossless).
89    pub transcript: Delivery,
90    /// Policy for thought-summary frames.
91    pub thought: Delivery,
92    /// Policy for VAD start/end frames.
93    pub vad: Delivery,
94    /// Policy for phase-changed frames.
95    pub phase: Delivery,
96}
97
98impl Default for DeliveryConfig {
99    fn default() -> Self {
100        Self {
101            audio: Delivery::Lossless,
102            text: Delivery::Lossless,
103            transcript: Delivery::Lossless,
104            thought: Delivery::Lossless,
105            vad: Delivery::Lossless,
106            phase: Delivery::Lossless,
107        }
108    }
109}
110
111impl DeliveryConfig {
112    /// A config with every class set to [`Delivery::Lossless`] (same as
113    /// [`Default`]).
114    pub fn lossless() -> Self {
115        Self::default()
116    }
117
118    /// Set the audio policy.
119    pub fn audio(mut self, d: Delivery) -> Self {
120        self.audio = d;
121        self
122    }
123
124    /// Set the text policy.
125    pub fn text(mut self, d: Delivery) -> Self {
126        self.text = d;
127        self
128    }
129
130    /// Set the transcript policy.
131    pub fn transcript(mut self, d: Delivery) -> Self {
132        self.transcript = d;
133        self
134    }
135
136    /// Set the thought policy.
137    pub fn thought(mut self, d: Delivery) -> Self {
138        self.thought = d;
139        self
140    }
141
142    /// Set the VAD policy.
143    pub fn vad(mut self, d: Delivery) -> Self {
144        self.vad = d;
145        self
146    }
147
148    /// Set the phase policy.
149    pub fn phase(mut self, d: Delivery) -> Self {
150        self.phase = d;
151        self
152    }
153}
154
155/// Per-class counters for fast-lane frames dropped under a lossy policy.
156///
157/// Incremented with a single relaxed atomic add on the router hot path when a
158/// [`Delivery::LossyDropNewest`] send overflows. Reads are for observability /
159/// tests and never gate the hot path.
160#[derive(Debug, Default)]
161pub(crate) struct DroppedFrames {
162    pub audio: AtomicU64,
163    pub text: AtomicU64,
164    pub transcript: AtomicU64,
165    pub thought: AtomicU64,
166    pub vad: AtomicU64,
167    pub phase: AtomicU64,
168}
169
170impl DroppedFrames {
171    /// Total dropped frames across all classes.
172    ///
173    /// Currently only consumed by tests; the per-class atomics are read
174    /// directly elsewhere. Kept test-gated until a handle accessor surfaces it.
175    #[cfg(test)]
176    pub fn total(&self) -> u64 {
177        self.audio.load(Ordering::Relaxed)
178            + self.text.load(Ordering::Relaxed)
179            + self.transcript.load(Ordering::Relaxed)
180            + self.thought.load(Ordering::Relaxed)
181            + self.vad.load(Ordering::Relaxed)
182            + self.phase.load(Ordering::Relaxed)
183    }
184}
185
186/// Forward one fast-lane frame according to its class delivery policy.
187///
188/// - [`Delivery::Lossless`]: `tx.send(ev).await` — awaits when the channel is
189///   full (identical to the pre-policy behavior).
190/// - [`Delivery::LossyDropNewest`]: `tx.try_send(ev)` — on a full channel, drop
191///   the frame and increment `dropped`.
192///
193/// Returns without ever blocking the router under a lossy policy.
194async fn deliver_fast(
195    tx: &mpsc::Sender<FastEvent>,
196    ev: FastEvent,
197    policy: Delivery,
198    dropped: &AtomicU64,
199) {
200    match policy {
201        Delivery::Lossless => {
202            let _ = tx.send(ev).await;
203        }
204        Delivery::LossyDropNewest => {
205            if let Err(mpsc::error::TrySendError::Full(_)) = tx.try_send(ev) {
206                dropped.fetch_add(1, Ordering::Relaxed);
207            }
208            // `TrySendError::Closed` is ignored, matching the `let _ = send`
209            // pattern used elsewhere (the consumer is gone; nothing to do).
210        }
211    }
212}
213
214/// Events routed to the fast lane (sync processing).
215pub(crate) enum FastEvent {
216    Audio(Bytes),
217    Text(String),
218    TextComplete(String),
219    InputTranscript(String),
220    OutputTranscript(String),
221    Thought(String),
222    VadStart,
223    VadEnd,
224    Phase(SessionPhase),
225    /// Interruption flag — tells fast lane to stop forwarding audio.
226    Interrupted,
227}
228
229/// Events routed to the control lane (async processing).
230pub(crate) enum ControlEvent {
231    ToolCall(Vec<gemini_genai_rs::prelude::FunctionCall>),
232    ToolCallCancelled(Vec<String>),
233    /// A background tool finished. Posted by the detached background task (which
234    /// can't reach the synchronous `FlowMonitor`) so the control lane can advance
235    /// the governed flow through the same gate as inline tools (#7).
236    ToolCompleted {
237        /// The tool call's correlation id (for once-per-call_id flow dedup).
238        call_id: String,
239        /// The tool name (matches `FunctionCall::name`).
240        name: String,
241        /// Whether the tool completed successfully.
242        ok: bool,
243    },
244    Interrupted,
245    TurnComplete,
246    /// Model finished generating (even if interrupted). Fires before TurnComplete.
247    GenerationComplete,
248    GoAway(Option<String>),
249    Connected,
250    Disconnected(Option<String>),
251    SessionResumeUpdate(gemini_genai_rs::session::ResumeInfo),
252    Error(String),
253    /// Transcript accumulation — pushed from router, exclusive to control lane.
254    InputTranscript(String),
255    OutputTranscript(String),
256}
257
258/// Shared state between the two lanes.
259pub(crate) struct SharedState {
260    /// When true, fast lane suppresses audio callbacks.
261    pub interrupted: AtomicBool,
262    /// Barge-in signal for in-flight inline tool dispatch.
263    ///
264    /// Cancelled by the ROUTER the moment an `Interrupted` event arrives,
265    /// then re-armed (replaced with a fresh token) by the control lane once
266    /// it has processed the interruption. The control lane races inline tool
267    /// dispatch against this token, so a user barge-in is never stuck waiting
268    /// behind a slow tool.
269    pub barge_in: parking_lot::Mutex<CancellationToken>,
270    /// Latest resume handle from server.
271    pub resume_handle: parking_lot::Mutex<Option<String>>,
272    /// Last instruction sent via instruction_template (for dedup).
273    pub last_instruction: parking_lot::Mutex<Option<String>>,
274    /// Steering lines projected on the previous turn boundary, for repeat
275    /// suppression.
276    ///
277    /// Context turns are *appended* to the server-side conversation and cannot
278    /// be retracted, so re-sending an unchanged steering line does not
279    /// re-emphasise it — it adds a second standing directive the model has to
280    /// weigh against everything that follows. Holding the previous turn's lines
281    /// lets an unchanged one be dropped while a changed one still lands.
282    pub last_context: parking_lot::Mutex<Vec<String>>,
283    /// Pending context buffer for deferred delivery (None when Immediate mode).
284    pub pending_context: Option<Arc<PendingContext>>,
285    /// Fast-lane delivery policy per event class.
286    pub delivery: DeliveryConfig,
287    /// Per-class counters for frames dropped under a lossy delivery policy.
288    pub dropped: DroppedFrames,
289}
290
291/// Runs the three-lane event processor.
292///
293/// Returns JoinHandles for the fast consumer and control processor tasks.
294/// The telemetry lane is spawned separately via [`spawn_telemetry_lane`].
295/// Configuration for the control plane's new capabilities.
296pub(crate) struct ControlPlaneConfig {
297    /// The connect-time system instruction, as text.
298    ///
299    /// What an `instruction_amendment` composes onto when no phase supplies a
300    /// base. Amendments are additive by contract, so they need something to be
301    /// added to; a session with no phase machine has no phase instruction, and
302    /// before this was carried the amendment was computed each turn and then
303    /// silently discarded. Sending the amendment alone was not an option —
304    /// under `SteeringMode::InstructionUpdate` that replaces the system
305    /// instruction, so it would have deleted the caller's own prompt.
306    pub base_instruction: Option<String>,
307    /// Soft turn detector for proactive silence awareness.
308    pub soft_turn: Option<SoftTurnDetector>,
309    /// Steering mode for phase instruction delivery.
310    pub steering_mode: SteeringMode,
311    /// When to deliver context turns to the wire.
312    /// Deferred = synchronize with user activity (speech, interruption);
313    /// Immediate = send during TurnComplete processing.
314    pub context_delivery: ContextDelivery,
315    /// Conversation repair tracker.
316    pub needs_fulfillment: Option<NeedsFulfillment>,
317    /// Session persistence backend.
318    pub persistence: Option<Arc<dyn SessionPersistence>>,
319    /// Session ID for persistence key.
320    pub session_id: Option<String>,
321    /// Whether to inject tool availability advisory on phase transitions.
322    pub tool_advisory: bool,
323    /// Shared pending context buffer for deferred delivery (None when Immediate).
324    /// Must be the same Arc given to the DeferredWriter so the control lane
325    /// can push context and the DeferredWriter can drain it.
326    pub pending_context: Option<Arc<PendingContext>>,
327    /// Middleware layers run around tool dispatch in the control lane
328    /// (`before_tool` / `after_tool` / `on_tool_error`).
329    pub middleware: Arc<crate::middleware::MiddlewareChain>,
330    /// Optional governed-flow monitor: gates tool calls, projects active-step
331    /// postures into steering, and drives repair from unmet requirements.
332    /// Shared (`Arc<Mutex<..>>`) so the [`LiveHandle`](super::handle::LiveHandle)
333    /// can snapshot `explain`/`why_blocked` while the control lane advances it.
334    /// Lock briefly; never hold the guard across an `await`.
335    pub flow: Option<crate::flow::SharedFlowMonitor>,
336    /// Fast-lane delivery (backpressure) policy per event class. Defaults to
337    /// all-`Lossless`, preserving the historical `send().await` behavior.
338    pub delivery: DeliveryConfig,
339}
340
341impl Default for ControlPlaneConfig {
342    fn default() -> Self {
343        Self {
344            base_instruction: None,
345            soft_turn: None,
346            steering_mode: SteeringMode::default(),
347            context_delivery: ContextDelivery::default(),
348            needs_fulfillment: None,
349            persistence: None,
350            session_id: None,
351            tool_advisory: true,
352            pending_context: None,
353            middleware: Arc::new(crate::middleware::MiddlewareChain::new()),
354            flow: None,
355            delivery: DeliveryConfig::default(),
356        }
357    }
358}
359
360#[allow(
361    clippy::too_many_arguments,
362    reason = "lane spawn site: parameters are the owned subsystem handles split between the fast and control lanes"
363)]
364pub(crate) fn spawn_event_processor(
365    mut event_rx: broadcast::Receiver<SessionEvent>,
366    callbacks: Arc<EventCallbacks>,
367    dispatcher: Option<Arc<ToolDispatcher>>,
368    writer: Arc<dyn SessionWriter>,
369    extractors: Vec<Arc<dyn TurnExtractor>>,
370    state: State,
371    computed: Option<ComputedRegistry>,
372    phase_machine: Option<tokio::sync::Mutex<PhaseMachine>>,
373    watchers: Option<WatcherRegistry>,
374    temporal: Option<Arc<TemporalRegistry>>,
375    background_tracker: Option<Arc<BackgroundToolTracker>>,
376    execution_modes: std::collections::HashMap<String, super::background_tool::ToolExecutionMode>,
377    control_plane: ControlPlaneConfig,
378    live_event_tx: broadcast::Sender<LiveEvent>,
379) -> (
380    tokio::task::JoinHandle<()>,
381    tokio::task::JoinHandle<()>,
382    mpsc::WeakSender<ControlEvent>,
383) {
384    let shared = Arc::new(SharedState {
385        interrupted: AtomicBool::new(false),
386        barge_in: parking_lot::Mutex::new(CancellationToken::new()),
387        resume_handle: parking_lot::Mutex::new(None),
388        last_instruction: parking_lot::Mutex::new(None),
389        last_context: parking_lot::Mutex::new(Vec::new()),
390        pending_context: control_plane.pending_context.clone(),
391        delivery: control_plane.delivery,
392        dropped: DroppedFrames::default(),
393    });
394
395    let timer_cancel = CancellationToken::new();
396
397    // Channels between router and lanes.
398    //
399    // The control channel matches the fast channel at 512: control events
400    // are routed with a lossless `send().await`, so a *full* control queue
401    // blocks the shared router — and a blocked router stops forwarding audio
402    // frames too, causing playback glitches. Transcript accumulation events
403    // (one per ASR chunk) flow through this channel, so a slow control-lane
404    // consumer (e.g. a blocking turn-complete pipeline) could realistically
405    // fill 64 slots; 512 gives the lane room to fall behind transiently
406    // without starving the fast lane.
407    let (fast_tx, fast_rx) = mpsc::channel::<FastEvent>(512);
408    let (ctrl_tx, ctrl_rx) = mpsc::channel::<ControlEvent>(512);
409
410    // Spawn the router task (reads broadcast, routes to lanes)
411    // NOTE: SessionSignals is NOT called here — it runs on the telemetry lane.
412    let fast_tx_clone = fast_tx.clone();
413    let ctrl_tx_clone = ctrl_tx.clone();
414    let shared_clone = shared.clone();
415    tokio::spawn(async move {
416        loop {
417            match event_rx.recv().await {
418                Ok(event) => {
419                    // `Disconnected` is terminal in L0 (the session loop returns
420                    // after emitting it), so the router exits after routing it.
421                    // Dropping the router's lane senders closes the fast/control
422                    // channels, letting both lanes drain their queues and shut
423                    // down gracefully (final persistence drain, etc.) instead of
424                    // idling forever on a broadcast channel that never closes
425                    // while the `SessionHandle` is alive.
426                    let terminal = matches!(event, SessionEvent::Disconnected(_));
427                    route_event(event, &fast_tx_clone, &ctrl_tx_clone, &shared_clone).await;
428                    if terminal {
429                        break;
430                    }
431                }
432                Err(broadcast::error::RecvError::Lagged(n)) => {
433                    #[cfg(feature = "tracing-support")]
434                    tracing::warn!(skipped = n, "Event processor lagged, skipped events");
435                    let _ = n;
436                }
437                Err(broadcast::error::RecvError::Closed) => break,
438            }
439        }
440    });
441
442    // Spawn fast consumer (no transcript buffer — transcripts are in control lane)
443    let fast_callbacks = callbacks.clone();
444    let fast_shared = shared.clone();
445    let fast_event_tx = live_event_tx.clone();
446    let fast_handle = tokio::spawn(async move {
447        run_fast_lane(fast_rx, fast_callbacks, fast_shared, fast_event_tx).await;
448    });
449
450    // Clone for the timer task (before moving into ctrl spawn)
451    let timer_temporal = temporal.clone();
452    let timer_state = state.clone();
453    let timer_writer = writer.clone();
454
455    // Spawn control processor (owns TranscriptBuffer exclusively — no mutex needed)
456    let ctrl_callbacks = callbacks;
457    let ctrl_shared = shared;
458    let ctrl_timer_cancel = timer_cancel.clone();
459    // Weak sender handed to the control lane so background tool tasks can post
460    // completions back without keeping the channel open on shutdown (the lane
461    // upgrades it per background spawn; the channel closes once the router and
462    // all in-flight background tasks drop their strong senders).
463    let ctrl_tx_weak = ctrl_tx.downgrade();
464    let ctrl_handle = tokio::spawn(async move {
465        run_control_lane(
466            ctrl_rx,
467            ctrl_tx_weak,
468            ctrl_callbacks,
469            dispatcher,
470            writer,
471            ctrl_shared,
472            extractors,
473            state,
474            computed,
475            phase_machine,
476            watchers,
477            temporal,
478            background_tracker,
479            execution_modes,
480            control_plane,
481            live_event_tx,
482        )
483        .await;
484        ctrl_timer_cancel.cancel();
485    });
486
487    // Optional timer task for sustained temporal patterns
488    if let Some(ref temporal_ref) = timer_temporal {
489        if temporal_ref.needs_timer() {
490            let t = temporal_ref.clone();
491            let cancel = timer_cancel.clone();
492            tokio::spawn(async move {
493                let mut interval = tokio::time::interval(Duration::from_millis(500));
494                loop {
495                    tokio::select! {
496                        _ = cancel.cancelled() => break,
497                        _ = interval.tick() => {
498                            for action in t.check_all(&timer_state, None, &timer_writer) {
499                                tokio::spawn(action);
500                            }
501                        }
502                    }
503                }
504            });
505        }
506    }
507
508    // Handed to `LiveHandle` so `send_text` can record a typed turn on the
509    // transcript. Deliberately *weak*: the control channel must still close when
510    // the router and all in-flight background tasks drop their strong senders,
511    // otherwise the lane would never drain and shut down (see `ctrl_tx_weak`).
512    let ctrl_tx_handle = ctrl_tx.downgrade();
513
514    (fast_handle, ctrl_handle, ctrl_tx_handle)
515}
516
517/// Spawns the telemetry lane — processes events on its own broadcast receiver.
518///
519/// SessionSignals + SessionTelemetry run here, off the router hot path.
520/// Derived timing signals (silence_ms, elapsed_ms, remaining_budget_ms)
521/// are flushed every 100ms via debounced timer.
522pub(crate) fn spawn_telemetry_lane(
523    mut telem_rx: broadcast::Receiver<SessionEvent>,
524    signals: SessionSignals,
525    telemetry: Arc<SessionTelemetry>,
526    cancel: CancellationToken,
527    on_usage: Option<super::callbacks::UsageCallback>,
528) -> tokio::task::JoinHandle<()> {
529    tokio::spawn(async move {
530        let mut debounce = tokio::time::interval(Duration::from_millis(100));
531        // Consume the first immediate tick
532        debounce.tick().await;
533        loop {
534            tokio::select! {
535                biased;
536                result = telem_rx.recv() => {
537                    match result {
538                        Ok(event) => {
539                            // SessionTelemetry: record atomic counters
540                            match &event {
541                                SessionEvent::AudioData(data) => {
542                                    telemetry.record_audio_out(data.len());
543                                }
544                                SessionEvent::TextDelta(_) => {
545                                    telemetry.record_text_out();
546                                }
547                                SessionEvent::VoiceActivityEnd => {
548                                    telemetry.record_vad_end();
549                                }
550                                SessionEvent::Interrupted => {
551                                    telemetry.record_interruption();
552                                }
553                                SessionEvent::TurnComplete => {
554                                    telemetry.record_turn_complete();
555                                }
556                                SessionEvent::VoiceActivityStart => {
557                                    telemetry.mark_turn_start();
558                                }
559                                SessionEvent::Usage(ref usage) => {
560                                    telemetry.record_usage(
561                                        usage.total_token_count,
562                                        usage.prompt_token_count,
563                                        usage.response_token_count,
564                                        usage.cached_content_token_count,
565                                        usage.thoughts_token_count,
566                                    );
567                                    if let Some(cb) = &on_usage {
568                                        cb(usage);
569                                    }
570                                }
571                                _ => {}
572                            }
573                            // SessionSignals: update state keys + atomic timestamps
574                            signals.on_event(&event);
575                        }
576                        Err(broadcast::error::RecvError::Lagged(n)) => {
577                            #[cfg(feature = "tracing-support")]
578                            tracing::warn!(skipped = n, "Telemetry lane lagged");
579                            let _ = n;
580                        }
581                        Err(broadcast::error::RecvError::Closed) => break,
582                    }
583                }
584                _ = debounce.tick() => {
585                    // Flush derived timing signals to state (debounced)
586                    signals.flush_timing();
587                }
588                _ = cancel.cancelled() => break,
589            }
590        }
591    })
592}
593
594/// Routes a SessionEvent to the appropriate lane.
595async fn route_event(
596    event: SessionEvent,
597    fast_tx: &mpsc::Sender<FastEvent>,
598    ctrl_tx: &mpsc::Sender<ControlEvent>,
599    shared: &SharedState,
600) {
601    let delivery = &shared.delivery;
602    let dropped = &shared.dropped;
603    match event {
604        // Fast lane events
605        SessionEvent::AudioData(data) => {
606            deliver_fast(
607                fast_tx,
608                FastEvent::Audio(data),
609                delivery.audio,
610                &dropped.audio,
611            )
612            .await;
613        }
614        SessionEvent::TextDelta(text) => {
615            deliver_fast(fast_tx, FastEvent::Text(text), delivery.text, &dropped.text).await;
616        }
617        SessionEvent::TextComplete(text) => {
618            deliver_fast(
619                fast_tx,
620                FastEvent::TextComplete(text),
621                delivery.text,
622                &dropped.text,
623            )
624            .await;
625        }
626        // Transcripts: fast lane for callbacks, control lane for accumulation.
627        // The control-lane accumulation send keeps its lossless `send().await`.
628        SessionEvent::InputTranscription(text) => {
629            deliver_fast(
630                fast_tx,
631                FastEvent::InputTranscript(text.clone()),
632                delivery.transcript,
633                &dropped.transcript,
634            )
635            .await;
636            let _ = ctrl_tx.send(ControlEvent::InputTranscript(text)).await;
637        }
638        SessionEvent::OutputTranscription(text) => {
639            deliver_fast(
640                fast_tx,
641                FastEvent::OutputTranscript(text.clone()),
642                delivery.transcript,
643                &dropped.transcript,
644            )
645            .await;
646            let _ = ctrl_tx.send(ControlEvent::OutputTranscript(text)).await;
647        }
648        SessionEvent::Thought(text) => {
649            deliver_fast(
650                fast_tx,
651                FastEvent::Thought(text),
652                delivery.thought,
653                &dropped.thought,
654            )
655            .await;
656        }
657        SessionEvent::VoiceActivityStart => {
658            deliver_fast(fast_tx, FastEvent::VadStart, delivery.vad, &dropped.vad).await;
659        }
660        SessionEvent::VoiceActivityEnd => {
661            deliver_fast(fast_tx, FastEvent::VadEnd, delivery.vad, &dropped.vad).await;
662        }
663        SessionEvent::PhaseChanged(phase) => {
664            deliver_fast(
665                fast_tx,
666                FastEvent::Phase(phase),
667                delivery.phase,
668                &dropped.phase,
669            )
670            .await;
671        }
672        SessionEvent::SessionResumeUpdate(info) => {
673            *shared.resume_handle.lock() = Some(info.handle.clone());
674            let _ = ctrl_tx.send(ControlEvent::SessionResumeUpdate(info)).await;
675        }
676        SessionEvent::GenerationComplete => {
677            let _ = ctrl_tx.send(ControlEvent::GenerationComplete).await;
678        }
679
680        // Control lane events
681        SessionEvent::ToolCall(calls) => {
682            let _ = ctrl_tx.send(ControlEvent::ToolCall(calls)).await;
683        }
684        SessionEvent::ToolCallCancelled(ids) => {
685            let _ = ctrl_tx.send(ControlEvent::ToolCallCancelled(ids)).await;
686        }
687        SessionEvent::Interrupted => {
688            // Signal BOTH lanes
689            shared.interrupted.store(true, Ordering::Release);
690            // Cancel any in-flight inline tool dispatch immediately: the
691            // control lane may be blocked awaiting a slow tool and would
692            // otherwise not see this interruption until the tool finished.
693            shared.barge_in.lock().cancel();
694            let _ = fast_tx.send(FastEvent::Interrupted).await;
695            let _ = ctrl_tx.send(ControlEvent::Interrupted).await;
696        }
697        SessionEvent::TurnComplete => {
698            let _ = ctrl_tx.send(ControlEvent::TurnComplete).await;
699        }
700        // Usage metadata is handled by the telemetry lane (SessionSignals)
701        SessionEvent::Usage(_) => {}
702        SessionEvent::GoAway(time_left) => {
703            let _ = ctrl_tx.send(ControlEvent::GoAway(time_left)).await;
704        }
705        SessionEvent::Connected => {
706            let _ = ctrl_tx.send(ControlEvent::Connected).await;
707        }
708        SessionEvent::Disconnected(reason) => {
709            let _ = ctrl_tx.send(ControlEvent::Disconnected(reason)).await;
710        }
711        SessionEvent::Error(err) => {
712            let _ = ctrl_tx.send(ControlEvent::Error(err)).await;
713        }
714        // SessionEvent is #[non_exhaustive]: future wire events the runtime
715        // doesn't understand yet are surfaced (not silently dropped) so
716        // applications on an older runtime can observe them.
717        other => {
718            #[cfg(feature = "tracing-support")]
719            tracing::debug!(?other, "unhandled SessionEvent variant (newer wire event?)");
720            #[cfg(not(feature = "tracing-support"))]
721            let _ = other;
722        }
723    }
724}
725
726/// Fast lane consumer — processes high-frequency events with sync callbacks.
727/// No transcript buffer — transcripts are accumulated exclusively in the control lane.
728async fn run_fast_lane(
729    mut rx: mpsc::Receiver<FastEvent>,
730    callbacks: Arc<EventCallbacks>,
731    shared: Arc<SharedState>,
732    event_tx: broadcast::Sender<LiveEvent>,
733) {
734    while let Some(event) = rx.recv().await {
735        match event {
736            FastEvent::Audio(data) => {
737                // Suppress audio during interruption
738                if !shared.interrupted.load(Ordering::Acquire) {
739                    if let Some(cb) = &callbacks.on_audio {
740                        cb(&data);
741                    }
742                    let _ = event_tx.send(LiveEvent::Audio(data));
743                }
744            }
745            FastEvent::Text(delta) => {
746                if let Some(cb) = &callbacks.on_text {
747                    cb(&delta);
748                }
749                let _ = event_tx.send(LiveEvent::TextDelta(delta));
750            }
751            FastEvent::TextComplete(text) => {
752                if let Some(cb) = &callbacks.on_text_complete {
753                    cb(&text);
754                }
755                let _ = event_tx.send(LiveEvent::TextComplete(text));
756            }
757            FastEvent::InputTranscript(text) => {
758                // Callback only — accumulation happens in control lane
759                if let Some(cb) = &callbacks.on_input_transcript {
760                    cb(&text, false);
761                }
762                let _ = event_tx.send(LiveEvent::InputTranscript {
763                    text,
764                    is_final: false,
765                });
766            }
767            FastEvent::OutputTranscript(text) => {
768                // Callback only — accumulation happens in control lane
769                if let Some(cb) = &callbacks.on_output_transcript {
770                    cb(&text, false);
771                }
772                let _ = event_tx.send(LiveEvent::OutputTranscript {
773                    text,
774                    is_final: false,
775                });
776            }
777            FastEvent::Thought(text) => {
778                if let Some(cb) = &callbacks.on_thought {
779                    cb(&text);
780                }
781                let _ = event_tx.send(LiveEvent::Thought(text));
782            }
783            FastEvent::VadStart => {
784                if let Some(cb) = &callbacks.on_vad_start {
785                    cb();
786                }
787                let _ = event_tx.send(LiveEvent::VadStart);
788            }
789            FastEvent::VadEnd => {
790                if let Some(cb) = &callbacks.on_vad_end {
791                    cb();
792                }
793                let _ = event_tx.send(LiveEvent::VadEnd);
794            }
795            FastEvent::Phase(phase) => {
796                if let Some(cb) = &callbacks.on_phase {
797                    cb(phase);
798                }
799                // Phase is L0-level wire event, not emitted as LiveEvent
800            }
801            FastEvent::Interrupted => {
802                // Audio already suppressed via shared.interrupted flag
803                // Interrupted LiveEvent is emitted from control lane
804            }
805        }
806    }
807}
808
809#[cfg(test)]
810mod tests {
811    use super::*;
812    use std::sync::atomic::AtomicUsize;
813
814    use crate::live::events::LiveEvent;
815    use gemini_genai_rs::prelude::FunctionResponse;
816
817    fn dummy_event_tx() -> broadcast::Sender<LiveEvent> {
818        broadcast::channel::<LiveEvent>(16).0
819    }
820
821    #[test]
822    fn delivery_config_default_is_all_lossless() {
823        let cfg = DeliveryConfig::default();
824        assert_eq!(cfg.audio, Delivery::Lossless);
825        assert_eq!(cfg.text, Delivery::Lossless);
826        assert_eq!(cfg.transcript, Delivery::Lossless);
827        assert_eq!(cfg.thought, Delivery::Lossless);
828        assert_eq!(cfg.vad, Delivery::Lossless);
829        assert_eq!(cfg.phase, Delivery::Lossless);
830        // The standalone Delivery default must also be Lossless.
831        assert_eq!(Delivery::default(), Delivery::Lossless);
832    }
833
834    #[tokio::test]
835    async fn lossy_drop_newest_does_not_block_and_counts_drops() {
836        // Capacity-1 channel that we fill, so the next send would block under
837        // Lossless. The receiver is held but never drains.
838        let (tx, _rx) = mpsc::channel::<FastEvent>(1);
839        tx.send(FastEvent::VadStart).await.unwrap(); // channel now full
840        let dropped = AtomicU64::new(0);
841
842        // Under LossyDropNewest this must return immediately (not block) and
843        // bump the counter. We bound it with a timeout to prove non-blocking.
844        let res = tokio::time::timeout(
845            Duration::from_millis(100),
846            deliver_fast(&tx, FastEvent::VadEnd, Delivery::LossyDropNewest, &dropped),
847        )
848        .await;
849        assert!(res.is_ok(), "deliver_fast blocked under LossyDropNewest");
850        assert_eq!(dropped.load(Ordering::Relaxed), 1);
851    }
852
853    #[tokio::test]
854    async fn lossless_delivers_on_non_full_channel() {
855        let (tx, mut rx) = mpsc::channel::<FastEvent>(4);
856        let dropped = AtomicU64::new(0);
857
858        deliver_fast(
859            &tx,
860            FastEvent::Text("hello".into()),
861            Delivery::Lossless,
862            &dropped,
863        )
864        .await;
865
866        // No drop, and the value arrives on the receiver.
867        assert_eq!(dropped.load(Ordering::Relaxed), 0);
868        match rx.recv().await {
869            Some(FastEvent::Text(s)) => assert_eq!(s, "hello"),
870            other => panic!("expected Text frame, got {:?}", other.is_some()),
871        }
872    }
873
874    #[test]
875    fn dropped_frames_total_sums_classes() {
876        let d = DroppedFrames::default();
877        d.audio.fetch_add(2, Ordering::Relaxed);
878        d.transcript.fetch_add(3, Ordering::Relaxed);
879        assert_eq!(d.total(), 5);
880    }
881
882    #[tokio::test]
883    async fn fast_lane_routes_audio() {
884        let count = Arc::new(AtomicUsize::new(0));
885        let count_clone = count.clone();
886
887        let callbacks = EventCallbacks {
888            on_audio: Some(Box::new(move |_| {
889                count_clone.fetch_add(1, Ordering::SeqCst);
890            })),
891            ..Default::default()
892        };
893        let callbacks = Arc::new(callbacks);
894
895        let (event_tx, _) = broadcast::channel(16);
896        let event_rx = event_tx.subscribe();
897
898        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
899
900        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
901            event_rx,
902            callbacks,
903            None,
904            writer,
905            vec![],
906            State::new(),
907            None,
908            None,
909            None,
910            None,
911            None,
912            std::collections::HashMap::new(),
913            ControlPlaneConfig::default(),
914            dummy_event_tx(),
915        );
916
917        // Send audio events
918        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"audio1")));
919        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"audio2")));
920
921        // Allow tasks to process
922        tokio::time::sleep(Duration::from_millis(50)).await;
923
924        assert_eq!(count.load(Ordering::SeqCst), 2);
925
926        // Cleanup
927        drop(event_tx);
928        let _ = fast_handle.await;
929        let _ = ctrl_handle.await;
930    }
931
932    #[tokio::test]
933    async fn interrupt_suppresses_audio() {
934        let count = Arc::new(AtomicUsize::new(0));
935        let count_clone = count.clone();
936
937        let callbacks = EventCallbacks {
938            on_audio: Some(Box::new(move |_| {
939                count_clone.fetch_add(1, Ordering::SeqCst);
940            })),
941            ..Default::default()
942        };
943        let callbacks = Arc::new(callbacks);
944
945        let (event_tx, _) = broadcast::channel(16);
946        let event_rx = event_tx.subscribe();
947
948        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
949
950        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
951            event_rx,
952            callbacks,
953            None,
954            writer,
955            vec![],
956            State::new(),
957            None,
958            None,
959            None,
960            None,
961            None,
962            std::collections::HashMap::new(),
963            ControlPlaneConfig::default(),
964            dummy_event_tx(),
965        );
966
967        // Send audio, then interrupt, then more audio
968        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"before")));
969        tokio::time::sleep(Duration::from_millis(20)).await;
970        let _ = event_tx.send(SessionEvent::Interrupted);
971        tokio::time::sleep(Duration::from_millis(20)).await;
972        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"during")));
973        tokio::time::sleep(Duration::from_millis(50)).await;
974
975        // At least the first audio was received
976        assert!(count.load(Ordering::SeqCst) >= 1);
977
978        drop(event_tx);
979        let _ = fast_handle.await;
980        let _ = ctrl_handle.await;
981    }
982
983    #[tokio::test]
984    async fn control_lane_routes_turn_complete() {
985        let called = Arc::new(AtomicBool::new(false));
986        let called_clone = called.clone();
987
988        let callbacks = EventCallbacks {
989            on_turn_complete: Some(Arc::new(move || {
990                let c = called_clone.clone();
991                Box::pin(async move {
992                    c.store(true, Ordering::SeqCst);
993                })
994            })),
995            ..Default::default()
996        };
997        let callbacks = Arc::new(callbacks);
998
999        let (event_tx, _) = broadcast::channel(16);
1000        let event_rx = event_tx.subscribe();
1001
1002        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1003
1004        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1005            event_rx,
1006            callbacks,
1007            None,
1008            writer,
1009            vec![],
1010            State::new(),
1011            None,
1012            None,
1013            None,
1014            None,
1015            None,
1016            std::collections::HashMap::new(),
1017            ControlPlaneConfig::default(),
1018            dummy_event_tx(),
1019        );
1020
1021        let _ = event_tx.send(SessionEvent::TurnComplete);
1022        tokio::time::sleep(Duration::from_millis(50)).await;
1023
1024        assert!(called.load(Ordering::SeqCst));
1025
1026        drop(event_tx);
1027        let _ = fast_handle.await;
1028        let _ = ctrl_handle.await;
1029    }
1030
1031    #[tokio::test]
1032    async fn transcript_accumulates_in_control_lane() {
1033        let callbacks = Arc::new(EventCallbacks::default());
1034
1035        let (event_tx, _) = broadcast::channel(16);
1036        let event_rx = event_tx.subscribe();
1037
1038        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1039
1040        let state = State::new();
1041        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1042            event_rx,
1043            callbacks,
1044            None,
1045            writer,
1046            vec![],
1047            state.clone(),
1048            None,
1049            None,
1050            None,
1051            None,
1052            None,
1053            std::collections::HashMap::new(),
1054            ControlPlaneConfig::default(),
1055            dummy_event_tx(),
1056        );
1057
1058        // Send transcripts
1059        let _ = event_tx.send(SessionEvent::InputTranscription("Hello ".to_string()));
1060        let _ = event_tx.send(SessionEvent::InputTranscription("world".to_string()));
1061        let _ = event_tx.send(SessionEvent::OutputTranscription("Hi there!".to_string()));
1062        tokio::time::sleep(Duration::from_millis(50)).await;
1063
1064        // End turn
1065        let _ = event_tx.send(SessionEvent::TurnComplete);
1066        tokio::time::sleep(Duration::from_millis(50)).await;
1067
1068        // Turn count should have been incremented
1069        let tc: u32 = state.session().get("turn_count").unwrap_or(0);
1070        assert_eq!(tc, 1);
1071
1072        drop(event_tx);
1073        let _ = fast_handle.await;
1074        let _ = ctrl_handle.await;
1075    }
1076
1077    #[tokio::test]
1078    async fn extractor_runs_on_turn_complete() {
1079        use crate::live::extractor::TurnExtractor;
1080        use crate::live::transcript::TranscriptTurn;
1081        use crate::llm::LlmError;
1082
1083        struct FixedExtractor;
1084
1085        #[async_trait::async_trait]
1086        impl TurnExtractor for FixedExtractor {
1087            fn name(&self) -> &str {
1088                "TestExtractor"
1089            }
1090            fn window_size(&self) -> usize {
1091                3
1092            }
1093            async fn extract(
1094                &self,
1095                _turns: &[TranscriptTurn],
1096            ) -> Result<serde_json::Value, LlmError> {
1097                Ok(serde_json::json!({"score": 0.9, "mood": "happy"}))
1098            }
1099        }
1100
1101        let callbacks = Arc::new(EventCallbacks::default());
1102
1103        let (event_tx, _) = broadcast::channel(16);
1104        let event_rx = event_tx.subscribe();
1105
1106        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1107
1108        let state = State::new();
1109
1110        let extractors: Vec<Arc<dyn TurnExtractor>> = vec![Arc::new(FixedExtractor)];
1111
1112        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1113            event_rx,
1114            callbacks,
1115            None,
1116            writer,
1117            extractors,
1118            state.clone(),
1119            None,
1120            None,
1121            None,
1122            None,
1123            None,
1124            std::collections::HashMap::new(),
1125            ControlPlaneConfig::default(),
1126            dummy_event_tx(),
1127        );
1128
1129        // Produce a turn with content
1130        let _ = event_tx.send(SessionEvent::InputTranscription("hi".to_string()));
1131        tokio::time::sleep(Duration::from_millis(20)).await;
1132        let _ = event_tx.send(SessionEvent::TurnComplete);
1133        tokio::time::sleep(Duration::from_millis(100)).await;
1134
1135        // Check extraction results
1136        let score: Option<f64> = state.get("score");
1137        assert_eq!(score, Some(0.9));
1138        let mood: Option<String> = state.get("mood");
1139        assert_eq!(mood, Some("happy".to_string()));
1140
1141        drop(event_tx);
1142        let _ = fast_handle.await;
1143        let _ = ctrl_handle.await;
1144    }
1145
1146    /// Text pushed through the returned control sender lands on the *user* side
1147    /// of the transcript, so extractors see a typed turn exactly as they see a
1148    /// spoken one.
1149    ///
1150    /// This is the plumbing `LiveHandle::send_text` uses. Before it existed the
1151    /// transcript's user side was written only by `SessionEvent::InputTranscription`
1152    /// (ASR of audio), so a text-driven session handed every extractor an empty
1153    /// user turn.
1154    #[tokio::test]
1155    async fn control_sender_records_text_on_transcript() {
1156        use crate::live::extractor::TurnExtractor;
1157        use crate::live::transcript::TranscriptTurn;
1158        use crate::llm::LlmError;
1159
1160        /// Records the user side of the window it was handed.
1161        struct WindowRecorder(Arc<parking_lot::Mutex<Vec<String>>>);
1162
1163        #[async_trait::async_trait]
1164        impl TurnExtractor for WindowRecorder {
1165            fn name(&self) -> &str {
1166                "WindowRecorder"
1167            }
1168            fn window_size(&self) -> usize {
1169                3
1170            }
1171            async fn extract(
1172                &self,
1173                turns: &[TranscriptTurn],
1174            ) -> Result<serde_json::Value, LlmError> {
1175                self.0.lock().extend(turns.iter().map(|t| t.user.clone()));
1176                Ok(serde_json::json!({}))
1177            }
1178        }
1179
1180        let seen = Arc::new(parking_lot::Mutex::new(Vec::new()));
1181
1182        let (event_tx, _) = broadcast::channel(16);
1183        let event_rx = event_tx.subscribe();
1184        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1185        let extractors: Vec<Arc<dyn TurnExtractor>> = vec![Arc::new(WindowRecorder(seen.clone()))];
1186
1187        let (fast_handle, ctrl_handle, ctrl_tx) = spawn_event_processor(
1188            event_rx,
1189            Arc::new(EventCallbacks::default()),
1190            None,
1191            writer,
1192            extractors,
1193            State::new(),
1194            None,
1195            None,
1196            None,
1197            None,
1198            None,
1199            std::collections::HashMap::new(),
1200            ControlPlaneConfig::default(),
1201            dummy_event_tx(),
1202        );
1203
1204        // No `InputTranscription` event at all — this is the text path.
1205        let tx = ctrl_tx.upgrade().expect("control lane is alive");
1206        tx.send(ControlEvent::InputTranscript("I am pescatarian".into()))
1207            .await
1208            .expect("control lane accepts the typed turn");
1209        drop(tx);
1210
1211        tokio::time::sleep(Duration::from_millis(20)).await;
1212        let _ = event_tx.send(SessionEvent::TurnComplete);
1213        tokio::time::sleep(Duration::from_millis(100)).await;
1214
1215        assert_eq!(
1216            seen.lock().as_slice(),
1217            ["I am pescatarian"],
1218            "typed turn must reach the extractor's transcript window"
1219        );
1220
1221        drop(event_tx);
1222        let _ = fast_handle.await;
1223        let _ = ctrl_handle.await;
1224    }
1225
1226    /// The control sender handed to `LiveHandle` must be *weak*: a strong one
1227    /// would keep the control channel open forever, so the lane would never
1228    /// drain and shut down when the router exits.
1229    #[tokio::test]
1230    async fn control_sender_does_not_keep_lane_alive() {
1231        let (event_tx, _) = broadcast::channel(16);
1232        let event_rx = event_tx.subscribe();
1233        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1234
1235        let (fast_handle, ctrl_handle, ctrl_tx) = spawn_event_processor(
1236            event_rx,
1237            Arc::new(EventCallbacks::default()),
1238            None,
1239            writer,
1240            vec![],
1241            State::new(),
1242            None,
1243            None,
1244            None,
1245            None,
1246            None,
1247            std::collections::HashMap::new(),
1248            ControlPlaneConfig::default(),
1249            dummy_event_tx(),
1250        );
1251
1252        // Hold the handle's sender across shutdown, as `LiveHandle` does.
1253        drop(event_tx);
1254
1255        tokio::time::timeout(Duration::from_secs(2), ctrl_handle)
1256            .await
1257            .expect("control lane must shut down while the handle's sender is held")
1258            .expect("control lane joined cleanly");
1259
1260        assert!(
1261            ctrl_tx.upgrade().is_none(),
1262            "control channel must be closed once the router is gone"
1263        );
1264
1265        let _ = fast_handle.await;
1266    }
1267
1268    #[tokio::test]
1269    async fn telemetry_lane_auto_collects() {
1270        let (event_tx, _) = broadcast::channel(16);
1271        let telem_rx = event_tx.subscribe();
1272
1273        let telemetry = Arc::new(SessionTelemetry::new());
1274        let signals = SessionSignals::new(State::new());
1275        let cancel = CancellationToken::new();
1276
1277        let telem_handle =
1278            spawn_telemetry_lane(telem_rx, signals, telemetry.clone(), cancel.clone(), None);
1279
1280        // Send events
1281        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"chunk1")));
1282        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"chunk2")));
1283        let _ = event_tx.send(SessionEvent::VoiceActivityEnd);
1284        tokio::time::sleep(Duration::from_millis(50)).await;
1285        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"response")));
1286        tokio::time::sleep(Duration::from_millis(50)).await;
1287
1288        let snap = telemetry.snapshot();
1289        assert_eq!(snap["audio_chunks_out"], 3);
1290        assert!(snap["response_count"].as_u64().unwrap() >= 1);
1291
1292        cancel.cancel();
1293        let _ = telem_handle.await;
1294    }
1295
1296    #[tokio::test]
1297    async fn background_tool_sends_ack_immediately() {
1298        use crate::live::background_tool::{BackgroundToolTracker, ToolExecutionMode};
1299        use crate::tool::{SimpleTool, ToolDispatcher};
1300
1301        // Create a slow tool
1302        let tool = SimpleTool::new(
1303            "slow_search",
1304            "A slow search tool",
1305            Some(serde_json::json!({"type": "object", "properties": {"q": {"type": "string"}}})),
1306            |_args| async move {
1307                tokio::time::sleep(Duration::from_millis(200)).await;
1308                Ok(serde_json::json!({"results": ["found"]}))
1309            },
1310        );
1311
1312        let mut dispatcher = ToolDispatcher::new();
1313        dispatcher.register(tool);
1314
1315        let mut execution_modes = std::collections::HashMap::new();
1316        execution_modes.insert(
1317            "slow_search".to_string(),
1318            ToolExecutionMode::Background {
1319                formatter: None,
1320                scheduling: None,
1321            },
1322        );
1323
1324        let sent = Arc::new(parking_lot::Mutex::new(Vec::<Vec<FunctionResponse>>::new()));
1325        let sent_clone = sent.clone();
1326
1327        // Use a writer that records sent tool responses
1328        struct RecordingWriter {
1329            sent: Arc<parking_lot::Mutex<Vec<Vec<FunctionResponse>>>>,
1330        }
1331
1332        #[async_trait::async_trait]
1333        impl SessionWriter for RecordingWriter {
1334            async fn send_audio(
1335                &self,
1336                _data: Vec<u8>,
1337            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1338                Ok(())
1339            }
1340            async fn send_text(
1341                &self,
1342                _text: String,
1343            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1344                Ok(())
1345            }
1346            async fn send_video(
1347                &self,
1348                _data: Vec<u8>,
1349            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1350                Ok(())
1351            }
1352            async fn send_tool_response(
1353                &self,
1354                responses: Vec<FunctionResponse>,
1355            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1356                self.sent.lock().push(responses);
1357                Ok(())
1358            }
1359            async fn update_instruction(
1360                &self,
1361                _instruction: String,
1362            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1363                Ok(())
1364            }
1365            async fn send_client_content(
1366                &self,
1367                _content: Vec<gemini_genai_rs::prelude::Content>,
1368                _turn_complete: bool,
1369            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1370                Ok(())
1371            }
1372            async fn signal_activity_start(
1373                &self,
1374            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1375                Ok(())
1376            }
1377            async fn signal_activity_end(
1378                &self,
1379            ) -> Result<(), gemini_genai_rs::session::SessionError> {
1380                Ok(())
1381            }
1382            async fn disconnect(&self) -> Result<(), gemini_genai_rs::session::SessionError> {
1383                Ok(())
1384            }
1385        }
1386
1387        let writer: Arc<dyn SessionWriter> = Arc::new(RecordingWriter { sent: sent_clone });
1388        let callbacks = Arc::new(EventCallbacks::default());
1389        let tracker = Arc::new(BackgroundToolTracker::new());
1390
1391        let (event_tx, _) = broadcast::channel(16);
1392        let event_rx = event_tx.subscribe();
1393
1394        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1395            event_rx,
1396            callbacks,
1397            Some(Arc::new(dispatcher)),
1398            writer,
1399            vec![],
1400            State::new(),
1401            None,
1402            None,
1403            None,
1404            None,
1405            Some(tracker.clone()),
1406            execution_modes,
1407            ControlPlaneConfig::default(),
1408            dummy_event_tx(),
1409        );
1410
1411        // Send a tool call
1412        let _ = event_tx.send(SessionEvent::ToolCall(vec![
1413            gemini_genai_rs::prelude::FunctionCall {
1414                name: "slow_search".to_string(),
1415                args: serde_json::json!({"q": "test"}),
1416                id: Some("fc_1".to_string()),
1417            },
1418        ]));
1419
1420        // Wait just enough for the ack (but not the full tool)
1421        tokio::time::sleep(Duration::from_millis(50)).await;
1422
1423        // Scope the guard so it is never held across an await point.
1424        {
1425            let responses = sent.lock();
1426            // First batch should be the ack
1427            assert!(!responses.is_empty(), "Should have sent ack immediately");
1428            assert_eq!(responses[0][0].response["status"], "running");
1429        }
1430
1431        // Wait for background tool to complete
1432        tokio::time::sleep(Duration::from_millis(300)).await;
1433
1434        {
1435            let responses = sent.lock();
1436            // Second batch should be the completed result
1437            assert!(
1438                responses.len() >= 2,
1439                "Should have sent result after completion"
1440            );
1441            assert_eq!(responses[1][0].response["status"], "completed");
1442        }
1443
1444        drop(event_tx);
1445        let _ = fast_handle.await;
1446        let _ = ctrl_handle.await;
1447    }
1448
1449    #[tokio::test]
1450    async fn callback_mode_blocking_awaits_inline() {
1451        use crate::live::callbacks::CallbackMode;
1452        use std::sync::atomic::AtomicU32;
1453
1454        let order = Arc::new(AtomicU32::new(0));
1455        let order_clone = order.clone();
1456
1457        let callbacks = EventCallbacks {
1458            // Blocking on_turn_complete sets order to 1
1459            on_turn_complete: Some(Arc::new(move || {
1460                let o = order_clone.clone();
1461                Box::pin(async move {
1462                    // Simulate brief work
1463                    tokio::time::sleep(Duration::from_millis(10)).await;
1464                    o.store(1, Ordering::SeqCst);
1465                })
1466            })),
1467            on_turn_complete_mode: CallbackMode::Blocking,
1468            ..Default::default()
1469        };
1470        let callbacks = Arc::new(callbacks);
1471
1472        let (event_tx, _) = broadcast::channel(16);
1473        let event_rx = event_tx.subscribe();
1474
1475        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1476
1477        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1478            event_rx,
1479            callbacks,
1480            None,
1481            writer,
1482            vec![],
1483            State::new(),
1484            None,
1485            None,
1486            None,
1487            None,
1488            None,
1489            std::collections::HashMap::new(),
1490            ControlPlaneConfig::default(),
1491            dummy_event_tx(),
1492        );
1493
1494        let _ = event_tx.send(SessionEvent::TurnComplete);
1495        tokio::time::sleep(Duration::from_millis(100)).await;
1496
1497        // Blocking mode: callback completed before control lane processed next event
1498        assert_eq!(order.load(Ordering::SeqCst), 1);
1499
1500        drop(event_tx);
1501        let _ = fast_handle.await;
1502        let _ = ctrl_handle.await;
1503    }
1504
1505    #[tokio::test]
1506    async fn interruption_beats_slow_inline_tool() {
1507        use crate::tool::{SimpleTool, ToolDispatcher};
1508
1509        // A slow inline tool that blocks the control lane for 5s.
1510        let mut dispatcher = ToolDispatcher::new();
1511        dispatcher.register(SimpleTool::new("slow", "slow", None, |_args| async move {
1512            tokio::time::sleep(Duration::from_secs(5)).await;
1513            Ok(serde_json::json!({"done": true}))
1514        }));
1515
1516        let interrupted_at = Arc::new(parking_lot::Mutex::new(None::<std::time::Instant>));
1517        let flag = interrupted_at.clone();
1518        let callbacks = EventCallbacks {
1519            on_interrupted: Some(Arc::new(move || {
1520                let flag = flag.clone();
1521                Box::pin(async move {
1522                    *flag.lock() = Some(std::time::Instant::now());
1523                })
1524            })),
1525            ..Default::default()
1526        };
1527
1528        let (event_tx, _) = broadcast::channel(16);
1529        let event_rx = event_tx.subscribe();
1530        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1531
1532        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1533            event_rx,
1534            Arc::new(callbacks),
1535            Some(Arc::new(dispatcher)),
1536            writer,
1537            vec![],
1538            State::new(),
1539            None,
1540            None,
1541            None,
1542            None,
1543            None,
1544            std::collections::HashMap::new(),
1545            ControlPlaneConfig::default(),
1546            dummy_event_tx(),
1547        );
1548
1549        // Tool call starts the 5s dispatch, then the user barges in.
1550        let start = std::time::Instant::now();
1551        let _ = event_tx.send(SessionEvent::ToolCall(vec![
1552            gemini_genai_rs::prelude::FunctionCall {
1553                name: "slow".to_string(),
1554                args: serde_json::json!({}),
1555                id: Some("fc_slow".to_string()),
1556            },
1557        ]));
1558        tokio::time::sleep(Duration::from_millis(50)).await;
1559        let _ = event_tx.send(SessionEvent::Interrupted);
1560
1561        // The interruption must be processed long before the tool's 5s —
1562        // before the fix it queued behind the blocking dispatch.
1563        let mut waited = Duration::ZERO;
1564        while interrupted_at.lock().is_none() && waited < Duration::from_secs(2) {
1565            tokio::time::sleep(Duration::from_millis(25)).await;
1566            waited += Duration::from_millis(25);
1567        }
1568        let fired = (*interrupted_at.lock()).expect("on_interrupted must fire");
1569        assert!(
1570            fired.duration_since(start) < Duration::from_secs(2),
1571            "interruption must not wait for the slow tool"
1572        );
1573
1574        drop(event_tx);
1575        let _ = fast_handle.await;
1576        let _ = ctrl_handle.await;
1577    }
1578
1579    #[tokio::test]
1580    async fn control_lane_exit_persists_final_snapshot_synchronously() {
1581        use crate::live::persistence::{MemoryPersistence, SessionPersistence};
1582
1583        let persistence = Arc::new(MemoryPersistence::new());
1584        let control_plane = ControlPlaneConfig {
1585            persistence: Some(persistence.clone()),
1586            session_id: Some("final-drain".to_string()),
1587            ..Default::default()
1588        };
1589
1590        let (event_tx, _) = broadcast::channel(16);
1591        let event_rx = event_tx.subscribe();
1592        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1593
1594        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1595            event_rx,
1596            Arc::new(EventCallbacks::default()),
1597            None,
1598            writer,
1599            vec![],
1600            State::new(),
1601            None,
1602            None,
1603            None,
1604            None,
1605            None,
1606            std::collections::HashMap::new(),
1607            control_plane,
1608            dummy_event_tx(),
1609        );
1610
1611        // Accumulate state mid-turn — but never reach a TurnComplete, so the
1612        // per-turn (spawn-and-forget) save never fires.
1613        let _ = event_tx.send(SessionEvent::InputTranscription("last words".to_string()));
1614        tokio::time::sleep(Duration::from_millis(50)).await;
1615
1616        // Session ends. The control lane must run a final synchronous save on
1617        // exit; before the fix nothing was ever persisted in this scenario.
1618        drop(event_tx);
1619        let _ = fast_handle.await;
1620        let _ = ctrl_handle.await;
1621
1622        let snap = persistence
1623            .load("final-drain")
1624            .await
1625            .unwrap()
1626            .expect("control-lane exit must persist a final snapshot");
1627        assert_eq!(snap.turn_count, 0);
1628    }
1629
1630    #[tokio::test]
1631    async fn lanes_exit_after_terminal_disconnected_event() {
1632        // The Disconnected event is terminal in L0; the router must exit after
1633        // routing it (dropping its lane senders) so the lanes can drain and
1634        // shut down gracefully — even though the broadcast sender stays alive
1635        // for the LiveHandle's lifetime.
1636        let callbacks = Arc::new(EventCallbacks::default());
1637        let (event_tx, _) = broadcast::channel(16);
1638        let event_rx = event_tx.subscribe();
1639        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1640
1641        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1642            event_rx,
1643            callbacks,
1644            None,
1645            writer,
1646            vec![],
1647            State::new(),
1648            None,
1649            None,
1650            None,
1651            None,
1652            None,
1653            std::collections::HashMap::new(),
1654            ControlPlaneConfig::default(),
1655            dummy_event_tx(),
1656        );
1657
1658        let _ = event_tx.send(SessionEvent::Disconnected(None));
1659
1660        // NOTE: event_tx is intentionally kept alive — before the fix the
1661        // router only exited on channel close, and both awaits below hung.
1662        let joined = tokio::time::timeout(Duration::from_secs(2), async {
1663            let _ = fast_handle.await;
1664            let _ = ctrl_handle.await;
1665        })
1666        .await;
1667        assert!(
1668            joined.is_ok(),
1669            "lanes must exit after the terminal Disconnected event"
1670        );
1671        drop(event_tx);
1672    }
1673
1674    #[tokio::test]
1675    async fn callback_mode_concurrent_spawns_task() {
1676        use crate::live::callbacks::CallbackMode;
1677
1678        let called = Arc::new(AtomicBool::new(false));
1679        let called_clone = called.clone();
1680
1681        let callbacks = EventCallbacks {
1682            on_turn_complete: Some(Arc::new(move || {
1683                let c = called_clone.clone();
1684                Box::pin(async move {
1685                    tokio::time::sleep(Duration::from_millis(10)).await;
1686                    c.store(true, Ordering::SeqCst);
1687                })
1688            })),
1689            on_turn_complete_mode: CallbackMode::Concurrent,
1690            ..Default::default()
1691        };
1692        let callbacks = Arc::new(callbacks);
1693
1694        let (event_tx, _) = broadcast::channel(16);
1695        let event_rx = event_tx.subscribe();
1696
1697        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
1698
1699        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
1700            event_rx,
1701            callbacks,
1702            None,
1703            writer,
1704            vec![],
1705            State::new(),
1706            None,
1707            None,
1708            None,
1709            None,
1710            None,
1711            std::collections::HashMap::new(),
1712            ControlPlaneConfig::default(),
1713            dummy_event_tx(),
1714        );
1715
1716        let _ = event_tx.send(SessionEvent::TurnComplete);
1717        // Give spawned task time to complete
1718        tokio::time::sleep(Duration::from_millis(100)).await;
1719
1720        // Concurrent mode: callback was spawned and eventually completed
1721        assert!(called.load(Ordering::SeqCst));
1722
1723        drop(event_tx);
1724        let _ = fast_handle.await;
1725        let _ = ctrl_handle.await;
1726    }
1727}
gemini_adk_rs/live/processor.rs

gemini_adk_rs/live/
processor.rs
