gemini_adk_fluent_rs/

conversation.rs

//! The conversation compiler (Phase 1 MVP).
//!
//! Authors describe a voice experience in terms of **stages** that *say* things,
//! *collect* slots, *commit* tools behind confirmation, and advance via *next*
//! transitions. A [`Conversation`] builder produces a serializable
//! [`ConversationSpec`] (the single source of truth), and [`Conversation::compile`]
//! lowers it to the governed [`CompiledFlow`] IR — so the high level is sugar over
//! the low level, never a parallel runtime (see
//! `docs/plans/2026-06-06-conversation-compiler-rfc.md`).
//!
//! ```no_run
//! # use gemini_adk_fluent_rs::prelude::*;
//! # use gemini_adk_fluent_rs::conversation::Conversation;
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let convo = Conversation::new("booking")
//!     .stage("collect")
//!         .instruction("Help the user book a table.")
//!         .collect(["party_size", "slot"])
//!         .next("check", Guard::captured(["party_size", "slot"]))
//!     .stage("check")
//!         .ground("Party of {party_size} at {slot}.")
//!         .next("confirm", Guard::is_true("availability_ok"))
//!     .stage("confirm")
//!         .commit("book", Guard::is_true("user_confirmed"))
//!         .next("done", Guard::called_ok("book"))
//!     .stage("done").terminal()
//!     .require(["done"])
//!     .compile()?;
//! let mut monitor = convo.monitor(Enforcement::Enforce);
//! # let _ = &mut monitor; Ok(())
//! # }
//! ```
//!
//! ### Lowering semantics (MVP)
//!
//! - A stage lowers to a Flow [`Step`](gemini_adk_rs::flow). `say` → posture,
//!   `ground` → grounding template, `allow` → tool whitelist.
//! - `commit(tool, when)` gates a confirm-before-act tool (`tool` is auto-allowed
//!   in the stage).
//! - `next(to, when)` adds a forward edge: `to` depends on this stage (`after`)
//!   and its activation gate is `when`. Multiple incoming edges are an **AND-join**
//!   (all predecessors must complete) — richer topologies are Phase 3.
//! - A non-terminal stage's completion is, in priority order: an explicit
//!   `done`, else `captured(collect)` when it collects slots, else the disjunction
//!   of its `next` conditions.

use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use gemini_adk_rs::extract::Extract;
use gemini_adk_rs::flow::{CompiledFlow, Enforcement, Flow, FlowErrors, FlowMonitor, Guard, Pred};
use gemini_adk_rs::frame::{Frame, FrameSpec};

/// A boxed async fetcher: bind args (a JSON object) → resolved value.
type SlotFetch =
    Arc<dyn Fn(Value) -> Pin<Box<dyn Future<Output = Result<Value, String>> + Send>> + Send + Sync>;

/// A registry of named async resolvers, bound to a [`ConversationSpec`]'s
/// declared resolver slots at load time via
/// [`Conversation::from_spec_with_resolvers`].
///
/// This is what makes a JSON spec a complete deployable unit: the spec carries
/// the *declarations* (slot, resolver name, args, ttl) and the registry supplies
/// the *implementations*.
///
/// ```no_run
/// # use gemini_adk_fluent_rs::conversation::{Conversation, ConversationSpec, ResolverRegistry};
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// # let spec = ConversationSpec::default();
/// let registry = ResolverRegistry::new()
///     .with("availability", |args| async move {
///         let _ = args;
///         Ok(serde_json::json!({ "open": true }))
///     });
/// let convo = Conversation::from_spec_with_resolvers(spec, &registry)?;
/// # let _ = convo; Ok(())
/// # }
/// ```
#[derive(Clone, Default)]
pub struct ResolverRegistry {
    fetchers: BTreeMap<String, SlotFetch>,
}

impl ResolverRegistry {
    /// An empty registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Register an async resolver under `name` (builder style).
    pub fn with<F, Fut>(mut self, name: impl Into<String>, fetch: F) -> Self
    where
        F: Fn(Value) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<Value, String>> + Send + 'static,
    {
        self.add(name, fetch);
        self
    }

    /// Register an async resolver under `name`.
    pub fn add<F, Fut>(&mut self, name: impl Into<String>, fetch: F)
    where
        F: Fn(Value) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<Value, String>> + Send + 'static,
    {
        let fetch = Arc::new(fetch);
        self.fetchers.insert(
            name.into(),
            Arc::new(move |v| {
                let fetch = fetch.clone();
                Box::pin(async move { fetch(v).await })
            }),
        );
    }

    /// Whether a resolver named `name` is registered.
    pub fn contains(&self, name: &str) -> bool {
        self.fetchers.contains_key(name)
    }

    /// A registry that stubs every resolver declared anywhere in `spec` (stages
    /// and overlays) with a no-op returning JSON `null`.
    ///
    /// For **structural validation and deterministic simulation**, the resolver
    /// *implementations* are irrelevant — a resolver is an external fetch, so in
    /// a model-free test its output is supplied via a scenario `set` step (like
    /// a tool result via `tool_ok`), and the stub is never actually invoked when
    /// the slot is pre-set. Real implementations bind at deploy time via
    /// [`Conversation::from_spec_with_resolvers`].
    pub fn stubbing(spec: &ConversationSpec) -> Self {
        let mut reg = Self::new();
        let stages = spec
            .stages
            .iter()
            .chain(spec.overlays.iter().flat_map(|o| o.stages.iter()));
        for stage in stages {
            for r in &stage.resolve {
                let name = r.resolver_name().to_string();
                if !reg.contains(&name) {
                    reg.add(name, |_args| async { Ok(serde_json::Value::Null) });
                }
            }
        }
        reg
    }

    fn get(&self, name: &str) -> Option<SlotFetch> {
        self.fetchers.get(name).cloned()
    }
}

/// A resolver binding attached to a stage by [`Conversation::resolve_slot`]. The
/// closure lives only in the builder (it is not serializable); the serializable
/// [`ConversationSpec`] is unaffected.
#[derive(Clone)]
struct StageResolver {
    stage: String,
    name: String,
    args: Vec<String>,
    ttl: Option<Duration>,
    fetch: SlotFetch,
}

/// A serializable declaration that a slot is filled by a *named* async resolver.
///
/// The resolver's implementation (the async fetch) is bound at load time from a
/// [`ResolverRegistry`] — so a `ConversationSpec` carrying these is a complete,
/// JSON-deployable unit once paired with a registry. This is the data half of
/// [`Conversation::resolve_slot`]; the closure half lives only in the builder.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ResolveSpec {
    /// The slot (state key) the resolver fills. Added to the stage's `collect`.
    pub slot: String,
    /// The registered resolver name to bind (defaults to `slot` if omitted).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resolver: Option<String>,
    /// State keys passed to the resolver as a JSON object argument.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub args: Vec<String>,
    /// Optional memoization TTL in seconds.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ttl_secs: Option<u64>,
}

impl ResolveSpec {
    /// The resolver name to look up in the registry (the explicit `resolver`, or
    /// the slot name as a default).
    fn resolver_name(&self) -> &str {
        self.resolver.as_deref().unwrap_or(&self.slot)
    }
}

/// A transition: advance to `to` when `when` holds.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct TransitionSpec {
    /// Target stage id.
    pub to: String,
    /// The (serializable) guard that fires this transition.
    pub when: Guard,
}

/// A confirm-before-act tool, gated by `when`.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct CommitSpec {
    /// The committing tool (e.g. `book`, `charge_card`).
    pub tool: String,
    /// The guard that must hold before the tool is admitted.
    pub when: Guard,
}

/// One authored conversation stage.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct StageSpec {
    /// Stage id, unique across the main flow and every overlay.
    pub id: String,
    /// Instruction projected as steering while the stage is active. Serialized
    /// as `say` for compatibility; `instruction` is accepted on input.
    #[serde(
        default,
        alias = "instruction",
        skip_serializing_if = "Option::is_none"
    )]
    pub say: Option<String>,
    /// Grounding template projected while active (`{key}` interpolation).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ground: Option<String>,
    /// Slots to collect; drives the default completion (`captured`).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub collect: Vec<String>,
    /// The frame whose slots this stage collects, if set via `collect_frame`. Its
    /// recognizer-bearing slots lower to an extractor that fills the slots.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub frame: Option<FrameSpec>,
    /// Tools available while this stage is active.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub allow: Vec<String>,
    /// Explicit completion guard (overrides the `collect`/`next` default).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub done: Option<Guard>,
    /// A confirm-before-act tool committed in this stage.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub commit: Option<CommitSpec>,
    /// Forward transitions out of this stage.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub next: Vec<TransitionSpec>,
    /// Explicit dependency stage ids (in addition to `next`-derived edges).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub after: Vec<String>,
    /// Whether this is a terminal stage.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub terminal: bool,
    /// Repair policy for this stage (reprompt/escalate on stalling).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub repair: Option<RepairPolicy>,
    /// Voice pacing while this stage is active: reprompt on silence, filler
    /// cues for slow tools, holding the floor, endpointing, context delivery.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timing: Option<VoiceTiming>,
    /// Text the model must say word for word in this stage (strict canned
    /// mode). The stage completes only once the output transcript matches it;
    /// see [`gemini_adk_rs::flow::verbatim`]. The model holds the floor
    /// unless `timing` says otherwise.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub verbatim: Option<String>,
    /// Named-resolver slot declarations. Bound to implementations at load via a
    /// [`ResolverRegistry`]; the data lives in the spec so it round-trips JSON.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub resolve: Vec<ResolveSpec>,
}

// The digression/repair vocabulary is owned by the runtime: the authoring layer
// lowers into it and never reimplements it. Re-exported here so existing
// `conversation::{Resume, RepairPolicy, FlowStack}` paths keep working.
pub use gemini_adk_rs::flow::{FlowStack, RepairPolicy, Resume, VoiceTiming};
use gemini_adk_rs::flow::{Overlay, correction_flag, escalate_flag, verbatim_flag};

/// A digression (overlay): a named sub-flow that suspends the main flow when its
/// `trigger` holds, runs to completion, then resumes per `resume`.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct OverlaySpec {
    /// Overlay name, unique across the conversation. The safety-handoff policy
    /// generates an overlay named `safety`.
    pub name: String,
    /// The guard that activates this overlay (e.g. an `intent:*` flag).
    pub trigger: Guard,
    /// The overlay's own stages.
    #[serde(default)]
    pub stages: Vec<StageSpec>,
    /// Stages required for the overlay to be considered complete.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub require: Vec<String>,
    /// What the main flow does once the overlay completes.
    #[serde(default)]
    pub resume: Resume,
}

/// The serializable authoring spec — the single source of truth from which the
/// typed builder, YAML, and (later) codegen all derive.
#[derive(Debug, Clone, Default, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ConversationSpec {
    /// Conversation name.
    pub name: String,
    /// The authored stages.
    #[serde(default)]
    pub stages: Vec<StageSpec>,
    /// Stages that must be done for the conversation to be complete.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub require: Vec<String>,
    /// Digressions/overlays that can suspend and resume the main flow.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub overlays: Vec<OverlaySpec>,
    /// Cross-cutting policy aspects (safety/redaction/commit governance).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub policies: Vec<crate::policy::Policy>,
}

/// Error compiling a [`ConversationSpec`] into a [`CompiledConversation`].
#[derive(Debug)]
pub enum ConversationError {
    /// The spec has no stages.
    Empty,
    /// An authoring-level error (e.g. a transition to an unknown stage).
    Spec(String),
    /// The lowered flow failed referential/acyclicity validation.
    Flow(Vec<String>),
    /// The lowered flow failed to compile (unreachable steps, unguarded commit…).
    Compile(FlowErrors),
}

impl std::fmt::Display for ConversationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ConversationError::Empty => write!(f, "conversation has no stages"),
            ConversationError::Spec(m) => write!(f, "conversation spec error: {m}"),
            ConversationError::Flow(errs) => {
                write!(f, "lowered flow is invalid: {}", errs.join("; "))
            }
            ConversationError::Compile(e) => write!(f, "lowered flow failed to compile: {e}"),
        }
    }
}

impl std::error::Error for ConversationError {}

/// The JSON Schema for a [`ConversationSpec`], as a pretty-printed string.
///
/// This is the machine-readable authoring contract: a web form, an IDE, or an
/// LLM/skill drafting a spec targets this schema. Generated from the same
/// `#[derive(JsonSchema)]` types the runtime compiles, so it cannot drift.
pub fn conversation_spec_schema() -> String {
    let schema = schemars::schema_for!(ConversationSpec);
    serde_json::to_string_pretty(&schema).expect("schema serialization is infallible")
}

impl serde::Serialize for ConversationError {
    /// A machine-readable diagnostic so authoring tools (web/CLI/skills) can
    /// render structured errors. Shape:
    /// `{ "kind": "compile", "errors": [ { "kind": "unreachable_step", ... } ] }`.
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeMap;
        let mut m = s.serialize_map(None)?;
        m.serialize_entry("message", &self.to_string())?;
        match self {
            ConversationError::Empty => {
                m.serialize_entry("kind", "empty")?;
            }
            ConversationError::Spec(msg) => {
                m.serialize_entry("kind", "spec")?;
                m.serialize_entry("detail", msg)?;
            }
            ConversationError::Flow(errs) => {
                m.serialize_entry("kind", "flow")?;
                m.serialize_entry("errors", errs)?;
            }
            ConversationError::Compile(e) => {
                m.serialize_entry("kind", "compile")?;
                m.serialize_entry("errors", &e.0)?;
            }
        }
        m.end()
    }
}

/// A compiled digression: its trigger, lowered flow, extractors, and resume policy.
#[derive(Clone)]
pub struct CompiledOverlay {
    /// Overlay name.
    pub name: String,
    /// Guard that activates the overlay.
    pub trigger: Guard,
    /// The overlay's lowered governance flow.
    pub flow: CompiledFlow,
    /// Extractors that fill the overlay's frame slots.
    pub extractors: Vec<Extract>,
    /// What the main flow does once this overlay completes.
    pub resume: Resume,
}

impl CompiledOverlay {
    /// The runtime [`Overlay`] this digression installs: trigger, governed flow
    /// and resume policy. Extractors are registered separately (see
    /// [`CompiledConversation::all_extractors`]).
    pub fn to_runtime(&self) -> Overlay {
        Overlay::new(
            self.name.clone(),
            self.trigger.clone(),
            self.flow.clone(),
            self.resume,
        )
    }
}

/// A compiled conversation: the validated main [`CompiledFlow`], the extractors
/// that fill its frames' slots, any digressions, and the source spec.
#[derive(Clone)]
pub struct CompiledConversation {
    flow: CompiledFlow,
    extractors: Vec<Extract>,
    overlays: Vec<CompiledOverlay>,
    repair: BTreeMap<String, RepairPolicy>,
    timing: BTreeMap<String, VoiceTiming>,
    corrections: BTreeMap<String, Vec<String>>,
    verbatim: BTreeMap<String, String>,
    policies: Vec<crate::policy::Policy>,
    spec: ConversationSpec,
}

// Manual: the lowered `Extract`s hold recognizer/resolver closures (not `Debug`),
// so they are summarized by count.
impl std::fmt::Debug for CompiledConversation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CompiledConversation")
            .field("flow", &self.flow)
            .field("extractors", &self.extractors.len())
            .field("overlays", &self.overlays.len())
            .field("policies", &self.policies)
            .field("spec", &self.spec)
            .finish()
    }
}

impl CompiledConversation {
    /// The governed flow IR this conversation lowered to.
    pub fn flow(&self) -> &CompiledFlow {
        &self.flow
    }
    /// The extractors lowered from `collect_frame` stages — register these on the
    /// live session so each turn fills the frames' slots from the transcript.
    pub fn extractors(&self) -> &[Extract] {
        &self.extractors
    }
    /// The compiled digressions/overlays.
    pub fn overlays(&self) -> &[CompiledOverlay] {
        &self.overlays
    }
    /// The cross-cutting policy aspects attached to this conversation.
    pub fn policies(&self) -> &[crate::policy::Policy] {
        &self.policies
    }
    /// The set of state keys marked for redaction by `Policy::redact`.
    pub fn redacted_fields(&self) -> BTreeSet<String> {
        self.policies
            .iter()
            .flat_map(|p| p.redacted_keys().iter().cloned())
            .collect()
    }
    /// Every extractor (main + overlays) — what [`Live::converse`](crate::live::Live)
    /// registers so slots fill whether the main flow or a digression is active.
    pub fn all_extractors(&self) -> Vec<Extract> {
        let mut all = self.extractors.clone();
        for ov in &self.overlays {
            all.extend(ov.extractors.iter().cloned());
        }
        all
    }
    /// Build the runtime [`FlowStack`] — the main flow plus its digressions and
    /// repair policies, with push-on-trigger / resume-on-completion.
    ///
    /// This is the artifact every execution path drives: the simulator
    /// ([`Sim`](crate::simulation::Sim)) and a live session
    /// ([`Live::converse`](crate::live::Live::converse)) both install exactly
    /// this stack, so what a definition does is the same in each.
    pub fn stack(&self, mode: Enforcement) -> FlowStack {
        FlowStack::new(self.flow.clone(), mode)
            .with_overlays(self.overlays.iter().map(CompiledOverlay::to_runtime))
            .with_repairs(self.repair.clone())
            .with_timings(self.timing.clone())
            .with_corrections(self.corrections.clone())
            .with_verbatims(self.verbatim.clone())
    }
    /// The per-stage repair policies the runtime applies to the main flow.
    pub fn repair_policies(&self) -> &BTreeMap<String, RepairPolicy> {
        &self.repair
    }
    /// The stages that must say their text word for word, keyed by stage id.
    pub fn verbatim_policies(&self) -> &BTreeMap<String, String> {
        &self.verbatim
    }
    /// The slots whose correction re-opens later stages, each with the state
    /// keys the correction clears (see
    /// [`FlowStack::with_correction`](gemini_adk_rs::flow::FlowStack::with_correction)).
    pub fn correction_policies(&self) -> &BTreeMap<String, Vec<String>> {
        &self.corrections
    }
    /// The per-stage voice timing, main flow and digressions, keyed by
    /// stage id.
    pub fn timing_policies(&self) -> &BTreeMap<String, VoiceTiming> {
        &self.timing
    }
    /// The authoring spec it was compiled from.
    pub fn spec(&self) -> &ConversationSpec {
        &self.spec
    }
    /// Render the lowered flow as a Mermaid diagram.
    pub fn to_mermaid(&self) -> String {
        self.flow.to_mermaid()
    }
    /// Build a [`FlowMonitor`] over the lowered flow.
    pub fn monitor(&self, mode: Enforcement) -> FlowMonitor {
        FlowMonitor::compiled(self.flow.clone(), mode)
    }
}

/// Fluent builder that produces a [`ConversationSpec`]; sugar over the spec.
///
/// (Not `Debug`: [`resolve_slot`](Conversation::resolve_slot) bindings hold async
/// closures. The serializable [`ConversationSpec`] is `Debug` via [`spec`](Conversation::spec).)
#[derive(Clone, Default)]
pub struct Conversation {
    spec: ConversationSpec,
    resolvers: Vec<StageResolver>,
    /// When `Some(i)`, stage setters target `spec.overlays[i]` instead of the main
    /// flow (between `.overlay(..)` and `.end_overlay()`).
    current_overlay: Option<usize>,
}

impl Conversation {
    /// Start a new conversation.
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            spec: ConversationSpec {
                name: name.into(),
                ..Default::default()
            },
            resolvers: Vec::new(),
            current_overlay: None,
        }
    }

    /// Begin authoring a new stage; subsequent setters apply to it. Routes to the
    /// active overlay when between `.overlay(..)` and `.end_overlay()`.
    pub fn stage(mut self, id: impl Into<String>) -> Self {
        let stage = StageSpec {
            id: id.into(),
            ..Default::default()
        };
        match self.current_overlay {
            Some(i) => self.spec.overlays[i].stages.push(stage),
            None => self.spec.stages.push(stage),
        }
        self
    }

    /// Begin authoring a digression/overlay; subsequent `.stage(..)` calls (until
    /// `.end_overlay()`) populate it. Set its activation guard with `.trigger(..)`
    /// (an overlay with no trigger never fires — fail-closed).
    pub fn overlay(mut self, name: impl Into<String>) -> Self {
        self.spec.overlays.push(OverlaySpec {
            name: name.into(),
            // Fail-closed default: never triggers until `.trigger(..)` is set.
            trigger: Guard::is_true("__overlay_never_triggers__"),
            stages: Vec::new(),
            require: Vec::new(),
            resume: Resume::Previous,
        });
        self.current_overlay = Some(self.spec.overlays.len() - 1);
        self
    }

    /// Set the activation guard of the overlay currently being authored.
    pub fn trigger(mut self, guard: Guard) -> Self {
        if let Some(i) = self.current_overlay {
            self.spec.overlays[i].trigger = guard;
        }
        self
    }

    /// Set the resume policy of the overlay currently being authored.
    pub fn resume(mut self, resume: Resume) -> Self {
        if let Some(i) = self.current_overlay {
            self.spec.overlays[i].resume = resume;
        }
        self
    }

    /// End overlay authoring; subsequent `.stage(..)` calls target the main
    /// flow again. (Not a sub-builder exit — the `Conversation` is the same
    /// value throughout — so it is named for what it does.)
    pub fn end_overlay(mut self) -> Self {
        self.current_overlay = None;
        self
    }

    /// Append a pre-built [`StageSpec`] (e.g. from a [`Motif`](crate::motifs::Motif)).
    /// Routes to the active overlay when authoring one, else the main flow.
    /// Subsequent stage setters (`next`, `say`, …) apply to it.
    pub fn add_stage(mut self, stage: StageSpec) -> Self {
        match self.current_overlay {
            Some(i) => self.spec.overlays[i].stages.push(stage),
            None => self.spec.stages.push(stage),
        }
        self
    }

    /// Append a pre-built [`OverlaySpec`] (e.g. a `Motif::faq_digression`). Leaves
    /// overlay-authoring mode (the overlay is already complete).
    pub fn add_overlay(mut self, overlay: OverlaySpec) -> Self {
        self.spec.overlays.push(overlay);
        self.current_overlay = None;
        self
    }

    /// Attach a cross-cutting [`Policy`](crate::policy::Policy) aspect.
    pub fn policy(mut self, policy: impl Into<crate::policy::Policy>) -> Self {
        self.spec.policies.push(policy.into());
        self
    }

    fn current(&mut self) -> &mut StageSpec {
        let stages = match self.current_overlay {
            Some(i) => &mut self.spec.overlays[i].stages,
            None => &mut self.spec.stages,
        };
        stages
            .last_mut()
            .expect("call .stage(..) before configuring a stage")
    }

    /// Set the stage's instruction: guidance projected to the model while the
    /// stage is active. It steers what the model says; it is not verbatim
    /// speech and carries no guarantee of exact wording.
    pub fn instruction(mut self, text: impl Into<String>) -> Self {
        self.current().say = Some(text.into());
        self
    }

    /// Alias of [`instruction`](Self::instruction). The name suggests exact
    /// speech, which this never was; prefer `instruction` in new code.
    pub fn say(self, text: impl Into<String>) -> Self {
        self.instruction(text)
    }

    /// Set the stage's grounding template.
    pub fn ground(mut self, template: impl Into<String>) -> Self {
        self.current().ground = Some(template.into());
        self
    }

    /// Collect the given slots in this stage (drives default completion).
    pub fn collect<I, S>(mut self, fields: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.current().collect = fields.into_iter().map(Into::into).collect();
        self
    }

    /// Collect the slots of a typed [`gemini_adk_rs::frame::Frame`] in this
    /// stage. The frame's slot state-keys drive the `captured` completion; its
    /// metadata (prompts/confirm/pii) is available via `F::frame()` for
    /// confirmation and repair.
    pub fn collect_frame<F: Frame>(mut self) -> Self {
        let spec = F::frame();
        let stage = self.current();
        stage.collect = spec.slot_keys();
        stage.frame = Some(spec);
        self
    }

    /// Allow the given tools while this stage is active.
    ///
    /// Calls to other governed tools are refused. Under
    /// [`SteeringMode::ContextUpdate`](gemini_adk_rs::live::SteeringMode::ContextUpdate)
    /// the model is also offered only the admitted tools while the stage is
    /// active.
    pub fn allow<I, S>(mut self, tools: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.current().allow = tools.into_iter().map(Into::into).collect();
        self
    }

    /// Set an explicit completion guard for this stage: the stage counts as
    /// complete when `guard` holds (default: all `collect`ed slots captured).
    pub fn complete_when(mut self, guard: Guard) -> Self {
        self.current().done = Some(guard);
        self
    }

    /// Commit a confirm-before-act tool in this stage, gated by `when`.
    pub fn commit(mut self, tool: impl Into<String>, when: Guard) -> Self {
        self.current().commit = Some(CommitSpec {
            tool: tool.into(),
            when,
        });
        self
    }

    /// Add a forward transition to `to` when `when` holds.
    pub fn next(mut self, to: impl Into<String>, when: Guard) -> Self {
        self.current().next.push(TransitionSpec {
            to: to.into(),
            when,
        });
        self
    }

    /// Fill a slot in the current stage from an **async resolver** — a tool call,
    /// HTTP fetch, MCP request, or agent. `args` names the `State` keys bound into
    /// the JSON object passed to `fetch`; the returned value fills `name`. With a
    /// `ttl`, results are memoized by `(field, canonical args)`.
    ///
    /// The slot is added to the stage's `collect`, so its `captured` completion
    /// waits for the resolution. The closure lives only in the builder; the
    /// serializable spec is unaffected.
    pub fn resolve_slot<I, S, F, Fut>(
        mut self,
        name: impl Into<String>,
        args: I,
        ttl: Option<Duration>,
        fetch: F,
    ) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
        F: Fn(Value) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<Value, String>> + Send + 'static,
    {
        let name = name.into();
        let stage = self.current().id.clone();
        let args: Vec<String> = args.into_iter().map(Into::into).collect();
        if !self.current().collect.contains(&name) {
            self.current().collect.push(name.clone());
        }
        // Record the serializable declaration too, so `into_spec()` carries the
        // resolver and the spec can later be re-bound from a `ResolverRegistry`.
        self.current().resolve.push(ResolveSpec {
            slot: name.clone(),
            resolver: None,
            args: args.clone(),
            ttl_secs: ttl.map(|d| d.as_secs()),
        });
        let fetch = Arc::new(fetch);
        self.resolvers.push(StageResolver {
            stage,
            name,
            args,
            ttl,
            fetch: Arc::new(move |v| {
                let fetch = fetch.clone();
                Box::pin(async move { fetch(v).await })
            }),
        });
        self
    }

    /// Add an explicit dependency on another stage.
    pub fn after(mut self, dep: impl Into<String>) -> Self {
        self.current().after.push(dep.into());
        self
    }

    /// Mark the current stage terminal.
    pub fn terminal(mut self) -> Self {
        self.current().terminal = true;
        self
    }

    /// Attach a [`RepairPolicy`] to the current stage.
    pub fn repair(mut self, policy: RepairPolicy) -> Self {
        self.current().repair = Some(policy);
        self
    }

    /// Require the current stage to say `text` word for word (strict canned
    /// mode): it completes only once the model's output transcript matches.
    /// The model holds the floor while it reads, unless the stage sets
    /// [`timing`](Self::timing). Needs output transcription on the session.
    pub fn verbatim(mut self, text: impl Into<String>) -> Self {
        self.current().verbatim = Some(text.into());
        self
    }

    /// Set the current stage's voice pacing; see [`VoiceTiming`].
    ///
    /// ```
    /// # use gemini_adk_fluent_rs::prelude::*;
    /// use std::time::Duration;
    /// Conversation::new("disclosure")
    ///     .stage("read_terms")
    ///     .say("Read the terms word for word.")
    ///     .timing(VoiceTiming::new().uninterruptible())
    ///     .stage("confirm")
    ///     .collect(["agreed"])
    ///     .timing(VoiceTiming::new().reprompt_after(Duration::from_secs(6)));
    /// ```
    pub fn timing(mut self, timing: VoiceTiming) -> Self {
        self.current().timing = Some(timing);
        self
    }

    /// Require these stages for completion (lowers to a Flow `require`). Targets
    /// the active overlay when authoring one, else the main flow.
    pub fn require<I, S>(mut self, steps: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let req: Vec<String> = steps.into_iter().map(Into::into).collect();
        match self.current_overlay {
            Some(i) => self.spec.overlays[i].require = req,
            None => self.spec.require = req,
        }
        self
    }

    /// The spec built so far.
    pub fn spec(&self) -> &ConversationSpec {
        &self.spec
    }

    /// Consume into the underlying spec.
    pub fn into_spec(self) -> ConversationSpec {
        self.spec
    }

    /// Compile from a [`ConversationSpec`] (e.g. parsed from JSON/YAML).
    ///
    /// If the spec declares any named-resolver slots (`stage.resolve`), this
    /// errors — those slots would be collected but never filled. Use
    /// [`Conversation::from_spec_with_resolvers`] to bind them.
    pub fn from_spec(spec: ConversationSpec) -> Result<CompiledConversation, ConversationError> {
        Self::from_spec_with_resolvers(spec, &ResolverRegistry::new())
    }

    /// Compile a [`ConversationSpec`] with its declared resolvers **stubbed**
    /// (each returns JSON `null`) — for structural validation, model-free
    /// simulation, and CI, where resolver outputs are supplied by scenario
    /// `set` steps rather than live fetches.
    ///
    /// Use this (not [`from_spec`](Self::from_spec)) when compiling a spec from
    /// untrusted JSON for testing/authoring; use
    /// [`from_spec_with_resolvers`](Self::from_spec_with_resolvers) at deploy
    /// time to bind real implementations.
    pub fn from_spec_stubbing_resolvers(
        spec: ConversationSpec,
    ) -> Result<CompiledConversation, ConversationError> {
        let registry = ResolverRegistry::stubbing(&spec);
        Self::from_spec_with_resolvers(spec, &registry)
    }

    /// Compile a [`ConversationSpec`] and bind its declared named-resolver slots
    /// from `registry`. This makes a JSON spec + a resolver registry a complete
    /// deployable unit.
    ///
    /// Errors if the spec references a resolver name absent from `registry`.
    pub fn from_spec_with_resolvers(
        spec: ConversationSpec,
        registry: &ResolverRegistry,
    ) -> Result<CompiledConversation, ConversationError> {
        let mut resolvers = Vec::new();
        let stages = spec
            .stages
            .iter()
            .chain(spec.overlays.iter().flat_map(|o| o.stages.iter()));
        for stage in stages {
            for r in &stage.resolve {
                let fetch = registry.get(r.resolver_name()).ok_or_else(|| {
                    ConversationError::Spec(format!(
                        "stage '{}' slot '{}' needs resolver '{}', which is not in the registry",
                        stage.id,
                        r.slot,
                        r.resolver_name()
                    ))
                })?;
                resolvers.push(StageResolver {
                    stage: stage.id.clone(),
                    name: r.slot.clone(),
                    args: r.args.clone(),
                    ttl: r.ttl_secs.map(Duration::from_secs),
                    fetch,
                });
            }
        }
        compile_spec(spec, resolvers)
    }

    /// Lower and validate into a [`CompiledConversation`].
    pub fn compile(self) -> Result<CompiledConversation, ConversationError> {
        compile_spec(self.spec, self.resolvers)
    }
}

impl crate::live::Live {
    /// Drive a [`Live`](crate::live::Live) session from a compiled conversation:
    /// **govern** with its lowered flow, **install** its digressions and repair
    /// policies on the session's [`FlowStack`], and **register** the extractors
    /// that fill its frames' slots each turn. The one-liner entrypoint for "run
    /// this conversation": the session drives the same stack
    /// [`CompiledConversation::stack`] builds for the simulator.
    ///
    /// ```no_run
    /// # use gemini_adk_fluent_rs::prelude::*;
    /// # use gemini_adk_fluent_rs::conversation::Conversation;
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// let convo = Conversation::new("booking")
    ///     .stage("done").terminal()
    ///     .require(["done"])
    ///     .compile()?;
    /// let handle = Live::builder()
    ///     .converse(&convo)
    ///     .connect_from_env()
    ///     .await?;
    /// # let _ = handle; Ok(())
    /// # }
    /// ```
    pub fn converse(self, convo: &CompiledConversation) -> Self {
        self.govern_compiled(convo.flow().clone())
            .install_conversation(convo)
    }

    /// Like [`converse`](Self::converse) but attaches the flow in **observe** mode
    /// (nothing blocked; deviations recorded) while still registering extractors.
    pub fn converse_observe(self, convo: &CompiledConversation) -> Self {
        self.observe_compiled(convo.flow().clone())
            .install_conversation(convo)
    }

    /// Everything a compiled conversation declares beyond its main flow: the
    /// digressions and repair policies (installed on the session's
    /// [`FlowStack`]) and the extractors (main + overlays). The main flow was
    /// attached by the caller in the chosen enforcement mode.
    fn install_conversation(mut self, convo: &CompiledConversation) -> Self {
        self.digressions = convo
            .overlays()
            .iter()
            .map(CompiledOverlay::to_runtime)
            .collect();
        self.repair_policies = convo.repair_policies().clone();
        // A timing set on the builder with `stage_timing` wins over the
        // conversation's, whichever was called first.
        for (step, timing) in convo.timing_policies() {
            self.stage_timings
                .entry(step.clone())
                .or_insert_with(|| timing.clone());
        }
        self.corrections = convo.correction_policies().clone();
        self.verbatims = convo.verbatim_policies().clone();
        // A safety hand-off is already lowered into the digressions above;
        // redaction and commit governance are enforced at connect. Policies
        // added with `Live::policy` are kept alongside.
        let mut policies: Vec<_> = convo
            .policies()
            .iter()
            .filter(|p| !matches!(p, crate::policy::Policy::SafetyHandoff { .. }))
            .cloned()
            .collect();
        policies.append(&mut self.policies);
        self.policies = policies;
        for extract in convo.all_extractors() {
            self = self.extract_record(extract);
        }
        self
    }
}

fn is_always(g: &Guard) -> bool {
    matches!(g, Guard::Spec(Pred::Always))
}

/// Combine guards into a single disjunction, collapsing trivial cases.
fn any_of(guards: Vec<Guard>) -> Option<Guard> {
    if guards.is_empty() {
        return None;
    }
    if guards.iter().any(is_always) {
        return Some(Guard::always());
    }
    if guards.len() == 1 {
        return guards.into_iter().next();
    }
    Some(Guard::any(guards))
}

/// Lower a set of stages (the main flow or an overlay) into a [`CompiledFlow`],
/// with conversation-level referential checks.
/// The stages an overlay without an explicit `require` must finish: its
/// terminal stages, else the stages no other stage follows.
fn overlay_ends(stages: &[StageSpec]) -> Vec<String> {
    let terminal: Vec<String> = stages
        .iter()
        .filter(|s| s.terminal)
        .map(|s| s.id.clone())
        .collect();
    if !terminal.is_empty() {
        return terminal;
    }
    let followed = |id: &str| stages.iter().any(|s| s.after.iter().any(|a| a == id));
    stages
        .iter()
        .filter(|s| s.next.is_empty() && !followed(&s.id))
        .map(|s| s.id.clone())
        .collect()
}

fn lower_flow(stages: &[StageSpec], require: &[String]) -> Result<CompiledFlow, ConversationError> {
    if stages.is_empty() {
        return Err(ConversationError::Empty);
    }
    let ids: BTreeSet<&str> = stages.iter().map(|s| s.id.as_str()).collect();
    for s in stages {
        for t in &s.next {
            if !ids.contains(t.to.as_str()) {
                return Err(ConversationError::Spec(format!(
                    "stage '{}' transitions to unknown stage '{}'",
                    s.id, t.to
                )));
            }
        }
        for d in &s.after {
            if !ids.contains(d.as_str()) {
                return Err(ConversationError::Spec(format!(
                    "stage '{}' depends on unknown stage '{}'",
                    s.id, d
                )));
            }
        }
        if let Some(target) = s.repair.as_ref().and_then(|r| r.escalate_to.as_ref())
            && !ids.contains(target.as_str())
        {
            return Err(ConversationError::Spec(format!(
                "stage '{}' escalates to unknown stage '{}'",
                s.id, target
            )));
        }
    }
    for r in require {
        if !ids.contains(r.as_str()) {
            return Err(ConversationError::Spec(format!(
                "require references unknown stage '{r}'"
            )));
        }
    }

    // Incoming edges: target -> [(source, when)].
    let mut incoming: BTreeMap<&str, Vec<(&str, Guard)>> = BTreeMap::new();
    for s in stages {
        for t in &s.next {
            incoming
                .entry(t.to.as_str())
                .or_default()
                .push((s.id.as_str(), t.when.clone()));
        }
        // Repair escalation is an extra edge gated on the escalate signal.
        if let Some(target) = s.repair.as_ref().and_then(|r| r.escalate_to.as_ref()) {
            incoming
                .entry(target.as_str())
                .or_default()
                .push((s.id.as_str(), Guard::is_true(escalate_flag(&s.id))));
        }
    }

    let mut fb = Flow::new();
    for s in stages {
        fb = fb.step(&s.id);

        let mut deps: BTreeSet<&str> = s.after.iter().map(String::as_str).collect();
        if let Some(inc) = incoming.get(s.id.as_str()) {
            for (src, _) in inc {
                deps.insert(src);
            }
        }
        for d in deps {
            fb = fb.after(d);
        }

        if let Some(inc) = incoming.get(s.id.as_str())
            && let Some(gate) = any_of(inc.iter().map(|(_, w)| w.clone()).collect())
        {
            fb = fb.gate(gate);
        }

        if let Some(posture) = stage_posture(s) {
            fb = fb.posture(posture);
        }
        if let Some(ground) = &s.ground {
            fb = fb.ground(ground.clone());
        }

        let mut allow: Vec<String> = s.allow.clone();
        if let Some(c) = &s.commit
            && !allow.contains(&c.tool)
        {
            allow.push(c.tool.clone());
        }
        if !allow.is_empty() {
            fb = fb.allow(allow);
        }
        if let Some(c) = &s.commit {
            fb = fb.commit(&c.tool, c.when.clone());
        }

        if s.terminal {
            fb = fb.terminal();
        } else {
            let done = stage_completion(s).ok_or_else(|| {
                ConversationError::Spec(format!(
                    "non-terminal stage '{}' has no completion (add collect, next, or done)",
                    s.id
                ))
            })?;
            fb = fb.done(done);
        }
    }

    // A corrected slot re-opens the stages downstream of it.
    for rule in reopen_rules(stages) {
        if let Some(when) = any_of(
            rule.slots
                .iter()
                .map(|slot| Guard::is_true(correction_flag(slot)))
                .collect(),
        ) {
            fb = fb.reset(rule.steps).when(when);
        }
    }

    if !require.is_empty() {
        fb = fb.require(require.to_vec());
    }

    let flow = fb.build().map_err(|e| ConversationError::Flow(e.issues))?;
    flow.compile().map_err(ConversationError::Compile)
}

/// The extractors lowered from a stage list's `collect_frame` frames.
fn frame_extractors(stages: &[StageSpec]) -> Vec<Extract> {
    stages
        .iter()
        .filter_map(|s| s.frame.as_ref().and_then(FrameSpec::to_extract))
        .collect()
}

fn compile_spec(
    mut spec: ConversationSpec,
    resolvers: Vec<StageResolver>,
) -> Result<CompiledConversation, ConversationError> {
    let authored_overlay_count = spec.overlays.len();
    // Apply cross-cutting policies. SafetyHandoff lowers to a `safety` digression
    // (terminate on intent); Redact/Commit are carried for the runtime.
    for policy in spec.policies.clone() {
        if let crate::policy::Policy::SafetyHandoff { intents } = policy
            && let Some(trigger) = any_of(
                intents
                    .iter()
                    .map(|i| Guard::is_true(format!("intent:{i}")))
                    .collect(),
            )
        {
            spec.overlays.push(OverlaySpec {
                name: "safety".into(),
                trigger,
                stages: vec![StageSpec {
                    id: "safety_handoff".into(),
                    say: Some("Safety concern detected — hand off to a human now.".into()),
                    terminal: true,
                    ..Default::default()
                }],
                require: Vec::new(),
                resume: Resume::Terminate,
            });
        }
    }

    // Stage ids also key resolver routing, timing, verbatim policies, and the
    // Studio graph. Their identity spans the whole conversation, not one layer.
    let mut overlay_names = BTreeSet::new();
    for overlay in &spec.overlays {
        if !overlay_names.insert(overlay.name.as_str()) {
            return Err(ConversationError::Spec(format!(
                "duplicate overlay name '{}'",
                overlay.name
            )));
        }
    }
    let mut stage_layers = BTreeMap::new();
    let layers = std::iter::once(("main flow".to_string(), spec.stages.as_slice())).chain(
        spec.overlays.iter().map(|overlay| {
            (
                format!("overlay '{}'", overlay.name),
                overlay.stages.as_slice(),
            )
        }),
    );
    for (layer, stages) in layers {
        for stage in stages {
            if let Some(previous) = stage_layers.insert(stage.id.as_str(), layer.clone()) {
                return Err(ConversationError::Spec(format!(
                    "duplicate stage ids: '{}' appears in {previous} and {layer}",
                    stage.id
                )));
            }
        }
    }

    // A declared resolver slot behaves like a collected slot: the stage's
    // implicit `captured` completion must wait for the resolution. The builder
    // path (`resolve_slot`) adds it eagerly; specs parsed from JSON get the
    // same normalization here, before lowering.
    for stage in spec
        .stages
        .iter_mut()
        .chain(spec.overlays.iter_mut().flat_map(|o| o.stages.iter_mut()))
    {
        for slot in stage
            .resolve
            .iter()
            .map(|r| r.slot.clone())
            .collect::<Vec<_>>()
        {
            if !stage.collect.contains(&slot) {
                stage.collect.push(slot);
            }
        }
    }

    // Main flow.
    let flow = lower_flow(&spec.stages, &spec.require)?;

    // Resolver bindings must reference a known, uniquely owned stage — main
    // or overlay, as validated above.
    let ids: BTreeSet<&str> = spec.stages.iter().map(|s| s.id.as_str()).collect();
    let mut overlay_of: BTreeMap<&str, usize> = BTreeMap::new();
    for (i, ov) in spec.overlays.iter().enumerate() {
        for s in &ov.stages {
            overlay_of.insert(s.id.as_str(), i);
        }
    }
    for r in &resolvers {
        if !ids.contains(r.stage.as_str()) && !overlay_of.contains_key(r.stage.as_str()) {
            return Err(ConversationError::Spec(format!(
                "resolver for slot '{}' references unknown stage '{}'",
                r.name, r.stage
            )));
        }
    }

    // Group resolver bindings per stage, routed to the main flow or the owning
    // overlay so overlay-declared resolvers actually fill their slots.
    let build_resolver_extractor = |stage: &str, binds: &[&StageResolver]| {
        let mut builder = Extract::record(format!("{}__{}_resolve", spec.name, stage));
        for r in binds {
            let fetch = r.fetch.clone();
            builder = builder.field_resolve(r.name.clone(), r.args.clone(), r.ttl, move |args| {
                let fetch = fetch.clone();
                async move { fetch(args).await }
            });
        }
        builder.build()
    };
    let mut by_stage: BTreeMap<&str, Vec<&StageResolver>> = BTreeMap::new();
    for r in &resolvers {
        by_stage.entry(r.stage.as_str()).or_default().push(r);
    }
    let mut overlay_extractors: BTreeMap<usize, Vec<Extract>> = BTreeMap::new();

    // Main extractors: frame recognizers + resolver-slot bindings.
    let mut extractors = frame_extractors(&spec.stages);
    for (stage, binds) in &by_stage {
        let extractor = build_resolver_extractor(stage, binds);
        if ids.contains(stage) {
            extractors.push(extractor);
        } else if let Some(&i) = overlay_of.get(stage) {
            overlay_extractors.entry(i).or_default().push(extractor);
        }
    }

    // Overlays: each lowers to its own validated flow + extractors. An overlay
    // with no explicit `require` is complete when its terminal stages are done,
    // or, with none marked terminal, its last stages (those nothing follows) —
    // so completion is meaningful (without it, `is_complete()` is trivially
    // true, and a handoff digression closed on the next turn whether or not its
    // transfer tool had run).
    let mut overlays = Vec::with_capacity(spec.overlays.len());
    for ov in &spec.overlays {
        let require = if ov.require.is_empty() {
            overlay_ends(&ov.stages)
        } else {
            ov.require.clone()
        };
        let ov_flow = lower_flow(&ov.stages, &require)?;
        let mut ov_extractors = frame_extractors(&ov.stages);
        if let Some(bound) = overlay_extractors.remove(&overlays.len()) {
            ov_extractors.extend(bound);
        }
        overlays.push(CompiledOverlay {
            name: ov.name.clone(),
            trigger: ov.trigger.clone(),
            flow: ov_flow,
            extractors: ov_extractors,
            resume: ov.resume,
        });
    }

    // Per-stage repair policies for the runtime to apply.
    let repair = spec
        .stages
        .iter()
        .filter_map(|s| s.repair.clone().map(|p| (s.id.clone(), p)))
        .collect();

    // Per-stage voice timing, main flow and digressions alike.
    let timing = spec
        .stages
        .iter()
        .chain(spec.overlays.iter().flat_map(|ov| ov.stages.iter()))
        .filter_map(|s| {
            let timing = s.timing.clone().or_else(|| {
                s.verbatim
                    .as_ref()
                    .map(|_| VoiceTiming::new().uninterruptible())
            });
            timing.map(|t| (s.id.clone(), t))
        })
        .collect();

    // Verbatim stages, main flow and digressions alike.
    let verbatim = spec
        .stages
        .iter()
        .chain(spec.overlays.iter().flat_map(|ov| ov.stages.iter()))
        .filter_map(|s| s.verbatim.clone().map(|t| (s.id.clone(), t)))
        .collect();

    // Slots whose correction re-opens downstream stages (lowered as resets
    // in the main flow above); the stack watches them.
    let mut corrections: BTreeMap<String, Vec<String>> = BTreeMap::new();
    for rule in reopen_rules(&spec.stages) {
        for slot in rule.slots {
            let clear = corrections.entry(slot).or_default();
            for key in &rule.clear {
                if !clear.contains(key) {
                    clear.push(key.clone());
                }
            }
        }
    }

    let policies = spec.policies.clone();
    // Generated overlays belong to the compiled runtime. Keep the stored spec
    // re-compilable: its policies will generate these overlays again.
    spec.overlays.truncate(authored_overlay_count);

    Ok(CompiledConversation {
        flow,
        extractors,
        overlays,
        repair,
        timing,
        corrections,
        verbatim,
        policies,
        spec,
    })
}

/// The completion guard for a non-terminal stage, by priority:
/// explicit `done` → `captured(collect)` → disjunction of `next` conditions.
/// When repair escalation is configured, the stage may also complete by escalating
/// (so a stalled stage can hand off even though its normal completion never fired).
/// How a correction to each collected slot re-opens the conversation: the
/// stages downstream of the stage that collected it, and the state keys to
/// clear so those stages run again (their commit confirmations).
struct Reopen {
    slots: Vec<String>,
    steps: Vec<String>,
    clear: Vec<String>,
}

/// For every stage that collects slots, the stages that depend on it
/// (through `after`, `next` or repair escalation, transitively) and would
/// have to run again if one of those slots were corrected.
fn reopen_rules(stages: &[StageSpec]) -> Vec<Reopen> {
    // Direct dependents: source -> stages that come after it.
    let mut dependents: BTreeMap<&str, BTreeSet<&str>> = BTreeMap::new();
    for s in stages {
        for d in &s.after {
            dependents
                .entry(d.as_str())
                .or_default()
                .insert(s.id.as_str());
        }
        for t in &s.next {
            dependents
                .entry(s.id.as_str())
                .or_default()
                .insert(t.to.as_str());
        }
        if let Some(target) = s.repair.as_ref().and_then(|r| r.escalate_to.as_ref()) {
            dependents
                .entry(s.id.as_str())
                .or_default()
                .insert(target.as_str());
        }
    }
    let collected: BTreeSet<&str> = stages
        .iter()
        .flat_map(|s| s.collect.iter().map(String::as_str))
        .collect();
    stages
        .iter()
        .filter(|s| !s.collect.is_empty())
        .filter_map(|s| {
            let mut downstream: BTreeSet<&str> = BTreeSet::new();
            let mut queue = vec![s.id.as_str()];
            while let Some(id) = queue.pop() {
                for next in dependents.get(id).into_iter().flatten() {
                    if *next != s.id && downstream.insert(next) {
                        queue.push(next);
                    }
                }
            }
            if downstream.is_empty() {
                return None;
            }
            let clear = stages
                .iter()
                .filter(|st| downstream.contains(st.id.as_str()))
                .filter_map(|st| st.commit.as_ref())
                .flat_map(|c| c.when.state_keys())
                .filter(|k| !collected.contains(k.as_str()))
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect();
            Some(Reopen {
                slots: s.collect.clone(),
                steps: downstream.into_iter().map(str::to_string).collect(),
                clear,
            })
        })
        .collect()
}

/// The posture a stage projects: its `say`, plus the exact-reading
/// instruction of a verbatim stage.
fn stage_posture(s: &StageSpec) -> Option<String> {
    let read = s.verbatim.as_ref().map(|text| {
        format!(
            "Say the following exactly as written, word for word. Do not paraphrase, \
             add to or leave out any part of it:\n\n{text}"
        )
    });
    match (&s.say, read) {
        (Some(say), Some(read)) => Some(format!("{say}\n\n{read}")),
        (say, read) => say.clone().or(read),
    }
}

fn stage_completion(s: &StageSpec) -> Option<Guard> {
    let base = stage_base_completion(s);
    // A verbatim stage also waits for its text to have been said.
    match (base, s.verbatim.is_some()) {
        (base, false) => base,
        (None, true) => Some(Guard::is_true(verbatim_flag(&s.id))),
        (Some(base), true) => Some(Guard::all(vec![base, Guard::is_true(verbatim_flag(&s.id))])),
    }
}

fn stage_base_completion(s: &StageSpec) -> Option<Guard> {
    let base = if let Some(g) = &s.done {
        Some(g.clone())
    } else if !s.collect.is_empty() {
        Some(Guard::captured(s.collect.clone()))
    } else {
        any_of(s.next.iter().map(|t| t.when.clone()).collect())
    };
    if s.repair
        .as_ref()
        .and_then(|r| r.escalate_to.as_ref())
        .is_some()
    {
        let esc = Guard::is_true(escalate_flag(&s.id));
        return Some(match base {
            Some(b) => Guard::any(vec![b, esc]),
            None => esc,
        });
    }
    base
}

#[cfg(test)]
mod tests {
    use super::*;
    use gemini_adk_rs::flow::Enforcement;
    use gemini_adk_rs::state::State;

    fn booking() -> CompiledConversation {
        Conversation::new("booking")
            .stage("collect")
            .instruction("Help the user book a table.")
            .collect(["party_size", "slot"])
            .next("check", Guard::captured(["party_size", "slot"]))
            .stage("check")
            .ground("Party of {party_size} at {slot}.")
            .next("confirm", Guard::is_true("availability_ok"))
            .stage("confirm")
            .commit("book", Guard::is_true("user_confirmed"))
            .next("done", Guard::called_ok("book"))
            .stage("done")
            .terminal()
            .require(["done"])
            .compile()
            .expect("booking compiles")
    }

    #[test]
    fn compiles_to_a_governed_flow() {
        let convo = booking();
        // The commit tool is in the tool universe and gated.
        assert!(convo.flow().tool_surface().tools.contains("book"));
        assert_eq!(convo.flow().flow().steps.len(), 4);
    }

    #[test]
    fn lowered_flow_enforces_stage_order_and_commit() {
        let convo = booking();
        let mut mon = convo.monitor(Enforcement::Enforce);
        let state = State::new();

        // First stage active; book is blocked (not allowed here + not confirmed).
        let ex = mon.explain(&state);
        assert!(ex.active.contains(&"collect".to_string()));
        assert!(ex.blocked_tools.contains_key("book"));

        // Collect the slots → collect completes, check activates.
        let _ = state.set("party_size", 4u8);
        let _ = state.set("slot", "tomorrow 7pm");
        mon.on_turn(&state);
        assert!(mon.explain(&state).active.contains(&"check".to_string()));

        // Availability → confirm activates; book still needs confirmation.
        let _ = state.set("availability_ok", true);
        mon.on_turn(&state);
        assert!(mon.admits_tool("book", &state).is_err());

        // Confirm → book admitted; calling it completes the conversation.
        let _ = state.set("user_confirmed", true);
        assert!(mon.admits_tool("book", &state).is_ok());
        mon.on_tool_ok("book", &state);
        mon.on_turn(&state);
        assert!(mon.is_complete());
    }

    #[test]
    fn spec_round_trips_through_json() {
        let spec = booking().spec().clone();
        let json = serde_json::to_string(&spec).expect("serialize spec");
        let back: ConversationSpec = serde_json::from_str(&json).expect("deserialize spec");
        let recompiled = Conversation::from_spec(back).expect("recompile from spec");
        assert_eq!(recompiled.flow().flow().steps.len(), 4);
    }

    #[tokio::test]
    async fn named_resolver_spec_round_trips_and_binds_from_registry() {
        // A spec declaring a named-resolver slot (data only, no closure).
        let json = r#"
        {
          "name": "booking",
          "stages": [
            { "id": "check",
              "resolve": [{ "slot": "availability", "resolver": "avail", "args": ["party_size"] }],
              "next": [{ "to": "done", "when": { "captured": ["availability"] } }] },
            { "id": "done", "terminal": true }
          ],
          "require": ["done"]
        }
        "#;
        let spec: ConversationSpec = serde_json::from_str(json).expect("parse");
        // Round-trips losslessly.
        let reser = serde_json::to_string(&spec).unwrap();
        let back: ConversationSpec = serde_json::from_str(&reser).unwrap();
        assert_eq!(back.stages[0].resolve[0].resolver_name(), "avail");

        // Without a registry, an unbound resolver is a loud error (not a silently
        // unfillable slot).
        let err = Conversation::from_spec(back.clone()).expect_err("unbound resolver errors");
        assert!(matches!(err, ConversationError::Spec(m) if m.contains("avail")));

        // With a registry, it compiles and the resolver extractor is wired.
        let registry = ResolverRegistry::new().with("avail", |_args| async move {
            Ok(serde_json::json!({ "open": true }))
        });
        let convo =
            Conversation::from_spec_with_resolvers(back, &registry).expect("binds and compiles");
        // The resolver lowered to an extractor that fills the `availability` slot.
        assert!(
            convo.extractors().iter().any(|e| e
                .field_state_keys()
                .iter()
                .any(|(_, k)| k == "availability")),
            "a resolver extractor filling 'availability' was compiled in"
        );
    }

    #[tokio::test]
    async fn resolver_spec_is_validatable_and_simulatable_without_a_registry() {
        // The exact shape the CLI/Python/CI data-plane sees: a resolver slot
        // with no implementation available.
        let json = r#"{ "name": "r", "stages": [
            { "id": "check",
              "resolve": [{ "slot": "avail", "resolver": "lookup", "args": ["x"] }],
              "next": [{ "to": "done", "when": { "captured": ["avail"] } }] },
            { "id": "done", "terminal": true } ], "require": ["done"] }"#;
        let spec: ConversationSpec = serde_json::from_str(json).unwrap();

        // Strict path (no registry) errors — as it should at deploy time.
        assert!(Conversation::from_spec(spec.clone()).is_err());

        // Stubbing path compiles for structural validation + simulation.
        let convo = Conversation::from_spec_stubbing_resolvers(spec)
            .expect("stubbed resolvers compile for testing");

        // A scenario supplies the resolver's output via `set` (the stub fetch is
        // never invoked when the slot is pre-set), and the flow completes.
        use crate::simulation::Sim;
        let mut sim = Sim::new(&convo, gemini_adk_rs::flow::Enforcement::Enforce);
        sim.set("avail", serde_json::json!({ "open": true }));
        sim.turn();
        assert!(sim.is_complete(), "resolver slot supplied by set completes");
    }

    #[tokio::test]
    async fn overlay_resolver_binds_validates_and_fills_the_overlay() {
        // A resolver declared on an OVERLAY stage must be validated against the
        // registry and lowered into that overlay's extractors — not silently
        // ignored (which left the slot permanently unfilled at runtime).
        let json = r#"{ "name": "ov", "stages": [
            { "id": "main", "terminal": true } ],
          "require": ["main"],
          "overlays": [ { "name": "lookup", "trigger": { "is_true": "intent:lookup" },
            "stages": [
              { "id": "fetch", "resolve": [{ "slot": "balance", "resolver": "bal" }] },
              { "id": "ov_done", "terminal": true, "after": ["fetch"] } ] } ] }"#;
        let spec: ConversationSpec = serde_json::from_str(json).unwrap();

        // Unbound overlay resolver is a loud error, same as a main-stage one.
        let err = Conversation::from_spec(spec.clone()).expect_err("unbound overlay resolver");
        assert!(matches!(err, ConversationError::Spec(m) if m.contains("bal")));

        // Bound, it compiles and the extractor lands on the overlay.
        let registry =
            ResolverRegistry::new().with("bal", |_args| async move { Ok(serde_json::json!(42)) });
        let convo = Conversation::from_spec_with_resolvers(spec, &registry).expect("compiles");
        assert!(
            convo.overlays()[0]
                .extractors
                .iter()
                .any(|e| e.field_state_keys().iter().any(|(_, k)| k == "balance")),
            "the overlay carries the resolver extractor for 'balance'"
        );
    }

    #[tokio::test]
    async fn resolver_slot_counts_toward_stage_completion_from_json() {
        // A resolver-only stage from JSON must gain the slot in `collect` (the
        // builder path adds it eagerly), so its implicit captured completion
        // exists and lowering succeeds instead of failing with "no completion".
        let json = r#"{ "name": "r2", "stages": [
            { "id": "check", "resolve": [{ "slot": "avail", "resolver": "lookup" }] },
            { "id": "done", "terminal": true, "after": ["check"] } ],
          "require": ["done"] }"#;
        let spec: ConversationSpec = serde_json::from_str(json).unwrap();
        let convo =
            Conversation::from_spec_stubbing_resolvers(spec).expect("resolver-only stage compiles");
        assert!(
            convo.spec().stages[0]
                .collect
                .contains(&"avail".to_string()),
            "declared resolver slot normalized into collect"
        );
    }

    #[test]
    fn conversation_spec_schema_is_valid_json_with_expected_shape() {
        let schema_str = conversation_spec_schema();
        let schema: serde_json::Value =
            serde_json::from_str(&schema_str).expect("schema is valid JSON");
        // The root describes ConversationSpec; its `stages` property must exist
        // (proves the transitive JsonSchema derives wired through StageSpec).
        assert_eq!(schema["title"], "ConversationSpec");
        assert!(
            schema["properties"]["stages"].is_object(),
            "schema exposes the stages property: {schema}"
        );
        // The closed atom set (Pred, via Guard) must appear in $defs.
        assert!(
            schema["definitions"]["Pred"].is_object() || schema["$defs"]["Pred"].is_object(),
            "Pred atom schema is present"
        );
    }

    #[test]
    fn conversation_error_serializes_machine_readable() {
        // An unguarded commit is rejected at compile with a Compile error.
        let err = Conversation::new("x")
            .stage("s")
            .commit("pay", Guard::always())
            .complete_when(Guard::called_ok("pay"))
            .next("done", Guard::called_ok("pay"))
            .stage("done")
            .terminal()
            .compile()
            .expect_err("unguarded commit must fail");
        let json = serde_json::to_value(&err).expect("error serializes");
        assert_eq!(json["kind"], "compile");
        assert!(json["message"].is_string());
        // The structured per-error list carries the tagged FlowError diagnostics.
        let errors = json["errors"].as_array().expect("errors array");
        assert!(
            errors.iter().any(|e| e["kind"] == "unguarded_commit_tool"),
            "structured FlowError kinds present: {json}"
        );
    }

    #[test]
    fn collect_frame_uses_frame_slot_keys() {
        use gemini_adk_rs::frame::{Frame, FrameSpec, SlotSpec};

        struct Booking;
        impl Frame for Booking {
            fn frame() -> FrameSpec {
                FrameSpec {
                    name: "booking".into(),
                    slots: vec![SlotSpec::new("party_size"), SlotSpec::new("slot")],
                }
            }
        }

        let convo = Conversation::new("b")
            .stage("collect")
            .collect_frame::<Booking>()
            .next("done", Guard::captured(["party_size", "slot"]))
            .stage("done")
            .terminal()
            .compile()
            .expect("compiles");

        // The collect stage completes on the frame's slots being captured.
        let mut mon = convo.monitor(Enforcement::Enforce);
        let state = State::new();
        assert!(mon.explain(&state).active.contains(&"collect".to_string()));
        let _ = state.set("party_size", 2u8);
        let _ = state.set("slot", "noon");
        mon.on_turn(&state);
        // Frame slots captured -> collect completes and the (terminal) done latches.
        assert!(mon.marking().done.contains("collect"));
        assert!(mon.marking().done.contains("done"));
    }

    #[tokio::test]
    async fn collect_frame_extractor_fills_and_scores_slots() {
        use gemini_adk_rs::frame::{Frame, FrameSpec, SlotRecognizer, SlotSpec};
        use gemini_adk_rs::live::TranscriptTurn;

        struct Order;
        impl Frame for Order {
            fn frame() -> FrameSpec {
                FrameSpec {
                    name: "order".into(),
                    slots: vec![SlotSpec {
                        recognizer: Some(SlotRecognizer::OneOf(vec![
                            "pizza".into(),
                            "salad".into(),
                        ])),
                        ..SlotSpec::new("item")
                    }],
                }
            }
        }

        let convo = Conversation::new("o")
            .stage("collect")
            .collect_frame::<Order>()
            .next("done", Guard::captured(["item"]))
            .stage("done")
            .terminal()
            .compile()
            .expect("compiles");

        // The frame lowered to exactly one extractor.
        assert_eq!(convo.extractors().len(), 1);
        let extractor = convo.extractors()[0].clone().into_extractor();

        // Run it over a transcript turn against State — it fills the slot and
        // records confidence under `state_meta:` for evidence.
        let state = State::new();
        let window = vec![TranscriptTurn {
            turn_number: 0,
            user: "I'd like a large PIZZA".into(),
            model: String::new(),
            tool_calls: Vec::new(),
            timestamp: std::time::Instant::now(),
        }];
        let out = extractor.extract_with_state(&window, &state).await.unwrap();
        assert_eq!(out.get("item").and_then(|v| v.as_str()), Some("pizza"));

        let ev = state.evidence("item");
        assert_eq!(ev.source.as_deref(), Some("extraction"));
        assert!(ev.confidence.unwrap() > 0.0);
    }

    #[tokio::test]
    async fn validate_rejects_out_of_range_recognized_values() {
        use gemini_adk_rs::frame::{Frame, FrameSpec, SlotRecognizer, SlotSpec, SlotValidator};
        use gemini_adk_rs::live::TranscriptTurn;

        struct Party;
        impl Frame for Party {
            fn frame() -> FrameSpec {
                FrameSpec {
                    name: "party".into(),
                    slots: vec![SlotSpec {
                        recognizer: Some(SlotRecognizer::Integer),
                        validate: Some(SlotValidator::Range {
                            min: Some(1.0),
                            max: Some(12.0),
                        }),
                        ..SlotSpec::new("party_size")
                    }],
                }
            }
        }

        let convo = Conversation::new("p")
            .stage("collect")
            .collect_frame::<Party>()
            .next("done", Guard::captured(["party_size"]))
            .stage("done")
            .terminal()
            .compile()
            .expect("compiles");
        let extractor = convo.extractors()[0].clone().into_extractor();

        let run = |text: &str| {
            let extractor = extractor.clone();
            let text = text.to_string();
            async move {
                let state = State::new();
                let window = vec![TranscriptTurn {
                    turn_number: 0,
                    user: text,
                    model: String::new(),
                    tool_calls: Vec::new(),
                    timestamp: std::time::Instant::now(),
                }];
                let out = extractor.extract_with_state(&window, &state).await.unwrap();
                out.get("party_size").cloned()
            }
        };

        // In range -> filled; out of range -> rejected (no value promoted).
        assert_eq!(run("a table for 4").await, Some(serde_json::json!(4)));
        assert_eq!(run("a table for 40").await, None);
    }

    #[tokio::test]
    async fn resolve_slot_fills_from_async_fetch() {
        use gemini_adk_rs::live::TranscriptTurn;

        let convo = Conversation::new("c")
            .stage("check")
            .resolve_slot("availability", ["party_size"], None, |args| async move {
                // Echo an availability decision derived from the bound arg.
                let n = args
                    .get("party_size")
                    .and_then(serde_json::Value::as_i64)
                    .unwrap_or(0);
                Ok(serde_json::json!(n <= 8))
            })
            .next("done", Guard::is_set("availability"))
            .stage("done")
            .terminal()
            .compile()
            .expect("compiles");

        // The resolver lowered to an extractor.
        assert_eq!(convo.extractors().len(), 1);
        let extractor = convo.extractors()[0].clone().into_extractor();

        let state = State::new();
        let _ = state.set("party_size", 4i64);
        let window = vec![TranscriptTurn {
            turn_number: 0,
            user: "any".into(),
            model: String::new(),
            tool_calls: Vec::new(),
            timestamp: std::time::Instant::now(),
        }];
        let out = extractor.extract_with_state(&window, &state).await.unwrap();
        assert_eq!(out.get("availability"), Some(&serde_json::json!(true)));
    }

    #[test]
    fn converse_registers_flow_and_extractors() {
        // Smoke test: the one-liner entrypoint wires onto a Live builder.
        use gemini_adk_rs::frame::{Frame, FrameSpec, SlotRecognizer, SlotSpec};

        struct Order;
        impl Frame for Order {
            fn frame() -> FrameSpec {
                FrameSpec {
                    name: "order".into(),
                    slots: vec![SlotSpec {
                        recognizer: Some(SlotRecognizer::OneOf(vec!["pizza".into()])),
                        ..SlotSpec::new("item")
                    }],
                }
            }
        }
        let convo = Conversation::new("o")
            .stage("collect")
            .collect_frame::<Order>()
            .next("done", Guard::captured(["item"]))
            .stage("done")
            .terminal()
            .compile()
            .expect("compiles");

        // Builds without panic; converse is the one-liner that govern()s + registers.
        let _live = crate::live::Live::builder().converse(&convo);
    }

    #[test]
    fn overlay_suspends_main_then_resumes_previous() {
        // Main: a -> b. FAQ overlay triggered by intent:faq, single terminal stage,
        // resume Previous. While the overlay runs, the main marking is untouched.
        let convo = Conversation::new("support")
            .stage("a")
            .next("b", Guard::is_true("a_done"))
            .stage("b")
            .terminal()
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            // A gated answer stage so the overlay does not complete in one turn.
            .stage("answer")
            .complete_when(Guard::is_true("faq_answered"))
            .next("faq_end", Guard::is_true("faq_answered"))
            .stage("faq_end")
            .terminal()
            .resume(Resume::Previous)
            .end_overlay()
            .compile()
            .expect("compiles");

        assert_eq!(convo.overlays().len(), 1);
        let mut stack = convo.stack(Enforcement::Enforce);
        let state = State::new();

        // Main is active on `a`.
        assert!(stack.explain(&state).active.contains(&"a".to_string()));
        assert!(stack.active_overlay().is_none());

        // Intent fires -> digression suspends the main flow and stays active.
        let _ = state.set("intent:faq", true);
        stack.on_turn(&state);
        assert_eq!(stack.active_overlay(), Some("faq"));

        // Answer the FAQ and clear the intent so the overlay completes. It is
        // still the projected layer for this closing turn.
        let _ = state.set("faq_answered", true);
        let _ = state.set("intent:faq", false);
        stack.on_turn(&state);
        assert_eq!(stack.active_overlay(), Some("faq"));

        // Next boundary: main resumed exactly where it was, still on `a`.
        stack.on_turn(&state);
        assert!(stack.active_overlay().is_none());
        assert!(stack.explain(&state).active.contains(&"a".to_string()));

        // Main continues normally afterward.
        let _ = state.set("a_done", true);
        stack.on_turn(&state);
        assert!(stack.current().marking().done.contains("a"));
    }

    /// A handoff digression whose one stage waits for the transfer tool, with
    /// no `require` and no terminal stage (the shape `adk spec plan`'s
    /// `escalation` answer writes). In a live run the model offered the
    /// transfer instead of making it; the digression closed on the next turn
    /// and, resuming with `terminate`, ended the call untransferred.
    #[test]
    fn a_handoff_digression_waits_for_its_transfer() {
        let spec: ConversationSpec = serde_json::from_value(serde_json::json!({
            "name": "dental",
            "stages": [
                { "id": "book", "allow": ["book_cleaning"], "next": [{ "to": "done", "when": { "called_ok": "book_cleaning" } }] },
                { "id": "done", "terminal": true }
            ],
            "overlays": [{
                "name": "handoff",
                "trigger": { "is_true": "intent:urgent_care" },
                "stages": [{
                    "id": "transfer",
                    "allow": ["transfer_to_staff"],
                    "done": { "called_ok": "transfer_to_staff" }
                }],
                "resume": "terminate"
            }]
        }))
        .unwrap();
        let convo = Conversation::from_spec(spec).expect("compiles");
        let mut stack = convo.stack(Enforcement::Enforce);
        let state = State::new();

        let _ = state.set("intent:urgent_care", true);
        stack.on_turn(&state);
        assert_eq!(stack.active_overlay(), Some("handoff"));
        // The model talks instead of transferring, for two turns.
        stack.on_turn(&state);
        stack.on_turn(&state);
        assert_eq!(stack.active_overlay(), Some("handoff"));
        assert!(
            !stack.is_terminated(),
            "the call is still waiting on the transfer"
        );
        assert!(stack.admits_tool("transfer_to_staff", &state).is_ok());
        assert!(stack.admits_tool("book_cleaning", &state).is_err());

        stack.on_tool_ok("transfer_to_staff", &state);
        stack.on_turn(&state);
        stack.on_turn(&state);
        assert!(stack.is_terminated(), "terminated once the transfer ran");
    }

    /// The artifact `converse()` installs on a live session is the same stack
    /// the simulator drives. Build both from one compiled conversation, advance
    /// them over the same state and turn sequence, and require identical
    /// governance at every turn: active steps, admitted/blocked tools, the
    /// active digression, and repair signals.
    #[test]
    fn live_and_simulator_drive_the_same_stack() {
        use crate::simulation::Sim;

        let convo = Conversation::new("support")
            .stage("collect")
            .allow(["lookup"])
            .complete_when(Guard::is_true("info"))
            .next("done", Guard::is_true("info"))
            .repair(RepairPolicy::new(2, 3).escalate_to("handoff"))
            .stage("done")
            .terminal()
            .stage("handoff")
            .complete_when(Guard::is_true("handoff_complete"))
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            .stage("answer")
            .allow(["search_faq"])
            .complete_when(Guard::is_true("faq_answered"))
            .next("faq_end", Guard::is_true("faq_answered"))
            .stage("faq_end")
            .terminal()
            .resume(Resume::Previous)
            .end_overlay()
            .compile()
            .expect("compiles");

        // What a live session installs (the builder's view, no network).
        let live = crate::live::Live::builder().converse(&convo);
        assert_eq!(
            live.digressions()
                .iter()
                .map(gemini_adk_rs::flow::Overlay::name)
                .collect::<Vec<_>>(),
            ["faq"]
        );
        assert!(live.repair_policies().contains_key("collect"));
        let monitor = FlowMonitor::compiled(convo.flow().clone(), Enforcement::Enforce);
        let mut installed = crate::live::connect::assemble_stack(
            monitor,
            live.digressions().to_vec(),
            live.repair_policies().clone(),
            std::collections::BTreeMap::new(),
            std::collections::BTreeMap::new(),
            std::collections::BTreeMap::new(),
            &[],
        );

        // What the simulator drives.
        let mut sim = Sim::new(&convo, Enforcement::Enforce);
        // Both stacks read the same facts.
        let state = sim.state().clone();

        let script: [&[(&str, bool)]; 6] = [
            &[],
            &[("intent:faq", true)],
            &[("faq_answered", true), ("intent:faq", false)],
            &[],
            &[],
            &[],
        ];
        for facts in script {
            for (k, v) in facts {
                let _ = state.set(*k, *v);
            }
            sim.turn();
            installed.on_turn(&state);
            let a = sim.explain();
            let b = installed.explain(&state);
            assert_eq!(a.active, b.active);
            assert_eq!(a.allowed_tools, b.allowed_tools);
            assert_eq!(a.blocked_tools, b.blocked_tools);
            assert_eq!(sim.active_overlay(), installed.active_overlay());
            assert_eq!(sim.is_complete(), installed.is_complete());
        }
        // The script exercised a digression and a repair escalation.
        assert!(
            installed
                .explain(&state)
                .active
                .contains(&"handoff".to_string())
        );
        assert_eq!(state.get::<bool>("repair:collect:escalate"), Some(true));
    }

    /// Replacing the governing flow discards the conversation's digressions:
    /// they only mean something relative to the main flow they suspend.
    #[test]
    fn regoverning_drops_installed_digressions() {
        let convo = Conversation::new("s")
            .stage("a")
            .terminal()
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            .stage("x")
            .terminal()
            .end_overlay()
            .compile()
            .expect("compiles");
        let live = crate::live::Live::builder().converse(&convo);
        assert_eq!(live.digressions().len(), 1);
        let live = live.govern(Flow::new().step("only").terminal().build().expect("valid"));
        assert!(live.digressions().is_empty());
    }

    #[test]
    fn overlay_spec_round_trips_through_json() {
        let spec = Conversation::new("s")
            .stage("main")
            .terminal()
            .overlay("cancel")
            .trigger(Guard::is_true("intent:cancel"))
            .stage("confirm")
            .terminal()
            .resume(Resume::Terminate)
            .end_overlay()
            .into_spec();
        let json = serde_json::to_string(&spec).unwrap();
        let back: ConversationSpec = serde_json::from_str(&json).unwrap();
        assert_eq!(back.overlays.len(), 1);
        assert_eq!(back.overlays[0].resume, Resume::Terminate);
        // Recompiles from the round-tripped spec.
        assert!(Conversation::from_spec(back).is_ok());
    }

    #[tokio::test]
    async fn safety_policy_terminates_on_intent() {
        use crate::policy::Policy;
        use crate::simulation::Sim;

        let convo = Conversation::new("support")
            .policy(Policy::safety_handoff(["self_harm", "abuse"]))
            .policy(Policy::redact(["card_number"]))
            .stage("triage")
            .next("resolve", Guard::is_true("triaged"))
            .stage("resolve")
            .terminal()
            .require(["resolve"])
            .compile()
            .expect("compiles");

        // Redaction set is recorded for the runtime.
        assert!(convo.redacted_fields().contains("card_number"));
        // SafetyHandoff lowered to a `safety` digression.
        assert!(convo.overlays().iter().any(|o| o.name == "safety"));

        let mut sim = Sim::new(&convo, Enforcement::Enforce);
        assert!(sim.active().contains(&"triage".to_string()));
        assert!(!sim.is_complete());

        // A safety intent fires -> the `safety` digression drives this turn,
        // and its hand-off instruction is what the model is told.
        sim.set("intent:abuse", true);
        sim.turn();
        assert_eq!(sim.active_overlay(), Some("safety"));
        assert!(!sim.is_complete());
        assert!(
            sim.postures()
                .iter()
                .any(|p| p.contains("hand off to a human")),
            "{:?}",
            sim.postures()
        );

        // Next boundary: the conversation has terminated.
        sim.turn();
        assert!(sim.is_complete());
        assert!(sim.active().is_empty());
    }

    /// The user changes a detail after confirming: the confirmation is
    /// withdrawn and the commit is blocked until they confirm again.
    #[tokio::test]
    async fn a_corrected_slot_reopens_the_confirmation() {
        use crate::simulation::{Scenario, SimStep};
        use serde_json::json;
        let convo = Conversation::new("booking")
            .stage("collect")
            .collect(["party_size"])
            .stage("book")
            .after("collect")
            .commit("book_table", Guard::is_true("confirmed"))
            .complete_when(Guard::called_ok("book_table"))
            .stage("end")
            .after("book")
            .terminal()
            .compile()
            .unwrap();
        assert_eq!(
            convo.correction_policies()["party_size"],
            ["confirmed"],
            "a correction to party_size clears the booking confirmation"
        );

        let set = |key: &str, value: serde_json::Value| SimStep::Set {
            key: key.into(),
            value,
        };
        let scenario = Scenario {
            name: "correction".into(),
            steps: vec![
                set("party_size", json!(4)),
                SimStep::Turn,
                SimStep::ExpectActive(vec!["book".into()]),
                set("confirmed", json!(true)),
                SimStep::Turn,
                SimStep::ExpectAllowed("book_table".into()),
                // "Actually, make it five."
                set("party_size", json!(5)),
                SimStep::Turn,
                SimStep::ExpectDenied("book_table".into()),
                SimStep::ExpectActive(vec!["book".into()]),
                set("confirmed", json!(true)),
                SimStep::Turn,
                SimStep::ExpectAllowed("book_table".into()),
                SimStep::ToolOk("book_table".into()),
                SimStep::ExpectComplete,
            ],
        };
        scenario.run(&convo, Enforcement::Enforce).await.unwrap();
    }

    /// Repeated barge-ins escalate a stage to its hand-off.
    #[tokio::test]
    async fn repeated_barge_ins_escalate_to_the_handoff() {
        use crate::simulation::{Scenario, SimStep};
        let convo = Conversation::new("support")
            .stage("collect")
            .collect(["issue"])
            .repair(
                RepairPolicy::new(10, 10)
                    .escalate_after_interruptions(2)
                    .escalate_to("handoff"),
            )
            .stage("handoff")
            .terminal()
            .compile()
            .unwrap();
        let scenario = Scenario {
            name: "barge-ins".into(),
            steps: vec![
                SimStep::Turn,
                SimStep::ExpectActive(vec!["collect".into()]),
                SimStep::Interrupt,
                SimStep::Turn,
                SimStep::ExpectActive(vec!["collect".into()]),
                SimStep::Interrupt,
                SimStep::ExpectSlot {
                    key: escalate_flag("collect"),
                    value: serde_json::json!(true),
                },
                SimStep::Turn,
                // The terminal hand-off completes the conversation.
                SimStep::ExpectComplete,
            ],
        };
        scenario.run(&convo, Enforcement::Enforce).await.unwrap();
    }

    #[test]
    fn stage_timing_round_trips_and_reaches_the_stack() {
        use std::time::Duration;
        let convo = Conversation::new("pacing")
            .stage("ask")
            .collect(["name"])
            .timing(VoiceTiming::new().reprompt_after(Duration::from_secs(6)))
            .add_stage(crate::motifs::Motif::disclosure("terms", "terms_ack"))
            .compile()
            .unwrap();

        let json = serde_json::to_value(convo.spec()).unwrap();
        assert_eq!(json["stages"][0]["timing"]["reprompt_after_ms"], 6000);
        assert_eq!(json["stages"][1]["timing"]["interruptible"], false);
        let back: ConversationSpec = serde_json::from_value(json).unwrap();
        assert_eq!(
            back.stages[0].timing,
            Some(VoiceTiming::new().reprompt_after(Duration::from_secs(6)))
        );

        assert!(convo.timing_policies()["terms"].holds_floor());
        let stack = convo.stack(Enforcement::Enforce);
        assert_eq!(stack.timing_policies().len(), 2);
    }

    #[test]
    fn converse_keeps_timings_and_policies_set_on_the_builder() {
        use std::time::Duration;
        let convo = Conversation::new("pacing")
            .stage("ask")
            .collect(["name"])
            .timing(VoiceTiming::new().reprompt_after(Duration::from_secs(6)))
            .stage("done")
            .after("ask")
            .terminal()
            .policy(crate::policy::Policy::redact(["card"]))
            .compile()
            .unwrap();
        let mine = VoiceTiming::new().reprompt_after(Duration::from_secs(3));

        // Before and after `converse`, the builder's own settings survive.
        for live in [
            crate::live::Live::builder()
                .stage_timing("ask", mine.clone())
                .policy(crate::policy::Policy::redact(["ssn"]))
                .converse(&convo),
            crate::live::Live::builder()
                .converse(&convo)
                .stage_timing("ask", mine.clone())
                .policy(crate::policy::Policy::redact(["ssn"])),
        ] {
            assert_eq!(live.stage_timings["ask"], mine);
            assert_eq!(live.policies.len(), 2, "{:?}", live.policies);
        }
    }

    #[tokio::test]
    async fn repair_reprompts_then_escalates_to_handoff() {
        use crate::simulation::Sim;

        // `collect` needs `info`; if the user stalls, reprompt after 2 turns and
        // escalate (route to `handoff`) after 3.
        let convo = Conversation::new("support")
            .stage("collect")
            .complete_when(Guard::is_true("info"))
            .next("done", Guard::is_true("info"))
            .repair(RepairPolicy::new(2, 3).escalate_to("handoff"))
            .stage("done")
            .terminal()
            // Non-terminal so it stays *active* (terminal stages latch instantly).
            .stage("handoff")
            .complete_when(Guard::is_true("handoff_complete"))
            .compile()
            .expect("compiles");

        let mut sim = Sim::new(&convo, Enforcement::Enforce);
        assert!(sim.active().contains(&"collect".to_string()));

        // User stalls. Turn 1 active: no signal yet.
        sim.turn();
        assert_eq!(sim.slot::<bool>("repair:collect:reprompt"), None);
        // Turn 2 active: reprompt raised.
        sim.turn();
        assert_eq!(sim.slot::<bool>("repair:collect:reprompt"), Some(true));
        assert!(sim.active().contains(&"collect".to_string()));
        // Turn 3 active: escalate raised -> stage completes via escalation -> handoff.
        sim.turn();
        assert_eq!(sim.slot::<bool>("repair:collect:escalate"), Some(true));
        assert!(sim.active().contains(&"handoff".to_string()));
        assert!(!sim.active().contains(&"collect".to_string()));
    }

    #[tokio::test]
    async fn repair_signal_clears_when_stage_satisfied() {
        use crate::simulation::Sim;

        let convo = Conversation::new("s")
            .stage("collect")
            .complete_when(Guard::is_true("info"))
            .next("done", Guard::is_true("info"))
            .repair(RepairPolicy::new(1, 9))
            .stage("done")
            .terminal()
            .require(["done"])
            .compile()
            .expect("compiles");

        let mut sim = Sim::new(&convo, Enforcement::Enforce);
        sim.turn(); // active 1 turn -> reprompt (threshold 1)
        assert_eq!(sim.slot::<bool>("repair:collect:reprompt"), Some(true));

        // User provides info -> collect completes this turn; the signal clears the
        // following turn (once the stage is observed no longer active).
        sim.set("info", true);
        sim.turn();
        sim.turn();
        assert_eq!(sim.slot::<bool>("repair:collect:reprompt"), Some(false));
        assert!(sim.is_complete());
    }

    #[test]
    fn rejects_transition_to_unknown_stage() {
        let err = Conversation::new("x")
            .stage("a")
            .next("ghost", Guard::always())
            .stage("b")
            .terminal()
            .compile()
            .expect_err("unknown target must fail");
        assert!(matches!(err, ConversationError::Spec(_)));
    }

    #[test]
    fn rejects_unguarded_commit_via_flow_compile() {
        // commit guarded by Always is an unguarded commit — Flow::compile rejects it.
        let err = Conversation::new("x")
            .stage("s")
            .commit("pay", Guard::always())
            .complete_when(Guard::called_ok("pay"))
            .next("done", Guard::called_ok("pay"))
            .stage("done")
            .terminal()
            .compile()
            .expect_err("unguarded commit must fail");
        assert!(matches!(err, ConversationError::Compile(_)));
    }

    #[test]
    fn stage_ids_must_be_unique_across_main_and_overlays() {
        let error = Conversation::new("collision")
            .stage("shared")
            .terminal()
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            .stage("shared")
            .terminal()
            .end_overlay()
            .compile()
            .expect_err("a stage cannot have two owners");
        assert!(matches!(error, ConversationError::Spec(message)
            if message.contains("duplicate stage ids")
                && message.contains("shared")
                && message.contains("main flow")
                && message.contains("overlay 'faq'")));
    }

    #[test]
    fn stage_ids_must_be_unique_between_overlays() {
        let error = Conversation::new("collision")
            .stage("main")
            .terminal()
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            .stage("shared")
            .terminal()
            .end_overlay()
            .overlay("clarify")
            .trigger(Guard::is_true("intent:clarify"))
            .stage("shared")
            .terminal()
            .end_overlay()
            .compile()
            .expect_err("overlays cannot share stage identity");
        assert!(matches!(error, ConversationError::Spec(message)
            if message.contains("shared")
                && message.contains("overlay 'faq'")
                && message.contains("overlay 'clarify'")));
    }

    #[test]
    fn overlay_names_must_be_unique() {
        let error = Conversation::new("collision")
            .stage("main")
            .terminal()
            .overlay("faq")
            .trigger(Guard::is_true("intent:faq"))
            .stage("answer")
            .terminal()
            .end_overlay()
            .overlay("faq")
            .trigger(Guard::is_true("intent:other_faq"))
            .stage("other_answer")
            .terminal()
            .end_overlay()
            .compile()
            .expect_err("overlay names select active layers");
        assert!(matches!(error, ConversationError::Spec(message)
            if message == "duplicate overlay name 'faq'"));
    }

    #[test]
    fn generated_safety_overlay_cannot_collide_with_authored_names_or_stages() {
        use crate::policy::Policy;

        let error = Conversation::new("collision")
            .stage("main")
            .terminal()
            .overlay("safety")
            .trigger(Guard::is_true("intent:custom_safety"))
            .stage("custom_handoff")
            .terminal()
            .end_overlay()
            .policy(Policy::safety_handoff(["self_harm"]))
            .compile()
            .expect_err("generated safety overlay requires a unique name");
        assert!(matches!(error, ConversationError::Spec(message)
            if message == "duplicate overlay name 'safety'"));

        let error = Conversation::new("collision")
            .stage("safety_handoff")
            .terminal()
            .policy(Policy::safety_handoff(["self_harm"]))
            .compile()
            .expect_err("generated safety stage requires a unique id");
        assert!(matches!(error, ConversationError::Spec(message)
            if message.contains("safety_handoff")
                && message.contains("main flow")
                && message.contains("overlay 'safety'")));
    }

    #[test]
    fn duplicate_stage_ids_within_one_layer_remain_invalid() {
        let error = Conversation::new("collision")
            .stage("shared")
            .terminal()
            .stage("shared")
            .terminal()
            .compile()
            .expect_err("local duplicate stages remain invalid");
        assert!(matches!(error, ConversationError::Spec(message)
            if message.contains("duplicate stage ids")));
    }

    #[test]
    fn compiled_safety_spec_can_be_serialized_and_compiled_again() {
        use crate::policy::Policy;

        let compiled = Conversation::new("safety")
            .stage("main")
            .terminal()
            .policy(Policy::safety_handoff(["self_harm"]))
            .compile()
            .expect("compiles");
        assert_eq!(compiled.overlays().len(), 1);
        let value = serde_json::to_value(compiled.spec()).expect("serializes");
        let spec = serde_json::from_value(value).expect("parses");
        let recompiled = Conversation::from_spec(spec).expect("recompiles");
        assert_eq!(recompiled.overlays().len(), 1);
        assert_eq!(recompiled.overlays()[0].name, "safety");
    }
}