gemini_adk_fluent_rs/live/

mod.rs

//! `Live` — Fluent builder for callback-driven Gemini Live sessions.
//!
//! Wraps L1's `LiveSessionBuilder` with ergonomic callback registration
//! and integration with composition modules (M, T, P).
//!
//! # Callback Modes
//!
//! Control-lane callbacks support two execution modes via [`gemini_adk_rs::live::ExecutionMode`]:
//!
//! - **Default methods** (e.g., `.on_turn_complete()`) → [`gemini_adk_rs::live::ExecutionMode::Blocking`]
//! - **`_concurrent` methods** (e.g., `.on_turn_complete_concurrent()`) → [`gemini_adk_rs::live::ExecutionMode::Concurrent`]
//!
//! Use concurrent mode for fire-and-forget work (logging, analytics, webhook
//! dispatch). The lane rule for every callback is written once, at the top of
//! the callbacks module (see the `Live` callback setters).
//!
//! # Background Tool Execution
//!
//! Mark tools for background execution to eliminate dead air in voice sessions:
//!
//! ```no_run
//! # use gemini_adk_fluent_rs::prelude::*;
//! # async fn run(tools: gemini_adk_fluent_rs::compose::tools::ToolComposite) -> Result<(), AgentError> {
//! Live::builder()
//!     .tools(tools)
//!     .tool_background("search_kb")
//!     .connect_from_env()
//!     .await?;
//! # Ok(())
//! # }
//! ```

mod callbacks;
mod config;
pub use config::{
    DEFAULT_COMPRESSION_TARGET_TOKENS, DEFAULT_COMPRESSION_TRIGGER_TOKENS, InputAudioConfig,
    InputStage,
};
pub(crate) mod connect;
mod contract;
mod extraction;
mod introspect;

/// The ambient-tool merge `connect` performs, exposed so
/// [`check_live`](crate::testing::check_live) checks the same flow the session
/// will actually run rather than the one the caller wrote.
pub(crate) use connect::merge_ambient as merge_ambient_for_check;
mod phases;
pub(crate) mod scripted;

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

use gemini_adk_rs::State;
pub use gemini_adk_rs::live::extractor::TurnExtractor;
pub use gemini_adk_rs::live::needs::RepairConfig;
pub use gemini_adk_rs::live::persistence::{PersistenceError, SessionPersistence};
pub use gemini_adk_rs::live::steering::{ContextDelivery, SteeringMode};
pub use gemini_adk_rs::live::{
    ComputedRegistry, EventCallbacks, InstructionModifier, Phase, TemporalRegistry,
    ToolExecutionMode, WatcherRegistry,
};
use gemini_adk_rs::llm::BaseLlm;
use gemini_adk_rs::tool::ToolDispatcher;
use gemini_genai_rs::prelude::*;

// `gemini_adk_fluent_rs::live` is the curated home for the full
// Live control plane. The kernel `prelude` keeps only `Live` + the headline types;
// everything else (persistence, steering, repair, transcripts, extraction triggers,
// soft-turn, runtime contract, …) is re-exported here. (Explicit, rather than a
// glob, to avoid shadowing the L1/L2 private `callbacks`/`contract` modules.)
pub use gemini_adk_rs::live::{
    ActivityAuthority, BackendInputVad, BackendVadSnapshot, BackgroundAgentDispatcher,
    BackgroundToolTracker, ComputedContract, ComputedVar, ConsecutiveFailureDetector,
    ContextBuilder, ControlContract, DefaultResultFormatter, DeferredWriter, Delivery,
    DeliveryConfig, EffectPolicy, ExecutionMode, ExtractionTrigger, ExtractorContract,
    FieldPromotion, FsPersistence, InputAudioProcessor, LatencyBucket, LatencyStats, LiveEffect,
    LiveEffectExecutor, LiveEvent, LiveEventStream, LiveHandle, LiveReactor, LiveSessionBuilder,
    LlmExtractor, MemoryPersistence, MergePolicy, NeedsFulfillment, PatternDetector,
    PendingContext, PhaseContract, PhaseInstruction, PhaseMachine, PhasePreparation, PredicateFn,
    PreparationContract, PromotionContract, RateDetector, Reaction, ReactorEvent, ReactorRule,
    RepairAction, ResultFormatter, RuntimeContract, SessionHook, SessionSignals, SessionSnapshot,
    SessionTelemetry, SessionType, SoftTurnDetector, SustainedDetector, ToolCallSummary,
    ToolContract, TranscriptBuffer, TranscriptTurn, TranscriptWindow, Transition,
    TransitionContract, TransitionEvaluation, TransitionRecord, TransitionResult,
    TransitionTrigger, TurnCommitConfig, TurnCommitPolicy, TurnCountDetector, TurnSignal,
    VoiceRuntimeState, WatchPredicate, Watcher, WatcherContract,
};
// Offline record/replay harness (Milestone 7 determinism spine).
pub use gemini_adk_rs::live::replay::{
    ReplaySession, attach_session, collect_events_until_idle, replay_session,
};

/// A deferred agent tool registration (resolved at connect time when State is available).
pub(crate) struct DeferredAgentTool {
    pub(crate) name: String,
    pub(crate) description: String,
    pub(crate) agent: Arc<dyn gemini_adk_rs::text::TextAgent>,
}

/// Fluent builder for constructing and connecting Gemini Live sessions.
///
/// Accumulates model configuration, callbacks, extractors, phases, watchers,
/// temporal patterns, and tool execution modes, then connects via one of
/// the `connect_*` methods.
///
/// Control-lane callbacks can be registered with `_concurrent` suffixed
/// methods for fire-and-forget execution. Tools can be marked for background
/// execution via [`tool_background()`](Self::tool_background).
///
/// # Example
/// ```no_run
/// # use gemini_adk_fluent_rs::prelude::*;
/// # async fn run(tools: gemini_adk_fluent_rs::compose::tools::ToolComposite) -> Result<(), AgentError> {
/// let session = Live::builder()
///     .voice(Voice::Kore)
///     .instruction("You are a weather assistant")
///     .tools(tools)
///     .on_audio(|data| { let _ = data; })
///     .on_text(|t| print!("{t}"))
///     .on_interrupted(|| async { /* flush playback */ })
///     .connect_from_env()
///     .await?;
/// # let _ = session; Ok(())
/// # }
/// ```
///
/// # Extraction Pipeline
/// ```no_run
/// # use gemini_adk_fluent_rs::prelude::*;
/// # use std::sync::Arc;
/// # #[derive(serde::Deserialize, serde::Serialize, schemars::JsonSchema)]
/// # struct OrderState { items: Vec<String> }
/// # async fn run(flash_llm: Arc<dyn BaseLlm>) -> Result<(), AgentError> {
/// let handle = Live::builder()
///     .instruction("You are a restaurant order assistant")
///     .extract_turns::<OrderState>(
///         flash_llm,
///         "Extract: items ordered, quantities, modifications, order_phase",
///     )
///     .on_extracted(|name, value| async move {
///         println!("Extracted {name}: {value}");
///     })
///     .connect_from_env()
///     .await?;
///
/// // Read latest extraction from shared State at any time:
/// let order: Option<OrderState> = handle.extracted("OrderState");
/// # let _ = order; Ok(())
/// # }
/// ```
pub struct Live {
    pub(crate) config: SessionConfig,
    pub(crate) callbacks: EventCallbacks,
    pub(crate) dispatcher: Option<ToolDispatcher>,
    pub(crate) tasks: Option<gemini_adk_rs::tasks::TaskRuntime>,
    pub(crate) extractors: Vec<Arc<dyn TurnExtractor>>,
    pub(crate) decisions: Option<Arc<gemini_adk_rs::decision::Decisions>>,
    pub(crate) code_mode: Option<gemini_adk_rs::code_mode::CodeMode>,
    // L1 registries
    pub(crate) computed: ComputedRegistry,
    pub(crate) phases: Vec<Phase>,
    pub(crate) initial_phase: Option<String>,
    pub(crate) watchers: WatcherRegistry,
    pub(crate) temporal: TemporalRegistry,
    pub(crate) greeting: Option<String>,
    // Phase defaults: modifiers + prompt_on_enter inherited by all phases.
    pub(crate) phase_default_modifiers: Vec<InstructionModifier>,
    pub(crate) phase_default_prompt_on_enter: bool,
    // Per-tool execution modes (standard vs background).
    pub(crate) tool_execution_modes: HashMap<String, ToolExecutionMode>,
    // Deferred agent tools (resolved at connect time).
    pub(crate) deferred_agent_tools: Vec<DeferredAgentTool>,
    // Tools requiring async I/O to resolve (MCP/A2A/OpenAPI/Search),
    // resolved at connect time.
    pub(crate) deferred_tools: Vec<crate::compose::tools::DeferredTool>,
    // LLMs to warm up at connect time.
    pub(crate) warm_up_llms: Vec<Arc<dyn BaseLlm>>,
    // Control plane configuration.
    pub(crate) soft_turn_timeout: Option<Duration>,
    /// `None` until set, so a task session can steer by its foreground task.
    pub(crate) steering_mode: Option<SteeringMode>,
    pub(crate) context_delivery: ContextDelivery,
    pub(crate) delivery: gemini_adk_rs::live::DeliveryConfig,
    pub(crate) redactor: Option<gemini_adk_rs::live::redaction::TranscriptRedactor>,
    pub(crate) repair_config: Option<RepairConfig>,
    pub(crate) persistence: Option<Arc<dyn SessionPersistence>>,
    pub(crate) session_id: Option<String>,
    pub(crate) tool_advisory: bool,
    pub(crate) telemetry_interval: Option<Duration>,
    pub(crate) event_capacity: Option<usize>,
    pub(crate) clock: Option<gemini_adk_rs::clock::SharedClock>,
    // Middleware layers run around tool dispatch in the control lane.
    pub(crate) middleware_layers: Vec<Arc<dyn gemini_adk_rs::middleware::Middleware>>,
    // Confirmation provider consulted before running `T::confirm(..)` tools.
    pub(crate) confirmation_provider:
        Option<Arc<dyn gemini_adk_rs::confirmation::ConfirmationProvider>>,
    // Governed flow (DAG) + its enforcement mode.
    pub(crate) flow: Option<gemini_adk_rs::flow::Flow>,
    pub(crate) flow_mode: gemini_adk_rs::flow::Enforcement,
    /// Merged into the flow's own `ambient` list at connect, so an extension
    /// that registers cross-cutting tools composes with `govern` in either order.
    pub(crate) ambient_tools: Vec<String>,
    /// Set by `govern_compiled`/`observe_compiled`, whose documented contract is
    /// that a `CompiledFlow` already surfaced its diagnostics and connect will
    /// not re-check it. Connect validates the flow only when this is false.
    pub(crate) flow_precompiled: bool,
    /// Digressions installed on the session's `FlowStack` at connect. Set by
    /// `converse`; cleared whenever the governing flow is replaced, because a
    /// digression only means something relative to the main flow it suspends.
    pub(crate) digressions: Vec<gemini_adk_rs::flow::Overlay>,
    /// Per-step repair policies installed on the session's `FlowStack` at
    /// connect. Same lifecycle as `digressions`.
    pub(crate) repair_policies:
        std::collections::BTreeMap<String, gemini_adk_rs::flow::RepairPolicy>,
    /// Per-step voice timing, installed on the session's flow stack.
    pub(crate) stage_timings: std::collections::BTreeMap<String, gemini_adk_rs::flow::VoiceTiming>,
    /// Slots whose correction re-opens later stages, with the keys to clear.
    pub(crate) corrections: std::collections::BTreeMap<String, Vec<String>>,
    /// Text each step must say word for word.
    pub(crate) verbatims: std::collections::BTreeMap<String, String>,
    /// Redaction and commit-governance policies, enforced at connect.
    pub(crate) policies: Vec<crate::policy::Policy>,
    /// Caller-supplied session `State`, so tools and flow guards can share one.
    pub(crate) state: Option<State>,
    /// Input audio hardening: mic-chain stages, client input-VAD tuning, and
    /// interruption authority. Applied to the handle right after connect.
    pub(crate) input_audio: crate::live::config::InputAudioConfig,
    // Per-step on_enter actions: run an agent in a mode when a step activates.
    pub(crate) flow_actions: Vec<(
        String,
        Arc<dyn gemini_adk_rs::text::TextAgent>,
        gemini_adk_rs::orchestration::AgentMode,
    )>,
    // Wire-log path: a FileWireRecorder is created here at connect time.
    pub(crate) record_wire_path: Option<std::path::PathBuf>,
    /// Configuration problems found while building (e.g. a computed-variable
    /// dependency cycle). Builder setters cannot fail, so they are collected
    /// here and reported as one `AgentError::Config` at connect.
    pub(crate) config_errors: Vec<String>,
}

impl Live {
    /// Start building a Live session.
    ///
    /// # Examples
    ///
    /// Minimal live session setup:
    ///
    /// ```no_run
    /// # use gemini_adk_fluent_rs::prelude::*;
    /// # async fn run() -> Result<(), AgentError> {
    /// let handle = Live::builder()
    ///     .voice(Voice::Kore)
    ///     .instruction("You are a helpful assistant")
    ///     .greeting("Hello! How can I help?")
    ///     .on_audio(|data| { let _ = data; /* send to speaker */ })
    ///     .on_text(|t| print!("{t}"))
    ///     .connect_google_ai("API_KEY")
    ///     .await?;
    ///
    /// handle.send_text("What is the weather?").await?;
    /// handle.disconnect().await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// With phases and state-based transitions:
    ///
    /// ```no_run
    /// # use gemini_adk_fluent_rs::prelude::*;
    /// # async fn run() -> Result<(), AgentError> {
    /// let handle = Live::builder()
    ///     .phase("greeting")
    ///         .instruction("Welcome the user")
    ///         .transition("main", S::is_true("greeted"))
    ///         .done()
    ///     .phase("main")
    ///         .instruction("Help the user")
    ///         .terminal()
    ///         .done()
    ///     .initial_phase("greeting")
    ///     .connect_google_ai("API_KEY")
    ///     .await?;
    /// # let _ = handle; Ok(())
    /// # }
    /// ```
    pub fn builder() -> Self {
        Self {
            config: SessionConfig::from_endpoint(ApiEndpoint::google_ai(""))
                .context_window_trigger_tokens(config::DEFAULT_COMPRESSION_TRIGGER_TOKENS)
                .context_window_compression(config::DEFAULT_COMPRESSION_TARGET_TOKENS),
            callbacks: EventCallbacks::default(),
            dispatcher: None,
            tasks: None,
            extractors: Vec::new(),
            decisions: None,
            code_mode: None,
            computed: ComputedRegistry::new(),
            phases: Vec::new(),
            initial_phase: None,
            watchers: WatcherRegistry::new(),
            temporal: TemporalRegistry::new(),
            greeting: None,
            phase_default_modifiers: Vec::new(),
            phase_default_prompt_on_enter: false,
            tool_execution_modes: HashMap::new(),
            deferred_agent_tools: Vec::new(),
            deferred_tools: Vec::new(),
            warm_up_llms: Vec::new(),
            soft_turn_timeout: None,
            steering_mode: None,
            context_delivery: ContextDelivery::default(),
            delivery: gemini_adk_rs::live::DeliveryConfig::default(),
            redactor: None,
            repair_config: None,
            persistence: None,
            session_id: None,
            tool_advisory: true,
            telemetry_interval: None,
            event_capacity: None,
            clock: None,
            middleware_layers: Vec::new(),
            confirmation_provider: None,
            flow: None,
            flow_mode: gemini_adk_rs::flow::Enforcement::Enforce,
            ambient_tools: Vec::new(),
            flow_precompiled: false,
            digressions: Vec::new(),
            repair_policies: std::collections::BTreeMap::new(),
            stage_timings: std::collections::BTreeMap::new(),
            corrections: std::collections::BTreeMap::new(),
            policies: Vec::new(),
            verbatims: std::collections::BTreeMap::new(),
            state: None,
            flow_actions: Vec::new(),
            record_wire_path: None,
            config_errors: Vec::new(),
            input_audio: crate::live::config::InputAudioConfig::default(),
        }
    }

    /// Enforce a [`Policy`](crate::policy::Policy) on this session.
    ///
    /// - `Policy::redact(keys)`: the keys' values are masked wherever state
    ///   leaves the process (journal sink, persistence snapshots, extraction
    ///   events). See [`State::redact_keys`](gemini_adk_rs::State::redact_keys).
    /// - `Policy::commit(tool)`: the tool is wrapped in a
    ///   [`CommitGuard`](gemini_adk_rs::tool::CommitGuard) at connect, so a
    ///   repeated commit with the same idempotency key returns the first
    ///   result, and a failed one runs its compensating tool.
    /// - `Policy::safety_handoff(..)` is a digression; it needs the
    ///   conversation compiler, so attach it with `Conversation::policy`.
    ///   Here it is reported as a configuration error at connect.
    ///
    /// [`converse`](Self::converse) installs a conversation's policies.
    pub fn policy(mut self, policy: impl Into<crate::policy::Policy>) -> Self {
        let policy = policy.into();
        if matches!(policy, crate::policy::Policy::SafetyHandoff { .. }) {
            self.config_errors.push(
                "Policy::safety_handoff is lowered to a digression by the conversation \
                 compiler: attach it with Conversation::policy and converse(..)"
                    .into(),
            );
        } else {
            self.policies.push(policy);
        }
        self
    }

    /// Pace step `step` of the governed flow (or of a digression): reprompt
    /// on silence, filler cues for slow tools, holding the floor, endpointing
    /// and context delivery. See [`VoiceTiming`](gemini_adk_rs::flow::VoiceTiming).
    ///
    /// A [`Conversation`](crate::conversation::Conversation) carries its own
    /// per-stage timing; [`converse`](Self::converse) installs it.
    pub fn stage_timing(
        mut self,
        step: impl Into<String>,
        timing: gemini_adk_rs::flow::VoiceTiming,
    ) -> Self {
        self.stage_timings.insert(step.into(), timing);
        self
    }

    /// Govern the session with a [`Flow`](gemini_adk_rs::flow::Flow) DAG and
    /// **enforce** it: inadmissible tool calls are blocked and active-step
    /// postures steer the model at each turn boundary.
    pub fn govern(mut self, flow: gemini_adk_rs::flow::Flow) -> Self {
        self.flow = Some(flow);
        self.flow_mode = gemini_adk_rs::flow::Enforcement::Enforce;
        self.flow_precompiled = false;
        self.digressions.clear();
        self.repair_policies.clear();
        self
    }

    /// Use a `State` you already hold as the session's state.
    ///
    /// Without this a tool closure captures whatever `State` the caller made,
    /// the session runs on a different one, and the two never meet — so a tool
    /// that writes `identity_verified` and a `Guard::is_true("identity_verified")`
    /// that reads it are talking about different maps. The guard never fires,
    /// the flow never advances, and every subsequent tool is refused by a gate
    /// whose condition was in fact satisfied.
    ///
    /// That is the ordinary shape of a governed flow — tools write the facts,
    /// guards read them — so this is how you make it work:
    ///
    /// ```no_run
    /// # use gemini_adk_fluent_rs::live::Live;
    /// # use gemini_adk_rs::State;
    /// let state = State::new();
    /// Live::builder()
    ///     .state(state.clone())   // the session runs on this
    ///     .tools(my_tools(state)); // and so do the tools
    /// # fn my_tools(_: State) -> gemini_adk_fluent_rs::compose::tools::ToolComposite { todo!() }
    /// ```
    ///
    /// `agent_tool` already shares state with the agents it wraps; this is the
    /// same guarantee for ordinary tools.
    pub fn state(mut self, state: State) -> Self {
        self.state = Some(state);
        self
    }

    /// Install task-private capabilities in this speaking session.
    /// A second installation is a configuration error reported at connect.
    pub fn tasks(mut self, runtime: gemini_adk_rs::tasks::TaskRuntime) -> Self {
        if self.tasks.is_some() {
            self.config_errors.push(
                "task capabilities are already installed; combine them in one catalog".into(),
            );
        } else {
            self.tasks = Some(runtime);
        }
        self
    }

    /// Install compiled skills; configuration failures surface at connect.
    pub fn skills(
        mut self,
        skills: impl IntoIterator<Item = gemini_adk_rs::tasks::CompiledSkill>,
    ) -> Self {
        if self.tasks.is_some() {
            self.config_errors.push(
                "task capabilities are already installed; combine them in one catalog".into(),
            );
            return self;
        }
        match gemini_adk_rs::tasks::TaskRuntime::new(skills.into_iter().collect()) {
            Ok(runtime) => self.tasks = Some(runtime),
            Err(error) => self.config_errors.push(error.to_string()),
        }
        self
    }

    /// Register cross-cutting tools as
    /// [ambient](gemini_adk_rs::flow::Flow::ambient): exempt from every step's
    /// `allow` whitelist, still bound by anything that names them.
    ///
    /// Merged into the governing flow at connect, so this composes with
    /// [`govern`](Self::govern) in **either order**. Without a flow it is inert.
    ///
    /// Extensions that install their own tools should call this rather than
    /// making the application remember to widen every step — `with_memory` does
    /// exactly that for `recall_context` and `manage_memory`.
    pub fn ambient_tools<I, S>(mut self, tools: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.ambient_tools.extend(tools.into_iter().map(Into::into));
        self
    }

    /// The cross-cutting tools registered via [`ambient_tools`](Self::ambient_tools).
    ///
    /// Introspection for extensions and tests: the flow's own `ambient` list is
    /// not included, because the two are only merged at connect.
    pub fn ambient_tool_names(&self) -> &[String] {
        &self.ambient_tools
    }

    /// Attach a [`Flow`](gemini_adk_rs::flow::Flow) in **observe** mode: nothing
    /// is blocked, but deviations are recorded for audit/analytics.
    pub fn observe(mut self, flow: gemini_adk_rs::flow::Flow) -> Self {
        self.flow = Some(flow);
        self.flow_mode = gemini_adk_rs::flow::Enforcement::Observe;
        self.flow_precompiled = false;
        self.digressions.clear();
        self.repair_policies.clear();
        self
    }

    /// Govern the session with a pre-compiled
    /// [`CompiledFlow`](gemini_adk_rs::flow::CompiledFlow) and **enforce** it.
    ///
    /// A `CompiledFlow` carries proof that
    /// [`Flow::compile`](gemini_adk_rs::flow::Flow::compile) (or
    /// [`Flow::compile_with_tools`](gemini_adk_rs::flow::Flow::compile_with_tools))
    /// already surfaced its diagnostics, so connect does **not** re-validate or
    /// re-compile it — compile once at load time, govern many sessions.
    pub fn govern_compiled(self, flow: gemini_adk_rs::flow::CompiledFlow) -> Self {
        let mut live = self.govern(flow.into_flow());
        live.flow_precompiled = true;
        live
    }

    /// Attach a pre-compiled
    /// [`CompiledFlow`](gemini_adk_rs::flow::CompiledFlow) in **observe** mode:
    /// nothing is blocked, but deviations are recorded for audit/analytics.
    /// Like [`govern_compiled`](Self::govern_compiled), the flow is not
    /// re-validated or re-compiled at connect.
    pub fn observe_compiled(self, flow: gemini_adk_rs::flow::CompiledFlow) -> Self {
        let mut live = self.observe(flow.into_flow());
        live.flow_precompiled = true;
        live
    }

    /// Run an agent the first time the named flow step becomes active.
    ///
    /// The agent reads its inputs from `State` and its result lands in
    /// `{step}:result` ([`AgentMode::Call`] resolves inline at the turn boundary;
    /// [`AgentMode::Dispatch`]/[`AgentMode::Background`] run detached). A
    /// downstream step can then complete on it via `Guard::resolved(step)`. This
    /// is how a governed flow drives in-session orchestration. Requires a flow
    /// (`govern`/`observe`).
    ///
    /// [`AgentMode::Call`]: gemini_adk_rs::orchestration::AgentMode::Call
    /// [`AgentMode::Dispatch`]: gemini_adk_rs::orchestration::AgentMode::Dispatch
    /// [`AgentMode::Background`]: gemini_adk_rs::orchestration::AgentMode::Background
    pub fn on_step_enter(
        mut self,
        step: impl Into<String>,
        agent: Arc<dyn gemini_adk_rs::text::TextAgent>,
        mode: gemini_adk_rs::orchestration::AgentMode,
    ) -> Self {
        self.flow_actions.push((step.into(), agent, mode));
        self
    }

    /// Gate `T::confirm(..)` tools behind a confirmation provider.
    ///
    /// When set, any confirmation-gated tool is checked against `provider`
    /// before it runs; a denied decision returns an error to the model instead
    /// of executing the tool. Accepts any [`ConfirmationProvider`] — including a
    /// plain async closure of `Fn(ConfirmationRequest) -> impl Future<Output = ToolConfirmation>`.
    ///
    /// [`ConfirmationProvider`]: gemini_adk_rs::confirmation::ConfirmationProvider
    /// [`ConfirmationRequest`]: gemini_adk_rs::confirmation::ConfirmationRequest
    /// [`ToolConfirmation`]: gemini_adk_rs::confirmation::ToolConfirmation
    pub fn confirmation_provider(
        mut self,
        provider: Arc<dyn gemini_adk_rs::confirmation::ConfirmationProvider>,
    ) -> Self {
        self.confirmation_provider = Some(provider);
        self
    }

    /// Attach middleware — a [`MiddlewareComposite`](crate::compose::middleware::MiddlewareComposite)
    /// or a single `Arc<dyn Middleware>` — every layer runs around tool
    /// dispatch in the control lane (`before_tool` can veto a call,
    /// `after_tool` and `on_tool_error` observe results).
    ///
    /// Stack layers with `>>`, e.g. `M::log() >> M::latency()`.
    ///
    /// Note: model-level hooks (`before_model`/`after_model`) are TextAgent
    /// pipeline concepts and do not apply to a streaming Live session.
    pub fn middleware(
        mut self,
        middleware: impl Into<crate::compose::middleware::MiddlewareComposite>,
    ) -> Self {
        self.middleware_layers.extend(middleware.into().layers);
        self
    }

    /// Set the periodic telemetry emission interval.
    ///
    /// When set, the processor emits `LiveEvent::Telemetry` snapshots
    /// and `LiveEvent::TurnMetrics` at this rate.
    pub fn telemetry_interval(mut self, interval: Duration) -> Self {
        self.telemetry_interval = Some(interval);
        self
    }

    /// Set how many events a [`LiveHandle::events`](gemini_adk_rs::live::LiveHandle::events)
    /// subscriber can fall behind before it skips the oldest. The default
    /// (512) suits subscribers that hand events on without waiting; raise
    /// it for one that does slow work inline. See
    /// [`LiveSessionBuilder::event_capacity`](gemini_adk_rs::live::LiveSessionBuilder::event_capacity)
    /// for the memory cost.
    pub fn event_capacity(mut self, capacity: usize) -> Self {
        self.event_capacity = Some(capacity);
        self
    }

    /// Read the time from `clock` instead of the system clock.
    ///
    /// Temporal patterns, phase durations, resolver cache expiry, the
    /// `session:` timing signals and journal timestamps all follow it. Pass a
    /// [`ManualClock`](gemini_adk_rs::clock::ManualClock) to make timing
    /// decisions reproducible in a test.
    pub fn clock(mut self, clock: gemini_adk_rs::clock::SharedClock) -> Self {
        self.clock = Some(clock);
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::time::Duration;

    #[test]
    fn repeated_task_installation_reports_configuration_error() {
        let first = gemini_adk_rs::tasks::TaskRuntime::new(Vec::new()).expect("empty catalog");
        let second = gemini_adk_rs::tasks::TaskRuntime::new(Vec::new()).expect("empty catalog");
        let live = Live::builder().tasks(first).tasks(second);
        assert_eq!(live.config_errors.len(), 1);
        let live = Live::builder().skills(Vec::new()).skills(Vec::new());
        assert_eq!(live.config_errors.len(), 1);
    }

    #[test]
    fn builder_chain_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .voice(Voice::Kore)
            .instruction("Test")
            .temperature(0.7)
            .google_search()
            .transcription()
            .affective_dialog()
            .session_resume()
            .no_tool_advisory()
            .context_compression(4000, 2000)
            .on_audio(|_data| {})
            .on_text(|_t| {})
            .on_vad_start(|| {})
            .on_interrupted(|| async {})
            .on_turn_complete(|| async {})
            .on_go_away(|_d| async {})
            .on_connected(|_writer| async {})
            .on_disconnected(|_r| async {})
            .on_error(|_e| async {});
        // Just verify the builder chain compiles
    }

    #[test]
    fn govern_compiled_attaches_precompiled_flow_without_recompiling() {
        use gemini_adk_rs::flow::{Enforcement, Flow, Guard};

        let compiled = Flow::new()
            .step("greet")
            .done(Guard::is_true("greeted"))
            .step("end")
            .after("greet")
            .terminal()
            .build()
            .expect("valid flow")
            .compile()
            .expect("flow compiles");

        // Enforce mode.
        let live = Live::builder().govern_compiled(compiled.clone());
        assert!(live.flow.is_some(), "compiled flow attached");
        assert_eq!(live.flow_mode, Enforcement::Enforce);

        // Observe mode.
        let live = Live::builder().observe_compiled(compiled);
        assert!(live.flow.is_some(), "compiled flow attached");
        assert_eq!(live.flow_mode, Enforcement::Observe);
    }

    #[test]
    fn builder_with_extraction_compiles() {
        use gemini_adk_rs::llm::{BaseLlm, LlmError, LlmRequest, LlmResponse};
        use schemars::JsonSchema;

        #[derive(serde::Deserialize, serde::Serialize, JsonSchema)]
        struct OrderState {
            phase: String,
            items: Vec<String>,
        }

        struct FakeLlm;

        #[async_trait::async_trait]
        impl BaseLlm for FakeLlm {
            fn model_id(&self) -> &str {
                "fake"
            }
            async fn generate(&self, _req: LlmRequest) -> Result<LlmResponse, LlmError> {
                Err(LlmError::RequestFailed("FakeLlm never generates".into()))
            }
        }

        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .instruction("Restaurant order assistant")
            .extract_turns::<OrderState>(
                Arc::new(FakeLlm),
                "Extract order state: items, quantities, phase",
            )
            .on_extracted(|name, value| async move {
                let _ = (name, value);
            })
            // Outbound interceptors
            .before_tool_response(|responses, _state| async move {
                responses // pass through
            })
            .on_turn_boundary(|_state, _writer| async move {
                // inject context
            })
            .instruction_template(|state| {
                let phase: String = state.get("phase").unwrap_or_default();
                match phase.as_str() {
                    "ordering" => Some("Take orders accurately.".into()),
                    _ => None,
                }
            });
        // Just verify the builder chain with all features compiles
    }

    #[test]
    fn builder_with_computed_state_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .instruction("Test computed state")
            .computed("doubled", &["app:count"], |state| {
                let count: i64 = state.get("app:count")?;
                Some(serde_json::json!(count * 2))
            })
            .computed("level", &["app:score"], |state| {
                let score: f64 = state.get("app:score")?;
                if score > 0.5 {
                    Some(serde_json::json!("high"))
                } else {
                    Some(serde_json::json!("low"))
                }
            });
    }

    #[test]
    fn builder_with_phases_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .phase("greeting")
            .instruction("Welcome the user warmly")
            .transition("main", |s| s.get::<bool>("greeted").unwrap_or(false))
            .on_enter(|state, _writer| async move {
                let _ = state.set("entered_greeting", true);
            })
            .done()
            .phase("main")
            .dynamic_instruction(|s| {
                let topic: String = s.get("topic").unwrap_or_default();
                format!("Discuss {topic}")
            })
            .tools(vec!["search".into(), "lookup".into()])
            .transition("farewell", |s| s.get::<bool>("done").unwrap_or(false))
            .done()
            .phase("farewell")
            .instruction("Say goodbye")
            .terminal()
            .done()
            .initial_phase("greeting");
    }

    #[test]
    fn builder_with_phase_guard_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .phase("start")
            .instruction("Begin")
            .transition("secure", |_| true)
            .done()
            .phase("secure")
            .instruction("Secure area")
            .guard(|s| s.get::<bool>("verified").unwrap_or(false))
            .on_exit(|state, _writer| async move {
                let _ = state.set("left_secure", true);
            })
            .terminal()
            .done()
            .initial_phase("start");
    }

    #[test]
    fn builder_with_watchers_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .watch("app:score")
            .crossed_above(0.9)
            .then(|_old, _new, state| async move {
                let _ = state.set("high_score_alert", true);
            })
            .watch("app:status")
            .changed_to(serde_json::json!("complete"))
            .blocking()
            .then(|_old, _new, _state| async move {
                // blocking action
            })
            .watch("app:flag")
            .became_true()
            .then(|_old, _new, _state| async move {
                // flag became true
            });
    }

    #[test]
    fn builder_with_temporal_patterns_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .when_sustained(
                "user_confused",
                |s| s.get::<bool>("confused").unwrap_or(false),
                Duration::from_secs(30),
                |_state, _writer| async move {
                    // offer help
                },
            )
            .when_rate(
                "rapid_errors",
                |evt| matches!(evt, SessionEvent::TextDelta(_)),
                5,
                Duration::from_secs(10),
                |_state, _writer| async move {
                    // throttle
                },
            )
            .when_turns(
                "stuck_in_loop",
                |s| s.get::<bool>("repeating").unwrap_or(false),
                3,
                |_state, _writer| async move {
                    // break loop
                },
            );
    }

    #[test]
    fn builder_full_l1_chain_compiles() {
        // Full chain combining all L1 features in a single builder
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .voice(Voice::Kore)
            .instruction("Full featured agent")
            // Computed state
            .computed("sentiment_level", &["app:sentiment_score"], |state| {
                let score: f64 = state.get("app:sentiment_score")?;
                if score > 0.7 {
                    Some(serde_json::json!("positive"))
                } else if score < 0.3 {
                    Some(serde_json::json!("negative"))
                } else {
                    Some(serde_json::json!("neutral"))
                }
            })
            // Phases
            .phase("greeting")
            .instruction("Greet the user")
            .transition("help", |s| s.get::<bool>("needs_help").unwrap_or(false))
            .done()
            .phase("help")
            .instruction("Help the user")
            .terminal()
            .done()
            .initial_phase("greeting")
            // Watchers
            .watch("app:sentiment_score")
            .crossed_below(0.2)
            .then(|_old, _new, state| async move {
                let _ = state.set("alert:low_sentiment", true);
            })
            // Temporal
            .when_turns(
                "repeated_confusion",
                |s| s.get::<bool>("confused").unwrap_or(false),
                3,
                |_state, _writer| async move {},
            )
            // Standard callbacks
            .on_audio(|_data| {})
            .on_text(|_t| {})
            .on_turn_complete(|| async {});
    }

    #[test]
    fn builder_with_callback_modes_compiles() {
        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .on_turn_complete_concurrent(|| async {})
            .on_error_concurrent(|_e| async {})
            .on_extracted_concurrent(|_name, _val| async {})
            .on_extraction_error_concurrent(|_name, _err| async {})
            .on_connected_concurrent(|_w| async {})
            .on_disconnected_concurrent(|_r| async {})
            .on_go_away_concurrent(|_d| async {});
    }

    #[test]
    fn builder_with_background_tools_compiles() {
        use gemini_adk_rs::live::DefaultResultFormatter;

        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .tool_background("search_kb")
            .tool_background_with_formatter("analyze_document", Arc::new(DefaultResultFormatter));
    }

    #[test]
    fn builder_mixed_callback_modes_and_bg_tools() {
        use gemini_adk_rs::live::DefaultResultFormatter;

        let _live = Live::builder()
            .model(ModelId::LIVE_2_5_FLASH_NATIVE_AUDIO)
            .voice(Voice::Kore)
            .instruction("Full featured agent")
            .tool_background("slow_tool")
            .tool_background_with_formatter("kb_search", Arc::new(DefaultResultFormatter))
            .on_turn_complete_concurrent(|| async {})
            .on_extracted_concurrent(|_name, _val| async {})
            .on_audio(|_data| {})
            .on_text(|_t| {})
            .on_interrupted(|| async {});
    }
}