gemini_adk_rs/live/

processor.rs

//! Three-lane event processor for Live sessions.
//!
//! **Fast lane**: audio, text, VAD (sync callbacks, never blocks)
//! **Control lane**: tool calls, interruptions, lifecycle, transcript accumulation,
//!   extractors, phases, watchers (async callbacks, can block)
//! **Telemetry lane**: SessionSignals state writes, the usage callback and the
//!   100 ms timing flush. The router records SessionTelemetry's atomic counters
//!   inline and forwards only the events that write state, so audio and text
//!   deltas never wake this lane.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;

use bytes::Bytes;
use tokio::sync::{broadcast, mpsc};
use tokio_util::sync::CancellationToken;
use tracing::Instrument;

use gemini_genai_rs::prelude::{SessionEvent, SessionPhase};
use gemini_genai_rs::session::SessionWriter;

use crate::state::State;
use crate::tool::ToolDispatcher;

use super::background_tool::BackgroundToolTracker;
use super::callbacks::EventCallbacks;
use super::computed::ComputedRegistry;
use super::context_writer::PendingContext;
use super::control_plane::run_control_lane;
use super::events::LiveEvent;
use super::extractor::TurnExtractor;
use super::needs::NeedsFulfillment;
use super::persistence::SessionPersistence;
use super::phase::PhaseMachine;
use super::session_signals::SessionSignals;
use super::soft_turn::SoftTurnDetector;
use super::steering::{ContextDelivery, SteeringMode};
use super::telemetry::SessionTelemetry;
use super::temporal::TemporalRegistry;
use super::watcher::WatcherRegistry;

/// Backpressure (delivery) policy for a single class of fast-lane events.
///
/// The event router forwards fast-lane frames (audio, text, transcripts,
/// thoughts, VAD, phase) over a bounded channel to the fast-lane consumer. When
/// that consumer falls behind and the channel fills, the policy decides what the
/// router does — and crucially, whether the router *blocks*. Because the router
/// is shared by both the fast lane and the control lane, a blocking fast-lane
/// send stalls routing for *all* events, including control-lane lifecycle and
/// tool events. The policy lets callers trade frame durability for router
/// responsiveness on a per-class basis.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Delivery {
    /// Never drop a frame: `tx.send(ev).await` — the router awaits when the
    /// channel is full. This is the historical (and default) behavior and is
    /// byte-for-byte identical to the pre-policy code path. Use it when every
    /// frame matters and a slow consumer applying backpressure to the router is
    /// acceptable.
    #[default]
    Lossless,
    /// Drop the *newest* frame on overflow: `tx.try_send(ev)` and, on
    /// [`TrySendError::Full`](tokio::sync::mpsc::error::TrySendError::Full),
    /// discard the just-produced frame and bump a dropped-frame counter. The
    /// router never blocks on this class, so a slow fast-lane consumer can no
    /// longer stall control-lane routing. Use it for high-frequency, loss-
    /// tolerant streams (e.g. partial transcripts, thoughts) where freshness of
    /// already-queued frames matters less than keeping the router moving.
    ///
    /// A drop-oldest / latest-only variant is intentionally *not* provided:
    /// tokio's `mpsc` has no clean "evict the oldest queued item" primitive, so
    /// implementing it correctly would require a custom ring buffer. That is
    /// left as future work rather than shipped half-working.
    LossyDropNewest,
}

/// Per-event-class delivery (backpressure) policy for the fast lane.
///
/// Each fast-lane event class carries its own [`Delivery`] policy. The
/// [`Default`] impl sets **every** class to [`Delivery::Lossless`], which makes
/// the whole feature behavior-preserving: with the default config the router
/// uses the same `send().await` path it always has. Callers opt into lossy
/// behavior per class via the builder setters.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DeliveryConfig {
    /// Policy for raw PCM audio frames.
    pub audio: Delivery,
    /// Policy for non-audio media chunks (Live Avatar video).
    pub media: Delivery,
    /// Policy for incremental text deltas (and text-complete frames).
    pub text: Delivery,
    /// Policy for input/output transcript frames (fast-lane callback copy only;
    /// control-lane accumulation is unaffected and always lossless).
    pub transcript: Delivery,
    /// Policy for thought-summary frames.
    pub thought: Delivery,
    /// Policy for VAD start/end frames.
    pub vad: Delivery,
    /// Policy for phase-changed frames.
    pub phase: Delivery,
}

impl Default for DeliveryConfig {
    fn default() -> Self {
        Self {
            audio: Delivery::Lossless,
            media: Delivery::Lossless,
            text: Delivery::Lossless,
            transcript: Delivery::Lossless,
            thought: Delivery::Lossless,
            vad: Delivery::Lossless,
            phase: Delivery::Lossless,
        }
    }
}

impl DeliveryConfig {
    /// A config with every class set to [`Delivery::Lossless`] (same as
    /// [`Default`]).
    pub fn lossless() -> Self {
        Self::default()
    }

    /// Set the audio policy.
    pub fn audio(mut self, d: Delivery) -> Self {
        self.audio = d;
        self
    }

    /// Set the media (Live Avatar video) policy.
    pub fn media(mut self, d: Delivery) -> Self {
        self.media = d;
        self
    }

    /// Set the text policy.
    pub fn text(mut self, d: Delivery) -> Self {
        self.text = d;
        self
    }

    /// Set the transcript policy.
    pub fn transcript(mut self, d: Delivery) -> Self {
        self.transcript = d;
        self
    }

    /// Set the thought policy.
    pub fn thought(mut self, d: Delivery) -> Self {
        self.thought = d;
        self
    }

    /// Set the VAD policy.
    pub fn vad(mut self, d: Delivery) -> Self {
        self.vad = d;
        self
    }

    /// Set the phase policy.
    pub fn phase(mut self, d: Delivery) -> Self {
        self.phase = d;
        self
    }
}

/// Per-class counters for fast-lane frames dropped under a lossy policy.
///
/// Incremented with a single relaxed atomic add on the router hot path when a
/// [`Delivery::LossyDropNewest`] send overflows. Reads are for observability /
/// tests and never gate the hot path.
#[derive(Debug, Default)]
pub(crate) struct DroppedFrames {
    pub audio: AtomicU64,
    pub media: AtomicU64,
    pub text: AtomicU64,
    pub transcript: AtomicU64,
    pub thought: AtomicU64,
    pub vad: AtomicU64,
    pub phase: AtomicU64,
}

impl DroppedFrames {
    /// Total dropped frames across all classes.
    ///
    /// Currently only consumed by tests; the per-class atomics are read
    /// directly elsewhere. Kept test-gated until a handle accessor surfaces it.
    #[cfg(test)]
    pub(crate) fn total(&self) -> u64 {
        self.audio.load(Ordering::Relaxed)
            + self.media.load(Ordering::Relaxed)
            + self.text.load(Ordering::Relaxed)
            + self.transcript.load(Ordering::Relaxed)
            + self.thought.load(Ordering::Relaxed)
            + self.vad.load(Ordering::Relaxed)
            + self.phase.load(Ordering::Relaxed)
    }
}

/// Forward one fast-lane frame according to its class delivery policy.
///
/// - [`Delivery::Lossless`]: `tx.send(ev).await` — awaits when the channel is
///   full (identical to the pre-policy behavior).
/// - [`Delivery::LossyDropNewest`]: `tx.try_send(ev)` — on a full channel, drop
///   the frame and increment `dropped`.
///
/// Returns without ever blocking the router under a lossy policy.
async fn deliver_fast(
    tx: &mpsc::Sender<FastEvent>,
    ev: FastEvent,
    policy: Delivery,
    dropped: &AtomicU64,
) {
    match policy {
        Delivery::Lossless => {
            let _ = tx.send(ev).await;
        }
        Delivery::LossyDropNewest => {
            if let Err(mpsc::error::TrySendError::Full(_)) = tx.try_send(ev) {
                dropped.fetch_add(1, Ordering::Relaxed);
            }
            // `TrySendError::Closed` is ignored, matching the `let _ = send`
            // pattern used elsewhere (the consumer is gone; nothing to do).
        }
    }
}

/// Events routed to the fast lane (sync processing).
pub(crate) enum FastEvent {
    Audio(Bytes),
    Media(gemini_genai_rs::session::InlineMedia),
    Text(String),
    TextComplete(String),
    InputTranscript(String),
    OutputTranscript(String),
    Thought(String),
    VadStart,
    VadEnd,
    Phase(SessionPhase),
    /// Interruption flag — tells fast lane to stop forwarding audio.
    Interrupted,
    /// Replay lockstep: answered once everything queued before it is handled.
    Barrier(tokio::sync::oneshot::Sender<()>),
}

/// Events routed to the control lane (async processing).
pub(crate) enum ControlEvent {
    /// Replay lockstep: answered once everything queued before it is handled.
    Barrier(tokio::sync::oneshot::Sender<()>),
    ToolCall(Vec<gemini_genai_rs::prelude::FunctionCall>),
    TaskCommand {
        command: crate::tasks::TaskCommand,
        reply: tokio::sync::oneshot::Sender<Result<crate::tasks::TaskSessionSnapshot, String>>,
    },
    TaskCompleted(crate::tasks::InvocationCompletion),
    TaskServicesCompleted(crate::tasks::TaskServiceCompletion),
    TaskTimer,
    ShutdownTasks(tokio::sync::oneshot::Sender<()>),
    ToolCallCancelled(Vec<String>),
    /// A background tool finished. Posted by the detached background task (which
    /// can't reach the synchronous `FlowMonitor`) so the control lane can advance
    /// the governed flow through the same gate as inline tools (#7).
    ToolCompleted {
        /// The tool call's correlation id (for once-per-call_id flow dedup).
        call_id: String,
        /// The tool name (matches `FunctionCall::name`).
        name: String,
        /// Whether the tool completed successfully.
        ok: bool,
    },
    /// The model was interrupted. `heard_chars`: how many of the turn's
    /// output transcript characters the listener heard, when it can be told.
    Interrupted {
        heard_chars: Option<usize>,
    },
    TurnComplete,
    /// Model finished generating (even if interrupted). Fires before TurnComplete.
    GenerationComplete,
    GoAway(Option<std::time::Duration>),
    Connected,
    Disconnected(Option<String>),
    SessionResumeUpdate(gemini_genai_rs::session::ResumeInfo),
    Error(String),
    /// Transcript accumulation — pushed from router, exclusive to control lane.
    InputTranscript(String),
    OutputTranscript(String),
}

/// Shared state between the two lanes.
pub(crate) struct SharedState {
    /// When true, fast lane suppresses audio callbacks.
    pub interrupted: AtomicBool,
    /// Barge-in signal for in-flight inline tool dispatch.
    ///
    /// Cancelled by the ROUTER the moment an `Interrupted` event arrives,
    /// then re-armed (replaced with a fresh token) by the control lane once
    /// it has processed the interruption. The control lane races inline tool
    /// dispatch against this token, so a user barge-in is never stuck waiting
    /// behind a slow tool.
    pub barge_in: parking_lot::Mutex<CancellationToken>,
    /// Latest resume handle from server.
    pub resume_handle: parking_lot::Mutex<Option<String>>,
    /// Last instruction sent via instruction_template (for dedup).
    pub last_instruction: parking_lot::Mutex<Option<String>>,
    /// Steering lines projected on the previous turn boundary, for repeat
    /// suppression.
    ///
    /// Context turns are *appended* to the server-side conversation and cannot
    /// be retracted, so re-sending an unchanged steering line does not
    /// re-emphasise it — it adds a second standing directive the model has to
    /// weigh against everything that follows. Holding the previous turn's lines
    /// lets an unchanged one be dropped while a changed one still lands.
    pub last_context: parking_lot::Mutex<Vec<String>>,
    /// Pending context buffer for deferred delivery (None when Immediate mode).
    pub pending_context: Option<Arc<PendingContext>>,
    /// Fast-lane delivery policy per event class.
    pub delivery: DeliveryConfig,
    /// Per-class counters for frames dropped under a lossy delivery policy.
    pub dropped: DroppedFrames,
    /// Transcript redaction, applied at the router before either lane sees
    /// the text. `None` = pass through untouched.
    pub redactor: Option<Arc<crate::live::redaction::TranscriptRedactor>>,
    /// Where the listener's playback is reported.
    pub playback: super::playback::PlaybackClock,
    /// The current model turn's audio and transcript, for cutting the
    /// transcript to what was heard on an interruption.
    pub speech: parking_lot::Mutex<super::playback::TurnSpeech>,
}

/// Runs the three-lane event processor.
///
/// Returns JoinHandles for the fast consumer and control processor tasks.
/// The telemetry lane is spawned separately via [`spawn_telemetry_lane`] and
/// fed by the router through a [`TelemetryTap`].
/// Configuration for the control plane's new capabilities.
pub(crate) struct ControlPlaneConfig {
    /// The connect-time system instruction, as text.
    ///
    /// What an `instruction_amendment` composes onto when no phase supplies a
    /// base. Amendments are additive by contract, so they need something to be
    /// added to; a session with no phase machine has no phase instruction, and
    /// before this was carried the amendment was computed each turn and then
    /// silently discarded. Sending the amendment alone was not an option —
    /// under `SteeringMode::InstructionUpdate` that replaces the system
    /// instruction, so it would have deleted the caller's own prompt.
    pub base_instruction: Option<String>,
    /// Soft turn detector for proactive silence awareness.
    pub soft_turn: Option<SoftTurnDetector>,
    /// Steering mode for phase instruction delivery.
    pub steering_mode: SteeringMode,
    /// When to deliver context turns to the wire.
    /// Deferred = synchronize with user activity (speech, interruption);
    /// Immediate = send during TurnComplete processing.
    pub context_delivery: ContextDelivery,
    /// Conversation repair tracker.
    pub needs_fulfillment: Option<NeedsFulfillment>,
    /// Session persistence backend.
    pub persistence: Option<Arc<dyn SessionPersistence>>,
    /// Session ID for persistence key.
    pub session_id: Option<String>,
    /// Whether to inject tool availability advisory on phase transitions.
    pub tool_advisory: bool,
    /// Every tool declaration and the names declared now, under
    /// [`SteeringMode::ContextUpdate`] on a model that accepts it. `None`
    /// otherwise: then tools are declared once, at connect.
    pub tool_scope: Option<Arc<super::tool_scope::ToolScope>>,
    /// Every task tool declaration and the names declared now, in a task
    /// session on a model that accepts `contextUpdate`. `None` otherwise:
    /// then every skill's tools are declared at connect.
    pub task_scope: Option<Arc<super::tool_scope::ToolScope>>,
    /// The session's code tool: the runtime runs its programs, each call
    /// down the governed path. `None` without code mode.
    pub code_mode: Option<Arc<crate::code_mode::CodeMode>>,
    /// Shared pending context buffer for deferred delivery (None when Immediate).
    /// Must be the same Arc given to the DeferredWriter so the control lane
    /// can push context and the DeferredWriter can drain it.
    pub pending_context: Option<Arc<PendingContext>>,
    /// Middleware layers run around tool dispatch in the control lane
    /// (`before_tool` / `after_tool` / `on_tool_error`).
    pub middleware: Arc<crate::middleware::MiddlewareChain>,
    /// Optional governed-flow stack (main flow plus digressions): gates tool
    /// calls, projects the active layer's postures into steering, drives
    /// repair from unmet requirements, and suspends/resumes the main flow
    /// around digressions. Shared (`Arc<Mutex<..>>`) so the
    /// [`LiveHandle`](super::handle::LiveHandle) can snapshot `explain` while
    /// the control lane advances it. Lock briefly; never hold the guard across
    /// an `await`.
    pub flow: Option<crate::flow::SharedFlowStack>,
    /// Questions a decision model answers about the conversation, asked at
    /// the turn boundary and before tools are admitted.
    pub decisions: Option<Arc<crate::decision::Decisions>>,
    /// Task runtime, exclusively advanced by the control lane.
    pub tasks: Option<crate::tasks::TaskRuntime>,
    /// Read-only observation cache shared with LiveHandle.
    pub task_status: Option<Arc<parking_lot::Mutex<crate::tasks::TaskSessionSnapshot>>>,
    /// Fast-lane delivery (backpressure) policy per event class. Defaults to
    /// all-`Lossless`, preserving the historical `send().await` behavior.
    pub delivery: DeliveryConfig,
    /// Transcript redaction applied at the router, before either lane sees
    /// the text. `None` = pass through.
    pub redactor: Option<Arc<crate::live::redaction::TranscriptRedactor>>,
    /// Replay only: the router waits for both lanes to finish each event and
    /// counts it here. See [`Lockstep`].
    pub lockstep: Option<Arc<Lockstep>>,
    /// The session's playback clock, shared with its
    /// [`LiveHandle`](super::handle::LiveHandle).
    pub playback: super::playback::PlaybackClock,
}

/// Counts the session events the lanes have fully handled, for an offline
/// replay that must not move its clock past a frame the session is still
/// processing. With lockstep on, the router waits for both lanes to drain
/// each event before taking the next, which a live session never should.
#[derive(Debug, Default)]
pub(crate) struct Lockstep {
    settled: AtomicU64,
    notify: tokio::sync::Notify,
}

impl Lockstep {
    /// Events the lanes have finished with (or the router skipped).
    pub(crate) fn settled(&self) -> u64 {
        self.settled.load(Ordering::Acquire)
    }

    fn settle(&self, n: u64) {
        self.settled.fetch_add(n, Ordering::AcqRel);
        self.notify.notify_waiters();
    }

    /// Wait until at least `count` events are settled, or `timeout` passes.
    pub(crate) async fn wait_for(&self, count: u64, timeout: std::time::Duration) {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            let notified = self.notify.notified();
            if self.settled() >= count {
                return;
            }
            if tokio::time::timeout_at(deadline, notified).await.is_err() {
                return;
            }
        }
    }
}

impl Default for ControlPlaneConfig {
    fn default() -> Self {
        Self {
            base_instruction: None,
            soft_turn: None,
            steering_mode: SteeringMode::default(),
            context_delivery: ContextDelivery::default(),
            needs_fulfillment: None,
            persistence: None,
            session_id: None,
            tool_advisory: true,
            tool_scope: None,
            task_scope: None,
            code_mode: None,
            pending_context: None,
            middleware: Arc::new(crate::middleware::MiddlewareChain::new()),
            flow: None,
            decisions: None,
            tasks: None,
            task_status: None,
            delivery: DeliveryConfig::default(),
            redactor: None,
            lockstep: None,
            playback: super::playback::PlaybackClock::new(crate::clock::system_clock()),
        }
    }
}

#[allow(
    clippy::too_many_arguments,
    reason = "lane spawn site: parameters are the owned subsystem handles split between the fast and control lanes"
)]
pub(crate) fn spawn_event_processor(
    mut event_rx: broadcast::Receiver<SessionEvent>,
    callbacks: Arc<EventCallbacks>,
    dispatcher: Option<Arc<ToolDispatcher>>,
    writer: Arc<dyn SessionWriter>,
    extractors: Vec<Arc<dyn TurnExtractor>>,
    state: State,
    computed: Option<ComputedRegistry>,
    phase_machine: Option<tokio::sync::Mutex<PhaseMachine>>,
    watchers: Option<WatcherRegistry>,
    temporal: Option<Arc<TemporalRegistry>>,
    background_tracker: Option<Arc<BackgroundToolTracker>>,
    execution_modes: std::collections::HashMap<String, super::background_tool::ToolExecutionMode>,
    control_plane: ControlPlaneConfig,
    live_event_tx: broadcast::Sender<LiveEvent>,
    telemetry: Option<TelemetryTap>,
) -> (
    tokio::task::JoinHandle<()>,
    tokio::task::JoinHandle<()>,
    mpsc::WeakSender<ControlEvent>,
) {
    let shared = Arc::new(SharedState {
        interrupted: AtomicBool::new(false),
        barge_in: parking_lot::Mutex::new(CancellationToken::new()),
        resume_handle: parking_lot::Mutex::new(None),
        last_instruction: parking_lot::Mutex::new(None),
        // The opening steering was sent at connect: the first turn boundary
        // must not send it again.
        last_context: parking_lot::Mutex::new(super::control_plane::opening_steering(
            &control_plane.flow,
            &state,
        )),
        pending_context: control_plane.pending_context.clone(),
        delivery: control_plane.delivery,
        dropped: DroppedFrames::default(),
        redactor: control_plane.redactor.clone(),
        playback: control_plane.playback.clone(),
        speech: parking_lot::Mutex::default(),
    });

    let timer_cancel = CancellationToken::new();

    // Channels between router and lanes.
    //
    // The control channel matches the fast channel at 512: control events
    // are routed with a lossless `send().await`, so a *full* control queue
    // blocks the shared router — and a blocked router stops forwarding audio
    // frames too, causing playback glitches. Transcript accumulation events
    // (one per ASR chunk) flow through this channel, so a slow control-lane
    // consumer (e.g. a blocking turn-complete pipeline) could realistically
    // fill 64 slots; 512 gives the lane room to fall behind transiently
    // without starving the fast lane.
    let (fast_tx, fast_rx) = mpsc::channel::<FastEvent>(512);
    let (ctrl_tx, ctrl_rx) = mpsc::channel::<ControlEvent>(512);

    // Spawn the router task (reads broadcast, records telemetry counters,
    // routes to lanes). SessionSignals' state writes stay on the telemetry lane.
    let mut telemetry = telemetry;
    let fast_tx_clone = fast_tx.clone();
    let ctrl_tx_clone = ctrl_tx.clone();
    let shared_clone = shared.clone();
    let lockstep = control_plane.lockstep.clone();
    tokio::spawn(
        async move {
            // One span per turn; see `turn_trace` for why each lane keeps its own.
            let mut turn = super::turn_trace::TurnTrace::new();
            loop {
                match event_rx.recv().await {
                    Ok(event) => {
                        // `Disconnected` is terminal in L0 (the session loop returns
                        // after emitting it), so the router exits after routing it.
                        // Dropping the router's lane senders closes the fast/control
                        // channels, letting both lanes drain their queues and shut
                        // down gracefully (final persistence drain, etc.) instead of
                        // idling forever on a broadcast channel that never closes
                        // while the `SessionHandle` is alive.
                        let terminal = matches!(event, SessionEvent::Disconnected(_));
                        let boundary = matches!(event, SessionEvent::TurnComplete);
                        if let Some(tap) = telemetry.as_mut() {
                            turn.span().in_scope(|| tap.observe(&event));
                        }
                        route_event(event, &fast_tx_clone, &ctrl_tx_clone, &shared_clone)
                            .instrument(turn.span())
                            .await;
                        if let Some(lockstep) = &lockstep {
                            let (fast_done, fast_rx) = tokio::sync::oneshot::channel();
                            let (ctrl_done, ctrl_rx) = tokio::sync::oneshot::channel();
                            let _ = fast_tx_clone.send(FastEvent::Barrier(fast_done)).await;
                            let _ = ctrl_tx_clone.send(ControlEvent::Barrier(ctrl_done)).await;
                            let _ = fast_rx.await;
                            let _ = ctrl_rx.await;
                            lockstep.settle(1);
                        }
                        if boundary {
                            turn.advance();
                            tracing::debug!(parent: &turn.span(), "turn started");
                        }
                        if terminal {
                            break;
                        }
                    }
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(skipped = n, "Event processor lagged, skipped events");
                        if let Some(lockstep) = &lockstep {
                            lockstep.settle(n);
                        }
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                }
            }
        }
        .in_current_span(),
    );

    // Spawn fast consumer (no transcript buffer — transcripts are in control lane)
    let fast_callbacks = callbacks.clone();
    let fast_shared = shared.clone();
    let fast_event_tx = live_event_tx.clone();
    let fast_handle = tokio::spawn(
        async move {
            run_fast_lane(fast_rx, fast_callbacks, fast_shared, fast_event_tx).await;
        }
        .in_current_span(),
    );

    // Clone for the timer task (before moving into ctrl spawn)
    let timer_temporal = temporal.clone();
    let task_timer = control_plane.tasks.is_some();
    // Reprompt-on-silence runs only when some stage asks for it.
    let reprompt = control_plane.flow.as_ref().is_some_and(|stack| {
        stack
            .lock()
            .timing_policies()
            .values()
            .any(|t| t.reprompt_after_ms.is_some())
    });
    let reprompt_inputs = reprompt.then(|| (state.clone(), writer.clone(), live_event_tx.clone()));
    let timer_state = state.clone();
    let timer_writer = writer.clone();

    // Spawn control processor (owns TranscriptBuffer exclusively — no mutex needed)
    let ctrl_callbacks = callbacks;
    let ctrl_shared = shared;
    let ctrl_timer_cancel = timer_cancel.clone();
    // Weak sender handed to the control lane so background tool tasks can post
    // completions back without keeping the channel open on shutdown (the lane
    // upgrades it per background spawn; the channel closes once the router and
    // all in-flight background tasks drop their strong senders).
    let ctrl_tx_weak = ctrl_tx.downgrade();
    let ctrl_handle = tokio::spawn(
        async move {
            run_control_lane(
                ctrl_rx,
                ctrl_tx_weak,
                ctrl_callbacks,
                dispatcher,
                writer,
                ctrl_shared,
                extractors,
                state,
                computed,
                phase_machine,
                watchers,
                temporal,
                background_tracker,
                execution_modes,
                control_plane,
                live_event_tx,
            )
            .await;
            ctrl_timer_cancel.cancel();
        }
        .in_current_span(),
    );

    // Optional timer task for sustained temporal patterns
    if task_timer {
        let sender = ctrl_tx.downgrade();
        let cancel = timer_cancel.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(500));
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            interval.tick().await;
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = interval.tick() => {
                        let Some(sender) = sender.upgrade() else { break };
                        if sender.send(ControlEvent::TaskTimer).await.is_err() {
                            break;
                        }
                    }
                }
            }
        });
    }

    if let Some(ref temporal_ref) = timer_temporal
        && temporal_ref.needs_timer()
    {
        let t = temporal_ref.clone();
        let cancel = timer_cancel.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(500));
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = interval.tick() => {
                        for action in t.check_all(&timer_state, None, &timer_writer) {
                            tokio::spawn(action);
                        }
                    }
                }
            }
        });
    }

    if let Some((state, writer, events)) = reprompt_inputs {
        tokio::spawn(super::reprompt::run_reprompt_timer(
            state,
            writer,
            events,
            timer_cancel.clone(),
        ));
    }

    // Handed to `LiveHandle` so `send_text` can record a typed turn on the
    // transcript. Deliberately *weak*: the control channel must still close when
    // the router and all in-flight background tasks drop their strong senders,
    // otherwise the lane would never drain and shut down (see `ctrl_tx_weak`).
    let ctrl_tx_handle = ctrl_tx.downgrade();

    (fast_handle, ctrl_handle, ctrl_tx_handle)
}

/// What the router hands the telemetry lane.
pub(crate) enum TelemetryMsg {
    /// An event whose [`SessionSignals`] handling writes state.
    Event(SessionEvent),
    /// The turn's response latency, measured by the router on the model's
    /// first output.
    ResponseLatency(Duration),
}

/// How many forwarded events the telemetry lane can fall behind by before
/// the router drops them. Only low-rate events are forwarded.
const TELEMETRY_LANE_CAPACITY: usize = 256;

/// The router's side of telemetry.
///
/// [`SessionTelemetry`] is atomic counters, so the router records them inline,
/// in event order, at a cost of a few atomic operations per event. Events
/// whose [`SessionSignals`] handling writes state go on to the telemetry lane.
/// Audio and text, the high-rate events, only touch an atomic timestamp and
/// are not forwarded: before this, every audio frame woke a separate task,
/// which cost more CPU than the rest of the runtime's per-frame work.
pub(crate) struct TelemetryTap {
    telemetry: Arc<SessionTelemetry>,
    signals: Arc<SessionSignals>,
    lane: mpsc::Sender<TelemetryMsg>,
    /// The turn's latest usage report, added to the totals at its end.
    turn_usage: Option<gemini_genai_rs::prelude::UsageMetadata>,
}

impl TelemetryTap {
    /// A tap feeding a new telemetry lane, and the lane's receiving end for
    /// [`spawn_telemetry_lane`].
    pub(crate) fn new(
        telemetry: Arc<SessionTelemetry>,
        signals: Arc<SessionSignals>,
    ) -> (Self, mpsc::Receiver<TelemetryMsg>) {
        let (lane, rx) = mpsc::channel(TELEMETRY_LANE_CAPACITY);
        let tap = Self {
            telemetry,
            signals,
            lane,
            turn_usage: None,
        };
        (tap, rx)
    }

    /// Record one event's telemetry. Synchronous and cheap; never waits on
    /// the telemetry lane.
    pub(crate) fn observe(&mut self, event: &SessionEvent) {
        match event {
            SessionEvent::AudioData(data) => {
                if let Some(latency) = self.telemetry.record_audio_out(data.len()) {
                    tracing::info!(
                        latency_ms = latency.as_millis() as u64,
                        "first model audio after the user's turn"
                    );
                    self.forward(TelemetryMsg::ResponseLatency(latency));
                }
                self.signals.touch_activity();
                return;
            }
            SessionEvent::TextDelta(_) => {
                if let Some(latency) = self.telemetry.record_text_out() {
                    tracing::info!(
                        latency_ms = latency.as_millis() as u64,
                        "first model text after the user's turn"
                    );
                    self.forward(TelemetryMsg::ResponseLatency(latency));
                }
                self.signals.touch_activity();
                return;
            }
            SessionEvent::TextComplete(_) => {
                self.signals.touch_activity();
                return;
            }
            SessionEvent::VoiceActivityEnd => self.telemetry.record_vad_end(),
            SessionEvent::Interrupted => self.telemetry.record_interruption(),
            SessionEvent::TurnComplete => {
                self.telemetry.record_turn_complete();
                if let Some(usage) = self.turn_usage.take() {
                    self.telemetry.record_turn_usage(&usage);
                }
            }
            SessionEvent::VoiceActivityStart => self.telemetry.mark_turn_start(),
            SessionEvent::Usage(usage) => {
                self.telemetry.record_usage(
                    usage.total_token_count,
                    usage.prompt_token_count,
                    usage.response_token_count,
                    usage.cached_content_token_count,
                    usage.thoughts_token_count,
                );
                self.turn_usage = Some(usage.clone());
            }
            _ => {}
        }
        self.forward(TelemetryMsg::Event(event.clone()));
    }

    fn forward(&self, msg: TelemetryMsg) {
        // A closed lane means the session is shutting down.
        if let Err(mpsc::error::TrySendError::Full(_)) = self.lane.try_send(msg) {
            tracing::warn!("Telemetry lane lagged, dropped a session signal");
        }
    }
}

/// Spawns the telemetry lane: [`SessionSignals`] state writes for the events
/// a [`TelemetryTap`] forwards, the `on_usage` callback, and the derived
/// timing signals (`silence_ms`, `elapsed_ms`, `remaining_budget_ms`),
/// flushed to state every 100 ms.
pub(crate) fn spawn_telemetry_lane(
    mut rx: mpsc::Receiver<TelemetryMsg>,
    signals: Arc<SessionSignals>,
    cancel: CancellationToken,
    on_usage: Option<super::callbacks::UsageCallback>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut debounce = tokio::time::interval(Duration::from_millis(100));
        // Consume the first immediate tick
        debounce.tick().await;
        // One span per turn; see `turn_trace` for why each lane keeps its own.
        let mut turn = super::turn_trace::TurnTrace::new();
        loop {
            tokio::select! {
                biased;
                msg = rx.recv() => match msg {
                    Some(TelemetryMsg::ResponseLatency(latency)) => {
                        signals.record_response_latency(latency);
                    }
                    Some(TelemetryMsg::Event(event)) => {
                        turn.span().in_scope(|| {
                            if let (SessionEvent::Usage(usage), Some(cb)) = (&event, &on_usage) {
                                cb(usage);
                            }
                            signals.on_event(&event);
                        });
                        if matches!(event, SessionEvent::TurnComplete) {
                            turn.advance();
                        }
                    }
                    // The router is gone.
                    None => break,
                },
                _ = debounce.tick() => {
                    // Flush derived timing signals to state (debounced)
                    signals.flush_timing();
                }
                _ = cancel.cancelled() => break,
            }
        }
    }.in_current_span())
}

/// Routes a SessionEvent to the appropriate lane.
async fn route_event(
    event: SessionEvent,
    fast_tx: &mpsc::Sender<FastEvent>,
    ctrl_tx: &mpsc::Sender<ControlEvent>,
    shared: &SharedState,
) {
    let delivery = &shared.delivery;
    let dropped = &shared.dropped;
    match event {
        // Fast lane events
        SessionEvent::AudioData(data) => {
            shared.speech.lock().on_audio(data.len(), &shared.playback);
            deliver_fast(
                fast_tx,
                FastEvent::Audio(data),
                delivery.audio,
                &dropped.audio,
            )
            .await;
        }
        SessionEvent::Media(media) => {
            deliver_fast(
                fast_tx,
                FastEvent::Media(media),
                delivery.media,
                &dropped.media,
            )
            .await;
        }
        SessionEvent::TextDelta(text) => {
            deliver_fast(fast_tx, FastEvent::Text(text), delivery.text, &dropped.text).await;
        }
        SessionEvent::TextComplete(text) => {
            let text = match &shared.redactor {
                Some(redactor) => redactor.redact(text),
                None => text,
            };
            deliver_fast(
                fast_tx,
                FastEvent::TextComplete(text),
                delivery.text,
                &dropped.text,
            )
            .await;
        }
        // Transcripts: fast lane for callbacks, control lane for accumulation.
        // The control-lane accumulation send keeps its lossless `send().await`.
        SessionEvent::InputTranscription(text) => {
            let text = match &shared.redactor {
                Some(redactor) => redactor.redact(text),
                None => text,
            };
            deliver_fast(
                fast_tx,
                FastEvent::InputTranscript(text.clone()),
                delivery.transcript,
                &dropped.transcript,
            )
            .await;
            let _ = ctrl_tx.send(ControlEvent::InputTranscript(text)).await;
        }
        SessionEvent::OutputTranscription(text) => {
            let text = match &shared.redactor {
                Some(redactor) => redactor.redact(text),
                None => text,
            };
            shared.speech.lock().on_text(&text);
            deliver_fast(
                fast_tx,
                FastEvent::OutputTranscript(text.clone()),
                delivery.transcript,
                &dropped.transcript,
            )
            .await;
            let _ = ctrl_tx.send(ControlEvent::OutputTranscript(text)).await;
        }
        SessionEvent::Thought(text) => {
            deliver_fast(
                fast_tx,
                FastEvent::Thought(text),
                delivery.thought,
                &dropped.thought,
            )
            .await;
        }
        SessionEvent::VoiceActivityStart => {
            tracing::debug!("user started speaking");
            deliver_fast(fast_tx, FastEvent::VadStart, delivery.vad, &dropped.vad).await;
        }
        SessionEvent::VoiceActivityEnd => {
            tracing::debug!("user stopped speaking");
            deliver_fast(fast_tx, FastEvent::VadEnd, delivery.vad, &dropped.vad).await;
        }
        SessionEvent::PhaseChanged(phase) => {
            deliver_fast(
                fast_tx,
                FastEvent::Phase(phase),
                delivery.phase,
                &dropped.phase,
            )
            .await;
        }
        SessionEvent::SessionResumeUpdate(info) => {
            *shared.resume_handle.lock() = Some(info.handle.clone());
            let _ = ctrl_tx.send(ControlEvent::SessionResumeUpdate(info)).await;
        }
        SessionEvent::GenerationComplete => {
            let _ = ctrl_tx.send(ControlEvent::GenerationComplete).await;
        }

        // Control lane events
        SessionEvent::ToolCall(calls) => {
            let _ = ctrl_tx.send(ControlEvent::ToolCall(calls)).await;
        }
        SessionEvent::ToolCallCancelled(ids) => {
            let _ = ctrl_tx.send(ControlEvent::ToolCallCancelled(ids)).await;
        }
        SessionEvent::Interrupted => {
            tracing::debug!("user interrupted the model");
            // Signal BOTH lanes
            shared.interrupted.store(true, Ordering::Release);
            // Cancel any in-flight inline tool dispatch immediately: the
            // control lane may be blocked awaiting a slow tool and would
            // otherwise not see this interruption until the tool finished.
            shared.barge_in.lock().cancel();
            // Read the playback clock now, while the listener's speaker is
            // where the interruption found it.
            let heard_chars = shared.speech.lock().on_interrupted(&shared.playback);
            let _ = fast_tx.send(FastEvent::Interrupted).await;
            let _ = ctrl_tx
                .send(ControlEvent::Interrupted { heard_chars })
                .await;
        }
        SessionEvent::TurnComplete => {
            shared.speech.lock().on_turn_complete();
            let _ = ctrl_tx.send(ControlEvent::TurnComplete).await;
        }
        // Usage metadata is handled by the telemetry lane (SessionSignals)
        SessionEvent::Usage(_) => {}
        SessionEvent::GoAway(time_left) => {
            let _ = ctrl_tx.send(ControlEvent::GoAway(time_left)).await;
        }
        SessionEvent::Connected => {
            let _ = ctrl_tx.send(ControlEvent::Connected).await;
        }
        SessionEvent::Disconnected(reason) => {
            let _ = ctrl_tx.send(ControlEvent::Disconnected(reason)).await;
        }
        SessionEvent::Error(err) => {
            let _ = ctrl_tx.send(ControlEvent::Error(err.to_string())).await;
        }
        // SessionEvent is #[non_exhaustive]: future wire events the runtime
        // doesn't understand yet are surfaced (not silently dropped) so
        // applications on an older runtime can observe them.
        other => {
            tracing::debug!(?other, "unhandled SessionEvent variant (newer wire event?)");
        }
    }
}

/// Fast lane consumer — processes high-frequency events with sync callbacks.
/// No transcript buffer — transcripts are accumulated exclusively in the control lane.
async fn run_fast_lane(
    mut rx: mpsc::Receiver<FastEvent>,
    callbacks: Arc<EventCallbacks>,
    shared: Arc<SharedState>,
    event_tx: broadcast::Sender<LiveEvent>,
) {
    while let Some(event) = rx.recv().await {
        match event {
            FastEvent::Barrier(done) => {
                let _ = done.send(());
            }
            FastEvent::Audio(data) => {
                // Suppress audio during interruption
                if !shared.interrupted.load(Ordering::Acquire) {
                    if let Some(cb) = &callbacks.on_audio {
                        cb(&data);
                    }
                    let _ = event_tx.send(LiveEvent::Audio(data));
                }
            }
            FastEvent::Media(media) => {
                // Avatar video is speech made visible: stale after barge-in too.
                if !shared.interrupted.load(Ordering::Acquire) {
                    if let Some(cb) = &callbacks.on_media {
                        cb(&media);
                    }
                    let _ = event_tx.send(LiveEvent::Media(media));
                }
            }
            FastEvent::Text(delta) => {
                if let Some(cb) = &callbacks.on_text {
                    cb(&delta);
                }
                let _ = event_tx.send(LiveEvent::TextDelta(delta));
            }
            FastEvent::TextComplete(text) => {
                if let Some(cb) = &callbacks.on_text_complete {
                    cb(&text);
                }
                let _ = event_tx.send(LiveEvent::TextComplete(text));
            }
            FastEvent::InputTranscript(text) => {
                // Callback only — accumulation happens in control lane
                if let Some(cb) = &callbacks.on_input_transcript {
                    cb(&text, false);
                }
                let _ = event_tx.send(LiveEvent::InputTranscript {
                    text,
                    is_final: false,
                });
            }
            FastEvent::OutputTranscript(text) => {
                // Callback only — accumulation happens in control lane
                if let Some(cb) = &callbacks.on_output_transcript {
                    cb(&text, false);
                }
                let _ = event_tx.send(LiveEvent::OutputTranscript {
                    text,
                    is_final: false,
                });
            }
            FastEvent::Thought(text) => {
                if let Some(cb) = &callbacks.on_thought {
                    cb(&text);
                }
                let _ = event_tx.send(LiveEvent::Thought(text));
            }
            FastEvent::VadStart => {
                if let Some(cb) = &callbacks.on_vad_start {
                    cb();
                }
                let _ = event_tx.send(LiveEvent::VadStart);
            }
            FastEvent::VadEnd => {
                if let Some(cb) = &callbacks.on_vad_end {
                    cb();
                }
                let _ = event_tx.send(LiveEvent::VadEnd);
            }
            FastEvent::Phase(phase) => {
                if let Some(cb) = &callbacks.on_session_phase {
                    cb(phase);
                }
                // Phase is L0-level wire event, not emitted as LiveEvent
            }
            FastEvent::Interrupted => {
                // Audio already suppressed via shared.interrupted flag
                // Interrupted LiveEvent is emitted from control lane
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicUsize;

    use crate::live::events::LiveEvent;
    use gemini_genai_rs::prelude::FunctionResponse;

    fn dummy_event_tx() -> broadcast::Sender<LiveEvent> {
        broadcast::channel::<LiveEvent>(16).0
    }

    #[test]
    fn delivery_config_default_is_all_lossless() {
        let cfg = DeliveryConfig::default();
        assert_eq!(cfg.audio, Delivery::Lossless);
        assert_eq!(cfg.text, Delivery::Lossless);
        assert_eq!(cfg.transcript, Delivery::Lossless);
        assert_eq!(cfg.thought, Delivery::Lossless);
        assert_eq!(cfg.vad, Delivery::Lossless);
        assert_eq!(cfg.phase, Delivery::Lossless);
        // The standalone Delivery default must also be Lossless.
        assert_eq!(Delivery::default(), Delivery::Lossless);
    }

    #[tokio::test]
    async fn lossy_drop_newest_does_not_block_and_counts_drops() {
        // Capacity-1 channel that we fill, so the next send would block under
        // Lossless. The receiver is held but never drains.
        let (tx, _rx) = mpsc::channel::<FastEvent>(1);
        tx.send(FastEvent::VadStart).await.unwrap(); // channel now full
        let dropped = AtomicU64::new(0);

        // Under LossyDropNewest this must return immediately (not block) and
        // bump the counter. We bound it with a timeout to prove non-blocking.
        let res = tokio::time::timeout(
            Duration::from_millis(100),
            deliver_fast(&tx, FastEvent::VadEnd, Delivery::LossyDropNewest, &dropped),
        )
        .await;
        assert!(res.is_ok(), "deliver_fast blocked under LossyDropNewest");
        assert_eq!(dropped.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn lossless_delivers_on_non_full_channel() {
        let (tx, mut rx) = mpsc::channel::<FastEvent>(4);
        let dropped = AtomicU64::new(0);

        deliver_fast(
            &tx,
            FastEvent::Text("hello".into()),
            Delivery::Lossless,
            &dropped,
        )
        .await;

        // No drop, and the value arrives on the receiver.
        assert_eq!(dropped.load(Ordering::Relaxed), 0);
        match rx.recv().await {
            Some(FastEvent::Text(s)) => assert_eq!(s, "hello"),
            other => panic!("expected Text frame, got {:?}", other.is_some()),
        }
    }

    #[test]
    fn dropped_frames_total_sums_classes() {
        let d = DroppedFrames::default();
        d.audio.fetch_add(2, Ordering::Relaxed);
        d.transcript.fetch_add(3, Ordering::Relaxed);
        assert_eq!(d.total(), 5);
    }

    #[tokio::test]
    async fn fast_lane_routes_audio() {
        let count = Arc::new(AtomicUsize::new(0));
        let count_clone = count.clone();

        let callbacks = EventCallbacks {
            on_audio: Some(Box::new(move |_| {
                count_clone.fetch_add(1, Ordering::SeqCst);
            })),
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Send audio events
        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"audio1")));
        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"audio2")));

        // Allow tasks to process
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_eq!(count.load(Ordering::SeqCst), 2);

        // Cleanup
        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn media_chunks_reach_on_media_not_on_audio() {
        let count = Arc::new(AtomicUsize::new(0));
        let count_clone = count.clone();

        let callbacks = EventCallbacks {
            on_media: Some(Box::new(
                move |media: &gemini_genai_rs::session::InlineMedia| {
                    assert!(media.is_video());
                    count_clone.fetch_add(1, Ordering::SeqCst);
                },
            )),
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Live Avatar video chunks
        for _ in 0..2 {
            let _ = event_tx.send(SessionEvent::Media(gemini_genai_rs::session::InlineMedia {
                mime_type: "video/mp4".into(),
                data: Bytes::from_static(b"mp4"),
            }));
        }

        // Allow tasks to process
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_eq!(count.load(Ordering::SeqCst), 2);

        // Cleanup
        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn interrupt_suppresses_audio() {
        let count = Arc::new(AtomicUsize::new(0));
        let count_clone = count.clone();

        let callbacks = EventCallbacks {
            on_audio: Some(Box::new(move |_| {
                count_clone.fetch_add(1, Ordering::SeqCst);
            })),
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Send audio, then interrupt, then more audio
        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"before")));
        tokio::time::sleep(Duration::from_millis(20)).await;
        let _ = event_tx.send(SessionEvent::Interrupted);
        tokio::time::sleep(Duration::from_millis(20)).await;
        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from_static(b"during")));
        tokio::time::sleep(Duration::from_millis(50)).await;

        // At least the first audio was received
        assert!(count.load(Ordering::SeqCst) >= 1);

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn control_lane_routes_turn_complete() {
        let called = Arc::new(AtomicBool::new(false));
        let called_clone = called.clone();

        let callbacks = EventCallbacks {
            on_turn_complete: Some(Arc::new(move || {
                let c = called_clone.clone();
                Box::pin(async move {
                    c.store(true, Ordering::SeqCst);
                })
            })),
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert!(called.load(Ordering::SeqCst));

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn transcript_accumulates_in_control_lane() {
        let callbacks = Arc::new(EventCallbacks::default());

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let state = State::new();
        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            state.clone(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Send transcripts
        let _ = event_tx.send(SessionEvent::InputTranscription("Hello ".to_string()));
        let _ = event_tx.send(SessionEvent::InputTranscription("world".to_string()));
        let _ = event_tx.send(SessionEvent::OutputTranscription("Hi there!".to_string()));
        tokio::time::sleep(Duration::from_millis(50)).await;

        // End turn
        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Turn count should have been incremented
        let tc: u32 = state.session().get("turn_count").unwrap_or(0);
        assert_eq!(tc, 1);

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn an_interrupted_turn_keeps_only_what_was_heard() {
        let finals = Arc::new(parking_lot::Mutex::new(Vec::<String>::new()));
        let sink = finals.clone();
        let callbacks = Arc::new(EventCallbacks {
            on_output_transcript: Some(Box::new(move |text: &str, is_final: bool| {
                if is_final {
                    sink.lock().push(text.to_string());
                }
            })),
            ..EventCallbacks::default()
        });
        let (event_tx, _) = broadcast::channel(64);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
        let clock = Arc::new(crate::clock::ManualClock::new());
        let control_plane = ControlPlaneConfig {
            playback: crate::live::playback::PlaybackClock::new(clock.clone()),
            ..ControlPlaneConfig::default()
        };
        let playback = control_plane.playback.clone();
        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            control_plane,
            dummy_event_tx(),
            None,
        );

        // The model streams 4 s of audio (24 kHz PCM16) and its whole
        // sentence at once; the speaker starts playing it.
        let _ = event_tx.send(SessionEvent::AudioData(Bytes::from(vec![0u8; 4 * 48_000])));
        let _ = event_tx.send(SessionEvent::OutputTranscription(
            "Your table for four is booked for Friday at eight, and I have noted the allergy."
                .to_string(),
        ));
        tokio::time::sleep(Duration::from_millis(50)).await;
        playback.queued(Duration::from_secs(4));
        // The caller cuts in after 1.5 s: 24 characters at 16 a second.
        clock.advance(Duration::from_millis(1_500));
        let _ = event_tx.send(SessionEvent::Interrupted);
        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_eq!(*finals.lock(), ["Your table for four is"]);

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn extractor_runs_on_turn_complete() {
        use crate::live::extractor::TurnExtractor;
        use crate::live::transcript::TranscriptTurn;
        use crate::llm::LlmError;

        struct FixedExtractor;

        #[async_trait::async_trait]
        impl TurnExtractor for FixedExtractor {
            fn name(&self) -> &str {
                "TestExtractor"
            }
            fn window_size(&self) -> usize {
                3
            }
            async fn extract(
                &self,
                _turns: &[TranscriptTurn],
            ) -> Result<serde_json::Value, LlmError> {
                Ok(serde_json::json!({"score": 0.9, "mood": "happy"}))
            }
        }

        let callbacks = Arc::new(EventCallbacks::default());

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let state = State::new();

        let extractors: Vec<Arc<dyn TurnExtractor>> = vec![Arc::new(FixedExtractor)];

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            extractors,
            state.clone(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Produce a turn with content
        let _ = event_tx.send(SessionEvent::InputTranscription("hi".to_string()));
        tokio::time::sleep(Duration::from_millis(20)).await;
        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Check extraction results
        let score: Option<f64> = state.get("score");
        assert_eq!(score, Some(0.9));
        let mood: Option<String> = state.get("mood");
        assert_eq!(mood, Some("happy".to_string()));

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    /// Text pushed through the returned control sender lands on the *user* side
    /// of the transcript, so extractors see a typed turn exactly as they see a
    /// spoken one.
    ///
    /// This is the plumbing `LiveHandle::send_text` uses. Before it existed the
    /// transcript's user side was written only by `SessionEvent::InputTranscription`
    /// (ASR of audio), so a text-driven session handed every extractor an empty
    /// user turn.
    #[tokio::test]
    async fn control_sender_records_text_on_transcript() {
        use crate::live::extractor::TurnExtractor;
        use crate::live::transcript::TranscriptTurn;
        use crate::llm::LlmError;

        /// Records the user side of the window it was handed.
        struct WindowRecorder(Arc<parking_lot::Mutex<Vec<String>>>);

        #[async_trait::async_trait]
        impl TurnExtractor for WindowRecorder {
            fn name(&self) -> &str {
                "WindowRecorder"
            }
            fn window_size(&self) -> usize {
                3
            }
            async fn extract(
                &self,
                turns: &[TranscriptTurn],
            ) -> Result<serde_json::Value, LlmError> {
                self.0.lock().extend(turns.iter().map(|t| t.user.clone()));
                Ok(serde_json::json!({}))
            }
        }

        let seen = Arc::new(parking_lot::Mutex::new(Vec::new()));

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);
        let extractors: Vec<Arc<dyn TurnExtractor>> = vec![Arc::new(WindowRecorder(seen.clone()))];

        let (fast_handle, ctrl_handle, ctrl_tx) = spawn_event_processor(
            event_rx,
            Arc::new(EventCallbacks::default()),
            None,
            writer,
            extractors,
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // No `InputTranscription` event at all — this is the text path.
        let tx = ctrl_tx.upgrade().expect("control lane is alive");
        tx.send(ControlEvent::InputTranscript("I am pescatarian".into()))
            .await
            .expect("control lane accepts the typed turn");
        drop(tx);

        tokio::time::sleep(Duration::from_millis(20)).await;
        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert_eq!(
            seen.lock().as_slice(),
            ["I am pescatarian"],
            "typed turn must reach the extractor's transcript window"
        );

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    /// The control sender handed to `LiveHandle` must be *weak*: a strong one
    /// would keep the control channel open forever, so the lane would never
    /// drain and shut down when the router exits.
    #[tokio::test]
    async fn control_sender_does_not_keep_lane_alive() {
        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, ctrl_tx) = spawn_event_processor(
            event_rx,
            Arc::new(EventCallbacks::default()),
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Hold the handle's sender across shutdown, as `LiveHandle` does.
        drop(event_tx);

        tokio::time::timeout(Duration::from_secs(2), ctrl_handle)
            .await
            .expect("control lane must shut down while the handle's sender is held")
            .expect("control lane joined cleanly");

        assert!(
            ctrl_tx.upgrade().is_none(),
            "control channel must be closed once the router is gone"
        );

        let _ = fast_handle.await;
    }

    #[tokio::test]
    async fn telemetry_tap_counts_inline_and_forwards_only_state_events() {
        let telemetry = Arc::new(SessionTelemetry::new());
        let signals = Arc::new(SessionSignals::new(State::new()));
        let (mut tap, mut lane) = TelemetryTap::new(telemetry.clone(), signals);

        tap.observe(&SessionEvent::AudioData(Bytes::from_static(b"chunk1")));
        tap.observe(&SessionEvent::AudioData(Bytes::from_static(b"chunk2")));
        tap.observe(&SessionEvent::TextDelta("hi".into()));
        tap.observe(&SessionEvent::VoiceActivityEnd);
        tokio::time::sleep(Duration::from_millis(5)).await;
        tap.observe(&SessionEvent::AudioData(Bytes::from_static(b"response")));

        // Counters are recorded synchronously, in event order.
        let snap = telemetry.snapshot();
        assert_eq!(snap["audio_chunks_out"], 3);
        assert_eq!(snap["response_count"], 1);

        // Audio and text never reach the lane: only the state-writing VAD
        // edge and the measured latency do.
        assert!(matches!(
            lane.try_recv(),
            Ok(TelemetryMsg::Event(SessionEvent::VoiceActivityEnd))
        ));
        assert!(matches!(
            lane.try_recv(),
            Ok(TelemetryMsg::ResponseLatency(_))
        ));
        assert!(lane.try_recv().is_err());
    }

    #[tokio::test]
    async fn telemetry_lane_writes_signals_and_reports_usage() {
        let state = State::new();
        let telemetry = Arc::new(SessionTelemetry::new());
        let signals = Arc::new(SessionSignals::new(state.clone()));
        let (mut tap, lane) = TelemetryTap::new(telemetry.clone(), signals.clone());
        let usage_seen = Arc::new(AtomicU64::new(0));
        let seen = usage_seen.clone();
        let on_usage: super::super::callbacks::UsageCallback = Box::new(move |usage| {
            seen.store(
                usage.total_token_count.unwrap_or(0) as u64,
                Ordering::SeqCst,
            );
        });
        let cancel = CancellationToken::new();
        let handle = spawn_telemetry_lane(lane, signals, cancel.clone(), Some(on_usage));

        let usage: gemini_genai_rs::prelude::UsageMetadata =
            serde_json::from_value(serde_json::json!({ "totalTokenCount": 42 })).expect("usage");
        tap.observe(&SessionEvent::VoiceActivityStart);
        tap.observe(&SessionEvent::Usage(usage));
        tap.observe(&SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_eq!(state.session().get::<bool>("is_user_speaking"), Some(true));
        assert_eq!(state.session().get::<u64>("total_token_count"), Some(42));
        assert_eq!(usage_seen.load(Ordering::SeqCst), 42);
        assert_eq!(telemetry.snapshot()["total_token_count"], 42);

        // Dropping the router's tap ends the lane.
        drop(tap);
        tokio::time::timeout(Duration::from_secs(1), handle)
            .await
            .expect("lane exits once the router is gone")
            .expect("lane task");
    }

    #[tokio::test]
    async fn background_tool_sends_ack_immediately() {
        use crate::live::background_tool::{BackgroundToolTracker, ToolExecutionMode};
        use crate::tool::{SimpleTool, ToolDispatcher};

        // Create a slow tool
        let tool = SimpleTool::new(
            "slow_search",
            "A slow search tool",
            Some(serde_json::json!({"type": "object", "properties": {"q": {"type": "string"}}})),
            |_args| async move {
                tokio::time::sleep(Duration::from_millis(200)).await;
                Ok(serde_json::json!({"results": ["found"]}))
            },
        );

        let mut dispatcher = ToolDispatcher::new();
        dispatcher.register(tool);

        let mut execution_modes = std::collections::HashMap::new();
        execution_modes.insert(
            "slow_search".to_string(),
            ToolExecutionMode::Background {
                formatter: None,
                scheduling: None,
            },
        );

        let sent = Arc::new(parking_lot::Mutex::new(Vec::<Vec<FunctionResponse>>::new()));
        let sent_clone = sent.clone();

        // Use a writer that records sent tool responses
        struct RecordingWriter {
            sent: Arc<parking_lot::Mutex<Vec<Vec<FunctionResponse>>>>,
        }

        #[async_trait::async_trait]
        impl SessionWriter for RecordingWriter {
            async fn send_audio(
                &self,
                _data: bytes::Bytes,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn send_text(
                &self,
                _text: String,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn send_video(
                &self,
                _data: bytes::Bytes,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn send_tool_response(
                &self,
                responses: Vec<FunctionResponse>,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                self.sent.lock().push(responses);
                Ok(())
            }
            async fn update_instruction(
                &self,
                _instruction: String,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn send_client_content(
                &self,
                _content: Vec<gemini_genai_rs::prelude::Content>,
                _turn_complete: bool,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn signal_activity_start(
                &self,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn signal_activity_end(
                &self,
            ) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
            async fn disconnect(&self) -> Result<(), gemini_genai_rs::session::SessionError> {
                Ok(())
            }
        }

        let writer: Arc<dyn SessionWriter> = Arc::new(RecordingWriter { sent: sent_clone });
        let callbacks = Arc::new(EventCallbacks::default());
        let tracker = Arc::new(BackgroundToolTracker::new());

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            Some(Arc::new(dispatcher)),
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            Some(tracker.clone()),
            execution_modes,
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Send a tool call
        let _ = event_tx.send(SessionEvent::ToolCall(vec![
            gemini_genai_rs::prelude::FunctionCall {
                name: "slow_search".to_string(),
                args: serde_json::json!({"q": "test"}),
                id: Some("fc_1".to_string()),
            },
        ]));

        // Wait just enough for the ack (but not the full tool)
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Scope the guard so it is never held across an await point.
        {
            let responses = sent.lock();
            // First batch should be the ack
            assert!(!responses.is_empty(), "Should have sent ack immediately");
            assert_eq!(responses[0][0].response["status"], "running");
        }

        // Wait for background tool to complete
        tokio::time::sleep(Duration::from_millis(300)).await;

        {
            let responses = sent.lock();
            // Second batch should be the completed result
            assert!(
                responses.len() >= 2,
                "Should have sent result after completion"
            );
            assert_eq!(responses[1][0].response["status"], "completed");
        }

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn callback_mode_blocking_awaits_inline() {
        use crate::live::ExecutionMode;
        use std::sync::atomic::AtomicU32;

        let order = Arc::new(AtomicU32::new(0));
        let order_clone = order.clone();

        let callbacks = EventCallbacks {
            // Blocking on_turn_complete sets order to 1
            on_turn_complete: Some(Arc::new(move || {
                let o = order_clone.clone();
                Box::pin(async move {
                    // Simulate brief work
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    o.store(1, Ordering::SeqCst);
                })
            })),
            on_turn_complete_mode: ExecutionMode::Blocking,
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        let _ = event_tx.send(SessionEvent::TurnComplete);
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Blocking mode: callback completed before control lane processed next event
        assert_eq!(order.load(Ordering::SeqCst), 1);

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn interruption_beats_slow_inline_tool() {
        use crate::tool::{SimpleTool, ToolDispatcher};

        // A slow inline tool that blocks the control lane for 5s.
        let mut dispatcher = ToolDispatcher::new();
        dispatcher.register(SimpleTool::new("slow", "slow", None, |_args| async move {
            tokio::time::sleep(Duration::from_secs(5)).await;
            Ok(serde_json::json!({"done": true}))
        }));

        let interrupted_at = Arc::new(parking_lot::Mutex::new(None::<std::time::Instant>));
        let flag = interrupted_at.clone();
        let callbacks = EventCallbacks {
            on_interrupted: Some(Arc::new(move || {
                let flag = flag.clone();
                Box::pin(async move {
                    *flag.lock() = Some(std::time::Instant::now());
                })
            })),
            ..Default::default()
        };

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            Arc::new(callbacks),
            Some(Arc::new(dispatcher)),
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        // Tool call starts the 5s dispatch, then the user barges in.
        let start = std::time::Instant::now();
        let _ = event_tx.send(SessionEvent::ToolCall(vec![
            gemini_genai_rs::prelude::FunctionCall {
                name: "slow".to_string(),
                args: serde_json::json!({}),
                id: Some("fc_slow".to_string()),
            },
        ]));
        tokio::time::sleep(Duration::from_millis(50)).await;
        let _ = event_tx.send(SessionEvent::Interrupted);

        // The interruption must be processed long before the tool's 5s —
        // before the fix it queued behind the blocking dispatch.
        let mut waited = Duration::ZERO;
        while interrupted_at.lock().is_none() && waited < Duration::from_secs(2) {
            tokio::time::sleep(Duration::from_millis(25)).await;
            waited += Duration::from_millis(25);
        }
        let fired = (*interrupted_at.lock()).expect("on_interrupted must fire");
        assert!(
            fired.duration_since(start) < Duration::from_secs(2),
            "interruption must not wait for the slow tool"
        );

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }

    #[tokio::test]
    async fn control_lane_exit_persists_final_snapshot_synchronously() {
        use crate::live::persistence::{MemoryPersistence, SessionPersistence};

        let persistence = Arc::new(MemoryPersistence::new());
        let control_plane = ControlPlaneConfig {
            persistence: Some(persistence.clone()),
            session_id: Some("final-drain".to_string()),
            ..Default::default()
        };

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            Arc::new(EventCallbacks::default()),
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            control_plane,
            dummy_event_tx(),
            None,
        );

        // Accumulate state mid-turn — but never reach a TurnComplete, so the
        // per-turn (spawn-and-forget) save never fires.
        let _ = event_tx.send(SessionEvent::InputTranscription("last words".to_string()));
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Session ends. The control lane must run a final synchronous save on
        // exit; before the fix nothing was ever persisted in this scenario.
        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;

        let snap = persistence
            .load("final-drain")
            .await
            .unwrap()
            .expect("control-lane exit must persist a final snapshot");
        assert_eq!(snap.turn_count, 0);
    }

    #[tokio::test]
    async fn lanes_exit_after_terminal_disconnected_event() {
        // The Disconnected event is terminal in L0; the router must exit after
        // routing it (dropping its lane senders) so the lanes can drain and
        // shut down gracefully — even though the broadcast sender stays alive
        // for the LiveHandle's lifetime.
        let callbacks = Arc::new(EventCallbacks::default());
        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();
        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        let _ = event_tx.send(SessionEvent::Disconnected(None));

        // NOTE: event_tx is intentionally kept alive — before the fix the
        // router only exited on channel close, and both awaits below hung.
        let joined = tokio::time::timeout(Duration::from_secs(2), async {
            let _ = fast_handle.await;
            let _ = ctrl_handle.await;
        })
        .await;
        assert!(
            joined.is_ok(),
            "lanes must exit after the terminal Disconnected event"
        );
        drop(event_tx);
    }

    #[tokio::test]
    async fn callback_mode_concurrent_spawns_task() {
        use crate::live::ExecutionMode;

        let called = Arc::new(AtomicBool::new(false));
        let called_clone = called.clone();

        let callbacks = EventCallbacks {
            on_turn_complete: Some(Arc::new(move || {
                let c = called_clone.clone();
                Box::pin(async move {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    c.store(true, Ordering::SeqCst);
                })
            })),
            on_turn_complete_mode: ExecutionMode::Concurrent,
            ..Default::default()
        };
        let callbacks = Arc::new(callbacks);

        let (event_tx, _) = broadcast::channel(16);
        let event_rx = event_tx.subscribe();

        let writer: Arc<dyn SessionWriter> = Arc::new(crate::agent_session::NoOpSessionWriter);

        let (fast_handle, ctrl_handle, _ctrl_tx) = spawn_event_processor(
            event_rx,
            callbacks,
            None,
            writer,
            vec![],
            State::new(),
            None,
            None,
            None,
            None,
            None,
            std::collections::HashMap::new(),
            ControlPlaneConfig::default(),
            dummy_event_tx(),
            None,
        );

        let _ = event_tx.send(SessionEvent::TurnComplete);
        // Give spawned task time to complete
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Concurrent mode: callback was spawned and eventually completed
        assert!(called.load(Ordering::SeqCst));

        drop(event_tx);
        let _ = fast_handle.await;
        let _ = ctrl_handle.await;
    }
}