gemini_adk_rs/live/

builder.rs

//! LiveSessionBuilder — combines SessionConfig + callbacks + tools into one setup.

use std::collections::HashMap;
use std::sync::Arc;

use tokio_util::sync::CancellationToken;

use gemini_genai_rs::prelude::{ConnectBuilder, SessionConfig, SessionEvent, SessionPhase};
use gemini_genai_rs::session::{SessionError, SessionHandle, SessionWriter};
use gemini_genai_rs::transport::{Transport, TransportConfig};

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

use super::background_tool::{BackgroundToolTracker, ToolExecutionMode};
use super::callbacks::EventCallbacks;
use super::computed::ComputedRegistry;
use super::context_writer::{DeferredWriter, PendingContext};
use super::extractor::TurnExtractor;
use super::handle::LiveHandle;
use super::needs::{NeedsFulfillment, RepairConfig};
use super::persistence::SessionPersistence;
use super::phase::PhaseMachine;
use super::processor::{
    ControlPlaneConfig, TelemetryTap, spawn_event_processor, spawn_telemetry_lane,
};
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::tool_scope::{self, ToolScope};
use super::watcher::WatcherRegistry;

/// Builder for a callback-driven Live session.
///
/// Combines [`SessionConfig`], [`EventCallbacks`], tool dispatching, extractors,
/// computed state, phase machines, watchers, and temporal patterns into a
/// single connection setup. Call [`connect()`](Self::connect) to establish
/// the WebSocket connection and start the three-lane event processor.
///
/// For ergonomic usage, prefer the L2 `Live` builder from `gemini-adk-fluent-rs`
/// which wraps this with a fluent API.
pub struct LiveSessionBuilder {
    config: SessionConfig,
    callbacks: EventCallbacks,
    dispatcher: Option<Arc<ToolDispatcher>>,
    extractors: Vec<Arc<dyn TurnExtractor>>,
    computed: Option<ComputedRegistry>,
    phase_machine: Option<PhaseMachine>,
    watchers: Option<WatcherRegistry>,
    temporal: Option<TemporalRegistry>,
    greeting: Option<String>,
    state: Option<State>,
    execution_modes: HashMap<String, ToolExecutionMode>,
    // Control plane configuration
    soft_turn_timeout: Option<std::time::Duration>,
    steering_mode: SteeringMode,
    /// Whether `steering_mode` was set, rather than left at its default.
    steering_set: bool,
    context_delivery: ContextDelivery,
    delivery: super::processor::DeliveryConfig,
    repair_config: Option<RepairConfig>,
    persistence: Option<Arc<dyn SessionPersistence>>,
    session_id: Option<String>,
    tool_advisory: bool,
    telemetry_interval: Option<std::time::Duration>,
    middleware: Vec<Arc<dyn crate::middleware::Middleware>>,
    flow: Option<crate::flow::FlowStack>,
    decisions: Option<Arc<crate::decision::Decisions>>,
    code_mode: Option<Arc<crate::code_mode::CodeMode>>,
    tasks: Option<crate::tasks::TaskRuntime>,
    redactor: Option<Arc<super::redaction::TranscriptRedactor>>,
    clock: Option<crate::clock::SharedClock>,
    lockstep: Option<Arc<super::processor::Lockstep>>,
    event_capacity: usize,
}

impl LiveSessionBuilder {
    /// Create a new builder with the given session config.
    pub fn new(config: SessionConfig) -> Self {
        Self {
            config,
            callbacks: EventCallbacks::default(),
            dispatcher: None,
            extractors: Vec::new(),
            computed: None,
            phase_machine: None,
            watchers: None,
            temporal: None,
            greeting: None,
            state: None,
            execution_modes: HashMap::new(),
            soft_turn_timeout: None,
            steering_mode: SteeringMode::default(),
            steering_set: false,
            context_delivery: ContextDelivery::default(),
            delivery: super::processor::DeliveryConfig::default(),
            repair_config: None,
            persistence: None,
            session_id: None,
            tool_advisory: true,
            telemetry_interval: None,
            middleware: Vec::new(),
            flow: None,
            decisions: None,
            code_mode: None,
            tasks: None,
            redactor: None,
            clock: None,
            lockstep: None,
            event_capacity: super::events::DEFAULT_EVENT_CAPACITY,
        }
    }

    /// Install a catalog of independently owned tasks in this voice session.
    ///
    /// Task commands and completions run on the control lane. A task session
    /// cannot also install session-wide governance or state-changing pipelines.
    pub fn tasks(mut self, tasks: crate::tasks::TaskRuntime) -> Self {
        self.tasks = Some(tasks);
        self
    }

    /// Replay lockstep: the router waits for both lanes to handle each event
    /// and counts it in `lockstep`. Offline replay only.
    pub(crate) fn lockstep(mut self, lockstep: Arc<super::processor::Lockstep>) -> Self {
        self.lockstep = Some(lockstep);
        self
    }

    /// Install transcript redaction — see
    /// [`redaction::TranscriptRedactor`](super::redaction::TranscriptRedactor).
    ///
    /// Applied at the event router, before callbacks, the transcript buffer,
    /// extraction, or persistence see the text. An inactive redactor (no
    /// rules enabled) is dropped rather than installed.
    pub fn redaction(mut self, redactor: super::redaction::TranscriptRedactor) -> Self {
        self.redactor = redactor.is_active().then(|| Arc::new(redactor));
        self
    }

    /// Read the time from `clock` instead of the system clock.
    ///
    /// Installed on the session [`State`] (see
    /// [`State::set_clock`](crate::state::State::set_clock)), so temporal
    /// patterns, phase durations, resolver cache expiry, the `session:`
    /// timing signals and journal timestamps all follow it. Pass a
    /// [`ManualClock`](crate::clock::ManualClock) to make timing decisions
    /// reproducible in tests; replay installs one driven by the recording.
    pub fn clock(mut self, clock: crate::clock::SharedClock) -> Self {
        self.clock = Some(clock);
        self
    }

    /// Add a middleware layer.
    ///
    /// Layers run around tool dispatch in the control lane: `before_tool`
    /// (a returned error vetoes the call), `after_tool`, and `on_tool_error`.
    /// Multiple calls accumulate in order.
    pub fn middleware(mut self, layer: Arc<dyn crate::middleware::Middleware>) -> Self {
        self.middleware.push(layer);
        self
    }

    /// Attach a governed-flow monitor (built from a `Flow` + `Mode`). The
    /// monitor becomes the main layer of a [`FlowStack`](crate::flow::FlowStack)
    /// with no digressions.
    pub fn flow_monitor(self, monitor: crate::flow::FlowMonitor) -> Self {
        self.flow_stack(monitor.into_stack())
    }

    /// Attach a governed-flow stack: the main flow plus its digressions and
    /// repair policies. This is the one governance object the control plane
    /// drives; `flow_monitor` is the no-digression special case.
    pub fn flow_stack(mut self, stack: crate::flow::FlowStack) -> Self {
        self.flow = Some(stack);
        self
    }

    /// The questions a decision model answers about the conversation, for
    /// guards that use the `decided` atom (see [`crate::flow::decided`]).
    ///
    /// At each decision point (the caller's turn ends, or the model calls a
    /// tool) the runtime asks, in one request, the questions the flow can act
    /// on then, plus the standing ones, against the rolling conversation and
    /// the active stages' grounding lines. A tool call is admitted after its
    /// questions are answered for the caller's latest words, and a digression
    /// whose trigger they satisfy opens before the call is admitted.
    pub fn decisions(mut self, decisions: Arc<crate::decision::Decisions>) -> Self {
        self.decisions = Some(decisions);
        self
    }

    /// Provide a pre-created State to use for this session.
    ///
    /// If not set, a new State is created at connect time. Use this when
    /// the State needs to be shared with tools or other components before
    /// the session connects.
    pub fn state(mut self, state: State) -> Self {
        self.state = Some(state);
        self
    }

    /// Set a greeting prompt sent on connect to trigger the model to speak first.
    pub fn greeting(mut self, prompt: impl Into<String>) -> Self {
        self.greeting = Some(prompt.into());
        self
    }

    /// Give the model a code tool: it runs a short Python program whose
    /// functions are the session's tools, each call down the same governed
    /// path as a direct call. See [`code_mode`](crate::code_mode).
    pub fn code_mode(mut self, code: crate::code_mode::CodeMode) -> Self {
        self.code_mode = Some(Arc::new(code));
        self
    }

    /// Set the tool dispatcher for auto-dispatch of tool calls.
    pub fn dispatcher(mut self, dispatcher: ToolDispatcher) -> Self {
        // Add tool declarations to session config
        for tool in dispatcher.to_tool_declarations() {
            self.config = self.config.add_tool(tool);
        }
        self.dispatcher = Some(Arc::new(dispatcher));
        self
    }

    /// Set the event callbacks.
    pub fn callbacks(mut self, callbacks: EventCallbacks) -> Self {
        self.callbacks = callbacks;
        self
    }

    /// Add a turn extractor that runs between turns.
    pub fn extractor(mut self, extractor: Arc<dyn TurnExtractor>) -> Self {
        self.extractors.push(extractor);
        self
    }

    /// Set the computed variable registry for derived state.
    pub fn computed(mut self, registry: ComputedRegistry) -> Self {
        self.computed = Some(registry);
        self
    }

    /// Set the phase machine for declarative conversation phase management.
    pub fn phase_machine(mut self, machine: PhaseMachine) -> Self {
        self.phase_machine = Some(machine);
        self
    }

    /// Set the watcher registry for state change watchers.
    pub fn watchers(mut self, registry: WatcherRegistry) -> Self {
        self.watchers = Some(registry);
        self
    }

    /// Set the temporal pattern registry.
    pub fn temporal(mut self, registry: TemporalRegistry) -> Self {
        self.temporal = Some(registry);
        self
    }

    /// Set the execution mode for a named tool.
    ///
    /// Tools default to [`ToolExecutionMode::Standard`]. Set to
    /// [`ToolExecutionMode::Background`] for zero-dead-air execution.
    pub fn tool_execution_mode(
        mut self,
        tool_name: impl Into<String>,
        mode: ToolExecutionMode,
    ) -> Self {
        self.execution_modes.insert(tool_name.into(), mode);
        self
    }

    /// Enable soft turn detection for proactive silence awareness.
    ///
    /// When `proactiveAudio` is enabled, the model may choose not to respond.
    /// This sets a timeout after VAD end — if the model stays silent, a
    /// lightweight "soft turn" fires to keep state updated without forcing
    /// the model to speak.
    pub fn soft_turn_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.soft_turn_timeout = Some(timeout);
        self
    }

    /// Set the steering mode for how the phase machine delivers instructions.
    ///
    /// A task session (see [`tasks`](Self::tasks)) on a model that accepts
    /// `contextUpdate` declares only the foreground skill's tools unless this
    /// sets a mode other than [`SteeringMode::ContextUpdate`].
    pub fn steering_mode(mut self, mode: SteeringMode) -> Self {
        self.steering_mode = mode;
        self.steering_set = true;
        self
    }

    /// Set the context delivery timing.
    ///
    /// - `Immediate` (default): send batched context during TurnComplete.
    /// - `Deferred`: queue context and flush with next user send.
    pub fn context_delivery(mut self, mode: ContextDelivery) -> Self {
        self.context_delivery = mode;
        self
    }

    /// Set the fast-lane delivery (backpressure) policy per event class.
    ///
    /// Defaults to all-[`Lossless`](super::processor::Delivery::Lossless), which
    /// preserves the historical `send().await` routing behavior. Opt classes
    /// into [`LossyDropNewest`](super::processor::Delivery::LossyDropNewest) to
    /// keep the router from stalling when a fast-lane consumer falls behind.
    pub fn delivery(mut self, delivery: super::processor::DeliveryConfig) -> Self {
        self.delivery = delivery;
        self
    }

    /// Enable the conversation repair protocol.
    ///
    /// Tracks need fulfillment per phase and nudges the model when the
    /// conversation stalls on gathering required information.
    pub fn repair(mut self, config: RepairConfig) -> Self {
        self.repair_config = Some(config);
        self
    }

    /// Set a session persistence backend for surviving process restarts.
    pub fn persistence(mut self, backend: Arc<dyn SessionPersistence>) -> Self {
        self.persistence = Some(backend);
        self
    }

    /// Set the session ID for persistence.
    pub fn session_id(mut self, id: impl Into<String>) -> Self {
        self.session_id = Some(id.into());
        self
    }

    /// Enable or disable tool availability advisory on phase transitions.
    pub fn tool_advisory(mut self, enabled: bool) -> Self {
        self.tool_advisory = enabled;
        self
    }

    /// Set the periodic telemetry emission interval.
    ///
    /// When set, the processor periodically emits `LiveEvent::Telemetry`
    /// and `LiveEvent::TurnMetrics` to the event stream.
    pub fn telemetry_interval(mut self, interval: std::time::Duration) -> Self {
        self.telemetry_interval = Some(interval);
        self
    }

    /// Set how many [`LiveEvent`](super::events::LiveEvent)s a subscriber
    /// can fall behind before it starts skipping the oldest ones.
    ///
    /// The buffer behind [`LiveHandle::events`](super::handle::LiveHandle::events)
    /// is allocated in full when the session starts, at roughly 170 bytes
    /// per slot, whether or not anyone subscribes. The default,
    /// [`DEFAULT_EVENT_CAPACITY`](super::events::DEFAULT_EVENT_CAPACITY), is
    /// enough for a subscriber that hands each event on without waiting.
    /// Raise it when a subscriber does slow work inline, such as pacing
    /// audio to a device from its receive loop; a lagging subscriber gets
    /// `RecvError::Lagged` and skips events. Values below 1 are raised to 1.
    pub fn event_capacity(mut self, capacity: usize) -> Self {
        self.event_capacity = capacity.max(1);
        self
    }

    /// Connect to Gemini and start the three-lane event processor.
    ///
    /// This is a thin orchestrator over three explicit, behavior-preserving
    /// stages:
    ///
    /// 1. `into_plan` — pure derivation/validation of the resolved startup
    ///    configuration (`SessionPlan`); no I/O, no spawning.
    /// 2. Connect the L0 transport using the plan's resolved [`SessionConfig`].
    /// 3. `build_runtime` — assemble the runtime wiring (channels, shared
    ///    state, dispatcher, control plane) from the plan + connected session.
    /// 4. `spawn_lanes` — spawn the telemetry/event/tool lanes and return the
    ///    assembled [`LiveHandle`].
    pub async fn connect(self) -> Result<LiveHandle, AgentError> {
        let mut plan = self.into_plan()?;

        // Connect via L0 using the resolved config (taken out of the plan so
        // the rest of the plan can be moved into the runtime stage).
        let config = plan.config.take().expect("plan always carries a config");
        let session = ConnectBuilder::new(config)
            .connect()
            .await
            .map_err(AgentError::Session)?;
        finish_connect(plan, session).await
    }

    /// [`connect`](Self::connect) over `transport` instead of a WebSocket to
    /// Gemini: the whole runtime (phases, tools, extractors, watchers, flow
    /// governance) runs as it would live, against whatever the transport
    /// answers.
    ///
    /// Use it to test a session without a network or a credential, over a
    /// [`ReplayTransport`](gemini_genai_rs::transport::ReplayTransport)
    /// scripted with server frames. The transport is not reconnected if it
    /// closes.
    pub async fn connect_with_transport<T: Transport>(
        self,
        transport: T,
    ) -> Result<LiveHandle, AgentError> {
        let mut plan = self.into_plan()?;
        let config = plan.config.take().expect("plan always carries a config");
        let session = ConnectBuilder::new(config)
            .transport_config(TransportConfig {
                max_reconnect_attempts: 0,
                ..TransportConfig::default()
            })
            .transport(transport)
            .connect()
            .await
            .map_err(AgentError::Session)?;
        finish_connect(plan, session).await
    }
}

/// Wait for a freshly connected session to become active, then build and
/// start its runtime.
async fn finish_connect(
    plan: SessionPlan,
    session: SessionHandle,
) -> Result<LiveHandle, AgentError> {
    // Wait for Active phase — or for the session to give up trying.
    //
    // `wait_for_phase` alone waits forever: the L0 session loop retries the
    // setup handshake `max_reconnect_attempts` times, then emits
    // `Disconnected` and returns, but the phase watch stays alive because
    // the handle holds it. Active never arrives and nothing ever wakes the
    // waiter — so a permanently unacceptable setup (a retired model name is
    // the common one) wedges `connect()` with no error, forever. Racing the
    // terminal event turns that into a returned failure.
    let mut events = session.subscribe();
    tokio::select! {
        () = session.wait_for_phase(SessionPhase::Active) => {}
        failure = wait_for_connect_failure(&mut events) => {
            return Err(AgentError::Session(SessionError::SetupFailed(
                gemini_genai_rs::session::SetupError::ServerRejected {
                    code: None,
                    message: failure,
                },
            )));
        }
    }
    let runtime = build_runtime(plan, session);
    spawn_lanes(runtime).await
}

impl LiveSessionBuilder {
    /// Derive the resolved [`SessionPlan`] from this builder.
    ///
    /// This is a pure transformation: it runs build-time validations and
    /// resolves the startup configuration (notably applying `NonBlocking`
    /// behavior to background-tool declarations) without performing any I/O or
    /// spawning any tasks. It is unit-testable without a live connection.
    pub(crate) fn into_plan(self) -> Result<SessionPlan, AgentError> {
        // Build-time validations
        if self.tasks.is_some()
            && (self.flow.is_some()
                || self.phase_machine.is_some()
                || self.dispatcher.is_some()
                || !self.config.tools.is_empty()
                || !self.extractors.is_empty()
                || self.computed.is_some()
                || self.watchers.is_some()
                || self
                    .temporal
                    .as_ref()
                    .is_some_and(|temporal| !temporal.is_empty())
                || self.callbacks.on_tool_call.is_some()
                || !self.middleware.is_empty())
        {
            return Err(AgentError::Config(
                "task sessions own their tools, state and governance; configure these inside skills".into(),
            ));
        }
        if let Some(ref pm) = self.phase_machine {
            pm.validate()?;
        }
        if let Some(code) = &self.code_mode {
            if self.tasks.is_some() {
                return Err(AgentError::Config(
                    "a task session cannot have a session-wide code tool".into(),
                ));
            }
            let known: Vec<&str> = self
                .dispatcher
                .as_ref()
                .map(|d| d.names().collect())
                .unwrap_or_default();
            let missing: Vec<&str> = code
                .function_list()
                .iter()
                .map(|f| f.name.as_str())
                .filter(|name| !known.contains(name))
                .collect();
            if !missing.is_empty() {
                return Err(AgentError::Config(format!(
                    "the code tool exposes {} but the session has no tool of that name",
                    missing.join(", ")
                )));
            }
            if known.contains(&code.tool()) {
                return Err(AgentError::Config(format!(
                    "the code tool's name '{}' is also a tool of the session",
                    code.tool()
                )));
            }
        }
        if let Some(ref computed) = self.computed {
            computed.validate()?;
        }

        // Apply NON_BLOCKING behavior to tool declarations for background tools
        let mut config = self.config;
        if let Some(tasks) = &self.tasks {
            config.tools.extend(super::task_tools::declarations(tasks));
            let catalog = super::task_tools::catalog_instruction(tasks);
            let instruction = config.system_instruction.get_or_insert_with(|| {
                gemini_genai_rs::prelude::Content {
                    role: None,
                    parts: Vec::new(),
                }
            });
            instruction
                .parts
                .push(gemini_genai_rs::prelude::Part::Text { text: catalog });
        }
        for (tool_name, mode) in &self.execution_modes {
            if matches!(
                mode,
                super::background_tool::ToolExecutionMode::Background { .. }
            ) {
                for tool in &mut config.tools {
                    if let Some(ref mut decls) = tool.function_declarations {
                        for decl in decls {
                            if decl.name == *tool_name {
                                decl.behavior = Some(
                                    gemini_genai_rs::prelude::FunctionCallingBehavior::NonBlocking,
                                );
                            }
                        }
                    }
                }
            }
        }

        // The instruction the caller set at connect, kept as text so an
        // `instruction_amendment` has something to compose onto in a session
        // with no phase machine. Without it such an amendment was computed and
        // then dropped — see the note in `control_plane::lifecycle` step 10.
        let base_instruction = config.system_instruction.as_ref().map(|content| {
            content
                .parts
                .iter()
                .filter_map(|part| match part {
                    gemini_genai_rs::prelude::Part::Text { text } => Some(text.as_str()),
                    _ => None,
                })
                .collect::<Vec<_>>()
                .join("\n")
        });

        // `ContextUpdate` steering needs a model that accepts the message;
        // elsewhere it would close the session, so steer as the nearest mode.
        let steering_mode = match self.steering_mode {
            SteeringMode::ContextUpdate if !config.supports_context_update() => {
                tracing::warn!(
                    model = %config.resolved_model(),
                    "SteeringMode::ContextUpdate needs a model that accepts contextUpdate; steering as Hybrid"
                );
                SteeringMode::Hybrid
            }
            mode => mode,
        };

        // Under `ContextUpdate` steering the setup already declares only what
        // the initial phase and flow step admit, with the base instruction
        // followed by the initial phase's own. Later changes replace both
        // through `contextUpdate`, so the full declarations are kept here.
        let tool_scope = if steering_mode == SteeringMode::ContextUpdate {
            let state = self.state.clone().unwrap_or_default();
            let phase = self
                .phase_machine
                .as_ref()
                .and_then(PhaseMachine::current_phase);
            let admitted = tool_scope::offered_names(
                &config.tools,
                phase.and_then(|p| p.tools_enabled.as_deref()),
                self.flow.as_ref(),
                &state,
            );
            let scope = ToolScope::new(
                std::mem::take(&mut config.tools),
                base_instruction.clone(),
                &admitted,
            )
            .with_code(self.code_mode.clone());
            config.tools = scope.tools_for(&admitted);
            if let Some(phase) = phase {
                let instruction = scope.instruction_for(
                    &phase
                        .instruction
                        .resolve_with_modifiers(&state, &phase.modifiers),
                );
                if !instruction.is_empty() {
                    config = config.system_instruction(instruction);
                }
            }
            Some(Arc::new(scope))
        } else {
            // Declared once, at connect: the code tool lists every function
            // it exposes.
            if let Some(code) = &self.code_mode {
                config.tools = tool_scope::with_code(std::mem::take(&mut config.tools), code);
            }
            None
        };

        // A task session steers by the foreground task, not by phases: on a
        // model that accepts `contextUpdate` it declares `task_control` and
        // each skill's typed `start_{skill}` at connect, and the task lane
        // re-declares the foreground skill's tools as tasks start, finish and
        // advance.
        let task_scope = match &self.tasks {
            Some(tasks)
                if config.supports_context_update()
                    && (!self.steering_set
                        || self.steering_mode == SteeringMode::ContextUpdate) =>
            {
                let declared = super::task_tools::entry_names(tasks);
                let scope = ToolScope::new(
                    std::mem::take(&mut config.tools),
                    base_instruction.clone(),
                    &declared,
                );
                config.tools = scope.tools_for(&declared);
                Some(Arc::new(scope))
            }
            _ => None,
        };

        Ok(SessionPlan {
            config: Some(config),
            base_instruction,
            tool_scope,
            task_scope,
            code_mode: self.code_mode,
            callbacks: self.callbacks,
            dispatcher: self.dispatcher,
            extractors: self.extractors,
            computed: self.computed,
            phase_machine: self.phase_machine,
            watchers: self.watchers,
            temporal: self.temporal,
            greeting: self.greeting,
            state: self.state,
            execution_modes: self.execution_modes,
            soft_turn_timeout: self.soft_turn_timeout,
            steering_mode,
            context_delivery: self.context_delivery,
            delivery: self.delivery,
            repair_config: self.repair_config,
            persistence: self.persistence,
            session_id: self.session_id,
            tool_advisory: self.tool_advisory,
            telemetry_interval: self.telemetry_interval,
            middleware: self.middleware,
            decisions: {
                warn_unanswerable(self.flow.as_ref(), self.decisions.as_deref());
                self.decisions
            },
            flow: self.flow,
            tasks: self.tasks,
            redactor: self.redactor,
            clock: self.clock,
            lockstep: self.lockstep,
            event_capacity: self.event_capacity,
        })
    }
}

/// Warn about `decided` guards no decision service can answer: their guards
/// never hold.
fn warn_unanswerable(
    flow: Option<&crate::flow::FlowStack>,
    decisions: Option<&crate::decision::Decisions>,
) {
    let Some(flow) = flow else {
        return;
    };
    let mut named = flow.main().flow().decisions();
    for ov in flow.overlays() {
        named.extend(ov.trigger().decisions());
        named.extend(ov.flow().decisions());
    }
    let missing: Vec<&String> = named
        .iter()
        .filter(|q| decisions.is_none_or(|d| d.get(q).is_none()))
        .collect();
    if !missing.is_empty() {
        tracing::warn!(
            ?missing,
            "the flow's guards name decisions no decision service declares; those guards never hold"
        );
    }
}

/// The resolved startup configuration for a Live session.
///
/// Produced purely from a [`LiveSessionBuilder`] via
/// [`into_plan`](LiveSessionBuilder::into_plan) — no I/O, no task spawning. The
/// `config` is held in an `Option` so [`connect`](LiveSessionBuilder::connect)
/// can take it out to open the transport while moving the remaining plan into
/// [`build_runtime`].
pub(crate) struct SessionPlan {
    /// Resolved session config (background-tool `NonBlocking` already applied).
    /// `Some` until the transport is opened; `connect` takes it out.
    config: Option<SessionConfig>,
    /// The connect-time system instruction as text, for amendments to compose
    /// onto when there is no phase to supply a base.
    base_instruction: Option<String>,
    /// Every tool declaration, under `ContextUpdate` steering.
    tool_scope: Option<Arc<ToolScope>>,
    /// Every task tool declaration, in a task session on a model that
    /// accepts `contextUpdate`.
    task_scope: Option<Arc<ToolScope>>,
    /// The session's code tool.
    code_mode: Option<Arc<crate::code_mode::CodeMode>>,
    callbacks: EventCallbacks,
    dispatcher: Option<Arc<ToolDispatcher>>,
    extractors: Vec<Arc<dyn TurnExtractor>>,
    computed: Option<ComputedRegistry>,
    phase_machine: Option<PhaseMachine>,
    watchers: Option<WatcherRegistry>,
    temporal: Option<TemporalRegistry>,
    greeting: Option<String>,
    state: Option<State>,
    execution_modes: HashMap<String, ToolExecutionMode>,
    soft_turn_timeout: Option<std::time::Duration>,
    steering_mode: SteeringMode,
    context_delivery: ContextDelivery,
    delivery: super::processor::DeliveryConfig,
    repair_config: Option<RepairConfig>,
    persistence: Option<Arc<dyn SessionPersistence>>,
    session_id: Option<String>,
    tool_advisory: bool,
    telemetry_interval: Option<std::time::Duration>,
    middleware: Vec<Arc<dyn crate::middleware::Middleware>>,
    flow: Option<crate::flow::FlowStack>,
    decisions: Option<Arc<crate::decision::Decisions>>,
    tasks: Option<crate::tasks::TaskRuntime>,
    redactor: Option<Arc<super::redaction::TranscriptRedactor>>,
    clock: Option<crate::clock::SharedClock>,
    lockstep: Option<Arc<super::processor::Lockstep>>,
    event_capacity: usize,
}

/// Fully wired runtime for a connected Live session, ready for lane spawning.
///
/// Produced by [`build_runtime`] from a [`SessionPlan`] plus the connected
/// [`SessionHandle`]. Holds the channels, shared atomics/state, dispatcher,
/// control-plane config, and the resolved writers — but spawns nothing. The
/// final stage [`spawn_lanes`] consumes this to start the lanes and assemble
/// the [`LiveHandle`].
pub(crate) struct SessionRuntime {
    session: SessionHandle,
    callbacks: Arc<EventCallbacks>,
    dispatcher: Option<Arc<ToolDispatcher>>,
    extractors: Vec<Arc<dyn TurnExtractor>>,
    computed: Option<ComputedRegistry>,
    phase_machine: Option<tokio::sync::Mutex<PhaseMachine>>,
    watchers: Option<WatcherRegistry>,
    temporal: Option<Arc<TemporalRegistry>>,
    greeting: Option<String>,
    state: State,
    execution_modes: HashMap<String, ToolExecutionMode>,
    background_tracker: Arc<BackgroundToolTracker>,
    telemetry: Arc<SessionTelemetry>,
    telemetry_interval: Option<std::time::Duration>,
    control_plane: ControlPlaneConfig,
    pending_context: Option<Arc<PendingContext>>,
    /// Writer used by the processor for internal sends.
    writer: Arc<dyn SessionWriter>,
    /// User-facing writer handed to the `LiveHandle` (and used for greeting).
    user_writer: Arc<dyn SessionWriter>,
    event_rx: tokio::sync::broadcast::Receiver<gemini_genai_rs::prelude::SessionEvent>,
    on_usage_cb: Option<super::callbacks::UsageCallback>,
    live_event_tx: tokio::sync::broadcast::Sender<super::events::LiveEvent>,
    telem_cancel: CancellationToken,
    flow_monitor: Option<crate::flow::SharedFlowStack>,
}

/// Stage 3 input: construct the runtime wiring from a resolved plan and the
/// connected session. Builds channels, shared state, the dispatcher set, the
/// control-plane config (including deferred-context writer wrapping), and the
/// telemetry handle — but does not spawn any lanes.
pub(crate) fn build_runtime(plan: SessionPlan, session: SessionHandle) -> SessionRuntime {
    // Share the governed-flow stack between the control lane (which
    // advances it) and the LiveHandle (which snapshots explain).
    let flow_monitor = plan.flow.map(crate::flow::FlowStack::into_shared);
    let task_status = plan
        .tasks
        .as_ref()
        .map(|tasks| Arc::new(parking_lot::Mutex::new(tasks.snapshot())));
    let mut callbacks = plan.callbacks;
    let on_usage_cb = callbacks.on_usage.take();
    let callbacks = Arc::new(callbacks);
    let raw_writer: Arc<dyn SessionWriter> = Arc::new(session.clone());
    let state = plan.state.unwrap_or_default();
    if let Some(clock) = plan.clock {
        state.set_clock(clock);
    }
    if let Some(stack) = &flow_monitor {
        stack.lock().publish_timing(&state);
    }

    // The router is the runtime's only subscriber; it feeds the telemetry lane.
    let event_rx = session.subscribe();

    // Store initial phase's `needs` metadata for ContextBuilder.
    if let Some(ref pm) = plan.phase_machine {
        let _ = state.session().set("phase", pm.current());
        if let Some(phase) = pm.current_phase()
            && !phase.needs.is_empty()
        {
            let _ = state.set("session:phase_needs", phase.needs.clone());
        }
    }

    // What the setup message declares, for the application and Studio to read.
    if let Some(scope) = plan.tool_scope.as_ref().or(plan.task_scope.as_ref()) {
        let _ = state
            .session()
            .set(super::tool_scope::DECLARED_TOOLS_KEY, scope.declared());
    }

    let phase_machine_mutex = plan.phase_machine.map(|mut machine| {
        machine.set_clock(state.clock());
        tokio::sync::Mutex::new(machine)
    });
    let temporal_arc = plan.temporal.map(Arc::new);
    let background_tracker = Arc::new(BackgroundToolTracker::new());

    // Create telemetry (recorded by the router)
    let telemetry = Arc::new(SessionTelemetry::new());
    let telem_cancel = CancellationToken::new();

    // Build control plane config
    let mut control_plane = ControlPlaneConfig {
        soft_turn: plan.soft_turn_timeout.map(SoftTurnDetector::new),
        steering_mode: plan.steering_mode,
        context_delivery: plan.context_delivery,
        delivery: plan.delivery,
        needs_fulfillment: plan.repair_config.map(NeedsFulfillment::new),
        persistence: plan.persistence,
        session_id: plan.session_id,
        tool_advisory: plan.tool_advisory,
        tool_scope: plan.tool_scope,
        task_scope: plan.task_scope,
        code_mode: plan.code_mode,
        base_instruction: plan.base_instruction,
        pending_context: None, // set after PendingContext is created below
        middleware: {
            let mut chain = crate::middleware::MiddlewareChain::new();
            for layer in plan.middleware {
                chain.add(layer);
            }
            Arc::new(chain)
        },
        flow: flow_monitor.clone(),
        decisions: plan.decisions,
        tasks: plan.tasks,
        task_status,
        redactor: plan.redactor,
        lockstep: plan.lockstep,
        playback: super::playback::PlaybackClock::new(state.clock()),
    };

    // Create shared PendingContext for deferred delivery.
    // The SAME Arc is given to both the DeferredWriter (which drains it before
    // user sends) and the ControlPlaneConfig (which the processor uses to push
    // context turns from the control lane).
    // A stage whose voice timing defers context needs the queue even when the
    // session as a whole delivers immediately.
    let stage_defers = flow_monitor.as_ref().is_some_and(|stack| {
        stack
            .lock()
            .timing_policies()
            .values()
            .any(|t| t.context_delivery == Some(ContextDelivery::Deferred))
    });
    let pending_context = if plan.context_delivery == ContextDelivery::Deferred || stage_defers {
        Some(Arc::new(PendingContext::new()))
    } else {
        None
    };

    // Wrap writer in DeferredWriter if deferred context delivery is enabled.
    let (writer, user_writer) = if let Some(ref pending) = pending_context {
        let deferred: Arc<dyn SessionWriter> =
            Arc::new(DeferredWriter::new(raw_writer.clone(), pending.clone()));
        // Processor uses raw_writer for internal sends (lifecycle context
        // goes through PendingContext, not through the writer directly).
        // User-facing LiveHandle uses the DeferredWriter.
        (raw_writer, deferred)
    } else {
        (raw_writer.clone(), raw_writer)
    };

    // Pass shared pending context to control plane config
    control_plane.pending_context = pending_context.clone();

    // Create LiveEvent broadcast channel
    use super::events::LiveEvent;
    use tokio::sync::broadcast;
    let (live_event_tx, _) = broadcast::channel::<LiveEvent>(plan.event_capacity);

    SessionRuntime {
        session,
        callbacks,
        dispatcher: plan.dispatcher,
        extractors: plan.extractors,
        computed: plan.computed,
        phase_machine: phase_machine_mutex,
        watchers: plan.watchers,
        temporal: temporal_arc,
        greeting: plan.greeting,
        state,
        execution_modes: plan.execution_modes,
        background_tracker,
        telemetry,
        telemetry_interval: plan.telemetry_interval,
        control_plane,
        pending_context,
        writer,
        user_writer,
        event_rx,
        on_usage_cb,
        live_event_tx,
        telem_cancel,
        flow_monitor,
    }
}

/// Stage 4: spawn the telemetry lane, event processor, and periodic telemetry
/// emitter, send any greeting, and assemble the [`LiveHandle`].
/// Resolve once the session has definitively failed to come up, returning a
/// message that names the cause.
///
/// The L0 loop reports each failed setup attempt as `Error` and only says
/// `Disconnected` when it stops retrying, so the last `Error` carries the real
/// reason ("Setup failed: …") while `Disconnected` carries only "max attempts
/// exceeded". Both are reported: the terminal event proves it is over, the
/// retained error says why.
///
/// Pends forever if the session comes up normally — the caller races it against
/// the phase wait.
async fn wait_for_connect_failure(
    events: &mut tokio::sync::broadcast::Receiver<SessionEvent>,
) -> String {
    let mut last_error: Option<String> = None;
    loop {
        match events.recv().await {
            Ok(SessionEvent::Error(err)) => last_error = Some(err.to_string()),
            Ok(SessionEvent::Disconnected(reason)) => {
                let reason = reason.unwrap_or_else(|| "connection closed".to_string());
                return match last_error {
                    Some(err) => format!("{reason} ({err})"),
                    None => reason,
                };
            }
            Ok(_) => {}
            // The sender is gone, which is itself terminal.
            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                return last_error.unwrap_or_else(|| "session ended during setup".to_string());
            }
            Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
        }
    }
}

pub(crate) async fn spawn_lanes(rt: SessionRuntime) -> Result<LiveHandle, AgentError> {
    use super::events::LiveEvent;

    // One root span per session, so a trace backend shows every turn of a
    // conversation under it, keyed by the conversation id (the persistence
    // session id when set, else the transport's). The lanes are spawned
    // inside it and each turn span is its child.
    let conversation_id = rt
        .control_plane
        .session_id
        .clone()
        .unwrap_or_else(|| rt.session.session_id().to_string());
    let session_span = tracing::info_span!(
        "live_session",
        "gen_ai.operation.name" = "live",
        "gen_ai.system" = "gemini",
        "gen_ai.conversation.id" = %conversation_id,
    );
    let entered = session_span.enter();

    // Spawn the telemetry lane. The router records SessionTelemetry inline
    // and forwards only the events whose SessionSignals write state.
    let session_signals = Arc::new(SessionSignals::new(rt.state.clone()));
    let (telemetry_tap, telem_rx) =
        TelemetryTap::new(rt.telemetry.clone(), session_signals.clone());
    let _telem_handle = spawn_telemetry_lane(
        telem_rx,
        session_signals,
        rt.telem_cancel.clone(),
        rt.on_usage_cb,
    );

    // The opening steering goes out before the lanes start and before any
    // greeting, so the model's first turn already has the first stage's
    // posture. Deferred delivery queues it with the first user send instead.
    let opening = super::control_plane::opening_steering(&rt.control_plane.flow, &rt.state);
    if !opening.is_empty() {
        let contents: Vec<_> = opening
            .into_iter()
            .map(gemini_genai_rs::prelude::Content::model)
            .collect();
        let delivery = rt
            .state
            .get::<crate::flow::VoiceTiming>(crate::flow::VOICE_TIMING_KEY)
            .and_then(|t| t.context_delivery)
            .unwrap_or(rt.control_plane.context_delivery);
        match (&delivery, &rt.pending_context) {
            (ContextDelivery::Deferred, Some(pending)) => pending.extend(contents),
            _ => rt
                .writer
                .send_client_content(contents, false)
                .await
                .map_err(AgentError::Session)?,
        }
    }

    // Spawn fast + control lanes (no session_signals, no transcript mutex)
    let greeting_writer = rt.user_writer.clone();
    let playback = rt.control_plane.playback.clone();
    let task_status = rt.control_plane.task_status.clone();
    let (fast_handle, ctrl_handle, ctrl_tx) = spawn_event_processor(
        rt.event_rx,
        rt.callbacks,
        rt.dispatcher,
        rt.writer,
        rt.extractors,
        rt.state.clone(),
        rt.computed,
        rt.phase_machine,
        rt.watchers,
        rt.temporal,
        Some(rt.background_tracker.clone()),
        rt.execution_modes,
        rt.control_plane,
        rt.live_event_tx.clone(),
        Some(telemetry_tap),
    );
    drop(entered);

    // Spawn periodic telemetry emitter if interval is set
    if let Some(interval) = rt.telemetry_interval {
        let telem_tx = rt.live_event_tx.clone();
        let telem_ref = rt.telemetry.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(interval);
            let mut prev_turns = 0u64;
            loop {
                tick.tick().await;
                let snap = telem_ref.snapshot();
                if let Some(obj) = snap.as_object() {
                    let tc = obj
                        .get("turn_count")
                        .or_else(|| obj.get("response_count"))
                        .and_then(serde_json::Value::as_u64)
                        .unwrap_or(0);
                    if tc > prev_turns {
                        let latency = obj
                            .get("last_response_latency_ms")
                            .and_then(serde_json::Value::as_u64)
                            .unwrap_or(0) as u32;
                        let prompt = obj
                            .get("prompt_token_count")
                            .and_then(serde_json::Value::as_u64)
                            .unwrap_or(0) as u32;
                        let response = obj
                            .get("response_token_count")
                            .and_then(serde_json::Value::as_u64)
                            .unwrap_or(0) as u32;
                        let _ = telem_tx.send(LiveEvent::TurnMetrics {
                            turn: tc as u32,
                            latency_ms: latency,
                            prompt_tokens: prompt,
                            response_tokens: response,
                        });
                        prev_turns = tc;
                    }
                }
                if telem_tx.send(LiveEvent::Telemetry(snap)).is_err() {
                    break;
                }
            }
        });
    }

    // Send greeting prompt to trigger model-initiated conversation
    if let Some(greeting) = rt.greeting {
        greeting_writer
            .send_text(greeting)
            .await
            .map_err(AgentError::Session)?;
    }

    Ok(LiveHandle::new(
        rt.session,
        rt.user_writer,
        fast_handle,
        ctrl_handle,
        rt.state,
        rt.telemetry,
        rt.live_event_tx,
        rt.pending_context,
        rt.flow_monitor,
        rt.background_tracker,
        rt.telem_cancel,
    )
    .with_control_sender(ctrl_tx)
    .with_playback(playback)
    .with_task_status(task_status))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn builder_creates_with_defaults() {
        let config = SessionConfig::new("test-key");
        let builder = LiveSessionBuilder::new(config);
        assert!(builder.dispatcher.is_none());
        assert!(builder.computed.is_none());
        assert!(builder.phase_machine.is_none());
        assert!(builder.watchers.is_none());
        assert!(builder.temporal.is_none());
    }

    #[test]
    fn into_plan_derives_defaults() {
        let config = SessionConfig::new("test-key");
        let plan = LiveSessionBuilder::new(config)
            .into_plan()
            .expect("default builder should produce a plan");

        // Config is carried (taken out only at connect time).
        assert!(plan.config.is_some());
        // Defaults preserved.
        assert!(plan.dispatcher.is_none());
        assert!(plan.phase_machine.is_none());
        assert!(plan.persistence.is_none());
        assert!(plan.session_id.is_none());
        assert!(plan.greeting.is_none());
        assert!(plan.soft_turn_timeout.is_none());
        assert!(plan.telemetry_interval.is_none());
        assert!(plan.repair_config.is_none());
        assert!(plan.flow.is_none());
        assert!(plan.execution_modes.is_empty());
        assert!(plan.middleware.is_empty());
        assert_eq!(plan.steering_mode, SteeringMode::default());
        assert_eq!(plan.context_delivery, ContextDelivery::default());
        // Default builder enables tool advisory.
        assert!(plan.tool_advisory);
    }

    #[test]
    fn into_plan_carries_persistence_and_session_id() {
        let config = SessionConfig::new("test-key");
        let plan = LiveSessionBuilder::new(config)
            .session_id("user-123-session-456")
            .into_plan()
            .expect("plan derivation should succeed");

        assert_eq!(plan.session_id.as_deref(), Some("user-123-session-456"));
    }

    #[test]
    fn into_plan_carries_steering_and_context_delivery() {
        let config = SessionConfig::new("test-key");
        let plan = LiveSessionBuilder::new(config)
            .steering_mode(SteeringMode::ContextInjection)
            .context_delivery(ContextDelivery::Deferred)
            .tool_advisory(false)
            .into_plan()
            .expect("plan derivation should succeed");

        assert_eq!(plan.steering_mode, SteeringMode::ContextInjection);
        assert_eq!(plan.context_delivery, ContextDelivery::Deferred);
        assert!(!plan.tool_advisory);
    }

    fn two_phase_machine() -> PhaseMachine {
        use super::super::phase::Phase;
        let mut verify = Phase::new("verify", "Verify the caller.");
        verify.tools_enabled = Some(vec!["verify_identity".into()]);
        let mut serve = Phase::new("serve", "Serve the caller.");
        serve.tools_enabled = Some(vec!["get_balance".into()]);
        let mut machine = PhaseMachine::new("verify");
        machine.add_phase(verify);
        machine.add_phase(serve);
        machine
    }

    fn banking_config(model: gemini_genai_rs::prelude::ModelId) -> SessionConfig {
        use gemini_genai_rs::prelude::{FunctionDeclaration, Tool};
        let decl = |name: &str| FunctionDeclaration {
            name: name.into(),
            description: name.into(),
            parameters: None,
            behavior: None,
        };
        SessionConfig::new("test-key")
            .model(model)
            .system_instruction("You are a bank agent.")
            .add_tool(Tool::functions(vec![
                decl("verify_identity"),
                decl("get_balance"),
            ]))
    }

    fn declared(config: &SessionConfig) -> Vec<String> {
        config
            .tools
            .iter()
            .filter_map(|t| t.function_declarations.as_ref())
            .flatten()
            .map(|d| d.name.clone())
            .collect()
    }

    fn instruction_text(config: &SessionConfig) -> String {
        config
            .system_instruction
            .iter()
            .flat_map(|c| &c.parts)
            .filter_map(|p| match p {
                gemini_genai_rs::prelude::Part::Text { text } => Some(text.as_str()),
                _ => None,
            })
            .collect()
    }

    fn lookup_dispatcher() -> ToolDispatcher {
        let mut dispatcher = ToolDispatcher::new();
        for name in ["lookup", "book"] {
            dispatcher.register_function(Arc::new(crate::tool::SimpleTool::new(
                name,
                format!("{name} tool"),
                None,
                |_| async { Ok(serde_json::json!({})) },
            )));
        }
        dispatcher
    }

    #[test]
    fn a_code_tool_is_declared_at_setup_beside_the_tools() {
        let plan = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .dispatcher(lookup_dispatcher())
            .code_mode(crate::code_mode::CodeMode::new().function("lookup"))
            .into_plan()
            .expect("plan");
        let mut names = declared(plan.config.as_ref().unwrap());
        names.sort();
        assert_eq!(names, vec!["book", "lookup", "run_code"]);
        assert!(plan.code_mode.is_some());
    }

    #[test]
    fn a_code_tool_must_expose_tools_the_session_has() {
        let err = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .dispatcher(lookup_dispatcher())
            .code_mode(crate::code_mode::CodeMode::new().function("refund"))
            .into_plan()
            .err()
            .expect("an unknown function is refused");
        assert!(err.to_string().contains("refund"), "{err}");

        let err = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .dispatcher(lookup_dispatcher())
            .code_mode(
                crate::code_mode::CodeMode::new()
                    .function("lookup")
                    .tool_name("book"),
            )
            .into_plan()
            .err()
            .expect("a name clash is refused");
        assert!(err.to_string().contains("'book'"), "{err}");
    }

    #[test]
    fn context_update_steering_declares_only_the_initial_phase_at_setup() {
        use gemini_genai_rs::prelude::ModelId;
        let plan = LiveSessionBuilder::new(banking_config(ModelId::LIVE_3_8))
            .phase_machine(two_phase_machine())
            .steering_mode(SteeringMode::ContextUpdate)
            .into_plan()
            .expect("plan derivation should succeed");

        assert_eq!(plan.steering_mode, SteeringMode::ContextUpdate);
        let config = plan.config.as_ref().unwrap();
        assert_eq!(declared(config), vec!["verify_identity"]);
        assert_eq!(
            instruction_text(config),
            "You are a bank agent.\n\nVerify the caller."
        );
        // The full catalog stays with the scope for later phases.
        let scope = plan.tool_scope.as_ref().expect("scope under ContextUpdate");
        assert_eq!(scope.declared(), vec!["verify_identity"]);
        assert_eq!(
            plan.base_instruction.as_deref(),
            Some("You are a bank agent.")
        );
    }

    #[test]
    fn a_task_session_declares_only_task_control_where_context_update_is_accepted() {
        use gemini_genai_rs::prelude::ModelId;
        let session = |model, steering: Option<SteeringMode>| {
            let mut builder = LiveSessionBuilder::new(
                SessionConfig::new("test-key")
                    .model(model)
                    .system_instruction("You are a bank agent."),
            )
            .tasks(crate::tasks::two_skill_runtime());
            if let Some(mode) = steering {
                builder = builder.steering_mode(mode);
            }
            builder.into_plan().expect("plan derivation should succeed")
        };

        let plan = session(ModelId::LIVE_3_8, None);
        let config = plan.config.as_ref().unwrap();
        // `task_control` and each skill's typed entry point.
        assert_eq!(
            declared(config),
            vec!["task_control", "start_billing", "start_faq"]
        );
        let scope = plan.task_scope.as_ref().expect("a task scope on 3.8");
        assert_eq!(
            scope.declared(),
            vec!["start_billing", "start_faq", "task_control"]
        );
        assert!(
            instruction_text(config).contains("Installed skills"),
            "the catalog stays in the instruction"
        );
        assert!(
            plan.tool_scope.is_none(),
            "phases and flows are not involved"
        );

        let plan = session(ModelId::LIVE_3_8, Some(SteeringMode::ContextUpdate));
        assert!(plan.task_scope.is_some());

        // A model without `contextUpdate`, or a steering mode chosen against
        // it, keeps every skill's tools declared.
        for plan in [
            session(ModelId::FLASH_2_5_NATIVE_AUDIO_LATEST, None),
            session(ModelId::LIVE_3_8, Some(SteeringMode::Hybrid)),
        ] {
            assert!(plan.task_scope.is_none());
            assert_eq!(
                declared(plan.config.as_ref().unwrap()),
                vec![
                    "task_control",
                    "start_billing",
                    "start_faq",
                    "billing__verify",
                    "billing__pay",
                    "faq__answer"
                ]
            );
        }
    }

    #[test]
    fn context_update_steering_falls_back_to_hybrid_without_model_support() {
        use gemini_genai_rs::prelude::ModelId;
        let plan = LiveSessionBuilder::new(banking_config(ModelId::FLASH_2_5_NATIVE_AUDIO_LATEST))
            .phase_machine(two_phase_machine())
            .steering_mode(SteeringMode::ContextUpdate)
            .into_plan()
            .expect("plan derivation should succeed");

        assert_eq!(plan.steering_mode, SteeringMode::Hybrid);
        assert!(plan.tool_scope.is_none());
        let config = plan.config.as_ref().unwrap();
        assert_eq!(
            declared(config),
            vec!["verify_identity", "get_balance"],
            "every tool is declared, as under the other modes"
        );
        assert_eq!(instruction_text(config), "You are a bank agent.");
    }

    #[test]
    fn other_steering_modes_declare_every_tool_at_setup() {
        use gemini_genai_rs::prelude::ModelId;
        let plan = LiveSessionBuilder::new(banking_config(ModelId::LIVE_3_8))
            .phase_machine(two_phase_machine())
            .steering_mode(SteeringMode::Hybrid)
            .into_plan()
            .expect("plan derivation should succeed");
        assert!(plan.tool_scope.is_none());
        assert_eq!(
            declared(plan.config.as_ref().unwrap()),
            vec!["verify_identity", "get_balance"]
        );
    }

    #[test]
    fn into_plan_carries_greeting_and_telemetry_interval() {
        let config = SessionConfig::new("test-key");
        let plan = LiveSessionBuilder::new(config)
            .greeting("Hello there")
            .telemetry_interval(std::time::Duration::from_secs(5))
            .soft_turn_timeout(std::time::Duration::from_secs(2))
            .into_plan()
            .expect("plan derivation should succeed");

        assert_eq!(plan.greeting.as_deref(), Some("Hello there"));
        assert_eq!(
            plan.telemetry_interval,
            Some(std::time::Duration::from_secs(5))
        );
        assert_eq!(
            plan.soft_turn_timeout,
            Some(std::time::Duration::from_secs(2))
        );
    }

    #[test]
    fn into_plan_carries_event_capacity() {
        let default = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .into_plan()
            .expect("plan derivation should succeed");
        assert_eq!(
            default.event_capacity,
            super::super::events::DEFAULT_EVENT_CAPACITY
        );

        let custom = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .event_capacity(64)
            .into_plan()
            .expect("plan derivation should succeed");
        assert_eq!(custom.event_capacity, 64);

        // `broadcast::channel(0)` panics, so zero is raised to one.
        let zero = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .event_capacity(0)
            .into_plan()
            .expect("plan derivation should succeed");
        assert_eq!(zero.event_capacity, 1);
    }

    #[tokio::test]
    async fn event_capacity_bounds_how_far_a_subscriber_can_lag() {
        use super::super::events::LiveEvent;
        use gemini_genai_rs::session::SessionState;
        use tokio::sync::broadcast::error::RecvError;

        let (command_tx, _command_rx) = tokio::sync::mpsc::channel(8);
        let (event_tx, _) = tokio::sync::broadcast::channel(16);
        let (phase_tx, phase_rx) = tokio::sync::watch::channel(SessionPhase::Active);
        let state = Arc::new(SessionState::with_events(phase_tx, event_tx.clone()));
        let session = SessionHandle::new(command_tx, event_tx, state, phase_rx);

        let plan = LiveSessionBuilder::new(SessionConfig::new("test-key"))
            .event_capacity(4)
            .into_plan()
            .expect("plan derivation should succeed");
        let runtime = build_runtime(plan, session);
        let mut rx = runtime.live_event_tx.subscribe();
        for _ in 0..5 {
            let _ = runtime.live_event_tx.send(LiveEvent::TurnComplete);
        }
        // Five sends into four slots: the oldest is gone.
        assert!(matches!(rx.recv().await, Err(RecvError::Lagged(1))));
        for _ in 0..4 {
            assert!(matches!(rx.recv().await, Ok(LiveEvent::TurnComplete)));
        }
    }

    #[test]
    fn into_plan_validates_phase_machine() {
        // A PhaseMachine whose initial phase doesn't exist must fail validation
        // during plan derivation (no connection required).
        let config = SessionConfig::new("test-key");
        let pm = PhaseMachine::new("nonexistent");
        let result = LiveSessionBuilder::new(config)
            .phase_machine(pm)
            .into_plan();
        assert!(result.is_err(), "invalid phase machine should fail to plan");
    }

    #[test]
    fn into_plan_carries_valid_phase_machine_and_seeds_nothing() {
        // A valid phase machine is carried into the plan; into_plan does NOT
        // seed state (that happens in build_runtime), so this stays I/O-free.
        let config = SessionConfig::new("test-key");
        let mut pm = PhaseMachine::new("start");
        pm.add_phase(crate::live::phase::Phase::new("start", "Start phase"));
        let plan = LiveSessionBuilder::new(config)
            .phase_machine(pm)
            .into_plan()
            .expect("valid phase machine should plan");

        assert!(plan.phase_machine.is_some());
    }

    #[test]
    fn into_plan_applies_non_blocking_to_background_tools() {
        use gemini_genai_rs::prelude::{FunctionCallingBehavior, FunctionDeclaration, Tool};

        let decl = FunctionDeclaration {
            name: "search_kb".into(),
            description: "Search".into(),
            parameters: None,
            behavior: None,
        };
        let config = SessionConfig::new("test-key").add_tool(Tool::functions(vec![decl]));

        let plan = LiveSessionBuilder::new(config)
            .tool_execution_mode(
                "search_kb",
                ToolExecutionMode::Background {
                    formatter: None,
                    scheduling: None,
                },
            )
            .into_plan()
            .expect("plan derivation should succeed");

        let cfg = plan.config.expect("config carried");
        let decl = cfg.tools[0]
            .function_declarations
            .as_ref()
            .unwrap()
            .iter()
            .find(|d| d.name == "search_kb")
            .unwrap();
        assert_eq!(decl.behavior, Some(FunctionCallingBehavior::NonBlocking));
    }
}