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

gemini_adk_rs/live/
processor.rs
