gemini_adk_fluent_rs/spec/

mod.rs

//! `SessionSpec` — a whole Live session as one serializable JSON document.
//!
//! Where [`Flow`] made the *governance DAG* data, `SessionSpec` makes the
//! *application* data: session framing (instruction, greeting, modality),
//! declarative tool bindings (mock, HTTP, MCP), schema-as-JSON extraction that
//! fills the state guards read, data-driven phases and watchers over the same
//! closed [`Guard`] vocabulary, reusable flow fragments, and an embedded test
//! suite that replays scripted conversations through the real
//! [`FlowStack`](gemini_adk_rs::flow::FlowStack) offline.
//!
//! The invariants:
//! - **What serializes, runs.** [`SessionSpec::apply`] configures a
//!   [`Live`] builder from the document; nothing in the document needs Rust.
//! - **What can fail, fails at load time.** [`SessionSpec::validate`] runs the
//!   flow compiler, cross-checks tool names, and diffs the state keys guards
//!   *read* against the keys the session *writes* — the flow-level analogue of
//!   `compile_with_tools` for the dominant silent failure in data-authored
//!   flows (a guard waiting on a key nothing sets).
//! - **The escape hatches stay in code.** Custom closures (guards, tools,
//!   callbacks) are added on the returned builder after `apply`, exactly as
//!   before; the spec never pretends to serialize them.

pub mod authoring;
mod code;
mod codegen;
mod decisions;
pub mod project;
mod simulate;
mod skills;
pub mod store;

pub use code::{CodeLimitsSpec, CodeSpec, schema_shape, value_shape};
pub use decisions::{BooleanCriteriaSpec, DecisionKind, DecisionSpec};
pub use project::{ProjectFile, ProjectLanguage, ProjectOptions, SdkSource};
pub use store::{BundleRef, BundleStore, BundleVersion, StoreError, open_store};

pub use skills::{
    SkillSpec, SkillToolSpec, TaskExpectation, TaskObservationKind, TaskScenario, TaskStep,
    TaskTraceSnapshot,
};

pub use simulate::{
    SimEvent, SimSnapshot, SpecTest, TestExpectation, TestReport, TestStepResult, trace_test,
};

use std::collections::BTreeMap;
use std::sync::Arc;

use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

use gemini_adk_rs::expr::Expr;
use gemini_adk_rs::flow::{Constraint, Flow, Guard, Pred, Step};
use gemini_adk_rs::live::extractor::{ExtractionTrigger, FieldPromotion, LlmExtractor};
use gemini_adk_rs::live::{ContextDelivery, RepairConfig, SteeringMode};
use gemini_adk_rs::llm::BaseLlm;
use gemini_adk_rs::state::State;
use gemini_adk_rs::tool::{SimpleTool, ToolDispatcher, ToolFunction};
use gemini_genai_rs::prelude::{
    AutomaticActivityDetection, Content, FunctionResponseScheduling, Sensitivity, SessionWriter,
    Voice,
};

use crate::compose::tools::T;
use crate::live::Live;

/// Output modality for a spec-driven session.
#[derive(
    Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "lowercase")]
pub enum SpecModality {
    /// Text-only session (the default — no microphone needed).
    #[default]
    Text,
    /// Audio (voice) session.
    Audio,
}

/// An HTTP binding for a declared tool: the call is executed as an HTTP
/// request with `{args.field}` / `{state.key}` interpolation in the URL,
/// headers, and body strings, and the JSON response is returned to the model.
///
/// Requires the `http-tools` feature; without it, validation reports the
/// binding as unsupported instead of failing silently at call time.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct HttpBinding {
    /// HTTP method (GET, POST, PUT, PATCH, DELETE). Default GET.
    #[serde(default = "default_method")]
    pub method: String,
    /// Request URL, with `{args.*}`/`{state.*}` interpolation.
    pub url: String,
    /// Request headers, values interpolated.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub headers: BTreeMap<String, String>,
    /// JSON body; every string value is interpolated. Omit for no body.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub body: Option<Value>,
    /// Set by [`SessionSpec::sandboxed`]: the request URL (after
    /// interpolation) and every redirect target must stay inside it.
    #[serde(skip)]
    #[schemars(skip)]
    pub(crate) reach: Option<BindingAllowlist>,
}

fn default_method() -> String {
    "GET".to_string()
}

/// A declared tool. Without an `http` binding it is a **mock**: it returns
/// `response` (default `{"ok": true}`) and writes `set_state` — enough to
/// model, validate, and demo a governed conversation before any real tool
/// exists. With `http` it performs the request instead (and still applies
/// `set_state` afterwards, so guards latch identically) — swap a mock for a
/// binding without touching the flow.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ToolSpec {
    /// Tool (function) name the model calls.
    pub name: String,
    /// Description shown to the model.
    #[serde(default)]
    pub description: String,
    /// JSON Schema for the arguments (Gemini subset). `None` = no parameters.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parameters: Option<Value>,
    /// Canned JSON response (mock tools). Default `{"ok": true}`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub response: Option<Value>,
    /// JSON Schema of what the tool returns. Code mode shows it in the
    /// function's docstring, so a program reads the right keys; a mock tool
    /// without it shows the shape of its `response`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub returns: Option<Value>,
    /// State keys written when the tool runs — how a tool latches guards.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub set_state: BTreeMap<String, Value>,
    /// Store the tool's full response under this state key.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub save_response_as: Option<String>,
    /// Execute as an HTTP request instead of returning the canned response.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub http: Option<HttpBinding>,
    /// Execute by calling the tool of the same name on this MCP server: a
    /// command line (stdio) or an `http(s)://` URL. This is how a tool
    /// written in another language (a generated Python or Go tool server)
    /// implements a declared tool. The declaration here stays what the model
    /// sees, and `set_state` still applies.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub mcp: Option<String>,
    /// Run non-blocking: the model keeps speaking while the tool executes
    /// (`behavior: NonBlocking` on the wire; Google AI only, stripped on
    /// Vertex). Implied by `scheduling`.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub background: bool,
    /// How the async response is delivered (implies `background`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scheduling: Option<SchedulingSpec>,
}

/// Delivery mode for a background tool's response.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum SchedulingSpec {
    /// Halt current output and report immediately.
    Interrupt,
    /// Wait until the model finishes its current output.
    WhenIdle,
    /// Integrate silently without notifying the user.
    Silent,
}

impl SchedulingSpec {
    fn to_wire(self) -> FunctionResponseScheduling {
        match self {
            SchedulingSpec::Interrupt => FunctionResponseScheduling::Interrupt,
            SchedulingSpec::WhenIdle => FunctionResponseScheduling::WhenIdle,
            SchedulingSpec::Silent => FunctionResponseScheduling::Silent,
        }
    }
}

/// How extracted fields are promoted into bare state keys.
#[derive(
    Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum PromotePolicy {
    /// Write once, never overwrite a known value (default).
    #[default]
    KeepKnown,
    /// Always overwrite.
    Overwrite,
    /// Promote only when the extracted value is `true`.
    TrueOnly,
    /// Promote only non-empty values.
    NonEmpty,
}

/// Promote one extracted field into a session state key, where flow guards
/// (`captured`, `is_true`, …) read it.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct PromoteSpec {
    /// Field name in the extraction result.
    pub field: String,
    /// Target state key. Defaults to the field name.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub to: Option<String>,
    /// Merge policy.
    #[serde(default)]
    pub policy: PromotePolicy,
}

impl PromoteSpec {
    fn target(&self) -> &str {
        self.to.as_deref().unwrap_or(&self.field)
    }

    fn to_rule(&self) -> FieldPromotion {
        let rule = match self.policy {
            PromotePolicy::KeepKnown => FieldPromotion::keep_known(&self.field),
            PromotePolicy::Overwrite => FieldPromotion::overwrite(&self.field),
            PromotePolicy::TrueOnly => FieldPromotion::true_only(&self.field),
            PromotePolicy::NonEmpty => FieldPromotion::non_empty(&self.field),
        };
        match &self.to {
            Some(key) => rule.to(key),
            None => rule,
        }
    }
}

/// When an extractor runs.
#[derive(
    Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum TriggerSpec {
    /// After every turn (default).
    #[default]
    EveryTurn,
    /// After tool calls complete.
    AfterToolCall,
    /// On generation complete, before the turn completes.
    OnGenerationComplete,
    /// When a phase transition occurs.
    OnPhaseChange,
}

impl TriggerSpec {
    fn to_trigger(self) -> ExtractionTrigger {
        match self {
            TriggerSpec::EveryTurn => ExtractionTrigger::EveryTurn,
            TriggerSpec::AfterToolCall => ExtractionTrigger::AfterToolCall,
            TriggerSpec::OnGenerationComplete => ExtractionTrigger::OnGenerationComplete,
            TriggerSpec::OnPhaseChange => ExtractionTrigger::OnPhaseChange,
        }
    }
}

/// Schema-as-JSON out-of-band extraction: an OOB model fills `schema` from the
/// transcript, the result lands in state under `name`, and `promote` rules
/// write individual fields to bare keys — closing the loop that lets a flow
/// advance from *speech alone* (`captured` guards latch with no tool call).
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ExtractSpec {
    /// Extraction name — the state key the full result is stored under.
    pub name: String,
    /// Instruction for the extraction model.
    pub instruction: String,
    /// JSON Schema the extraction must satisfy.
    pub schema: Value,
    /// Transcript window in turns. Default 3.
    #[serde(default = "default_window")]
    pub window: usize,
    /// When to run.
    #[serde(default)]
    pub trigger: TriggerSpec,
    /// Field-promotion rules into bare state keys.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub promote: Vec<PromoteSpec>,
    /// The text model this entry runs on, such as `gemini-3.5-flash-lite`.
    /// Overrides `models.extraction`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    /// The thinking budget this entry's requests send. Overrides
    /// `models.extraction_thinking_budget`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub thinking_budget: Option<u32>,
}

/// The extraction model `adk spec check` suggests pinning, with
/// [`RECOMMENDED_EXTRACTION_THINKING_BUDGET`]. On the labelled pharmacy turn
/// it was correct in 60 of 60 rounds at 0.67 s median and 1.2 s at most, where
/// the rolling `gemini-flash-latest` default took 6.7 s at the median and up
/// to 62 s (`tests/extraction_latency.rs`). Without thinking it follows an
/// instruction literally: say what counts, not "leave it out until".
pub const RECOMMENDED_EXTRACTION_MODEL: &str = "gemini-3.5-flash-lite";

/// The thinking budget pinned with [`RECOMMENDED_EXTRACTION_MODEL`], which
/// rejects 0 and does not think below about 1024.
pub const RECOMMENDED_EXTRACTION_THINKING_BUDGET: u32 =
    gemini_adk_rs::live::extractor::FALLBACK_THINKING_BUDGET;

/// What validation says when extraction names no model.
const UNPINNED_EXTRACTION_WARNING: &str = "extraction names no model, so it runs on the host \
     default (gemini-flash-latest, a rolling alias whose model and latency change under you, \
     unless GEMINI_TEXT_MODEL is set); every turn waits for it: set models.extraction";

/// The models a spec runs on, by role.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ModelsSpec {
    /// The text model `extract` entries run on, such as
    /// `gemini-3.5-flash-lite`, unless an entry names its own. Name one: the
    /// host default is the rolling `gemini-flash-latest` alias, whose model
    /// (and latency) changes under you, and every turn waits for extraction.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub extraction: Option<String>,
    /// The thinking budget extraction requests send. The default, 0, keeps
    /// thinking off; a model that rejects 0 (`gemini-3.5-flash-lite` needs
    /// at least 1) is retried at 64, then with no budget. Set 64 for such a
    /// model to skip the rejected first request.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub extraction_thinking_budget: Option<u32>,
}

fn default_window() -> usize {
    3
}

/// A serializable side effect for phase entry, watcher, and pattern actions —
/// the closed-effect counterpart to [`Guard`]'s closed predicates. One
/// vocabulary, honored identically wherever effects fire.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum EffectSpec {
    /// Write these state keys.
    Set(BTreeMap<String, Value>),
    /// Inject a model-role context turn (steering text the model reads before
    /// its next response — it does not answer it directly).
    Context(String),
    /// Inject the text as a model-role turn **and** ask the model to respond
    /// now — the "make the model speak" effect (proactive check-in, nudge).
    Prompt(String),
    /// Durably remember a note (`{state.key}` templates interpolated) through
    /// the session's memory binding. Requires the spec's `memory` section.
    Remember(String),
}

/// A data-driven phase transition: fire `when` the guard holds over state.
///
/// Guards here evaluate against state alone (no flow marking), so
/// `called_ok`/`done` atoms are rejected by validation — use a state key a
/// tool or extractor writes instead.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct TransitionSpec {
    /// Target phase.
    pub to: String,
    /// Condition over session state.
    pub when: Guard,
    /// Optional description for navigation steering.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
}

/// A data-driven conversation phase.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct PhaseSpec {
    /// Phase name.
    pub name: String,
    /// Instruction while this phase is active.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub instruction: Option<String>,
    /// Tool names available in this phase.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tools: Vec<String>,
    /// Required state keys (drives conversation repair).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub needs: Vec<String>,
    /// Effects fired on phase entry.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub on_enter: Vec<EffectSpec>,
    /// Guarded transitions out of this phase.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub transitions: Vec<TransitionSpec>,
    /// Prompt the model on entry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prompt_on_enter: Option<bool>,
    /// Terminal phase.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub terminal: bool,
}

/// A data-driven state watcher condition.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum WatchCondition {
    /// Any change.
    Changed,
    /// Changed to exactly this value.
    ChangedTo(Value),
    /// Numeric value crossed above the threshold.
    CrossedAbove(f64),
    /// Numeric value crossed below the threshold.
    CrossedBelow(f64),
    /// Became `true`.
    BecameTrue,
    /// Became `false`.
    BecameFalse,
}

/// A data-driven state watcher: when `key` satisfies the condition, run the
/// effects. Watchers receive the live session writer, so the full
/// [`EffectSpec`] vocabulary applies — a watcher can set state, inject
/// context, prompt the model, or remember durably.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct WatchSpec {
    /// Watched state key.
    pub key: String,
    /// Trigger condition.
    pub condition: WatchCondition,
    /// State keys written when it fires (sugar for an [`EffectSpec::Set`]).
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub set: BTreeMap<String, Value>,
    /// Effects fired in order after `set`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub effects: Vec<EffectSpec>,
}

/// A data-driven temporal pattern: fire effects when a state condition holds
/// continuously — for a duration (`sustained_secs`) or a number of
/// consecutive turns (`turns`). Exactly one of the two must be set.
///
/// This is the "the caller has sounded confused for 30 seconds" /
/// "we've been stuck on this for 3 turns" reactor, as data.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct PatternSpec {
    /// Pattern name (diagnostic identity).
    pub name: String,
    /// Condition over session state (no marking atoms).
    pub when: Guard,
    /// Fire after the condition holds this many seconds.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sustained_secs: Option<u64>,
    /// Fire after the condition holds for this many consecutive turns.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub turns: Option<u32>,
    /// Effects when the pattern fires (`set` state and/or inject `context`).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub effects: Vec<EffectSpec>,
}

/// A computed (derived) state variable authored as data: `key` is written to
/// `derived:{key}` whenever the [`Expr`] evaluates to a value. Dependencies
/// are inferred from the expression's [`Expr::keys_read`], so the runtime's
/// dependency-ordered [`ComputedRegistry`](gemini_adk_rs::live::ComputedRegistry)
/// invariants hold with nothing extra to declare. Guards read the result by
/// its bare key (the `derived:` fallback).
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ComputedSpec {
    /// Result key (written as `derived:{key}`).
    pub key: String,
    /// The expression computing the value.
    pub from: Expr,
    /// Human-readable note.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
}

/// Declared JSON type of a state key.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum StateType {
    /// JSON boolean.
    Boolean,
    /// JSON number.
    Number,
    /// JSON string.
    String,
    /// JSON object.
    Object,
    /// JSON array.
    Array,
}

impl StateType {
    fn matches(self, value: &Value) -> bool {
        matches!(
            (self, value),
            (StateType::Boolean, Value::Bool(_))
                | (StateType::Number, Value::Number(_))
                | (StateType::String, Value::String(_))
                | (StateType::Object, Value::Object(_))
                | (StateType::Array, Value::Array(_))
        )
    }
}

/// One declared state key: its type, meaning, and optional starting value.
///
/// The `state` section is the session's data dictionary. Declaring it is
/// optional, but once present it powers editor autocomplete, key-existence
/// warnings for every guard/effect/tool reference, typed key constants in
/// generated code, and `default` seeding at connect.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct StateFieldSpec {
    /// Declared JSON type.
    #[serde(rename = "type", default, skip_serializing_if = "Option::is_none")]
    pub kind: Option<StateType>,
    /// What this key means.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
    /// Initial value seeded at connect when the key is unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub default: Option<Value>,
}

/// Project one remembered fact into a governed state slot: when memory holds
/// a value for `predicate`, it is written to the `to` state key — where
/// `needs`, `captured`, and every other guard reads it exactly as if the
/// caller had just said it.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct MemorySlotSpec {
    /// Memory predicate (e.g. `dietary_identity`).
    pub predicate: String,
    /// Target state key (must not be `derived:` — that prefix is read-only).
    pub to: String,
}

/// The session's durable-memory declaration. Installing it wires the memory
/// subsystem in through a [`MemoryBinding`] supplied in [`SpecResources`]:
/// the `recall_context` / `manage_memory` tools (ambient, so step `allow`
/// lists don't switch recall off), turn ingestion, end-of-session
/// reconciliation, and the slot projections below. [`EffectSpec::Remember`]
/// writes through the same binding.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct MemorySpec {
    /// Remembered facts projected into state slots.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub slots: Vec<MemorySlotSpec>,
}

/// Speech-detection sensitivity for [`VadSpec`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum SensitivitySpec {
    /// Fewer false positives; may miss soft speech.
    Low,
    /// Balanced.
    Medium,
    /// Catches everything; more false positives.
    High,
}

impl SensitivitySpec {
    fn to_wire(self) -> Sensitivity {
        match self {
            SensitivitySpec::Low => Sensitivity::SensitivityLow,
            SensitivitySpec::Medium => Sensitivity::SensitivityMedium,
            SensitivitySpec::High => Sensitivity::SensitivityHigh,
        }
    }
}

/// Voice-activity-detection tuning — the knobs that decide how eagerly the
/// session hears speech start and stop.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct VadSpec {
    /// Sensitivity for detecting speech onset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub start_sensitivity: Option<SensitivitySpec>,
    /// Sensitivity for detecting end of speech.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub end_sensitivity: Option<SensitivitySpec>,
    /// Milliseconds of audio kept before speech onset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prefix_padding_ms: Option<u32>,
    /// Milliseconds of silence before end-of-speech triggers.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub silence_duration_ms: Option<u32>,
}

/// Input-audio hardening: the measured mic chain (denoiser, noise gate),
/// client input-VAD tuning, and interruption authority. Lowers to
/// `Live::mic_denoise` / `mic_noise_gate` / `input_vad` /
/// `client_interruption_authority`; see the hardening chapter for the
/// benchmark behind each default.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct AudioSpec {
    /// Run the RNNoise speech enhancer over incoming user audio (requires
    /// the `denoise` feature; skipped with a validation warning otherwise).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub denoise: Option<bool>,
    /// Noise gate after the denoiser — silences frames below a level
    /// threshold (near-talker preference; rejects denoiser residue).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub noise_gate: Option<NoiseGateSpec>,
    /// Client input-VAD tuning (the detector driving speech edges in
    /// `send_audio`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub client_vad: Option<ClientVadSpec>,
    /// Who decides when user speech interrupts the model.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub authority: Option<AuthoritySpec>,
    /// Milliseconds to hold the turn-end marker during mid-turn pauses, suppressing
    /// false end-of-turn commits. Measured on TurnBench dev: 800 ms = 0.1-fp
    /// qualifying point (recall 0.798), 1600 ms = recall 0.508 (frontier).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub eot_hold_ms: Option<u32>,
    /// Milliseconds to suppress barge-in detection on backchannels and false
    /// interruptions. Measured on TurnBench dev: 1400 ms suppresses backchannel
    /// false positives (0.702 → 0.062), 2000 ms = maximum interruption match window.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub min_interruption_ms: Option<u32>,
}

/// Noise-gate stage parameters.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct NoiseGateSpec {
    /// RMS threshold in sample units (i16 full scale 32767; measured sweet
    /// spot 400–700 behind the denoiser).
    #[serde(default = "default_gate_threshold")]
    pub threshold_rms: f64,
    /// Quiet frames the gate stays open after the last loud one.
    #[serde(default = "default_gate_hold")]
    pub hold_frames: u32,
}

fn default_gate_threshold() -> f64 {
    700.0
}
fn default_gate_hold() -> u32 {
    3
}

/// Client input-VAD tuning: start from a preset, override individual knobs.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ClientVadSpec {
    /// Base preset the overrides apply to.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub preset: Option<ClientVadPreset>,
    /// Energy above the noise floor (dB) to open.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub start_threshold_db: Option<f64>,
    /// Energy above the noise floor (dB) to close.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stop_threshold_db: Option<f64>,
    /// Consecutive frames (30 ms each) to confirm onset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub min_speech_frames: Option<u32>,
    /// Frames of hangover before speech-end.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hangover_frames: Option<u32>,
}

/// Named client-VAD starting points.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ClientVadPreset {
    /// The library defaults (quiet environments).
    Default,
    /// The closed-loop-tuned noisy-environment profile — use behind
    /// `denoise: true` (see `VadConfig::noisy_street`).
    NoisyStreet,
}

impl ClientVadSpec {
    /// Lower to a wire `VadConfig`: preset base plus overrides.
    pub fn to_config(&self) -> gemini_genai_rs::vad::VadConfig {
        let mut config = match self.preset {
            Some(ClientVadPreset::NoisyStreet) => gemini_genai_rs::vad::VadConfig::noisy_street(),
            _ => gemini_genai_rs::vad::VadConfig::default(),
        };
        if let Some(v) = self.start_threshold_db {
            config.start_threshold_db = v;
        }
        if let Some(v) = self.stop_threshold_db {
            config.stop_threshold_db = v;
        }
        if let Some(v) = self.min_speech_frames {
            config.min_speech_frames = v;
        }
        if let Some(v) = self.hangover_frames {
            config.hangover_frames = v;
        }
        config
    }
}

/// Interruption authority (measured trade: client is ~2× faster to barge
/// in; server posted zero false interruptions in every benchmark run).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum AuthoritySpec {
    /// The Live API's automatic activity detection decides (default).
    Server,
    /// This client's input VAD decides: automatic detection is disabled and
    /// speech edges send activityStart/activityEnd.
    Client,
}

/// Input/output transcription toggles.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, schemars::JsonSchema)]
pub struct TranscriptionSpec {
    /// Transcribe the user's speech.
    #[serde(default = "default_true")]
    pub input: bool,
    /// Transcribe the model's audio output.
    #[serde(default = "default_true")]
    pub output: bool,
}

fn default_true() -> bool {
    true
}

/// How phase instructions are steered to the model.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum SteeringSpec {
    /// Replace the system instruction on phase transitions (default).
    InstructionUpdate,
    /// Set the instruction once; deliver phase steering as context turns.
    ContextInjection,
    /// Both.
    Hybrid,
    /// Replace the instruction and the declared tools with `contextUpdate` at
    /// each phase and flow-step change, so the model is offered only the
    /// current tools (Gemini 3.8 Live; `hybrid` on other models).
    ContextUpdate,
}

/// When batched context turns hit the wire.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ContextDeliverySpec {
    /// Send immediately during TurnComplete (default).
    Immediate,
    /// Queue and flush with the next user send (voice-glitch avoidance).
    Deferred,
}

/// Conversation-repair thresholds (unmet phase `needs`).
#[derive(Debug, Clone, Copy, Serialize, Deserialize, schemars::JsonSchema)]
pub struct RepairSpec {
    /// Turns without progress before the first nudge.
    pub nudge_after: u32,
    /// Turns without progress before escalation.
    pub escalate_after: u32,
}

/// Session persistence backend.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum PersistenceSpec {
    /// Filesystem snapshots under the given directory.
    Fs {
        /// Snapshot directory (created if missing).
        dir: String,
    },
    /// In-memory (tests, ephemeral sessions).
    Memory,
}

/// Control-plane and voice tuning — every session capability that is
/// configuration rather than conversation, in one section. Everything here
/// lowers to a `Live` builder setter; omitted fields keep the builder's
/// defaults.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct RuntimeSpec {
    /// Sampling temperature.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub temperature: Option<f32>,
    /// Thinking budget in tokens (Google AI only; stripped on Vertex).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub thinking_budget: Option<u32>,
    /// Receive thought summaries (requires `thinking_budget`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub include_thoughts: Option<bool>,
    /// Input/output transcription.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub transcription: Option<TranscriptionSpec>,
    /// Let the model choose to stay silent (proactive audio).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub proactive_audio: Option<bool>,
    /// Voice-activity-detection tuning.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub vad: Option<VadSpec>,
    /// Input-audio hardening: mic chain, client VAD, interruption authority.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub audio: Option<AudioSpec>,
    /// Fire a soft turn when the model stays silent this long after VAD end.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub soft_turn_timeout_ms: Option<u64>,
    /// How phase instructions reach the model.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub steering: Option<SteeringSpec>,
    /// When context turns hit the wire.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub context_delivery: Option<ContextDeliverySpec>,
    /// Conversation-repair thresholds.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub repair: Option<RepairSpec>,
    /// Session persistence backend.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub persistence: Option<PersistenceSpec>,
    /// Stable session id (resume across restarts; requires `persistence`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Drop audio chunks instead of applying backpressure when consumers lag.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub lossy_audio: bool,
    /// Drop transcript deltas instead of applying backpressure.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub lossy_transcript: bool,
}

/// Splice a named flow fragment into the session's flow under a namespace.
///
/// Every step id inside the fragment becomes `{namespace}/{id}`; internal
/// `after` edges, `done(step)` guard atoms, and `before`/`require` constraints
/// are rewritten to match. Fragment root steps (no `after`) gain this entry's
/// `after` dependencies, attaching the fragment into the outer DAG. Steps
/// outside the fragment reference its steps as `{namespace}/{id}`.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct UseFragment {
    /// Fragment name (a key in [`SessionSpec::fragments`]).
    pub fragment: String,
    /// Namespace prefix for the spliced step ids.
    pub namespace: String,
    /// Outer steps the fragment's roots depend on.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub after: Vec<String>,
}

/// A complete Live session as one JSON document. See the module docs.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct SessionSpec {
    /// App name (display and registry identity).
    #[serde(default)]
    pub name: String,
    /// Spec version tag (freeform, e.g. "1" or "2025-08-24").
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub version: String,
    /// Human-readable description.
    #[serde(default)]
    pub description: String,
    /// Base system instruction.
    #[serde(default)]
    pub instruction: String,
    /// Optional greeting prompt — the model speaks first on connect.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub greeting: Option<String>,
    /// Output modality. Defaults to text.
    #[serde(default)]
    pub modality: SpecModality,
    /// Voice name for audio sessions (e.g. "Puck").
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub voice: Option<String>,
    /// Declared tools (mock or HTTP-bound).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tools: Vec<ToolSpec>,
    /// MCP toolset connection strings — the whole MCP ecosystem as this app's
    /// tool library. Resolved at connect time.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub mcp: Vec<String>,
    /// Out-of-band extraction pipelines.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub extract: Vec<ExtractSpec>,
    /// The models the spec runs on, by role.
    #[serde(default, skip_serializing_if = "is_default_models")]
    pub models: ModelsSpec,
    /// Questions a decision model (such as Jev) answers about the
    /// conversation, by id. Guards name them with the `decided` atom, and the
    /// runtime asks the ones the flow can act on at each decision point.
    /// Requires [`SpecResources::decision_model`].
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub decisions: BTreeMap<String, DecisionSpec>,
    /// A code tool: the model runs a short Python program whose functions
    /// are the declared tools, each call governed like a direct call. Use it
    /// for reads the model would otherwise add up, sort or join itself.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub code: Option<CodeSpec>,
    /// Declared state keys — the session's data dictionary.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub state: BTreeMap<String, StateFieldSpec>,
    /// Computed (derived) state variables.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub computed: Vec<ComputedSpec>,
    /// Durable memory: slots projected into state, `remember` effects, and
    /// the ambient recall/manage tools. Requires a [`MemoryBinding`] in
    /// [`SpecResources`] at apply time.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub memory: Option<MemorySpec>,
    /// Control-plane and voice tuning.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runtime: Option<RuntimeSpec>,
    /// Conversation phases.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub phases: Vec<PhaseSpec>,
    /// Initial phase name (required when `phases` is non-empty).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub initial_phase: Option<String>,
    /// State watchers.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub watch: Vec<WatchSpec>,
    /// Temporal patterns (sustained / consecutive-turn conditions).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub patterns: Vec<PatternSpec>,
    /// Reusable flow fragments, spliced via `use_fragments`.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub fragments: BTreeMap<String, Flow>,
    /// Fragment splice directives.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub use_fragments: Vec<UseFragment>,
    /// The governed flow DAG (optional — a spec may be phases-only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub flow: Option<Flow>,
    /// The conversation: stages, slots, digressions, commits, timing and
    /// policies, compiled by the [conversation compiler](crate::conversation).
    /// The higher-level alternative to `flow`; set one or the other. Stage
    /// resolvers bind to the declared `tools` by name.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub conversation: Option<crate::conversation::ConversationSpec>,
    /// Reusable capabilities, each with task-private state and operations.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub skills: Vec<SkillSpec>,
    /// Model-free journeys through skill activation, approval and cancellation.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub task_scenarios: Vec<TaskScenario>,
    /// Scenarios run against `conversation` by
    /// [`SessionSpec::run_scenarios`]: model-free, deterministic.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub scenarios: Vec<crate::simulation::Scenario>,
    /// Embedded conformance tests, replayed offline by
    /// [`SessionSpec::run_tests`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tests: Vec<SpecTest>,
}

/// Structured result of validating a [`SessionSpec`].
#[derive(Debug, Clone, Serialize)]
pub struct SpecValidation {
    /// Whether the spec compiled cleanly.
    pub valid: bool,
    /// Errors (empty when valid).
    pub errors: Vec<String>,
    /// Non-fatal advisories.
    pub warnings: Vec<String>,
    /// Mermaid rendering of the effective (fragment-spliced) flow.
    pub mermaid: String,
    /// Every tool name the flow references.
    pub tools: Vec<String>,
    /// Steps in the effective flow.
    pub steps: usize,
}

/// The seam through which a memory engine plugs into a spec-driven session.
///
/// The spec's `memory` section is pure data; the engine that honors it lives
/// above this crate (`gemini-memory-rs` implements this trait over its
/// `MemorySession`). `apply` calls [`install`](Self::install) once to wire
/// tools/ingestion/slots, and routes every [`EffectSpec::Remember`] through
/// [`remember`](Self::remember) for a root conversation. Skills instead bind
/// owned services through [`task_memory`](Self::task_memory), contextual tools
/// through [`task_tools`](Self::task_tools), and one shared task lifecycle.
pub trait MemoryBinding: Send + Sync {
    /// Wire the memory subsystem onto the builder per the spec's declaration.
    fn install(&self, live: Live, memory: &MemorySpec) -> Live;
    /// Durably remember a note (fire-and-forget; implementations may commit
    /// asynchronously).
    fn remember(&self, note: String);
    /// Bind task-local projections to this session's existing memory backend.
    /// Implementations must not capture a task's authoritative state.
    fn task_memory(
        &self,
        _memory: &MemorySpec,
    ) -> Result<Arc<dyn gemini_adk_rs::tasks::TaskMemoryService>, String> {
        Err("this memory binding does not support task-owned memory services".into())
    }
    /// Existing memory operations, bound to owned task invocation contexts.
    /// The task compiler supplies their read/commit policy and ambient admission.
    fn task_tools(&self) -> Vec<Arc<dyn ToolFunction>> {
        Vec::new()
    }
    /// Install shared session teardown once, without global extractors or tools.
    /// Serialize teardown with started effects and reject writes after sealing.
    fn install_task_lifecycle(&self, live: Live) -> Live {
        live
    }
}

/// What a spec written by someone else may reach when it runs.
///
/// A spec can bind tools to the outside world: `mcp` entries start a local
/// command (or connect to a URL) and an `http` binding sends a request to any
/// URL. That is fine for a spec you wrote and deploy. It is not fine for a
/// spec a browser posts to a shared server: running it would let the author
/// execute commands on the host or reach internal addresses. Pass such a spec
/// through [`SessionSpec::sandboxed`] with an allowlist the *operator*
/// configured.
///
/// The default allows nothing: every binding becomes a mock.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct BindingAllowlist {
    http_prefixes: Vec<String>,
    mcp: Vec<String>,
}

impl BindingAllowlist {
    /// Allow HTTP bindings whose URL template starts with `prefix`, e.g.
    /// `https://api.example.com/`. A prefix without a path gets a trailing
    /// `/`, so interpolated arguments can never change the host.
    pub fn allow_http_prefix(mut self, prefix: impl Into<String>) -> Self {
        let mut prefix = prefix.into();
        let after_scheme = prefix.split_once("://").map_or("", |(_, rest)| rest);
        if !after_scheme.contains('/') {
            prefix.push('/');
        }
        self.http_prefixes.push(prefix);
        self
    }

    /// Allow one `mcp` entry, matched exactly.
    pub fn allow_mcp(mut self, params: impl Into<String>) -> Self {
        self.mcp.push(params.into());
        self
    }

    /// Whether `url` falls inside an allowed prefix. Both are parsed and
    /// normalized first, so `..` and `%2e%2e` segments cannot climb out of a
    /// path prefix, and scheme, host and port must match exactly. A template
    /// that interpolates into the host never matches.
    fn allows_http(&self, url: &str) -> bool {
        let Ok(target) = url::Url::parse(url) else {
            return false;
        };
        if !target.username().is_empty() || target.password().is_some() {
            return false;
        }
        self.http_prefixes.iter().any(|prefix| {
            url::Url::parse(prefix).is_ok_and(|prefix| {
                prefix.scheme() == target.scheme()
                    && prefix.host() == target.host()
                    && prefix.port_or_known_default() == target.port_or_known_default()
                    && target.path().starts_with(prefix.path())
            })
        })
    }

    fn allows_mcp(&self, params: &str) -> bool {
        self.mcp.iter().any(|m| m.trim() == params.trim())
    }
}

/// State-key prefixes the runtime itself writes (flow marking, conversation
/// signals), so a guard that reads one is not reading an orphan key.
const RUNTIME_WRITTEN: [&str; 6] = [
    "flow:",
    "session:",
    "verbatim:",
    "correction:",
    "repair:",
    "telephony:",
];

/// The outcome of one scenario in [`SessionSpec::run_scenarios`].
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ScenarioReport {
    /// The scenario's name.
    pub name: String,
    /// Whether every step held.
    pub passed: bool,
    /// The first step that failed, and why.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// Tool names a [`MemoryBinding`] installs (ambient on the flow).
pub const MEMORY_TOOL_NAMES: [&str; 2] = ["recall_context", "manage_memory"];

/// External resources a spec cannot carry: model handles and capability
/// bindings.
#[derive(Default, Clone)]
pub struct SpecResources {
    /// The OOB model backing `extract` entries that name no model (neither
    /// their own `model` nor `models.extraction`). Required when any do.
    pub extraction_llm: Option<Arc<dyn BaseLlm>>,
    /// Resolves the model names the spec writes (`models.extraction`, an
    /// extract entry's `model`) to models. A name it does not resolve that
    /// starts with `gemini` runs on [`GeminiLlm`](gemini_adk_rs::llm::GeminiLlm);
    /// register a prefix here to run another provider, or `""` to send every
    /// name to one model (a mock, in offline tests).
    pub models: Option<Arc<gemini_adk_rs::llm::LlmRegistry>>,
    /// The decision model answering the spec's `decisions`, such as
    /// `GatewayDecisionModel` (Jev on Vercel AI Gateway). Required when any
    /// are declared.
    pub decision_model: Option<Arc<dyn gemini_adk_rs::decision::DecisionModel>>,
    /// The memory engine honoring the spec's `memory` section. Required when
    /// that section is present.
    pub memory: Option<Arc<dyn MemoryBinding>>,
    /// Where the spec's `code` runs programs, such as a
    /// [`MontyPool`](gemini_adk_rs::code_mode) of worker processes. Unset:
    /// in this process, which cannot bound a program's memory.
    pub code_sandbox: Option<Arc<dyn gemini_adk_rs::code_mode::CodeSandbox>>,
    /// In-process implementations of declared tools, by name. A declared
    /// tool with an implementation calls it instead of its mock response,
    /// HTTP or MCP binding; its declaration (what the model sees) and its
    /// `set_state` stay as the spec says. Add with
    /// [`implement`](Self::implement).
    pub tools: BTreeMap<String, Arc<dyn ToolFunction>>,
}

impl SpecResources {
    /// Bind an implementation to one skill, even when skills reuse a tool name.
    pub fn implement_skill(mut self, skill: &str, tool: impl ToolFunction + 'static) -> Self {
        self.tools
            .insert(format!("{skill}/{}", tool.name()), Arc::new(tool));
        self
    }

    /// Implement the declared tool of the same name with `tool`.
    pub fn implement(mut self, tool: impl ToolFunction + 'static) -> Self {
        self.tools.insert(tool.name().to_string(), Arc::new(tool));
        self
    }
}

fn is_default_models(models: &ModelsSpec) -> bool {
    *models == ModelsSpec::default()
}

/// Resolve a named model: the host's registry first, then Gemini for
/// `gemini-*` names. One client per name.
fn resolve_model(
    name: &str,
    resources: &SpecResources,
    cache: &mut BTreeMap<String, Arc<dyn BaseLlm>>,
) -> Result<Arc<dyn BaseLlm>, String> {
    if let Some(llm) = cache.get(name) {
        return Ok(llm.clone());
    }
    let llm = match resources.models.as_ref().and_then(|r| r.resolve(name)) {
        Some(llm) => llm,
        None if name.starts_with("gemini") || name.starts_with("models/gemini") => Arc::new(
            gemini_adk_rs::llm::GeminiLlm::try_new(gemini_adk_rs::llm::GeminiLlmParams {
                model: Some(name.to_string()),
                ..Default::default()
            })
            .map_err(|e| format!("model '{name}': {e}"))?,
        )
            as Arc<dyn BaseLlm>,
        None => {
            return Err(format!(
                "model '{name}': nothing resolves it; register it in SpecResources.models"
            ));
        }
    };
    cache.insert(name.to_string(), llm.clone());
    Ok(llm)
}

/// The model an extract entry runs on: its own `model`, then the spec's
/// `models.extraction`, then the host's `extraction_llm`.
pub(crate) fn extraction_llm_for(
    entry: &ExtractSpec,
    default: Option<&str>,
    resources: &SpecResources,
    cache: &mut BTreeMap<String, Arc<dyn BaseLlm>>,
) -> Result<Arc<dyn BaseLlm>, String> {
    match entry.model.as_deref().or(default) {
        Some(name) => resolve_model(name, resources, cache)
            .map_err(|e| format!("extract '{}': {e}", entry.name)),
        None => resources.extraction_llm.clone().ok_or_else(|| {
            format!(
                "extract '{}' names no model and SpecResources.extraction_llm is not set; \
                 set models.extraction",
                entry.name
            )
        }),
    }
}

impl SessionSpec {
    /// Whether some extraction (the spec's or a skill's `extract` entries)
    /// names no model, so it runs on the host's `extraction_llm`.
    pub fn uses_default_extraction_model(&self) -> bool {
        let unnamed = |entries: &[ExtractSpec], default: Option<&str>| {
            default.is_none() && entries.iter().any(|e| e.model.is_none())
        };
        unnamed(&self.extract, self.models.extraction.as_deref())
            || self
                .skills
                .iter()
                .any(|skill| unnamed(&skill.extract, self.models.extraction.as_deref()))
    }

    /// Whether root or task definitions need an out-of-band extraction model.
    pub fn requires_extraction(&self) -> bool {
        !self.extract.is_empty()
            || self.conversation.as_ref().is_some_and(|conversation| {
                conversation
                    .stages
                    .iter()
                    .chain(
                        conversation
                            .overlays
                            .iter()
                            .flat_map(|overlay| &overlay.stages),
                    )
                    .any(|stage| stage.frame.is_some())
            })
            || self.skills.iter().any(|skill| {
                !skill.extract.is_empty()
                    || skill.conversation.as_ref().is_some_and(|conversation| {
                        conversation
                            .stages
                            .iter()
                            .chain(
                                conversation
                                    .overlays
                                    .iter()
                                    .flat_map(|overlay| &overlay.stages),
                            )
                            .any(|stage| stage.frame.is_some())
                    })
            })
    }

    /// Whether root or task definitions use the contextual memory binding.
    pub fn requires_memory(&self) -> bool {
        self.memory.is_some() || self.skills.iter().any(|skill| skill.memory.is_some())
    }

    /// Parse a spec from a JSON value. Accepts a full document or a *bare
    /// flow* (`{"steps": [...]}`), which is wrapped in a default spec.
    pub fn from_value(value: Value) -> Result<Self, String> {
        let is_bare_flow = value.get("flow").is_none() && value.get("steps").is_some();
        if is_bare_flow {
            let flow: Flow =
                serde_json::from_value(value).map_err(|e| format!("invalid flow JSON: {e}"))?;
            return Ok(Self {
                flow: Some(flow),
                ..Self::default()
            });
        }
        serde_json::from_value(value).map_err(|e| format!("invalid session spec JSON: {e}"))
    }

    /// The JSON Schema of the spec document itself — for editor autocomplete
    /// and for validating machine-authored specs at generation time.
    pub fn json_schema() -> Value {
        serde_json::to_value(schemars::schema_for!(SessionSpec)).unwrap_or_else(|_| json!({}))
    }

    /// Declared tool names (mock/HTTP; MCP names resolve at connect).
    pub fn tool_names(&self) -> Vec<String> {
        self.tools.iter().map(|t| t.name.clone()).collect()
    }

    /// The flow with every `use_fragments` directive spliced in.
    pub fn effective_flow(&self) -> Result<Flow, Vec<String>> {
        if let Some(conversation) = &self.conversation {
            if self.flow.is_some() || !self.use_fragments.is_empty() {
                return Err(vec![
                    "set either `conversation` or `flow` (with `use_fragments`), not both".into(),
                ]);
            }
            return crate::conversation::Conversation::from_spec_stubbing_resolvers(
                conversation.clone(),
            )
            .map(|compiled| compiled.flow().flow().clone())
            .map_err(|e| vec![format!("conversation: {e}")]);
        }
        let mut flow = self.flow.clone().unwrap_or_default();
        let mut errors = Vec::new();
        for use_frag in &self.use_fragments {
            match self.fragments.get(&use_frag.fragment) {
                Some(fragment) => {
                    splice_fragment(&mut flow, fragment, use_frag, &mut errors);
                }
                None => errors.push(format!(
                    "use_fragments references unknown fragment '{}'",
                    use_frag.fragment
                )),
            }
        }
        if errors.is_empty() {
            Ok(flow)
        } else {
            Err(errors)
        }
    }

    /// Every state key the session declares a *writer* for: tool `set_state`
    /// and `save_response_as`, extraction names and promotion targets, phase
    /// and watcher `set` effects.
    pub fn state_keys_written(&self) -> std::collections::BTreeSet<String> {
        let mut keys = std::collections::BTreeSet::new();
        for t in &self.tools {
            keys.extend(t.set_state.keys().cloned());
            keys.extend(t.save_response_as.iter().cloned());
        }
        for e in &self.extract {
            keys.insert(e.name.clone());
            for p in &e.promote {
                keys.insert(p.target().to_string());
            }
        }
        keys.extend(self.decisions.values().filter_map(|d| d.writes.clone()));
        for p in &self.phases {
            for eff in &p.on_enter {
                if let EffectSpec::Set(map) = eff {
                    keys.extend(map.keys().cloned());
                }
            }
        }
        for w in &self.watch {
            keys.extend(w.set.keys().cloned());
            for eff in &w.effects {
                if let EffectSpec::Set(map) = eff {
                    keys.extend(map.keys().cloned());
                }
            }
        }
        for p in &self.patterns {
            for eff in &p.effects {
                if let EffectSpec::Set(map) = eff {
                    keys.extend(map.keys().cloned());
                }
            }
        }
        for c in &self.computed {
            // A computed var writes `derived:{key}`, and guards read it by
            // either name thanks to the `derived:` fallback.
            keys.insert(c.key.clone());
            keys.insert(format!("derived:{}", c.key));
        }
        if let Some(memory) = &self.memory {
            for slot in &memory.slots {
                keys.extend([slot.to.clone()]);
            }
        }
        for (key, field) in &self.state {
            if field.default.is_some() {
                keys.insert(key.clone());
            }
        }
        // A conversation's slots are filled by its extractors and resolvers.
        for stage in self.conversation_stages() {
            keys.extend(stage.collect.iter().cloned());
            keys.extend(stage.resolve.iter().map(|r| r.slot.clone()));
            if let Some(frame) = &stage.frame {
                keys.extend(frame.slot_keys());
            }
        }
        keys
    }

    /// Every stage of the conversation, main flow and digressions.
    fn conversation_stages(&self) -> impl Iterator<Item = &crate::conversation::StageSpec> {
        self.conversation.iter().flat_map(|c| {
            c.stages
                .iter()
                .chain(c.overlays.iter().flat_map(|o| o.stages.iter()))
        })
    }

    /// Resolvers for the conversation's `resolve` slots, each bound to the
    /// declared tool of the same name (the resolver name, or the slot).
    fn resolver_registry(
        &self,
        tools: &[Arc<SimpleTool>],
    ) -> crate::conversation::ResolverRegistry {
        let mut registry = crate::conversation::ResolverRegistry::new();
        for stage in self.conversation_stages() {
            for resolve in &stage.resolve {
                let name = resolve
                    .resolver
                    .clone()
                    .unwrap_or_else(|| resolve.slot.clone());
                if let Some(tool) = tools
                    .iter()
                    .find(|t| ToolFunction::name(t.as_ref()) == name)
                {
                    let tool = tool.clone();
                    registry.add(name, move |args| {
                        let tool = tool.clone();
                        async move {
                            gemini_adk_rs::tool::ToolFunction::call(tool.as_ref(), args)
                                .await
                                .map_err(|e| e.to_string())
                        }
                    });
                }
            }
        }
        registry
    }

    /// Run [`scenarios`](Self::scenarios) against the conversation, in
    /// order. Resolvers are stubbed: a scenario supplies their values with
    /// `set` steps.
    pub async fn run_scenarios(&self) -> Vec<ScenarioReport> {
        let failed = |error: String| -> Vec<ScenarioReport> {
            self.scenarios
                .iter()
                .map(|s| ScenarioReport {
                    name: s.name.clone(),
                    passed: false,
                    error: Some(error.clone()),
                })
                .collect()
        };
        let Some(conversation) = &self.conversation else {
            return failed("the spec has no `conversation` to run against".into());
        };
        let compiled = match crate::conversation::Conversation::from_spec_stubbing_resolvers(
            conversation.clone(),
        ) {
            Ok(compiled) => compiled,
            Err(e) => return failed(format!("conversation: {e}")),
        };
        let bank: BTreeMap<String, gemini_adk_rs::decision::Decision> = self
            .decisions
            .iter()
            .map(|(id, d)| (id.clone(), d.compile()))
            .collect();
        let mut reports = Vec::with_capacity(self.scenarios.len());
        for scenario in &self.scenarios {
            let sim =
                crate::simulation::Sim::new(&compiled, gemini_adk_rs::flow::Enforcement::Enforce)
                    .with_decisions(bank.clone());
            let result = scenario.run_in(sim).await;
            reports.push(ScenarioReport {
                name: scenario.name.clone(),
                passed: result.is_ok(),
                error: result.err(),
            });
        }
        reports
    }

    /// Every [`EffectSpec`] anywhere in the document, with a location label.
    fn all_effects(&self) -> Vec<(String, &EffectSpec)> {
        let mut out = Vec::new();
        for p in &self.phases {
            for eff in &p.on_enter {
                out.push((format!("phase '{}'", p.name), eff));
            }
        }
        for w in &self.watch {
            for eff in &w.effects {
                out.push((format!("watch '{}'", w.key), eff));
            }
        }
        for p in &self.patterns {
            for eff in &p.effects {
                out.push((format!("pattern '{}'", p.name), eff));
            }
        }
        out
    }

    /// Re-evaluate every computed variable in dependency order, writing
    /// results to `derived:{key}`. Used by the offline simulator so guards
    /// over computed keys latch exactly as they do live. Validation forbids
    /// cycles, so iterating to a fixed point terminates.
    pub(crate) fn recompute_computed(&self, state: &State) {
        for _ in 0..self.computed.len() {
            let mut changed = false;
            for c in &self.computed {
                if let Some(value) = c.from.eval(state) {
                    let derived = format!("derived:{}", c.key);
                    if state.get_raw(&derived).as_ref() != Some(&value) {
                        let _ = state.set(&derived, value);
                        changed = true;
                    }
                }
            }
            if !changed {
                break;
            }
        }
    }

    /// Seed declared `state` defaults for keys not yet set.
    pub(crate) fn seed_state_defaults(&self, state: &State) {
        for (key, field) in &self.state {
            if let Some(default) = &field.default
                && state.get_raw(key).is_none()
            {
                let _ = state.set(key, default.clone());
            }
        }
    }

    /// Validate the whole document: flow compilation (with the declared tool
    /// registry), fragment splicing, phase-guard restrictions, HTTP-binding
    /// support, and the read/write state-key diff.
    pub fn validate(&self) -> SpecValidation {
        self.validate_with_http_support(true)
    }

    /// Validate for scripted offline replay, where declared tools use mock
    /// effects and never open HTTP or MCP connections.
    ///
    /// Checks the original document, including conflicting tool bindings.
    /// Only the requirement to compile HTTP transport support is omitted.
    pub fn validate_for_replay(&self) -> SpecValidation {
        self.validate_with_http_support(false)
    }

    fn validate_with_http_support(&self, require_http_support: bool) -> SpecValidation {
        self.validate_parts(require_http_support, false, &Default::default())
    }

    fn validate_definition(
        &self,
        require_http_support: bool,
        activation_inputs: &std::collections::BTreeSet<String>,
    ) -> SpecValidation {
        self.validate_parts(require_http_support, true, activation_inputs)
    }

    fn validate_parts(
        &self,
        require_http_support: bool,
        allow_ungoverned: bool,
        activation_inputs: &std::collections::BTreeSet<String>,
    ) -> SpecValidation {
        let mut errors = Vec::new();
        let mut warnings = Vec::new();
        let mut skill_tools = Vec::new();
        let mut skill_steps = 0;
        let mut written = self.state_keys_written();
        written.extend(activation_inputs.iter().cloned());

        if !self.skills.is_empty() {
            if self.flow.is_some()
                || self.conversation.is_some()
                || !self.phases.is_empty()
                || !self.tools.is_empty()
                || !self.mcp.is_empty()
                || !self.extract.is_empty()
                || !self.decisions.is_empty()
                || !self.computed.is_empty()
                || !self.watch.is_empty()
                || !self.patterns.is_empty()
                || !self.use_fragments.is_empty()
                || self.memory.is_some()
                || !self.state.is_empty()
            {
                errors.push("skills own their state, tools and governance; do not combine them with a session-level pipeline".into());
            }
            let mut names = std::collections::BTreeSet::new();
            for skill in &self.skills {
                if !names.insert(&skill.name) {
                    errors.push(format!(
                        "skill '{}' is installed more than once",
                        skill.name
                    ));
                }
                let validation = skill.validation(require_http_support);
                skill_steps += validation.steps;
                skill_tools.extend(
                    skill
                        .tools
                        .iter()
                        .map(|tool| format!("{}__{}", skill.name, tool.tool.name)),
                );
                errors.extend(
                    validation
                        .errors
                        .into_iter()
                        .map(|problem| format!("skill '{}': {problem}", skill.name)),
                );
                warnings.extend(
                    validation
                        .warnings
                        .into_iter()
                        // The session reports unpinned extraction once, with
                        // its own `models.extraction` taken into account.
                        .filter(|warning| warning != UNPINNED_EXTRACTION_WARNING)
                        .map(|warning| format!("skill '{}': {warning}", skill.name)),
                );
            }
        } else if !self.task_scenarios.is_empty() {
            errors.push("task_scenarios require skills".into());
        }

        // Fragment namespace validation: must be non-empty to create valid step ids.
        for use_frag in &self.use_fragments {
            if use_frag.namespace.is_empty() {
                errors.push(
                    "use_fragments directive has empty namespace — step ids would be malformed"
                        .into(),
                );
            }
        }

        let flow = match self.effective_flow() {
            Ok(flow) => flow,
            Err(errs) => {
                errors.extend(errs);
                self.flow.clone().unwrap_or_default()
            }
        };
        for stage in self.conversation_stages() {
            for resolve in &stage.resolve {
                let name = resolve.resolver.as_deref().unwrap_or(&resolve.slot);
                if !self.tools.iter().any(|t| t.name == name) {
                    errors.push(format!(
                        "stage '{}' resolves '{}' with '{name}', which is not a declared tool",
                        stage.id, resolve.slot
                    ));
                }
            }
        }
        if !self.scenarios.is_empty() && self.conversation.is_none() {
            errors.push("`scenarios` need a `conversation` to run against".into());
        }
        let mermaid = flow.to_mermaid();
        let steps = flow.steps.len();
        let has_flow = !flow.steps.is_empty();

        if !has_flow && self.phases.is_empty() && self.skills.is_empty() && !allow_ungoverned {
            errors.push("spec has neither a flow nor phases — nothing to run".into());
        }
        if !self.phases.is_empty() && self.initial_phase.is_none() {
            errors.push("phases are declared but initial_phase is not set".into());
        }
        if let Some(initial) = &self.initial_phase
            && !self.phases.iter().any(|p| &p.name == initial)
        {
            errors.push(format!("initial_phase '{initial}' is not a declared phase"));
        }
        // Check for duplicate phase names
        {
            let phase_names: std::collections::BTreeSet<&str> =
                self.phases.iter().map(|p| p.name.as_str()).collect();
            if phase_names.len() != self.phases.len() {
                let mut seen = std::collections::BTreeSet::new();
                for p in &self.phases {
                    if !seen.insert(p.name.as_str()) {
                        errors.push(format!("phase '{}' is declared more than once", p.name));
                    }
                }
            }
        }
        for pattern in &self.patterns {
            match (pattern.sustained_secs, pattern.turns) {
                (Some(_), Some(_)) | (None, None) => errors.push(format!(
                    "pattern '{}' must set exactly one of sustained_secs or turns",
                    pattern.name
                )),
                _ => {}
            }
            if guard_uses_marking(&pattern.when) {
                errors.push(format!(
                    "pattern '{}' uses a called_ok/done atom — pattern guards see state only",
                    pattern.name
                ));
            }
        }
        for p in &self.phases {
            for t in &p.transitions {
                if guard_uses_marking(&t.when) {
                    errors.push(format!(
                        "phase '{}' transition to '{}' uses a called_ok/done atom — phase guards \
                         see state only (no flow marking); latch a state key instead",
                        p.name, t.to
                    ));
                }
            }
        }
        for t in &self.tools {
            if t.http.is_some() && t.mcp.is_some() {
                errors.push(format!(
                    "tool '{}' has both an http and an mcp binding; keep one",
                    t.name
                ));
            }
        }
        if require_http_support && cfg!(not(feature = "http-tools")) {
            for t in &self.tools {
                if t.http.is_some() {
                    errors.push(format!(
                        "tool '{}' has an http binding but the `http-tools` feature is not \
                         enabled",
                        t.name
                    ));
                }
            }
        }

        // Computed variables: unique keys, no dependency cycles (Kahn over the
        // computed subset; a bare read and a `derived:` read are the same
        // dependency).
        {
            let computed_keys: std::collections::BTreeSet<&str> =
                self.computed.iter().map(|c| c.key.as_str()).collect();
            if computed_keys.len() != self.computed.len() {
                errors.push("computed variables declare a duplicate key".into());
            }
            let normalize = |k: &str| k.strip_prefix("derived:").unwrap_or(k).to_string();
            let mut in_degree: BTreeMap<&str, usize> =
                computed_keys.iter().map(|k| (*k, 0)).collect();
            let mut dependents: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
            for c in &self.computed {
                for dep in c.from.keys_read() {
                    let dep = normalize(&dep);
                    if dep != c.key && computed_keys.contains(dep.as_str()) {
                        let dep_key = *computed_keys.get(dep.as_str()).unwrap();
                        dependents.entry(dep_key).or_default().push(c.key.as_str());
                        *in_degree.entry(c.key.as_str()).or_default() += 1;
                    }
                    if dep == c.key {
                        errors.push(format!("computed '{}' reads its own key", c.key));
                    }
                }
            }
            let mut queue: Vec<&str> = in_degree
                .iter()
                .filter(|(_, d)| **d == 0)
                .map(|(k, _)| *k)
                .collect();
            let mut visited = 0usize;
            while let Some(key) = queue.pop() {
                visited += 1;
                for dependent in dependents.get(key).cloned().unwrap_or_default() {
                    let d = in_degree.get_mut(dependent).unwrap();
                    *d -= 1;
                    if *d == 0 {
                        queue.push(dependent);
                    }
                }
            }
            if visited != computed_keys.len() {
                let cycle: Vec<&str> = in_degree
                    .iter()
                    .filter(|(_, d)| **d > 0)
                    .map(|(k, _)| *k)
                    .collect();
                errors.push(format!(
                    "computed variables form a dependency cycle: {}",
                    cycle.join(", ")
                ));
            }
        }

        // Effects: `remember` needs the memory section; memory slots must not
        // target the read-only `derived:` scope.
        for (location, effect) in self.all_effects() {
            if matches!(effect, EffectSpec::Remember(_)) && self.memory.is_none() {
                errors.push(format!(
                    "{location} uses a `remember` effect but the spec has no `memory` section"
                ));
            }
        }
        if let Some(memory) = &self.memory {
            for slot in &memory.slots {
                if slot.to.starts_with("derived:") {
                    errors.push(format!(
                        "memory slot '{}' targets read-only key '{}' — the `derived:` scope \
                         belongs to computed variables",
                        slot.predicate, slot.to
                    ));
                }
            }
        }

        // Declared state dictionary: defaults must match their declared type;
        // once a dictionary exists, undeclared keys are worth flagging.
        for (key, field) in &self.state {
            if let (Some(kind), Some(default)) = (field.kind, &field.default)
                && !kind.matches(default)
            {
                warnings.push(format!(
                    "state key '{key}' declares type {kind:?} but its default is {default}"
                ));
            }
        }
        if !self.state.is_empty() {
            let declared: std::collections::BTreeSet<&str> =
                self.state.keys().map(String::as_str).collect();
            let mut undeclared = std::collections::BTreeSet::new();
            for key in &written {
                let bare = key.strip_prefix("derived:").unwrap_or(key);
                if !declared.contains(bare) && !self.computed.iter().any(|c| c.key == bare) {
                    undeclared.insert(key.clone());
                }
            }
            for key in &undeclared {
                warnings.push(format!(
                    "state key '{key}' is written but not declared in the `state` section"
                ));
            }
        }

        // Computed inputs: like guard reads, a dependency nothing writes can
        // never produce a value.
        {
            for c in &self.computed {
                for dep in c.from.keys_read() {
                    let bare = dep.strip_prefix("derived:").unwrap_or(&dep);
                    if !written.contains(&dep)
                        && !written.contains(bare)
                        && !dep.ends_with(":result")
                    {
                        warnings.push(format!(
                            "computed '{}' reads state key '{dep}' but nothing writes it",
                            c.key
                        ));
                    }
                }
            }
        }

        // Runtime coherence.
        if let Some(runtime) = &self.runtime {
            if runtime.include_thoughts == Some(true) && runtime.thinking_budget.is_none() {
                warnings.push(
                    "runtime.include_thoughts is set without runtime.thinking_budget — no \
                     thoughts will arrive"
                        .into(),
                );
            }
            if let Some(audio) = &runtime.audio {
                if audio.denoise == Some(true) && !cfg!(feature = "denoise") {
                    warnings.push(
                        "runtime.audio.denoise requires building with the `denoise` feature — \
                         the stage will be skipped"
                            .into(),
                    );
                }
                if audio.authority == Some(AuthoritySpec::Client) && audio.denoise != Some(true) {
                    warnings.push(
                        "runtime.audio.authority=client without denoise — in noise the raw \
                         energy VAD latches open and will drive interruptions falsely \
                         (measured); enable denoise or expect spurious barge-ins"
                            .into(),
                    );
                }
                if audio.noise_gate.is_some() && audio.denoise != Some(true) {
                    warnings.push(
                        "runtime.audio.noise_gate without denoise — the gate calibrates on \
                         noisy levels; chain it behind denoise so it gates clean audio"
                            .into(),
                    );
                }
                if audio.authority == Some(AuthoritySpec::Client) && runtime.vad.is_some() {
                    warnings.push(
                        "runtime.audio.authority=client disables the server's automatic \
                         activity detection — runtime.vad sensitivities will have no effect"
                            .into(),
                    );
                }
                if let Some(eot_ms) = audio.eot_hold_ms
                    && eot_ms > 1600
                {
                    warnings.push(
                        "runtime.audio.eot_hold_ms exceeds the measured frontier (1600 ms): \
                             recall fell to 0.508 at 1600ms on TurnBench dev — values beyond this \
                             may cause missed turn-end detection"
                            .into(),
                    );
                }
                if let Some(min_int_ms) = audio.min_interruption_ms
                    && min_int_ms > 2000
                {
                    warnings.push(
                        "runtime.audio.min_interruption_ms exceeds the interruption match \
                             window (2000 ms) — commits may land too late to count"
                            .into(),
                    );
                }
            }
            if runtime.session_id.is_some() && runtime.persistence.is_none() {
                warnings.push(
                    "runtime.session_id is set without runtime.persistence — nothing will be \
                     snapshotted"
                        .into(),
                );
            }
        }

        // Flow compilation, with declared tools as the registry. MCP tool
        // names are unknown until connect, so their presence downgrades the
        // unknown-tool check to a warning-free plain compile.
        let (valid_flow, referenced) = if has_flow {
            let compile_result = if self.tools.is_empty() || !self.mcp.is_empty() {
                if !self.mcp.is_empty() && !self.tools.is_empty() {
                    warnings.push(
                        "MCP toolsets resolve at connect time, so tool-name checking against \
                         the flow is skipped"
                            .into(),
                    );
                }
                flow.clone().compile()
            } else {
                let mut names = self.tool_names();
                if self.memory.is_some() {
                    // The memory binding installs its tools at connect; the
                    // flow may reference or gate them.
                    names.extend(
                        MEMORY_TOOL_NAMES
                            .iter()
                            .map(std::string::ToString::to_string),
                    );
                }
                let refs: Vec<&str> = names.iter().map(String::as_str).collect();
                flow.clone().compile_with_tools(&refs)
            };
            match compile_result {
                Ok(compiled) => (
                    true,
                    compiled
                        .tool_surface()
                        .tools
                        .iter()
                        .cloned()
                        .collect::<Vec<_>>(),
                ),
                Err(errs) => {
                    errors.extend(errs.0.iter().map(std::string::ToString::to_string));
                    (false, Vec::new())
                }
            }
        } else {
            (true, Vec::new())
        };

        if valid_flow && has_flow {
            // Read/write state-key diff — the guard-key analogue of the
            // unknown-tool check.
            for key in flow.state_keys_read() {
                if written.contains(&key) || RUNTIME_WRITTEN.iter().any(|p| key.starts_with(p)) {
                    continue;
                }
                // `{name}:result` keys are written by on_enter orchestration.
                if key.ends_with(":result") {
                    continue;
                }
                let hint = written
                    .iter()
                    .filter(|w| levenshtein(&key, w) <= 2)
                    .cloned()
                    .collect::<Vec<_>>();
                let suffix = if hint.is_empty() {
                    String::new()
                } else {
                    format!(" — did you mean {}?", hint.join(" / "))
                };
                warnings.push(format!(
                    "a guard reads state key '{key}' but no tool, extractor, phase, or watcher \
                     writes it (it can never latch){suffix}"
                ));
            }
            // Digressions are not part of the lowered main flow; the tools
            // their stages admit or commit are referenced too.
            let overlay_tools: std::collections::BTreeSet<&str> = self
                .conversation
                .iter()
                .flat_map(|c| c.overlays.iter().flat_map(|o| o.stages.iter()))
                .flat_map(|s| {
                    s.allow
                        .iter()
                        .map(String::as_str)
                        .chain(s.commit.iter().map(|c| c.tool.as_str()))
                })
                .collect();
            for t in &self.tools {
                if !referenced.contains(&t.name) && !overlay_tools.contains(t.name.as_str()) {
                    warnings.push(format!(
                        "tool '{}' is declared but no step or constraint references it \
                         (it will be denied whenever a step with an `allow` list is active \
                         unless you add it to `ambient`)",
                        t.name
                    ));
                }
            }
            for s in &flow.steps {
                if !s.terminal && s.posture.is_none() {
                    warnings.push(format!(
                        "step '{}' has no posture — the model gets no steering while it is active",
                        s.id
                    ));
                }
            }
        }

        // Extraction models.
        if self.uses_default_extraction_model() {
            warnings.push(UNPINNED_EXTRACTION_WARNING.into());
        }
        for (place, name) in self
            .extract
            .iter()
            .filter_map(|e| {
                e.model
                    .as_deref()
                    .map(|m| (format!("extract '{}'", e.name), m))
            })
            .chain(
                self.models
                    .extraction
                    .as_deref()
                    .map(|m| ("models.extraction".into(), m)),
            )
        {
            if name.trim().is_empty() {
                errors.push(format!("{place}: model is empty"));
            }
        }

        // Decisions.
        let (decision_errors, decision_warnings) = self.decision_problems();
        errors.extend(decision_errors);
        warnings.extend(decision_warnings);

        // Code mode.
        let (code_errors, code_warnings) = self.code_problems();
        errors.extend(code_errors);
        warnings.extend(code_warnings);

        SpecValidation {
            valid: errors.is_empty(),
            errors,
            warnings,
            mermaid,
            tools: referenced.into_iter().chain(skill_tools).collect(),
            steps: steps + skill_steps,
        }
    }

    /// Build the dispatcher of declared tools bound to `state`.
    pub fn build_dispatcher(&self, state: &State) -> ToolDispatcher {
        self.build_dispatcher_with(state, &SpecResources::default())
    }

    /// Build the dispatcher of declared tools bound to `state`, calling the
    /// in-process implementations in `resources` where it has them. Tools
    /// bound to the same MCP server share one connection.
    pub fn build_dispatcher_with(
        &self,
        state: &State,
        resources: &SpecResources,
    ) -> ToolDispatcher {
        let mut dispatcher = ToolDispatcher::new();
        for tool in self.bound_tools(state, resources) {
            dispatcher.register(tool);
        }
        dispatcher
    }

    /// Every declared tool, bound to its call: the in-process implementation
    /// in `resources`, else its MCP server, else its HTTP binding or mock.
    fn bound_tools(&self, state: &State, resources: &SpecResources) -> Vec<Arc<SimpleTool>> {
        let mut servers: BTreeMap<String, Arc<gemini_adk_rs::tools::mcp::McpSessionManager>> =
            BTreeMap::new();
        let mut tools = Vec::new();
        for tool in &self.tools {
            let call = match (resources.tools.get(&tool.name), &tool.mcp) {
                (Some(implementation), _) => ToolCall::Code(implementation.clone()),
                (None, Some(params)) => ToolCall::Mcp(
                    servers
                        .entry(params.clone())
                        .or_insert_with(|| {
                            Arc::new(gemini_adk_rs::tools::mcp::McpSessionManager::new(
                                crate::live::connect::parse_mcp_params(params),
                            ))
                        })
                        .clone(),
                ),
                (None, None) => ToolCall::Declared,
            };
            tools.push(Arc::new(build_tool(tool, state, call)));
        }
        tools
    }

    /// Apply the mock semantics of a declared tool to `state` (its
    /// `set_state` writes). Used by the offline simulator.
    pub(crate) fn apply_tool_state(&self, name: &str, state: &State) {
        if let Some(tool) = self.tools.iter().find(|t| t.name == name) {
            for (key, value) in &tool.set_state {
                let _ = state.set(key, value.clone());
            }
        }
    }

    /// Run the embedded test suite offline (no model, no network) — scripted
    /// events replayed through the real [`FlowStack`](gemini_adk_rs::flow::FlowStack).
    /// Includes installed skill workflow tests, named `skill/test`.
    pub fn run_tests(&self) -> Vec<TestReport> {
        let mut reports = simulate::run_tests(self);
        for skill in &self.skills {
            reports.extend(skill.run_tests().into_iter().map(|mut report| {
                report.name = format!("{}/{}", skill.name, report.name);
                report
            }));
        }
        reports
    }

    /// A copy of this spec that reaches only what `allow` permits, and one
    /// note per binding it disabled.
    ///
    /// `mcp` entries not in the allowlist are removed. A tool whose `http`
    /// binding is not allowed becomes a mock: it returns its canned
    /// `response` and still writes `set_state`, so the flow behaves the same
    /// way offline. Use this before [`apply`](Self::apply) on any spec you did
    /// not write, such as one posted by a browser.
    pub fn sandboxed(&self, allow: &BindingAllowlist) -> (SessionSpec, Vec<String>) {
        let mut spec = self.clone();
        let mut notes = Vec::new();
        for skill in &mut spec.skills {
            let (sandboxed, disabled) = skill.sandboxed(allow);
            *skill = sandboxed;
            notes.extend(disabled);
        }
        spec.mcp.retain(|params| {
            let keep = allow.allows_mcp(params);
            if !keep {
                notes.push(format!(
                    "mcp entry `{params}` is not allowed on this server; removed"
                ));
            }
            keep
        });
        for tool in &mut spec.tools {
            if let Some(params) = &tool.mcp
                && !allow.allows_mcp(params)
            {
                notes.push(format!(
                    "tool `{}`: MCP binding `{params}` is not allowed on this server; it runs as a mock",
                    tool.name
                ));
                tool.mcp = None;
            }
            let Some(binding) = &mut tool.http else {
                continue;
            };
            if allow.allows_http(&binding.url) {
                // Arguments are interpolated and servers redirect at call
                // time, so the request is checked again then.
                binding.reach = Some(allow.clone());
            } else {
                notes.push(format!(
                    "tool `{}`: HTTP binding to `{}` is not allowed on this server; it runs as a mock",
                    tool.name, binding.url
                ));
                tool.http = None;
            }
        }
        (spec, notes)
    }

    /// Configure a [`Live`] builder from this spec.
    ///
    /// `state` is the session state the declared tools bind to — pass the
    /// same one via `.state` (this method does). Returns an error when
    /// the spec fails validation or requires a resource `resources` lacks.
    /// Everything code-only (callbacks, custom guards, middleware) is added on
    /// the returned builder afterwards.
    ///
    /// Tool bindings run as written. For a spec you did not write, call
    /// [`sandboxed`](Self::sandboxed) first.
    pub fn apply(
        &self,
        live: Live,
        state: &State,
        resources: &SpecResources,
    ) -> Result<Live, String> {
        let validation = self.validate();
        if !validation.valid {
            return Err(format!(
                "spec failed validation: {}",
                validation.errors.join("; ")
            ));
        }
        if self.requires_extraction()
            && self.uses_default_extraction_model()
            && resources.extraction_llm.is_none()
        {
            return Err(
                "spec declares extraction that names no model, and SpecResources.extraction_llm \
                 is not set; set models.extraction"
                    .into(),
            );
        }
        if self.requires_memory() && resources.memory.is_none() {
            return Err("spec declares memory but SpecResources.memory is not set".into());
        }
        if !self.decisions.is_empty() && resources.decision_model.is_none() {
            return Err(
                "spec declares decisions but SpecResources.decision_model is not set \
                 (for Jev on Vercel AI Gateway: GatewayDecisionModel::from_env(), which reads \
                 AI_GATEWAY_API_KEY)"
                    .into(),
            );
        }
        if let Some(name) = resources.tools.keys().find(|name| {
            !self.tools.iter().any(|t| &&t.name == name)
                && !self.skills.iter().any(|skill| {
                    skill.tools.iter().any(|tool| {
                        *name == &tool.tool.name
                            || **name == format!("{}/{}", skill.name, tool.tool.name)
                    })
                })
        }) {
            return Err(format!(
                "SpecResources implements tool '{name}', which the spec does not declare"
            ));
        }

        // Seed declared defaults before anything reads state.
        self.seed_state_defaults(state);

        let mut live = live
            .state(state.clone())
            .instruction(if self.instruction.is_empty() {
                "Follow the conversation flow you are given.".to_string()
            } else {
                self.instruction.clone()
            });

        if let Some(greeting) = &self.greeting {
            live = live.greeting(greeting.clone());
        }
        live = match self.modality {
            SpecModality::Text => live.text_only(),
            SpecModality::Audio => live.voice(resolve_voice(self.voice.as_deref())),
        };

        if !self.skills.is_empty() {
            let runtime = self.compile_tasks(resources)?;
            live = live.tasks(runtime);
            if self.skills.iter().any(|skill| skill.memory.is_some())
                && let Some(binding) = &resources.memory
            {
                live = binding.install_task_lifecycle(live);
            }
        }

        // Tools: declared (mock/HTTP) via the dispatcher, MCP merged on top.
        // A conversation's resolvers call the same bound tools.
        let tools = self.bound_tools(state, resources);
        if !tools.is_empty() {
            let mut dispatcher = ToolDispatcher::new();
            for tool in &tools {
                dispatcher.register(tool.clone());
            }
            live = live.dispatcher(dispatcher);
        }
        for params in &self.mcp {
            live = live.tools(T::mcp(params.clone()));
        }

        // Governance: a conversation compiles to a governed stack with its
        // extractors, timing and policies; a bare flow is governed as is.
        if let Some(conversation) = &self.conversation {
            let compiled = crate::conversation::Conversation::from_spec_with_resolvers(
                conversation.clone(),
                &self.resolver_registry(&tools),
            )
            .map_err(|e| format!("conversation: {e}"))?;
            live = live.converse(&compiled);
        } else {
            let flow = self.effective_flow().map_err(|e| e.join("; "))?;
            if !flow.steps.is_empty() {
                live = live.govern(flow);
            }
        }

        // Computed (derived) state variables — dependencies inferred from the
        // expression, so the registry's topological ordering holds.
        for c in &self.computed {
            let expr = c.from.clone();
            let deps: Vec<String> = expr.keys_read().into_iter().collect();
            let dep_refs: Vec<&str> = deps.iter().map(String::as_str).collect();
            live = live.computed(c.key.clone(), &dep_refs, move |s| expr.eval(s));
        }

        // Memory: install through the binding (tools, ingestion, slots).
        let memory_binding = resources.memory.clone();
        if let (Some(memory), Some(binding)) = (&self.memory, &memory_binding) {
            live = binding.install(live, memory);
        }

        // Background tools (async function calling).
        for tool in &self.tools {
            match (tool.background, tool.scheduling) {
                (_, Some(scheduling)) => {
                    live = live
                        .tool_background_with_scheduling(tool.name.clone(), scheduling.to_wire());
                }
                (true, None) => {
                    live = live.tool_background(tool.name.clone());
                }
                (false, None) => {}
            }
        }

        // Control-plane and voice tuning.
        if let Some(runtime) = &self.runtime {
            live = apply_runtime(live, runtime);
        }

        // Extraction, each entry on its model.
        let mut models = BTreeMap::new();
        for e in &self.extract {
            let llm =
                extraction_llm_for(e, self.models.extraction.as_deref(), resources, &mut models)?;
            live = live.extractor(Arc::new(compile_extractor(e, llm, &self.models)));
        }
        if let Some(model) = &resources.decision_model
            && !self.decisions.is_empty()
        {
            live = live.decisions(self.compile_decisions(model.clone()));
        }
        if let Some(code) = self.compile_code_mode(resources.code_sandbox.clone()) {
            live = live.code_mode(code);
        }

        // Phases.
        for p in &self.phases {
            let mut builder = live.phase(p.name.clone());
            if let Some(instruction) = &p.instruction {
                builder = builder.instruction(instruction.clone());
            }
            if !p.tools.is_empty() {
                builder = builder.tools(p.tools.clone());
            }
            if !p.needs.is_empty() {
                let refs: Vec<&str> = p.needs.iter().map(String::as_str).collect();
                builder = builder.needs(&refs);
            }
            if p.prompt_on_enter == Some(true) {
                builder = builder.prompt_on_enter();
            }
            if p.terminal {
                builder = builder.terminal();
            }
            for t in &p.transitions {
                let guard = t.when.clone();
                let predicate = move |s: &State| guard.eval_state(s);
                builder = match &t.description {
                    Some(desc) => builder.transition_with(&t.to, predicate, desc.clone()),
                    None => builder.transition(&t.to, predicate),
                };
            }
            if !p.on_enter.is_empty() {
                let effects = p.on_enter.clone();
                let memory = memory_binding.clone();
                builder = builder.on_enter(move |state, writer| {
                    let effects = effects.clone();
                    let memory = memory.clone();
                    async move {
                        run_effects(&effects, &state, &writer, memory.as_ref()).await;
                    }
                });
            }
            live = builder.done();
        }
        if let Some(initial) = &self.initial_phase {
            live = live.initial_phase(initial.clone());
        }

        // Temporal patterns.
        for pattern in &self.patterns {
            let guard = pattern.when.clone();
            let condition = move |s: &State| guard.eval_state(s);
            let effects = pattern.effects.clone();
            let memory = memory_binding.clone();
            let action = move |state: State, writer: Arc<dyn SessionWriter>| {
                let effects = effects.clone();
                let memory = memory.clone();
                async move {
                    run_effects(&effects, &state, &writer, memory.as_ref()).await;
                }
            };
            if let Some(secs) = pattern.sustained_secs {
                live = live.when_sustained(
                    pattern.name.clone(),
                    condition,
                    std::time::Duration::from_secs(secs),
                    action,
                );
            } else if let Some(turns) = pattern.turns {
                live = live.when_turns(pattern.name.clone(), condition, turns, action);
            }
        }

        // Watchers.
        for w in &self.watch {
            let builder = live.watch(w.key.clone());
            let builder = match &w.condition {
                WatchCondition::Changed => builder.changed(),
                WatchCondition::ChangedTo(v) => builder.changed_to(v.clone()),
                WatchCondition::CrossedAbove(t) => builder.crossed_above(*t),
                WatchCondition::CrossedBelow(t) => builder.crossed_below(*t),
                WatchCondition::BecameTrue => builder.became_true(),
                WatchCondition::BecameFalse => builder.became_false(),
            };
            let sets = w.set.clone();
            let effects = w.effects.clone();
            let memory = memory_binding.clone();
            live = builder.then_with_writer(move |_old, _new, state, writer| {
                let sets = sets.clone();
                let effects = effects.clone();
                let memory = memory.clone();
                async move {
                    for (key, value) in &sets {
                        let _ = state.set(key, value.clone());
                    }
                    run_effects(&effects, &state, &writer, memory.as_ref()).await;
                }
            });
        }

        Ok(live)
    }
}

/// Shared extractor lowering for session and task-owned services. `models`
/// is the session's, for the thinking budget an entry does not set.
fn compile_extractor(e: &ExtractSpec, llm: Arc<dyn BaseLlm>, models: &ModelsSpec) -> LlmExtractor {
    let mut extractor = LlmExtractor::new(e.name.clone(), llm, e.instruction.clone(), e.window)
        .with_schema(e.schema.clone())
        .with_min_words(3)
        .with_trigger(e.trigger.to_trigger());
    if let Some(budget) = e.thinking_budget.or(models.extraction_thinking_budget) {
        extractor = extractor.with_thinking_budget(Some(budget));
    }
    if e.promote.is_empty() {
        extractor
    } else {
        extractor.with_promotions(e.promote.iter().map(PromoteSpec::to_rule).collect())
    }
}

/// Execute a list of closed effects — the one executor behind phase
/// `on_enter`, watcher, and pattern actions, so every surface honors the
/// vocabulary identically.
async fn run_effects(
    effects: &[EffectSpec],
    state: &State,
    writer: &Arc<dyn SessionWriter>,
    memory: Option<&Arc<dyn MemoryBinding>>,
) {
    for effect in effects {
        match effect {
            EffectSpec::Set(map) => {
                for (key, value) in map {
                    let _ = state.set(key, value.clone());
                }
            }
            EffectSpec::Context(text) => {
                let _ = writer
                    .send_client_content(vec![Content::model(text.clone())], false)
                    .await;
            }
            EffectSpec::Prompt(text) => {
                let _ = writer
                    .send_client_content(vec![Content::model(text.clone())], true)
                    .await;
            }
            EffectSpec::Remember(template) => {
                if let Some(binding) = memory {
                    binding.remember(interpolate(template, &Value::Null, state));
                }
            }
        }
    }
}

/// Lower the turn-commit tuning knobs onto the Live builder. A knob left
/// unset keeps the default from
/// [`TurnCommitConfig::responsive()`](gemini_adk_rs::live::TurnCommitConfig::responsive).
fn apply_turn_commit(
    mut live: Live,
    eot_hold_ms: Option<u32>,
    min_interruption_ms: Option<u32>,
) -> Live {
    if let Some(ms) = eot_hold_ms {
        live = live.turn_commit_eot_hold_ms(u64::from(ms));
    }
    if let Some(ms) = min_interruption_ms {
        live = live.turn_commit_min_interruption_ms(u64::from(ms));
    }
    live
}

/// Lower the `runtime` section onto the builder.
fn apply_runtime(mut live: Live, runtime: &RuntimeSpec) -> Live {
    if let Some(t) = runtime.temperature {
        live = live.temperature(t);
    }
    if let Some(budget) = runtime.thinking_budget {
        live = live.thinking(budget);
    }
    if runtime.include_thoughts == Some(true) {
        live = live.include_thoughts();
    }
    if let Some(t) = runtime.transcription {
        if t.input {
            live = live.input_transcription();
        }
        if t.output {
            live = live.output_transcription();
        }
    }
    if runtime.proactive_audio == Some(true) {
        live = live.proactive_audio();
    }
    if let Some(vad) = &runtime.vad {
        live = live.vad(AutomaticActivityDetection {
            disabled: None,
            start_of_speech_sensitivity: vad.start_sensitivity.map(SensitivitySpec::to_wire),
            end_of_speech_sensitivity: vad.end_sensitivity.map(SensitivitySpec::to_wire),
            prefix_padding_ms: vad.prefix_padding_ms,
            silence_duration_ms: vad.silence_duration_ms,
        });
    }
    if let Some(audio) = &runtime.audio {
        #[cfg(feature = "denoise")]
        if audio.denoise == Some(true) {
            live = live.mic_denoise();
        }
        if let Some(gate) = &audio.noise_gate {
            live = live.mic_noise_gate(gate.threshold_rms, gate.hold_frames);
        }
        if let Some(vad) = &audio.client_vad {
            live = live.input_vad(vad.to_config());
        }
        if audio.authority == Some(AuthoritySpec::Client) {
            live = live.client_interruption_authority();
        }
        // Turn-commit tuning knobs integration seam: if either field is set,
        // the Live builder's turn_commit(...) method will wire them through.
        live = apply_turn_commit(live, audio.eot_hold_ms, audio.min_interruption_ms);
    }
    if let Some(ms) = runtime.soft_turn_timeout_ms {
        live = live.soft_turn_timeout(std::time::Duration::from_millis(ms));
    }
    if let Some(steering) = runtime.steering {
        live = live.steering_mode(match steering {
            SteeringSpec::InstructionUpdate => SteeringMode::InstructionUpdate,
            SteeringSpec::ContextInjection => SteeringMode::ContextInjection,
            SteeringSpec::Hybrid => SteeringMode::Hybrid,
            SteeringSpec::ContextUpdate => SteeringMode::ContextUpdate,
        });
    }
    if let Some(delivery) = runtime.context_delivery {
        live = live.context_delivery(match delivery {
            ContextDeliverySpec::Immediate => ContextDelivery::Immediate,
            ContextDeliverySpec::Deferred => ContextDelivery::Deferred,
        });
    }
    if let Some(repair) = runtime.repair {
        live = live.repair(RepairConfig {
            nudge_after: repair.nudge_after,
            escalate_after: repair.escalate_after,
        });
    }
    if let Some(persistence) = &runtime.persistence {
        live = match persistence {
            PersistenceSpec::Fs { dir } => {
                live.persistence(Arc::new(gemini_adk_rs::live::FsPersistence::new(dir)))
            }
            PersistenceSpec::Memory => {
                live.persistence(Arc::new(gemini_adk_rs::live::MemoryPersistence::new()))
            }
        };
    }
    if let Some(id) = &runtime.session_id {
        live = live.session_id(id.clone());
    }
    if runtime.lossy_audio {
        live = live.lossy_audio();
    }
    if runtime.lossy_transcript {
        live = live.lossy_transcript();
    }
    live
}

/// How a declared tool's call is carried out.
enum ToolCall {
    /// Its `http` binding, or its canned response.
    Declared,
    /// An in-process implementation.
    Code(Arc<dyn ToolFunction>),
    /// The tool of the same name on an MCP server.
    Mcp(Arc<gemini_adk_rs::tools::mcp::McpSessionManager>),
}

/// The value of an MCP `tools/call` result: its structured content, else
/// its single text part parsed as JSON, else the text itself.
fn mcp_value(result: Value) -> Value {
    if let Some(structured) = result.get("structuredContent") {
        return structured.clone();
    }
    let texts: Vec<&str> = result["content"]
        .as_array()
        .into_iter()
        .flatten()
        .filter_map(|part| part["text"].as_str())
        .collect();
    match texts.as_slice() {
        [text] => serde_json::from_str(text).unwrap_or_else(|_| json!({ "output": text })),
        [] => result,
        many => json!({ "output": many.join("\n") }),
    }
}

/// Build one declared tool, carrying out calls with `call`. Whatever carries
/// it out, the declaration and the state effects are the spec's.
fn build_tool(tool: &ToolSpec, state: &State, call: ToolCall) -> SimpleTool {
    let description = if tool.description.is_empty() {
        format!("Tool '{}'", tool.name)
    } else {
        tool.description.clone()
    };
    let response = tool.response.clone().unwrap_or_else(|| json!({"ok": true}));
    let sets = tool.set_state.clone();
    let save_as = tool.save_response_as.clone();
    let http = tool.http.clone();
    let st = state.clone();
    let name = tool.name.clone();
    let call = Arc::new(call);
    SimpleTool::new(
        &tool.name,
        description,
        tool.parameters.clone(),
        move |args| {
            let response = response.clone();
            let sets = sets.clone();
            let save_as = save_as.clone();
            let http = http.clone();
            let st = st.clone();
            let name = name.clone();
            let call = call.clone();
            async move {
                let result = match (call.as_ref(), &http) {
                    (ToolCall::Code(implementation), _) => implementation.call(args).await?,
                    (ToolCall::Mcp(server), _) => {
                        mcp_value(server.call_tool(&name, args).await.map_err(|e| {
                            gemini_adk_rs::error::ToolError::ExecutionFailed(format!("{name}: {e}"))
                        })?)
                    }
                    (ToolCall::Declared, Some(binding)) => {
                        execute_http(binding, &args, &st).await.map_err(|e| {
                            gemini_adk_rs::error::ToolError::Other(format!("{name}: {e}"))
                        })?
                    }
                    (ToolCall::Declared, None) => response,
                };
                for (key, value) in &sets {
                    let _ = st.set(key, value.clone());
                }
                if let Some(key) = &save_as {
                    let _ = st.set(key, result.clone());
                }
                Ok(result)
            }
        },
    )
}

/// Interpolate `{args.field}` and `{state.key}` templates in a string.
fn interpolate(template: &str, args: &Value, state: &State) -> String {
    let mut out = String::with_capacity(template.len());
    let mut rest = template;
    while let Some(open) = rest.find('{') {
        out.push_str(&rest[..open]);
        let after = &rest[open + 1..];
        let Some(close) = after.find('}') else {
            out.push_str(&rest[open..]);
            return out;
        };
        let expr = after[..close].trim();
        let value = if let Some(field) = expr.strip_prefix("args.") {
            args.get(field).cloned()
        } else if let Some(key) = expr.strip_prefix("state.") {
            state.get::<Value>(key)
        } else {
            None
        };
        match value {
            Some(Value::String(s)) => out.push_str(&s),
            Some(v) => out.push_str(&v.to_string()),
            None => {}
        }
        rest = &after[close + 1..];
    }
    out.push_str(rest);
    out
}

/// Walk a JSON value, interpolating every string.
#[cfg_attr(not(feature = "http-tools"), allow(dead_code))]
fn interpolate_value(value: &Value, args: &Value, state: &State) -> Value {
    match value {
        Value::String(s) => Value::String(interpolate(s, args, state)),
        Value::Array(items) => Value::Array(
            items
                .iter()
                .map(|v| interpolate_value(v, args, state))
                .collect(),
        ),
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(k, v)| (k.clone(), interpolate_value(v, args, state)))
                .collect(),
        ),
        other => other.clone(),
    }
}

#[cfg(feature = "http-tools")]
async fn execute_http(binding: &HttpBinding, args: &Value, state: &State) -> Result<Value, String> {
    let url = interpolate(&binding.url, args, state);
    let mut client = reqwest::Client::builder();
    if let Some(reach) = binding.reach.clone() {
        if !reach.allows_http(&url) {
            return Err(format!("{url} is outside this server's HTTP allowlist"));
        }
        client = client.redirect(reqwest::redirect::Policy::custom(move |attempt| {
            if attempt.previous().len() >= 10 {
                attempt.error("too many redirects")
            } else if reach.allows_http(attempt.url().as_str()) {
                attempt.follow()
            } else {
                let error = format!(
                    "redirect to {} is outside this server's HTTP allowlist",
                    attempt.url()
                );
                attempt.error(error)
            }
        }));
    }
    let client = client.build().map_err(|e| e.to_string())?;
    let method = reqwest::Method::from_bytes(binding.method.to_uppercase().as_bytes())
        .map_err(|_| format!("invalid HTTP method '{}'", binding.method))?;
    let mut request = client.request(method, &url);
    for (name, value) in &binding.headers {
        request = request.header(name, interpolate(value, args, state));
    }
    if let Some(body) = &binding.body {
        request = request.json(&interpolate_value(body, args, state));
    }
    let response = request.send().await.map_err(|e| e.to_string())?;
    let status = response.status().as_u16();
    let text = response.text().await.map_err(|e| e.to_string())?;
    Ok(serde_json::from_str(&text).unwrap_or_else(|_| json!({ "status": status, "body": text })))
}

#[cfg(not(feature = "http-tools"))]
#[allow(
    clippy::unused_async,
    reason = "same signature as the http-tools implementation so the call site is feature-agnostic"
)]
async fn execute_http(
    _binding: &HttpBinding,
    _args: &Value,
    _state: &State,
) -> Result<Value, String> {
    Err("http tool bindings require the `http-tools` feature".to_string())
}

/// Splice `fragment` into `flow` under the directive's namespace.
fn splice_fragment(
    flow: &mut Flow,
    fragment: &Flow,
    directive: &UseFragment,
    errors: &mut Vec<String>,
) {
    let ns = &directive.namespace;
    let prefix = |id: &str| format!("{ns}/{id}");
    let internal: std::collections::BTreeSet<&str> =
        fragment.steps.iter().map(|s| s.id.as_str()).collect();

    for step in &fragment.steps {
        let new_id = prefix(&step.id);
        if flow.steps.iter().any(|s| s.id == new_id) {
            errors.push(format!(
                "fragment splice '{ns}' collides with existing step '{new_id}'"
            ));
            continue;
        }
        let mut after: Vec<gemini_adk_rs::flow::Edge> = step
            .after
            .iter()
            .map(|d| gemini_adk_rs::flow::Edge {
                step: if internal.contains(d.step.as_str()) {
                    prefix(&d.step)
                } else {
                    d.step.clone()
                },
                when: d
                    .when
                    .clone()
                    .map(|g| rewrite_guard_steps(g, &internal, ns)),
            })
            .collect();
        if step.after.is_empty() {
            after.extend(
                directive
                    .after
                    .iter()
                    .cloned()
                    .map(gemini_adk_rs::flow::Edge::to),
            );
        }
        flow.steps.push(Step {
            id: new_id,
            after,
            join: step.join,
            gate: step
                .gate
                .clone()
                .map(|g| rewrite_guard_steps(g, &internal, ns)),
            done: step
                .done
                .clone()
                .map(|g| rewrite_guard_steps(g, &internal, ns)),
            posture: step.posture.clone(),
            ground: step.ground.clone(),
            allow: step.allow.clone(),
            deny: step.deny.clone(),
            terminal: step.terminal,
        });
    }
    for constraint in &fragment.constraints {
        flow.constraints.push(match constraint {
            Constraint::Once(t) => Constraint::Once(t.clone()),
            Constraint::Before(a, b) => Constraint::Before(
                if internal.contains(a.as_str()) {
                    prefix(a)
                } else {
                    a.clone()
                },
                if internal.contains(b.as_str()) {
                    prefix(b)
                } else {
                    b.clone()
                },
            ),
            Constraint::NeverUntil { tool, until } => Constraint::NeverUntil {
                tool: tool.clone(),
                until: rewrite_guard_steps(until.clone(), &internal, ns),
            },
            Constraint::Require(rs) => Constraint::Require(
                rs.iter()
                    .map(|r| {
                        if internal.contains(r.as_str()) {
                            prefix(r)
                        } else {
                            r.clone()
                        }
                    })
                    .collect(),
            ),
            Constraint::Reset { steps, when } => Constraint::Reset {
                steps: steps
                    .iter()
                    .map(|r| {
                        if internal.contains(r.as_str()) {
                            prefix(r)
                        } else {
                            r.clone()
                        }
                    })
                    .collect(),
                when: rewrite_guard_steps(when.clone(), &internal, ns),
            },
        });
    }
    for tool in &fragment.ambient {
        if !flow.ambient.contains(tool) {
            flow.ambient.push(tool.clone());
        }
    }
    for tool in &fragment.confirm_tools {
        if !flow.confirm_tools.contains(tool) {
            flow.confirm_tools.push(tool.clone());
        }
    }
}

/// Rewrite `done(step)` atoms that reference fragment-internal steps.
fn rewrite_guard_steps(
    guard: Guard,
    internal: &std::collections::BTreeSet<&str>,
    ns: &str,
) -> Guard {
    fn rewrite(pred: Pred, internal: &std::collections::BTreeSet<&str>, ns: &str) -> Pred {
        match pred {
            Pred::Done(s) if internal.contains(s.as_str()) => Pred::Done(format!("{ns}/{s}")),
            Pred::All(ps) => Pred::All(ps.into_iter().map(|p| rewrite(p, internal, ns)).collect()),
            Pred::Any(ps) => Pred::Any(ps.into_iter().map(|p| rewrite(p, internal, ns)).collect()),
            Pred::Not(p) => Pred::Not(Box::new(rewrite(*p, internal, ns))),
            other => other,
        }
    }
    match guard {
        Guard::Spec(p) => Guard::Spec(rewrite(p, internal, ns)),
        custom => custom,
    }
}

/// Whether a guard contains `called_ok`/`done` atoms, which need a flow
/// marking and therefore cannot back phase transitions.
fn guard_uses_marking(guard: &Guard) -> bool {
    fn walk(pred: &Pred) -> bool {
        match pred {
            Pred::CalledOk(_) | Pred::Done(_) => true,
            Pred::All(ps) | Pred::Any(ps) => ps.iter().any(walk),
            Pred::Not(p) => walk(p),
            _ => false,
        }
    }
    match guard {
        Guard::Spec(p) => walk(p),
        Guard::Custom(_) => false,
    }
}

/// Resolve a voice name to the `Voice` enum.
fn resolve_voice(name: Option<&str>) -> Voice {
    match name {
        Some("Aoede") => Voice::Aoede,
        Some("Charon") => Voice::Charon,
        Some("Fenrir") => Voice::Fenrir,
        Some("Kore") => Voice::Kore,
        Some("Puck") | None => Voice::Puck,
        Some(other) => Voice::Custom(other.to_string()),
    }
}

/// Edit distance for the did-you-mean hint on unmatched state keys.
fn levenshtein(a: &str, b: &str) -> usize {
    let a: Vec<char> = a.chars().collect();
    let b: Vec<char> = b.chars().collect();
    let mut prev: Vec<usize> = (0..=b.len()).collect();
    let mut current = vec![0; b.len() + 1];
    for (i, ca) in a.iter().enumerate() {
        current[0] = i + 1;
        for (j, cb) in b.iter().enumerate() {
            let cost = usize::from(ca != cb);
            current[j + 1] = (prev[j] + cost).min(prev[j + 1] + 1).min(current[j] + 1);
        }
        std::mem::swap(&mut prev, &mut current);
    }
    prev[b.len()]
}

#[cfg(test)]
mod tests {

    #[test]
    fn audio_spec_lowers_and_validates() {
        let spec: SessionSpec = serde_json::from_str(
            r#"{
                "name": "noisy",
                "instruction": "hi",
                "runtime": {
                    "audio": {
                        "denoise": true,
                        "noise_gate": { "threshold_rms": 700.0 },
                        "client_vad": { "preset": "noisy_street", "hangover_frames": 12 },
                        "authority": "client"
                    }
                }
            }"#,
        )
        .unwrap();
        let audio = spec.runtime.as_ref().unwrap().audio.as_ref().unwrap();
        assert_eq!(audio.noise_gate.as_ref().unwrap().hold_frames, 3); // serde default
        let config = audio.client_vad.as_ref().unwrap().to_config();
        assert_eq!(config.start_threshold_db, 21.0); // preset
        assert_eq!(config.hangover_frames, 12); // override wins
        assert_eq!(audio.authority, Some(AuthoritySpec::Client));
        // Round-trips through JSON unchanged.
        let json = serde_json::to_value(&spec).unwrap();
        let back: SessionSpec = serde_json::from_value(json).unwrap();
        assert_eq!(
            back.runtime
                .unwrap()
                .audio
                .unwrap()
                .client_vad
                .unwrap()
                .hangover_frames,
            Some(12)
        );
    }

    #[test]
    fn audio_spec_warns_on_risky_combinations() {
        let spec: SessionSpec = serde_json::from_str(
            r#"{
                "name": "risky",
                "instruction": "hi",
                "runtime": { "audio": { "authority": "client", "noise_gate": {} } }
            }"#,
        )
        .unwrap();
        let validation = spec.validate();
        assert!(
            validation
                .warnings
                .iter()
                .any(|w| w.contains("authority=client without denoise")),
            "expected client-authority warning, got {:?}",
            validation.warnings
        );
        assert!(
            validation
                .warnings
                .iter()
                .any(|w| w.contains("noise_gate without denoise")),
            "expected gate warning, got {:?}",
            validation.warnings
        );
    }

    #[test]
    fn turn_commit_tuning_knobs_serialize_and_round_trip() {
        let spec: SessionSpec = serde_json::from_str(
            r#"{
                "name": "tune",
                "instruction": "hi",
                "runtime": {
                    "audio": {
                        "eot_hold_ms": 800,
                        "min_interruption_ms": 1400
                    }
                }
            }"#,
        )
        .unwrap();
        let audio = spec.runtime.as_ref().unwrap().audio.as_ref().unwrap();
        assert_eq!(audio.eot_hold_ms, Some(800));
        assert_eq!(audio.min_interruption_ms, Some(1400));
        // Round-trips through JSON unchanged.
        let json = serde_json::to_value(&spec).unwrap();
        let back: SessionSpec = serde_json::from_value(json).unwrap();
        let audio_back = back.runtime.unwrap().audio.unwrap();
        assert_eq!(audio_back.eot_hold_ms, Some(800));
        assert_eq!(audio_back.min_interruption_ms, Some(1400));
    }

    #[test]
    fn turn_commit_tuning_knobs_validate_thresholds() {
        // Warn on eot_hold_ms > 1600
        let spec: SessionSpec = serde_json::from_str(
            r#"{
                "name": "frontier",
                "instruction": "hi",
                "runtime": { "audio": { "eot_hold_ms": 1700 } }
            }"#,
        )
        .unwrap();
        let validation = spec.validate();
        assert!(
            validation
                .warnings
                .iter()
                .any(|w| w.contains("eot_hold_ms") && w.contains("1600") && w.contains("frontier")),
            "expected eot_hold_ms frontier warning, got {:?}",
            validation.warnings
        );

        // Warn on min_interruption_ms > 2000
        let spec2: SessionSpec = serde_json::from_str(
            r#"{
                "name": "window",
                "instruction": "hi",
                "runtime": { "audio": { "min_interruption_ms": 2100 } }
            }"#,
        )
        .unwrap();
        let validation2 = spec2.validate();
        assert!(
            validation2
                .warnings
                .iter()
                .any(|w| w.contains("min_interruption_ms")
                    && w.contains("2000")
                    && w.contains("window")),
            "expected min_interruption_ms window warning, got {:?}",
            validation2.warnings
        );
    }
    use super::*;

    fn collections_spec() -> SessionSpec {
        SessionSpec::from_value(json!({
            "name": "collections",
            "instruction": "Collect payments.",
            "tools": [
                {"name": "verify_identity", "set_state": {"identity_verified": true}},
                {"name": "charge_card", "response": {"charged": true}}
            ],
            "extract": [{
                "name": "ptp",
                "instruction": "Extract the promise to pay.",
                "schema": {"type": "object", "properties": {
                    "ptp_amount": {"type": "number"}, "ptp_date": {"type": "string"}}},
                "promote": [
                    {"field": "ptp_amount"},
                    {"field": "ptp_date", "policy": "overwrite"}
                ]
            }],
            "flow": {
                "steps": [
                    {"id": "verify", "posture": "Verify the caller.",
                     "allow": ["verify_identity"],
                     "done": {"is_true": "identity_verified"}},
                    {"id": "pay", "after": ["verify"], "posture": "Take payment.",
                     "allow": ["charge_card"],
                     "gate": {"captured": ["ptp_amount", "ptp_date"]},
                     "done": {"called_ok": "charge_card"}}
                ]
            }
        }))
        .expect("spec parses")
    }

    #[test]
    fn extraction_promotions_satisfy_guard_reads() {
        let v = collections_spec().validate();
        assert!(v.valid, "errors: {:?}", v.errors);
        assert!(
            v.warnings.iter().all(|w| !w.contains("can never latch")),
            "promoted keys cover the guard reads: {:?}",
            v.warnings
        );
    }

    #[test]
    fn unwritten_guard_key_warns_with_suggestion() {
        let mut spec = collections_spec();
        // Typo in the guard key relative to the tool's write.
        spec.flow.as_mut().unwrap().steps[0].done = Some(Guard::is_true("identity_verifed"));
        let v = spec.validate();
        assert!(v.valid);
        let warning = v
            .warnings
            .iter()
            .find(|w| w.contains("identity_verifed"))
            .expect("warns about the unwritten key");
        assert!(
            warning.contains("identity_verified"),
            "suggests the fix: {warning}"
        );
    }

    #[test]
    fn phase_guard_rejects_marking_atoms() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "phases": [{"name": "a", "transitions": [
                {"to": "b", "when": {"called_ok": "some_tool"}}]},
                {"name": "b"}],
            "initial_phase": "a"
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(!v.valid);
        assert!(v.errors.iter().any(|e| e.contains("called_ok")));
    }

    #[test]
    fn fragments_splice_with_namespacing() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "tools": [
                {"name": "check_id", "set_state": {"id_ok": true}},
                {"name": "book", "response": {}}
            ],
            "fragments": {
                "verify": {"steps": [
                    {"id": "ask", "posture": "Ask for ID.", "allow": ["check_id"],
                     "done": {"is_true": "id_ok"}},
                    {"id": "confirm", "after": ["ask"], "terminal": true,
                     "gate": {"done": "ask"}}
                ]}
            },
            "use_fragments": [{"fragment": "verify", "namespace": "v"}],
            "flow": {"steps": [
                {"id": "book_step", "after": ["v/confirm"], "allow": ["book"],
                 "done": {"called_ok": "book"}}
            ]}
        }))
        .expect("parses");
        let flow = spec.effective_flow().expect("splices");
        let ids: Vec<&str> = flow.steps.iter().map(|s| s.id.as_str()).collect();
        assert!(ids.contains(&"v/ask") && ids.contains(&"v/confirm"));
        // Internal done() atom rewritten to the namespaced id.
        let confirm = flow.steps.iter().find(|s| s.id == "v/confirm").unwrap();
        assert_eq!(
            serde_json::to_value(confirm.gate.as_ref().unwrap()).unwrap(),
            json!({"done": "v/ask"})
        );
        let v = spec.validate();
        assert!(v.valid, "errors: {:?}", v.errors);
    }

    #[test]
    fn bare_flow_round_trips() {
        let spec = SessionSpec::from_value(json!({
            "steps": [{"id": "only", "terminal": true}]
        }))
        .expect("parses");
        assert!(spec.validate().valid);
    }

    #[test]
    fn empty_fragment_namespace_is_rejected() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "fragments": {
                "verify": {"steps": [
                    {"id": "ask", "terminal": true}
                ]}
            },
            "use_fragments": [{"fragment": "verify", "namespace": ""}],
            "flow": {"steps": [
                {"id": "start", "terminal": true}
            ]}
        }))
        .expect("parses");
        let v = spec.validate();
        // Empty namespace should be a validation error
        assert!(
            !v.valid,
            "empty namespace should fail validation, got errors: {:?}",
            v.errors
        );
        assert!(
            v.errors.iter().any(|e| e.contains("namespace")),
            "should mention namespace issue: {:?}",
            v.errors
        );
    }

    #[test]
    fn duplicate_computed_keys_are_rejected() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "computed": [
                {"key": "risk", "from": {"key": "score"}},
                {"key": "risk", "from": {"key": "other_score"}}
            ]
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(!v.valid, "duplicate computed keys should fail validation");
        assert!(
            v.errors.iter().any(|e| e.contains("duplicate")),
            "should mention duplicate computed key: {:?}",
            v.errors
        );
    }

    #[test]
    fn duplicate_phase_names_are_rejected() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "phases": [
                {"name": "greet", "instruction": "Welcome."},
                {"name": "greet", "instruction": "Hi again."}
            ],
            "initial_phase": "greet"
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(!v.valid, "duplicate phase names should fail validation");
        assert!(
            v.errors.iter().any(|e| e.contains("phase")),
            "should mention duplicate phase: {:?}",
            v.errors
        );
    }

    #[test]
    fn tool_background_false_round_trips() {
        // Test that explicitly setting background: false round-trips correctly
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "tools": [
                {"name": "search", "background": false},
                {"name": "log", "background": true}
            ],
            "flow": {"steps": [{"id": "only", "terminal": true}]}
        }))
        .expect("parses");

        // After serialization
        let serialized = serde_json::to_value(&spec).unwrap();
        let tools = serialized["tools"].as_array().unwrap();

        // background: false is skipped in serialization (optimization), but
        // when deserialized, missing field defaults to false ✓
        assert!(
            tools[0].get("background").is_none(),
            "background: false is optimized away"
        );

        // background: true is serialized
        assert_eq!(tools[1].get("background"), Some(&json!(true)));

        // Round-trip test
        let back: SessionSpec = serde_json::from_value(serialized).unwrap();
        assert!(!back.tools[0].background);
        assert!(back.tools[1].background);
    }

    #[test]
    fn promotion_spec_with_no_to_field_uses_field_name() {
        // Test that promotion without explicit "to" uses the field name as target
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "tools": [{"name": "extract", "set_state": {"test": true}}],
            "extract": [{
                "name": "data",
                "instruction": "Extract.",
                "schema": {"type": "object"},
                "promote": [
                    {"field": "amount"},
                    {"field": "date", "to": "extracted_date"}
                ]
            }],
            "flow": {"steps": [{"id": "only", "terminal": true}]}
        }))
        .expect("parses");

        let promote = &spec.extract[0].promote;
        assert_eq!(promote[0].target(), "amount");
        assert_eq!(promote[1].target(), "extracted_date");
    }

    #[test]
    fn spec_schema_publishes() {
        let schema = SessionSpec::json_schema().to_string();
        for token in [
            "is_true",
            "never_until",
            "set_state",
            "use_fragments",
            "promote",
        ] {
            assert!(schema.contains(token), "schema missing {token}");
        }
    }

    #[test]
    fn interpolation_reads_args_and_state() {
        let state = State::new();
        let _ = state.set("city", "Paris");
        let args = json!({"guests": 4});
        assert_eq!(
            interpolate("book/{state.city}/{args.guests}/{missing}", &args, &state),
            "book/Paris/4/"
        );
    }

    #[tokio::test]
    async fn declared_tools_write_state() {
        let spec = collections_spec();
        let state = State::new();
        let dispatcher = spec.build_dispatcher(&state);
        let out = dispatcher
            .call_function("verify_identity", json!({}))
            .await
            .expect("tool runs");
        assert_eq!(out, json!({"ok": true}));
        assert_eq!(state.get::<bool>("identity_verified"), Some(true));
    }

    #[test]
    fn computed_cycles_are_load_time_errors() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "computed": [
                {"key": "a", "from": {"add": [{"key": "b"}, {"const": 1}]}},
                {"key": "b", "from": {"add": [{"key": "derived:a"}, {"const": 1}]}}
            ]
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(!v.valid);
        assert!(v.errors.iter().any(|e| e.contains("dependency cycle")));

        let self_read = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "computed": [{"key": "a", "from": {"key": "a"}}]
        }))
        .expect("parses");
        assert!(
            self_read
                .validate()
                .errors
                .iter()
                .any(|e| e.contains("reads its own key"))
        );
    }

    #[test]
    fn computed_keys_satisfy_guard_reads_and_deps_are_checked() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "tools": [{"name": "record", "set_state": {"score": 0.8}}],
            "computed": [{"key": "high_risk",
                          "from": {"gt": [{"key": "score"}, {"const": 0.5}]}}],
            "flow": {"steps": [
                {"id": "assess", "posture": "Assess.", "allow": ["record"],
                 "done": {"is_true": "high_risk"}}
            ]}
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(v.valid, "errors: {:?}", v.errors);
        assert!(
            v.warnings.iter().all(|w| !w.contains("can never latch")),
            "computed key covers the guard read: {:?}",
            v.warnings
        );

        let mut dangling = spec.clone();
        dangling.computed[0].from =
            serde_json::from_value(json!({"gt": [{"key": "scoer"}, {"const": 0.5}]})).unwrap();
        let v = dangling.validate();
        assert!(
            v.warnings
                .iter()
                .any(|w| w.contains("computed 'high_risk' reads state key 'scoer'"))
        );
    }

    #[test]
    fn remember_requires_the_memory_section() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "patterns": [{"name": "note", "when": {"is_true": "flag"}, "turns": 2,
                          "effects": [{"remember": "caller likes {state.thing}"}]}]
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(!v.valid);
        assert!(v.errors.iter().any(|e| e.contains("no `memory` section")));
    }

    #[test]
    fn memory_slots_join_written_keys_and_reject_derived_targets() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "memory": {"slots": [{"predicate": "dietary_identity", "to": "user:diet"}]},
            "flow": {"steps": [
                {"id": "plan", "posture": "Plan dinner.",
                 "done": {"is_set": "user:diet"}}
            ]}
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(v.valid, "errors: {:?}", v.errors);
        assert!(v.warnings.iter().all(|w| !w.contains("can never latch")));

        let mut bad = spec.clone();
        bad.memory.as_mut().unwrap().slots[0].to = "derived:diet".into();
        assert!(
            bad.validate()
                .errors
                .iter()
                .any(|e| e.contains("read-only key"))
        );
    }

    #[test]
    fn memory_section_requires_a_binding_at_apply() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "memory": {},
            "flow": {"steps": [{"id": "only", "terminal": true}]}
        }))
        .expect("parses");
        let err = spec
            .apply(Live::builder(), &State::new(), &SpecResources::default())
            .err()
            .expect("memory binding required");
        assert!(err.contains("SpecResources.memory"));

        struct NullBinding;
        impl MemoryBinding for NullBinding {
            fn install(&self, live: Live, _memory: &MemorySpec) -> Live {
                live
            }
            fn remember(&self, _note: String) {}
        }
        let resources = SpecResources {
            memory: Some(Arc::new(NullBinding)),
            ..Default::default()
        };
        assert!(
            spec.apply(Live::builder(), &State::new(), &resources)
                .is_ok()
        );
    }

    #[test]
    fn state_dictionary_seeds_defaults_and_flags_undeclared_writes() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "state": {
                "attempts": {"type": "number", "default": 0,
                             "description": "Verification attempts so far."},
                "verified": {"type": "boolean", "default": "yes"}
            },
            "tools": [{"name": "verify", "set_state": {"verified": true, "vip": true}}],
            "flow": {"steps": [
                {"id": "v", "posture": "Verify.", "allow": ["verify"],
                 "done": {"is_true": "verified"}}
            ]}
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(v.valid, "errors: {:?}", v.errors);
        assert!(
            v.warnings
                .iter()
                .any(|w| w.contains("'verified' declares type Boolean")),
            "type-mismatched default warns: {:?}",
            v.warnings
        );
        assert!(
            v.warnings
                .iter()
                .any(|w| w.contains("'vip' is written but not declared")),
            "undeclared write warns: {:?}",
            v.warnings
        );

        let state = State::new();
        spec.seed_state_defaults(&state);
        assert_eq!(state.get::<i64>("attempts"), Some(0));
    }

    #[test]
    fn runtime_section_lowers_onto_the_builder() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "runtime": {
                "temperature": 0.4,
                "thinking_budget": 1024,
                "include_thoughts": true,
                "transcription": {"input": true, "output": false},
                "proactive_audio": true,
                "vad": {"start_sensitivity": "high", "silence_duration_ms": 400},
                "soft_turn_timeout_ms": 1500,
                "steering": "context_injection",
                "context_delivery": "deferred",
                "repair": {"nudge_after": 2, "escalate_after": 5},
                "persistence": "memory",
                "session_id": "user-1",
                "lossy_audio": true
            }
        }))
        .expect("parses");
        let v = spec.validate();
        assert!(v.valid, "errors: {:?}", v.errors);
        assert!(
            spec.apply(Live::builder(), &State::new(), &SpecResources::default())
                .is_ok()
        );

        let mut incoherent = spec.clone();
        incoherent.runtime.as_mut().unwrap().thinking_budget = None;
        assert!(
            incoherent
                .validate()
                .warnings
                .iter()
                .any(|w| w.contains("include_thoughts"))
        );
    }

    #[test]
    fn context_update_steering_lowers_onto_the_builder() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "flow": {"steps": [{"id": "only", "terminal": true}]},
            "runtime": {"steering": "context_update"}
        }))
        .expect("parses");
        assert!(spec.validate().valid);
        let live = spec
            .apply(Live::builder(), &State::new(), &SpecResources::default())
            .expect("applies");
        assert_eq!(
            live.describe_contract().controls.steering_mode,
            "ContextUpdate"
        );
        assert!(
            spec.to_rust()
                .contains(".steering_mode(SteeringMode::ContextUpdate)"),
            "codegen emits the same mode"
        );
    }

    #[test]
    fn background_tools_and_scheduling_parse_and_apply() {
        let spec = SessionSpec::from_value(json!({
            "instruction": "x",
            "tools": [
                {"name": "search_kb", "background": true},
                {"name": "log_event", "scheduling": "silent"}
            ],
            "flow": {"steps": [
                {"id": "s", "posture": "Serve.", "allow": ["search_kb", "log_event"],
                 "done": {"called_ok": "search_kb"}}
            ]}
        }))
        .expect("parses");
        assert!(spec.validate().valid);
        assert!(
            spec.apply(Live::builder(), &State::new(), &SpecResources::default())
                .is_ok()
        );
    }

    #[test]
    fn a_sandboxed_spec_reaches_only_what_the_operator_allows() {
        let spec = SessionSpec::from_value(json!({
            "name": "posted",
            "mcp": ["rm -rf /", "https://mcp.example.com/sse"],
            "tools": [
                { "name": "ok", "http": { "url": "https://api.example.com/v1/{args.id}" } },
                { "name": "internal", "http": { "url": "http://169.254.169.254/latest/meta-data" } },
                { "name": "host_from_args", "http": { "url": "https://{args.host}/x" } },
                { "name": "mock", "response": { "ok": true } }
            ]
        }))
        .unwrap();

        let (nothing, notes) = spec.sandboxed(&BindingAllowlist::default());
        assert!(nothing.mcp.is_empty());
        assert!(nothing.tools.iter().all(|t| t.http.is_none()));
        assert_eq!(notes.len(), 5, "{notes:?}");

        let allow = BindingAllowlist::default()
            .allow_http_prefix("https://api.example.com")
            .allow_mcp("https://mcp.example.com/sse");
        let (some, notes) = spec.sandboxed(&allow);
        assert_eq!(some.mcp, ["https://mcp.example.com/sse"]);
        let bound: Vec<&str> = some
            .tools
            .iter()
            .filter(|t| t.http.is_some())
            .map(|t| t.name.as_str())
            .collect();
        assert_eq!(bound, ["ok"]);
        assert_eq!(notes.len(), 3, "{notes:?}");

        // A prefix without a path cannot be stretched to another host.
        let tricky = BindingAllowlist::default().allow_http_prefix("https://api.example.com");
        assert!(!tricky.allows_http("https://api.example.com.evil.net/x"));
        assert!(!tricky.allows_http("https://api.example.com@evil.net/x"));
        assert!(!tricky.allows_http("https://api.example.com:8443/x"));
        assert!(!tricky.allows_http("http://api.example.com/x"));

        // Dot segments are resolved before a path prefix is checked.
        let subtree = BindingAllowlist::default().allow_http_prefix("https://api.example.com/v2/");
        assert!(subtree.allows_http("https://API.example.com/v2/orders/{args.id}"));
        assert!(!subtree.allows_http("https://api.example.com/v2/../admin"));
        assert!(!subtree.allows_http("https://api.example.com/v2/%2e%2e/admin"));
        assert!(!subtree.allows_http("https://api.example.com/v2/%2E%2E/admin"));
    }

    #[cfg(feature = "http-tools")]
    #[tokio::test]
    async fn a_sandboxed_binding_is_checked_again_when_it_runs() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        // A local server that redirects /v2/hop to /admin and answers
        // anything else with its path.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let origin = format!("http://{}", listener.local_addr().unwrap());
        tokio::spawn(async move {
            while let Ok((mut socket, _)) = listener.accept().await {
                let mut buf = vec![0u8; 4096];
                let n = socket.read(&mut buf).await.unwrap_or(0);
                let request = String::from_utf8_lossy(&buf[..n]).to_string();
                let path = request.split_whitespace().nth(1).unwrap_or("/").to_string();
                let response = if path == "/v2/hop" {
                    "HTTP/1.1 302 Found\r\nlocation: /admin\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".to_string()
                } else {
                    let body = json!({ "path": path }).to_string();
                    format!(
                        "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                        body.len()
                    )
                };
                let _ = socket.write_all(response.as_bytes()).await;
            }
        });

        let spec = SessionSpec::from_value(json!({
            "name": "posted",
            "tools": [{ "name": "get", "http": { "url": format!("{origin}/v2/{{args.path}}") } }]
        }))
        .unwrap();
        let allow = BindingAllowlist::default().allow_http_prefix(format!("{origin}/v2/"));
        let (sandboxed, notes) = spec.sandboxed(&allow);
        assert!(notes.is_empty(), "{notes:?}");
        let state = State::new();
        let dispatcher = sandboxed.build_dispatcher(&state);

        let ok = dispatcher
            .call_function("get", json!({ "path": "orders" }))
            .await
            .unwrap();
        assert_eq!(ok["path"], "/v2/orders");
        // An argument cannot climb out of the prefix...
        let climbed = dispatcher
            .call_function("get", json!({ "path": "../admin" }))
            .await;
        assert!(climbed.is_err(), "{climbed:?}");
        // ...and neither can a redirect.
        let redirected = dispatcher
            .call_function("get", json!({ "path": "hop" }))
            .await;
        assert!(redirected.is_err(), "{redirected:?}");

        // The same spec, not sandboxed, follows the redirect as written.
        let trusted = spec.build_dispatcher(&state);
        let followed = trusted
            .call_function("get", json!({ "path": "hop" }))
            .await
            .unwrap();
        assert_eq!(followed["path"], "/admin");
    }

    /// The repo's booking conversation, as one spec with its tools and
    /// scenarios.
    fn booking_bundle() -> SessionSpec {
        let conversation: Value =
            serde_json::from_str(include_str!("../../../../conversations/booking.spec.json"))
                .unwrap();
        let scenario = |file: &str| -> Value { serde_json::from_str(file).unwrap() };
        SessionSpec::from_value(json!({
            "name": "booking",
            "modality": "audio",
            "tools": [{
                "name": "book",
                "description": "Book the table",
                "response": { "confirmation": "B-1" }
            }],
            "conversation": conversation,
            "scenarios": [
                scenario(include_str!("../../../../conversations/booking.happy.scenario.json")),
                scenario(include_str!(
                    "../../../../conversations/booking.no_book_without_confirm.scenario.json"
                )),
            ]
        }))
        .unwrap()
    }

    #[test]
    fn tools_used_only_in_a_digression_are_referenced() {
        let mut spec = booking_bundle();
        spec.tools
            .push(serde_json::from_value(json!({ "name": "transfer" })).unwrap());
        let unreferenced = |spec: &SessionSpec| {
            spec.validate()
                .warnings
                .iter()
                .any(|w| w.contains("'transfer' is declared but no step"))
        };
        assert!(unreferenced(&spec));
        spec.conversation.as_mut().unwrap().overlays = serde_json::from_value(json!([{
            "name": "handoff",
            "trigger": { "is_true": "intent:human_agent" },
            "stages": [{ "id": "transfer", "allow": ["transfer"], "done": { "called_ok": "transfer" } }],
            "resume": "terminate"
        }]))
        .unwrap();
        assert!(!unreferenced(&spec));
    }

    #[test]
    fn a_conversation_spec_validates_like_a_flow() {
        let spec = booking_bundle();
        let report = spec.validate();
        assert!(report.valid, "{:?}", report.errors);
        // Collected slots are written by the conversation's extractors, so
        // only the genuinely unwritten key is reported.
        let unwritten: Vec<&String> = report
            .warnings
            .iter()
            .filter(|w| w.contains("no tool, extractor"))
            .collect();
        assert_eq!(unwritten.len(), 1, "{:?}", report.warnings);
        assert!(unwritten[0].contains("user_confirmed"));
        assert!(report.mermaid.contains("confirm"), "{}", report.mermaid);
        assert!(report.tools.contains(&"book".to_string()));

        // Conversation and flow are alternatives.
        let mut both = spec.clone();
        both.flow = Some(Flow::default());
        assert!(!both.validate().valid);

        // A resolver must be a declared tool.
        let mut resolving = spec;
        resolving.conversation.as_mut().unwrap().stages[0]
            .resolve
            .push(crate::conversation::ResolveSpec {
                slot: "availability".into(),
                resolver: None,
                args: vec![],
                ttl_secs: None,
            });
        let report = resolving.validate();
        assert!(
            report.errors.iter().any(|e| e.contains("availability")),
            "{:?}",
            report.errors
        );
    }

    #[tokio::test]
    async fn a_conversation_spec_runs_its_scenarios_and_applies() {
        let spec = booking_bundle();
        let reports = spec.run_scenarios().await;
        assert_eq!(reports.len(), 2);
        assert!(reports.iter().all(|r| r.passed), "{reports:?}");

        let mut broken = spec.clone();
        broken.scenarios[0]
            .steps
            .push(crate::simulation::SimStep::ExpectActive(vec![
                "nowhere".into(),
            ]));
        assert!(!broken.run_scenarios().await[0].passed);

        spec.apply(Live::builder(), &State::new(), &SpecResources::default())
            .expect("a conversation spec applies");
    }

    fn booking_tool(extra: Value) -> SessionSpec {
        let mut tool = json!({
            "name": "book",
            "description": "Book the table",
            "response": { "confirmation": "MOCK" },
            "set_state": { "booked": true },
            "save_response_as": "booking"
        });
        tool.as_object_mut()
            .unwrap()
            .extend(extra.as_object().unwrap().clone());
        SessionSpec::from_value(json!({
            "name": "t",
            "tools": [tool],
            "flow": { "steps": [{ "id": "s", "allow": ["book"] }] }
        }))
        .unwrap()
    }

    #[tokio::test]
    async fn an_implementation_replaces_the_call_not_the_declaration() {
        let spec = booking_tool(json!({}));
        let state = State::new();
        let resources = SpecResources::default().implement(gemini_adk_rs::tool::SimpleTool::new(
            "book",
            "real booking",
            None,
            |_| async { Ok(json!({ "confirmation": "REAL-7" })) },
        ));
        let dispatcher = spec.build_dispatcher_with(&state, &resources);
        let out = dispatcher.call_function("book", json!({})).await.unwrap();
        assert_eq!(out["confirmation"], "REAL-7");
        // The spec's state effects still apply.
        assert_eq!(state.get::<bool>("booked"), Some(true));
        assert_eq!(
            state.get::<Value>("booking").unwrap()["confirmation"],
            "REAL-7"
        );
        // And the model still sees the spec's declaration.
        assert_eq!(
            dispatcher.to_tool_declarations()[0]
                .function_declarations
                .as_ref()
                .unwrap()[0]
                .description,
            "Book the table"
        );

        // An implementation for a tool the spec does not declare is refused.
        let stray = SpecResources::default().implement(gemini_adk_rs::tool::SimpleTool::new(
            "refund",
            "",
            None,
            |_| async { Ok(json!({})) },
        ));
        assert!(spec.apply(Live::builder(), &state, &stray).is_err());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn an_mcp_binding_calls_the_tool_on_that_server() {
        // A minimal stdio MCP server: answers initialize and tools/call.
        let dir = std::env::temp_dir().join(format!("mcp-bind-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let script = dir.join("server.sh");
        std::fs::write(
            &script,
            r#"while IFS= read -r line; do
  id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p')
  [ -z "$id" ] && continue
  case "$line" in
    *'"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":"2024-11-05","capabilities":{},"serverInfo":{"name":"t","version":"1"}}}\n' "$id" ;;
    *'"tools/call"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"content":[{"type":"text","text":"{\\"confirmation\\":\\"MCP-1\\"}"}]}}\n' "$id" ;;
    *) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
  esac
done
"#,
        )
        .unwrap();
        let spec = booking_tool(json!({ "mcp": format!("sh {}", script.display()) }));
        let state = State::new();
        let dispatcher = spec.build_dispatcher_with(&state, &SpecResources::default());
        let out = dispatcher
            .call_function("book", json!({ "party": 2 }))
            .await
            .unwrap();
        assert_eq!(out, json!({ "confirmation": "MCP-1" }));
        assert_eq!(state.get::<bool>("booked"), Some(true));

        // Both bindings on one tool is a validation error.
        let both = booking_tool(json!({
            "mcp": "x",
            "http": { "url": "https://api.example.com/book" }
        }));
        assert!(!both.validate().valid);

        // A sandbox without that entry allowed turns the tool into its mock.
        let (sandboxed, notes) = spec.sandboxed(&BindingAllowlist::default());
        assert!(sandboxed.tools[0].mcp.is_none());
        assert_eq!(notes.len(), 1, "{notes:?}");
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn mcp_results_become_plain_values() {
        assert_eq!(
            mcp_value(json!({ "structuredContent": { "a": 1 }, "content": [] })),
            json!({ "a": 1 })
        );
        assert_eq!(
            mcp_value(json!({ "content": [{ "type": "text", "text": "{\"a\":2}" }] })),
            json!({ "a": 2 })
        );
        assert_eq!(
            mcp_value(json!({ "content": [{ "type": "text", "text": "done" }] })),
            json!({ "output": "done" })
        );
    }

    #[test]
    fn extraction_runs_on_the_model_the_spec_names() {
        use gemini_adk_rs::llm::{LlmRegistry, MockLlm};
        let mut spec = collections_spec();
        let mut second = spec.extract[0].clone();
        second.name = "second".into();
        spec.extract.push(second);
        spec.models.extraction = Some("fast-lite".into());
        spec.extract[0].model = Some("fast-full".into());
        assert!(!spec.uses_default_extraction_model());
        assert!(
            spec.validate()
                .warnings
                .iter()
                .all(|w| !w.contains("models.extraction")),
            "{:?}",
            spec.validate().warnings
        );

        let mut registry = LlmRegistry::new();
        registry.register("fast", |name| {
            Arc::new(MockLlm::text("{}").with_model_id(name)) as Arc<dyn BaseLlm>
        });
        let resources = SpecResources {
            models: Some(Arc::new(registry)),
            ..SpecResources::default()
        };
        let mut cache = BTreeMap::new();
        let ids: Vec<String> = spec
            .extract
            .iter()
            .map(|e| {
                extraction_llm_for(e, spec.models.extraction.as_deref(), &resources, &mut cache)
                    .unwrap()
                    .model_id()
                    .to_string()
            })
            .collect();
        assert_eq!(ids[0], "fast-full", "an entry's model wins");
        assert!(ids[1..].iter().all(|id| id == "fast-lite"), "{ids:?}");
        assert_eq!(cache.len(), 2, "one model per name");

        // Named models need no extraction_llm.
        spec.apply(Live::builder(), &State::new(), &resources)
            .expect("applies on the named models");
    }

    #[tokio::test]
    async fn an_entry_or_the_spec_sets_the_thinking_budget() {
        use gemini_adk_rs::live::TurnExtractor;
        use gemini_adk_rs::llm::{LlmResponse, MockLlm};
        let budgets = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let seen = budgets.clone();
        let llm: Arc<dyn BaseLlm> = Arc::new(MockLlm::from_fn(move |req| {
            seen.lock().push(req.thinking_budget);
            Ok(LlmResponse::from_text("{}"))
        }));
        let window = [gemini_adk_rs::live::TranscriptTurn {
            turn_number: 0,
            user: "I need a refill of my lisinopril.".into(),
            model: String::new(),
            tool_calls: Vec::new(),
            timestamp: std::time::Instant::now(),
        }];
        let entry = collections_spec().extract[0].clone();
        let models = ModelsSpec {
            extraction: None,
            extraction_thinking_budget: Some(64),
        };
        let mut own = entry.clone();
        own.thinking_budget = Some(8);
        for (e, m) in [
            (&entry, &models),
            (&own, &models),
            (&entry, &ModelsSpec::default()),
        ] {
            compile_extractor(e, llm.clone(), m)
                .extract(&window)
                .await
                .unwrap();
        }
        assert_eq!(*budgets.lock(), [Some(64), Some(8), Some(0)]);
    }

    #[test]
    fn unnamed_extraction_warns_and_an_unknown_model_fails() {
        let unnamed = collections_spec();
        assert!(unnamed.uses_default_extraction_model());
        assert!(
            unnamed
                .validate()
                .warnings
                .iter()
                .any(|w| w.contains("set models.extraction")),
            "{:?}",
            unnamed.validate().warnings
        );

        let mut unknown = collections_spec();
        unknown.models.extraction = Some("gemma-decide".into());
        let err = unknown
            .apply(Live::builder(), &State::new(), &SpecResources::default())
            .err()
            .expect("nothing resolves gemma-decide");
        assert!(err.contains("register it in SpecResources.models"), "{err}");

        let mut empty = collections_spec();
        empty.models.extraction = Some(" ".into());
        assert!(!empty.validate().valid);
    }

    #[test]
    fn apply_configures_a_builder() {
        let spec = collections_spec();
        let state = State::new();
        // No extraction LLM provided → apply must refuse (extraction declared).
        let err = spec
            .apply(Live::builder(), &state, &SpecResources::default())
            .err()
            .expect("requires extraction llm");
        assert!(err.contains("extraction_llm"));

        // Without the extraction entries it applies clean.
        let mut no_extract = spec.clone();
        no_extract.extract.clear();
        assert!(
            no_extract
                .apply(Live::builder(), &state, &SpecResources::default())
                .is_ok()
        );
    }
}