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::CallbackMode`]:
//!
//! - **Default methods** (e.g., `.on_turn_complete()`) → [`gemini_adk_rs::live::CallbackMode::Blocking`]
//! - **`_concurrent` methods** (e.g., `.on_turn_complete_concurrent()`) → [`gemini_adk_rs::live::CallbackMode::Concurrent`]
//!
//! Use concurrent mode for fire-and-forget work (logging, analytics, webhook dispatch).
//!
//! # Background Tool Execution
//!
//! Mark tools for background execution to eliminate dead air in voice sessions:
//!
//! ```rust,ignore
//! Live::builder()
//!     .tools(dispatcher)
//!     .tool_background("search_kb")
//!     .connect_vertex(project, location, token)
//!     .await?;
//! ```

mod callbacks;
mod config;
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;

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

pub use gemini_adk_rs::live::extractor::TurnExtractor;
pub use gemini_adk_rs::live::needs::RepairConfig;
pub use gemini_adk_rs::live::persistence::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_adk_rs::State;
use gemini_genai_rs::prelude::*;

// Carve (gap #9): `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::{
    BackendInputVad, BackendVadSnapshot, BackgroundAgentDispatcher, BackgroundToolTracker,
    CallbackMode, ComputedContract, ComputedVar, ConsecutiveFailureDetector, ContextBuilder,
    ControlContract, DefaultResultFormatter, DeferredWriter, EffectMode, EffectPolicy,
    ExtractionTrigger, ExtractorContract, FieldPromotion, FsPersistence, LiveEffect,
    LiveEffectExecutor, LiveEvent, LiveEventStream, LiveHandle, LiveReactor, LiveSessionBuilder,
    LlmExtractor, MemoryPersistence, MergePolicy, NeedsFulfillment, PatternDetector,
    PendingContext, PhaseContract, PhaseInstruction, PhaseMachine, PhasePreparation,
    PhaseTransition, PredicateFn, PreparationContract, PromotionContract, RateDetector, Reaction,
    ReactorEvent, ReactorRule, RepairAction, ResultFormatter, RuntimeContract, SessionSignals,
    SessionSnapshot, SessionTelemetry, SessionType, SoftTurnDetector, SustainedDetector,
    ToolCallSummary, ToolContract, TranscriptBuffer, TranscriptTurn, TranscriptWindow, Transition,
    TransitionContract, TransitionEvaluation, TransitionResult, TransitionTrigger,
    TurnCountDetector, VoiceRuntimeState, WatchPredicate, Watcher, WatcherContract,
};
// Offline record/replay harness (Milestone 7 determinism spine).
pub use gemini_adk_rs::live::replay::{
    attach_session, collect_events_until_idle, replay_session, ReplaySession,
};

/// 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
/// ```ignore
/// let session = Live::builder()
///     .model(GeminiModel::Gemini2_0FlashLive)
///     .voice(Voice::Kore)
///     .instruction("You are a weather assistant")
///     .tools(dispatcher)
///     .on_audio(|data| playback_tx.send(data.clone()).ok())
///     .on_text(|t| print!("{t}"))
///     .on_interrupted(|| async { playback.flush().await; })
///     .connect_vertex("project", "us-central1", token)
///     .await?;
/// ```
///
/// # Extraction Pipeline
/// ```ignore
/// let handle = Live::builder()
///     .model(GeminiModel::Gemini2_0FlashLive)
///     .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_vertex(project, location, token)
///     .await?;
///
/// // Read latest extraction from shared State at any time:
/// let order: Option<OrderState> = handle.extracted("OrderState");
/// ```
pub struct Live {
    pub(crate) config: SessionConfig,
    pub(crate) callbacks: EventCallbacks,
    pub(crate) dispatcher: Option<ToolDispatcher>,
    pub(crate) extractors: Vec<Arc<dyn TurnExtractor>>,
    // 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>,
    pub(crate) steering_mode: SteeringMode,
    pub(crate) context_delivery: ContextDelivery,
    pub(crate) delivery: gemini_adk_rs::live::DeliveryConfig,
    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>,
    // 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,
    /// Caller-supplied session `State`, so tools and flow guards can share one.
    pub(crate) state: Option<State>,
    // 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::Mode,
    )>,
    // Wire-log path: a FileWireRecorder is created here at connect time.
    pub(crate) record_wire_path: Option<std::path::PathBuf>,
}

impl Live {
    /// Start building a Live session.
    ///
    /// # Examples
    ///
    /// Minimal live session setup:
    ///
    /// ```rust,ignore
    /// use gemini_adk_fluent_rs::prelude::*;
    ///
    /// let handle = Live::builder()
    ///     .model(GeminiModel::Gemini2_0FlashLive)
    ///     .voice(Voice::Kore)
    ///     .instruction("You are a helpful assistant")
    ///     .greeting("Hello! How can I help?")
    ///     .on_audio(|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?;
    /// ```
    ///
    /// With phases and state-based transitions:
    ///
    /// ```rust,ignore
    /// let handle = Live::builder()
    ///     .model(GeminiModel::Gemini2_0FlashLive)
    ///     .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?;
    /// ```
    pub fn builder() -> Self {
        Self {
            config: SessionConfig::from_endpoint(ApiEndpoint::google_ai("")),
            callbacks: EventCallbacks::default(),
            dispatcher: None,
            extractors: Vec::new(),
            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: SteeringMode::default(),
            context_delivery: ContextDelivery::default(),
            delivery: gemini_adk_rs::live::DeliveryConfig::default(),
            repair_config: None,
            persistence: None,
            session_id: None,
            tool_advisory: true,
            telemetry_interval: 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,
            state: None,
            flow_actions: Vec::new(),
            record_wire_path: None,
        }
    }

    /// 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
    }

    /// 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()
    ///     .with_state(state.clone())   // the session runs on this
    ///     .with_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 with_state(mut self, state: State) -> Self {
        self.state = Some(state);
        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
    }

    /// 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::Mode::Call
    /// [`AgentMode::Dispatch`]: gemini_adk_rs::orchestration::Mode::Dispatch
    /// [`AgentMode::Background`]: gemini_adk_rs::orchestration::Mode::Background
    pub fn on_enter(
        mut self,
        step: impl Into<String>,
        agent: Arc<dyn gemini_adk_rs::text::TextAgent>,
        mode: gemini_adk_rs::orchestration::Mode,
    ) -> 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 a [`MiddlewareComposite`](crate::compose::middleware::MiddlewareComposite)
    /// — every layer runs around tool
    /// dispatch in the control lane (`before_tool` can veto a call,
    /// `after_tool` and `on_tool_error` observe results).
    ///
    /// Compose 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,
        composite: crate::compose::middleware::MiddlewareComposite,
    ) -> Self {
        self.middleware_layers.extend(composite.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
    }
}

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

    #[test]
    fn builder_chain_compiles() {
        let _live = Live::builder()
            .model(GeminiModel::Gemini2_0FlashLive)
            .voice(Voice::Kore)
            .instruction("Test")
            .temperature(0.7)
            .google_search()
            .transcription(true, true)
            .affective_dialog(true)
            .session_resume(true)
            .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> {
                unimplemented!()
            }
        }

        let _live = Live::builder()
            .model(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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(GeminiModel::Gemini2_0FlashLive)
            .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 {});
    }
}