ai_agents_runtime/

runtime.rs

use async_trait::async_trait;
use futures::stream::{Stream, StreamExt};
use parking_lot::RwLock;
use serde_json::Value;
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Instant;
use tracing::{debug, error, info, instrument, warn};

use ai_agents_context::{ContextManager, ContextProvider, TemplateRenderer};
use ai_agents_core::{
    AgentError, AgentSnapshot, AgentStorage, ChatMessage, FinishReason, LLMProvider, LLMResponse,
    Result, ToolResult,
};
use ai_agents_disambiguation::{
    DisambiguationConfig, DisambiguationContext, DisambiguationManager, DisambiguationResult,
};
use ai_agents_hitl::{
    ApprovalHandler, ApprovalResult, ApprovalTrigger, HITLCheckResult, HITLEngine,
    RejectAllHandler, TimeoutAction,
};
use ai_agents_hooks::{AgentHooks, NoopHooks};
use ai_agents_llm::LLMRegistry;
use ai_agents_memory::{
    CompressResult, EvictionReason, Memory, MemoryBudgetEvent, MemoryCompressEvent,
    MemoryEvictEvent, MemoryTokenBudget, OverflowStrategy,
};
use ai_agents_process::{ProcessData, ProcessProcessor};
use ai_agents_reasoning::{
    CriterionResult, EvaluationResult, Plan, PlanAction, PlanStatus, PlanStep, ReasoningConfig,
    ReasoningMetadata, ReasoningMode, ReasoningOutput, ReflectionAttempt, ReflectionConfig,
    ReflectionMetadata, StepFailureAction,
};
use ai_agents_recovery::{
    ByRoleFilter, ContextOverflowAction, FilterConfig, IntoClassifiedError, KeepRecentFilter,
    LLMFailureAction, MessageFilter, RecoveryManager, RetryConfig, SkipPatternFilter,
    ToolFailureAction,
};
use ai_agents_skills::{SkillDefinition, SkillExecutor, SkillRouter};
use ai_agents_state::{
    PromptMode, StateAction, StateMachine, StateMachineSnapshot, StateTransitionEvent, ToolRef,
    TransitionContext, TransitionEvaluator,
};
use ai_agents_storage::{StorageConfig as StorageStorageConfig, create_storage};
use ai_agents_tools::{
    ConditionEvaluator, EvaluationContext, LLMGetter, SecurityCheckResult, ToolCallRecord,
    ToolRegistry, ToolSecurityEngine,
};

use super::{
    Agent, AgentInfo, AgentResponse, ParallelToolsConfig, StreamChunk, StreamingConfig, ToolCall,
};
use crate::spec::StorageConfig;

/// Outcome of processing tool calls within the agent loop.
enum ToolCallOutcome {
    /// Tools executed successfully, continue the LLM loop for the next iteration.
    Continue,
    /// A state transition fired during tool call handling, continue the loop.
    TransitionFired,
    /// HITL rejected a tool call, return this response immediately.
    Rejected(AgentResponse),
}

/// Outcome of skill routing — used by `try_skill_route`.
enum SkillRouteResult {
    /// No skill matched, continue to normal LLM chat.
    NoMatch,
    /// Skill executed successfully.
    Response(String),
    /// Skill matched but needs disambiguation first.
    NeedsClarification(AgentResponse),
}

/// Outcome of post_loop_processing - drives the caller's next step.
enum PostLoopResult {
    /// No transition fired. Content is the LLM response for this turn.
    NoTransition(String),
    /// Transition fired. Content is from plain post-transition re-generation.
    Transitioned(String),
    /// Transition fired into a state that requires full dispatch.
    /// Caller re-enters run_loop_internal to apply the correct handler.
    NeedsRedispatch,
}

pub struct RuntimeAgent {
    info: AgentInfo,
    llm_registry: Arc<LLMRegistry>,
    memory: Arc<dyn Memory>,
    tools: Arc<ToolRegistry>,
    skills: Vec<SkillDefinition>,
    skill_router: Option<SkillRouter>,
    skill_executor: Option<SkillExecutor>,
    base_system_prompt: String,
    max_iterations: u32,
    iteration_count: RwLock<u32>,
    max_context_tokens: u32,
    memory_token_budget: Option<MemoryTokenBudget>,
    recovery_manager: RecoveryManager,
    tool_security: ToolSecurityEngine,
    process_processor: Option<ProcessProcessor>,
    message_filters: RwLock<HashMap<String, Arc<dyn MessageFilter>>>,
    state_machine: Option<Arc<StateMachine>>,
    transition_evaluator: Option<Arc<dyn TransitionEvaluator>>,
    context_manager: Arc<ContextManager>,
    template_renderer: TemplateRenderer,
    tool_call_history: RwLock<Vec<ToolCallRecord>>,
    parallel_tools: ParallelToolsConfig,
    streaming: StreamingConfig,
    hooks: Arc<dyn AgentHooks>,
    hitl_engine: Option<HITLEngine>,
    approval_handler: Arc<dyn ApprovalHandler>,
    storage_config: StorageConfig,
    storage: RwLock<Option<Arc<dyn AgentStorage>>>,
    reasoning_config: ReasoningConfig,
    reflection_config: ReflectionConfig,
    disambiguation_manager: Option<DisambiguationManager>,
    /// Structured persona manager for identity, evolution, and secrets.
    persona_manager: Option<Arc<ai_agents_persona::PersonaManager>>,
    /// Skill ID that triggered the current pending disambiguation.
    /// Set by try_skill_route() when skill-level disambiguation triggers clarification.
    /// Read by run_loop() when clarification resolves to route directly to the skill.
    pending_skill_id: RwLock<Option<String>>,
    current_plan: RwLock<Option<Plan>>,
    /// Tool IDs declared in the top-level `tools:` spec.
    declared_tool_ids: Option<Vec<String>>,
    /// Whether the context manager has been initialized (defaults loaded, env resolved, etc.)
    context_initialized: AtomicBool,
    /// Spawner for dynamic agent creation (set when YAML has a spawner: section).
    spawner: Option<Arc<crate::spawner::AgentSpawner>>,
    /// Registry tracking spawned agents (set when YAML has a spawner: section).
    spawner_registry: Option<Arc<crate::spawner::AgentRegistry>>,
    /// Re-dispatch depth for post-transition full dispatch.
    /// 0 = not re-dispatching. > 0 = user message already in memory, skip re-adding.
    redispatch_depth: RwLock<u32>,
}

impl std::fmt::Debug for RuntimeAgent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RuntimeAgent")
            .field("info", &self.info)
            .field("base_system_prompt", &self.base_system_prompt)
            .field("max_iterations", &self.max_iterations)
            .field("skills_count", &self.skills.len())
            .field("max_context_tokens", &self.max_context_tokens)
            .field("has_state_machine", &self.state_machine.is_some())
            .field("parallel_tools", &self.parallel_tools)
            .field("streaming", &self.streaming)
            .field("has_hooks", &true)
            .field("has_hitl", &self.hitl_engine.is_some())
            .field("storage_type", &self.storage_config.storage_type())
            .field("reasoning_mode", &self.reasoning_config.mode)
            .field("reflection_enabled", &self.reflection_config.enabled)
            .field("declared_tool_ids", &self.declared_tool_ids)
            .field("has_persona", &self.persona_manager.is_some())
            .finish_non_exhaustive()
    }
}

struct RegistryLLMGetter {
    registry: Arc<LLMRegistry>,
}

impl LLMGetter for RegistryLLMGetter {
    fn get_llm(&self, alias: &str) -> Option<Arc<dyn LLMProvider>> {
        self.registry.get(alias).ok()
    }
}

impl RuntimeAgent {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        info: AgentInfo,
        llm_registry: Arc<LLMRegistry>,
        memory: Arc<dyn Memory>,
        tools: Arc<ToolRegistry>,
        skills: Vec<SkillDefinition>,
        system_prompt: String,
        max_iterations: u32,
    ) -> Self {
        let (skill_router, skill_executor) = if !skills.is_empty() {
            let router_llm = llm_registry.router().ok();
            let router = router_llm.map(|llm| SkillRouter::new(llm, skills.clone()));
            let executor = SkillExecutor::new(llm_registry.clone(), tools.clone());
            (router, Some(executor))
        } else {
            (None, None)
        };

        let context_manager =
            ContextManager::new(HashMap::new(), info.name.clone(), info.version.clone());

        Self {
            info,
            llm_registry,
            memory,
            tools,
            skills,
            skill_router,
            skill_executor,
            base_system_prompt: system_prompt,
            max_iterations,
            iteration_count: RwLock::new(0),
            max_context_tokens: 4096,
            memory_token_budget: None,
            recovery_manager: RecoveryManager::default(),
            tool_security: ToolSecurityEngine::default(),
            process_processor: None,
            message_filters: RwLock::new(HashMap::new()),
            state_machine: None,
            transition_evaluator: None,
            context_manager: Arc::new(context_manager),
            template_renderer: TemplateRenderer::new(),
            tool_call_history: RwLock::new(Vec::new()),
            parallel_tools: ParallelToolsConfig::default(),
            streaming: StreamingConfig::default(),
            hooks: Arc::new(NoopHooks),
            hitl_engine: None,
            approval_handler: Arc::new(RejectAllHandler::new()),
            storage_config: StorageConfig::default(),
            storage: RwLock::new(None),
            reasoning_config: ReasoningConfig::default(),
            reflection_config: ReflectionConfig::default(),
            disambiguation_manager: None,
            persona_manager: None,
            pending_skill_id: RwLock::new(None),
            current_plan: RwLock::new(None),
            declared_tool_ids: None,
            context_initialized: AtomicBool::new(false),
            spawner: None,
            spawner_registry: None,
            redispatch_depth: RwLock::new(0),
        }
    }

    pub fn with_declared_tool_ids(mut self, ids: Option<Vec<String>>) -> Self {
        self.declared_tool_ids = ids;
        self
    }

    pub fn with_storage_config(mut self, config: StorageConfig) -> Self {
        self.storage_config = config;
        self
    }

    pub fn with_storage(self, storage: Arc<dyn AgentStorage>) -> Self {
        *self.storage.write() = Some(storage);
        self
    }

    pub fn with_reasoning(mut self, config: ReasoningConfig) -> Self {
        self.reasoning_config = config;
        self
    }

    pub fn with_reflection(mut self, config: ReflectionConfig) -> Self {
        self.reflection_config = config;
        self
    }

    pub fn reasoning_config(&self) -> &ReasoningConfig {
        &self.reasoning_config
    }

    pub fn reflection_config(&self) -> &ReflectionConfig {
        &self.reflection_config
    }

    pub fn with_persona(mut self, manager: Arc<ai_agents_persona::PersonaManager>) -> Self {
        self.persona_manager = Some(manager);
        self
    }

    pub fn persona_manager(&self) -> Option<&Arc<ai_agents_persona::PersonaManager>> {
        self.persona_manager.as_ref()
    }

    pub fn with_disambiguation(mut self, config: DisambiguationConfig) -> Self {
        if config.is_enabled() {
            self.disambiguation_manager = Some(DisambiguationManager::new(
                config,
                Arc::clone(&self.llm_registry),
            ));
        }
        self
    }

    pub fn disambiguation_manager(&self) -> Option<&DisambiguationManager> {
        self.disambiguation_manager.as_ref()
    }

    pub fn has_disambiguation(&self) -> bool {
        self.disambiguation_manager
            .as_ref()
            .is_some_and(|m| m.is_enabled())
    }

    pub async fn init_storage(&self) -> Result<()> {
        if self.storage_config.is_none() {
            return Ok(());
        }
        if self.storage.read().is_some() {
            return Ok(());
        }
        let storage_config = self.convert_storage_config();
        let storage = create_storage(&storage_config).await?;
        *self.storage.write() = storage;
        Ok(())
    }

    fn convert_storage_config(&self) -> StorageStorageConfig {
        crate::spec::storage::to_storage_config(&self.storage_config)
    }

    pub fn storage(&self) -> Option<Arc<dyn AgentStorage>> {
        self.storage.read().clone()
    }

    pub fn storage_config(&self) -> &StorageConfig {
        &self.storage_config
    }

    /// Returns the spawner if configured via a spawner: YAML section.
    pub fn spawner(&self) -> Option<&Arc<crate::spawner::AgentSpawner>> {
        self.spawner.as_ref()
    }

    /// Returns the agent registry if configured via a spawner: YAML section.
    pub fn spawner_registry(&self) -> Option<&Arc<crate::spawner::AgentRegistry>> {
        self.spawner_registry.as_ref()
    }

    pub fn has_spawner(&self) -> bool {
        self.spawner_registry.is_some()
    }

    pub fn with_spawner_handles(
        mut self,
        spawner: Arc<crate::spawner::AgentSpawner>,
        registry: Arc<crate::spawner::AgentRegistry>,
    ) -> Self {
        self.spawner = Some(spawner);
        self.spawner_registry = Some(registry);
        self
    }

    pub fn with_hooks(mut self, hooks: Arc<dyn AgentHooks>) -> Self {
        self.hooks = hooks;
        self
    }

    pub fn with_parallel_tools(mut self, config: ParallelToolsConfig) -> Self {
        self.parallel_tools = config;
        self
    }

    pub fn with_streaming(mut self, config: StreamingConfig) -> Self {
        self.streaming = config;
        self
    }

    pub fn with_hitl(mut self, engine: HITLEngine, handler: Arc<dyn ApprovalHandler>) -> Self {
        self.hitl_engine = Some(engine);
        self.approval_handler = handler;
        self
    }

    pub fn with_max_context_tokens(mut self, tokens: u32) -> Self {
        self.max_context_tokens = tokens;
        self
    }

    pub fn with_memory_token_budget(mut self, budget: MemoryTokenBudget) -> Self {
        self.memory_token_budget = Some(budget);
        self
    }

    pub fn with_recovery_manager(mut self, manager: RecoveryManager) -> Self {
        self.recovery_manager = manager;
        self
    }

    pub fn with_tool_security(mut self, engine: ToolSecurityEngine) -> Self {
        self.tool_security = engine;
        self
    }

    pub fn with_process_processor(mut self, processor: ProcessProcessor) -> Self {
        self.process_processor = Some(processor);
        self
    }

    pub fn with_state_machine(
        mut self,
        state_machine: Arc<StateMachine>,
        evaluator: Arc<dyn TransitionEvaluator>,
    ) -> Self {
        self.state_machine = Some(state_machine);
        self.transition_evaluator = Some(evaluator);
        self
    }

    pub fn with_context_manager(mut self, manager: Arc<ContextManager>) -> Self {
        self.context_manager = manager;
        self
    }

    pub fn register_message_filter(&self, name: impl Into<String>, filter: Arc<dyn MessageFilter>) {
        self.message_filters.write().insert(name.into(), filter);
    }

    pub fn set_context(&self, key: &str, value: Value) -> Result<()> {
        self.context_manager.set(key, value)
    }

    pub fn update_context(&self, path: &str, value: Value) -> Result<()> {
        self.context_manager.update(path, value)
    }

    pub fn get_context(&self) -> HashMap<String, Value> {
        self.context_manager.get_all()
    }

    pub fn remove_context(&self, key: &str) -> Option<Value> {
        self.context_manager.remove(key)
    }

    pub async fn refresh_context(&self, key: &str) -> Result<()> {
        self.context_manager.refresh(key).await
    }

    pub fn register_context_provider(&self, name: &str, provider: Arc<dyn ContextProvider>) {
        self.context_manager.register_provider(name, provider);
    }

    pub fn current_state(&self) -> Option<String> {
        self.state_machine.as_ref().map(|sm| sm.current())
    }

    pub async fn transition_to(&self, state: &str) -> Result<()> {
        if let Some(ref sm) = self.state_machine {
            let from_state = sm.current();
            self.execute_state_exit_actions(&from_state).await;
            sm.transition_to(state, "manual transition")?;
            self.execute_state_enter_actions(state).await;
            info!(to = %state, "Manual state transition");
        }
        Ok(())
    }

    pub fn state_history(&self) -> Vec<StateTransitionEvent> {
        self.state_machine
            .as_ref()
            .map(|sm| sm.history())
            .unwrap_or_default()
    }

    pub async fn save_state(&self) -> Result<AgentSnapshot> {
        let memory_snapshot = self.memory.snapshot().await?;
        let state_machine_snapshot = self.state_machine.as_ref().map(|sm| sm.snapshot());
        let context_snapshot = self.context_manager.snapshot();

        let mut snapshot = AgentSnapshot::new(self.info.id.clone())
            .with_memory(memory_snapshot)
            .with_context(context_snapshot)
            .with_state_machine(
                state_machine_snapshot.unwrap_or_else(|| StateMachineSnapshot {
                    current_state: String::new(),
                    previous_state: None,
                    turn_count: 0,
                    no_transition_count: 0,
                    history: vec![],
                }),
            );

        if let Some(ref persona) = self.persona_manager {
            snapshot.persona = Some(persona.snapshot_as_value()?);
        }

        Ok(snapshot)
    }

    /// Save state including spawned agents manifest for session persistence.
    pub async fn save_state_full(&self) -> Result<AgentSnapshot> {
        let mut snapshot = self.save_state().await?;
        if let Some(ref registry) = self.spawner_registry {
            let entries = registry.list_with_specs();
            if !entries.is_empty() {
                snapshot = snapshot.with_spawned_agents(entries);
            }
        }
        Ok(snapshot)
    }

    pub async fn restore_state(&self, snapshot: AgentSnapshot) -> Result<()> {
        self.memory.restore(snapshot.memory).await?;

        if let (Some(sm), Some(sm_snapshot)) = (&self.state_machine, snapshot.state_machine) {
            if !sm_snapshot.current_state.is_empty() {
                sm.restore(sm_snapshot)?;
            }
        }

        self.context_manager.restore(snapshot.context);

        if let (Some(persona_value), Some(persona_manager)) =
            (snapshot.persona, &self.persona_manager)
        {
            persona_manager.restore_from_value(persona_value)?;
        }

        info!(agent_id = %snapshot.agent_id, "State restored");
        Ok(())
    }

    pub async fn save_to(&self, storage: &dyn AgentStorage, session_id: &str) -> Result<()> {
        let snapshot = self.save_state().await?;
        storage.save(session_id, &snapshot).await
    }

    pub async fn load_from(&self, storage: &dyn AgentStorage, session_id: &str) -> Result<bool> {
        if let Some(snapshot) = storage.load(session_id).await? {
            self.restore_state(snapshot).await?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    pub async fn save_session(&self, session_id: &str) -> Result<()> {
        let storage = self.storage.read().clone();
        match storage {
            Some(s) => self.save_to(s.as_ref(), session_id).await,
            None => Err(AgentError::Config(
                "No storage configured. Use with_storage_config() or with_storage() first".into(),
            )),
        }
    }

    pub async fn load_session(&self, session_id: &str) -> Result<bool> {
        let storage = self.storage.read().clone();
        match storage {
            Some(s) => self.load_from(s.as_ref(), session_id).await,
            None => Err(AgentError::Config(
                "No storage configured. Use with_storage_config() or with_storage() first".into(),
            )),
        }
    }

    pub async fn delete_session(&self, session_id: &str) -> Result<()> {
        let storage = self.storage.read().clone();
        match storage {
            Some(s) => s.delete(session_id).await,
            None => Err(AgentError::Config(
                "No storage configured. Use with_storage_config() or with_storage() first".into(),
            )),
        }
    }

    pub async fn list_sessions(&self) -> Result<Vec<String>> {
        let storage = self.storage.read().clone();
        match storage {
            Some(s) => s.list_sessions().await,
            None => Err(AgentError::Config(
                "No storage configured. Use with_storage_config() or with_storage() first".into(),
            )),
        }
    }

    fn estimate_tokens(&self, text: &str) -> u32 {
        (text.len() as f32 / 4.0).ceil() as u32
    }

    fn estimate_total_tokens(&self, messages: &[ChatMessage]) -> u32 {
        messages
            .iter()
            .map(|m| self.estimate_tokens(&m.content))
            .sum()
    }

    fn truncate_context(&self, messages: &mut Vec<ChatMessage>, keep_recent: usize) {
        if messages.len() <= keep_recent + 1 {
            return;
        }
        let system_msg = messages.remove(0);
        let to_remove = messages.len().saturating_sub(keep_recent);
        messages.drain(..to_remove);
        messages.insert(0, system_msg);
    }

    fn get_filter(&self, config: &FilterConfig) -> Arc<dyn MessageFilter> {
        match config {
            FilterConfig::KeepRecent(n) => Arc::new(KeepRecentFilter::new(*n)),
            FilterConfig::ByRole { keep_roles } => Arc::new(ByRoleFilter::new(keep_roles.clone())),
            FilterConfig::SkipPattern { skip_if_contains } => {
                Arc::new(SkipPatternFilter::new(skip_if_contains.clone()))
            }
            FilterConfig::Custom { name } => {
                let filters = self.message_filters.read();
                filters
                    .get(name)
                    .cloned()
                    .unwrap_or_else(|| Arc::new(KeepRecentFilter::new(10)))
            }
        }
    }

    async fn summarize_context(
        &self,
        messages: &mut Vec<ChatMessage>,
        summarizer_llm: Option<&str>,
        max_summary_tokens: u32,
        custom_prompt: Option<&str>,
        keep_recent: usize,
        filter: Option<&FilterConfig>,
    ) -> Result<()> {
        let system_msg = messages.remove(0);

        let to_summarize_count = messages.len().saturating_sub(keep_recent);
        if to_summarize_count == 0 {
            messages.insert(0, system_msg);
            return Ok(());
        }

        let recent_msgs: Vec<ChatMessage> = messages.drain(to_summarize_count..).collect();
        let mut to_summarize = std::mem::take(messages);

        if let Some(filter_config) = filter {
            let filter = self.get_filter(filter_config);
            to_summarize = filter.filter(to_summarize);
        }

        if to_summarize.is_empty() {
            *messages = recent_msgs;
            messages.insert(0, system_msg);
            return Ok(());
        }

        let conversation_text = to_summarize
            .iter()
            .map(|m| format!("{:?}: {}", m.role, m.content))
            .collect::<Vec<_>>()
            .join("\n");

        let default_prompt = format!(
            "Summarize the following conversation in under {} tokens, preserving key information:\n\n{}",
            max_summary_tokens, conversation_text
        );

        let summary_prompt = custom_prompt
            .map(|p| format!("{}\n\n{}", p, conversation_text))
            .unwrap_or(default_prompt);

        let summarizer = if let Some(alias) = summarizer_llm {
            self.llm_registry
                .get(alias)
                .map_err(|e| AgentError::Config(e.to_string()))?
        } else {
            self.llm_registry
                .router()
                .or_else(|_| self.llm_registry.default())
                .map_err(|e| AgentError::Config(e.to_string()))?
        };

        let summary_msgs = vec![ChatMessage::user(&summary_prompt)];
        let response = summarizer.complete(&summary_msgs, None).await?;

        let summary_message = ChatMessage::system(&format!(
            "[Previous conversation summary]\n{}",
            response.content
        ));

        *messages = vec![system_msg, summary_message];
        messages.extend(recent_msgs);

        debug!(
            summarized_count = to_summarize_count,
            kept_recent = keep_recent,
            "Context summarized"
        );

        Ok(())
    }

    fn render_system_prompt(&self) -> Result<String> {
        let context = self.context_manager.get_all();
        self.template_renderer
            .render(&self.base_system_prompt, &context)
    }

    async fn get_available_tool_ids(&self) -> Result<Vec<String>> {
        match self.get_current_tool_refs() {
            // State explicitly sets tools (including empty = no tools)
            Some(tool_refs) => {
                if tool_refs.is_empty() {
                    // Explicitly empty: no tools available in this state
                    return Ok(Vec::new());
                }

                let eval_ctx = self.build_evaluation_context().await?;
                let llm_getter = RegistryLLMGetter {
                    registry: self.llm_registry.clone(),
                };
                let evaluator = ConditionEvaluator::new(llm_getter);

                let mut available = Vec::new();
                for tool_ref in &tool_refs {
                    let tool_id = tool_ref.id();

                    if self.tools.get(tool_id).is_none() {
                        continue;
                    }

                    if let Some(condition) = tool_ref.condition() {
                        match evaluator.evaluate(condition, &eval_ctx).await {
                            Ok(true) => {
                                available.push(tool_id.to_string());
                            }
                            Ok(false) => {
                                debug!(tool = tool_id, "Tool condition not met, skipping");
                            }
                            Err(e) => {
                                warn!(tool = tool_id, error = %e, "Error evaluating tool condition");
                            }
                        }
                    } else {
                        available.push(tool_id.to_string());
                    }
                }

                Ok(available)
            }
            // State doesn't specify tools: fallback to agent-level
            None => {
                match &self.declared_tool_ids {
                    // tools: [...] — specific tools listed
                    Some(ids) if !ids.is_empty() => Ok(ids
                        .iter()
                        .filter(|id| self.tools.get(id).is_some())
                        .cloned()
                        .collect()),
                    // tools: [] — explicitly no tools
                    Some(_) => Ok(Vec::new()),
                    // tools: not specified — all registered tools available
                    None => Ok(self.tools.list_ids()),
                }
            }
        }
    }

    /// Returns `Some(tools)` if the current state explicitly declares tools
    /// (including `Some([])` for "no tools"), or `None` if the state doesn't
    /// specify tools (meaning: fall back to agent-level declared_tool_ids).
    fn get_current_tool_refs(&self) -> Option<Vec<ToolRef>> {
        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                let parent_def = sm.get_parent_definition();
                if let Some(effective) = state_def.get_effective_tools(parent_def.as_ref()) {
                    return Some(effective.into_iter().cloned().collect());
                }
            }
        }
        None
    }

    async fn build_evaluation_context(&self) -> Result<EvaluationContext> {
        let context = self.context_manager.get_all();
        let messages = self.memory.get_messages(Some(10)).await?;
        let tool_history = self.tool_call_history.read().clone();

        let (state_name, turn_count, previous_state) = if let Some(ref sm) = self.state_machine {
            (Some(sm.current()), sm.turn_count(), sm.previous())
        } else {
            (None, 0, None)
        };

        Ok(EvaluationContext::default()
            .with_context(context)
            .with_state(state_name, turn_count, previous_state)
            .with_called_tools(tool_history)
            .with_messages(messages))
    }

    fn record_tool_call(&self, tool_id: &str, result: Value) {
        self.tool_call_history.write().push(ToolCallRecord {
            tool_id: tool_id.to_string(),
            result,
            timestamp: chrono::Utc::now(),
        });
    }

    async fn get_effective_system_prompt(&self) -> Result<String> {
        let rendered_base = self.render_system_prompt()?;

        // Render persona prompt (if configured) and fire hooks for newly revealed secrets.
        let persona_prefix = if let Some(ref persona) = self.persona_manager {
            let context = self.context_manager.get_all();
            let render_result = persona.render_prompt(&context)?;

            // Fire on_secret_revealed for each newly revealed secret.
            for content in &render_result.newly_revealed {
                self.hooks.on_secret_revealed(content).await;
            }

            render_result.prompt
        } else {
            String::new()
        };

        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                let state_prompt = if let Some(ref prompt) = state_def.prompt {
                    let context = self.context_manager.get_all();
                    self.template_renderer.render_with_state(
                        prompt,
                        &context,
                        &sm.current(),
                        sm.previous().as_deref(),
                        sm.turn_count(),
                        state_def.max_turns,
                    )?
                } else {
                    String::new()
                };

                let combined = match state_def.prompt_mode {
                    PromptMode::Append => {
                        if state_prompt.is_empty() {
                            rendered_base
                        } else {
                            format!(
                                "{}\n\n[Current State: {}]\n{}",
                                rendered_base,
                                sm.current(),
                                state_prompt
                            )
                        }
                    }
                    PromptMode::Replace => {
                        if state_prompt.is_empty() {
                            rendered_base
                        } else {
                            state_prompt
                        }
                    }
                    PromptMode::Prepend => {
                        if state_prompt.is_empty() {
                            rendered_base
                        } else {
                            format!("{}\n\n{}", state_prompt, rendered_base)
                        }
                    }
                };

                // Persona always prepended regardless of prompt_mode.
                let with_persona = if persona_prefix.is_empty() {
                    combined
                } else {
                    format!("{}\n\n{}", persona_prefix, combined)
                };

                let available_tool_ids = self.get_available_tool_ids().await?;
                // Only add tools prompt if tools are available.
                // When tools: [] is set (explicitly empty), show NO tools to the LLM.
                if !available_tool_ids.is_empty() {
                    let tools_prompt = self.tools.generate_filtered_prompt_with_parallel(
                        &available_tool_ids,
                        self.parallel_tools.enabled,
                    );
                    if !tools_prompt.is_empty() {
                        return Ok(format!("{}\n\n{}", with_persona, tools_prompt));
                    }
                }
                return Ok(with_persona);
            }
        }

        // No state machine - prepend persona to base.
        let with_persona = if persona_prefix.is_empty() {
            rendered_base
        } else {
            format!("{}\n\n{}", persona_prefix, rendered_base)
        };

        let tools_prompt = match &self.declared_tool_ids {
            Some(ids) if !ids.is_empty() => self
                .tools
                .generate_filtered_prompt_with_parallel(ids, self.parallel_tools.enabled),
            Some(_) => {
                // tools: [] - explicitly no tools, empty prompt
                String::new()
            }
            None => {
                // tools: not specified - all registered tools
                self.tools
                    .generate_tools_prompt_with_parallel(self.parallel_tools.enabled)
            }
        };
        if !tools_prompt.is_empty() {
            Ok(format!("{}\n\n{}", with_persona, tools_prompt))
        } else {
            Ok(with_persona)
        }
    }

    fn get_state_llm(&self) -> Result<Arc<dyn LLMProvider>> {
        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                if let Some(ref llm_alias) = state_def.llm {
                    return self
                        .llm_registry
                        .get(llm_alias)
                        .map_err(|e| AgentError::Config(e.to_string()));
                }
            }
        }
        self.llm_registry
            .default()
            .map_err(|e| AgentError::Config(e.to_string()))
    }

    fn get_effective_reasoning_config(&self) -> ReasoningConfig {
        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                if let Some(ref state_reasoning) = state_def.reasoning {
                    return state_reasoning.clone();
                }
            }
        }
        self.reasoning_config.clone()
    }

    fn get_effective_reflection_config(&self) -> ReflectionConfig {
        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                if let Some(ref state_reflection) = state_def.reflection {
                    return state_reflection.clone();
                }
            }
        }
        self.reflection_config.clone()
    }

    fn get_skill_reasoning_config(&self, skill: &SkillDefinition) -> ReasoningConfig {
        skill
            .reasoning
            .clone()
            .unwrap_or_else(|| self.get_effective_reasoning_config())
    }

    fn get_skill_reflection_config(&self, skill: &SkillDefinition) -> ReflectionConfig {
        skill
            .reflection
            .clone()
            .unwrap_or_else(|| self.get_effective_reflection_config())
    }

    async fn build_disambiguation_context(&self) -> Result<DisambiguationContext> {
        let recent_messages: Vec<String> = self
            .memory
            .get_messages(Some(5))
            .await?
            .iter()
            .rev()
            .map(|m| format!("{:?}: {}", m.role, m.content))
            .collect();

        let current_state = self.current_state().map(|s| s.to_string());

        // Include the current state's prompt text so the detector understands
        // what kind of input is expected (e.g., "Ask for the order number").
        let state_prompt: Option<String> = self
            .state_machine
            .as_ref()
            .and_then(|sm| sm.current_definition())
            .and_then(|def| def.prompt.clone());

        let available_tools: Vec<String> = self
            .get_available_tool_ids()
            .await
            .unwrap_or_else(|_| self.tools.list_ids());

        let available_skills: Vec<String> = self.skills.iter().map(|s| s.id.clone()).collect();

        let user_context = self.context_manager.get_all();

        // Extract canonical intent labels from current state's transitions
        let available_intents: Vec<String> = if let Some(ref sm) = self.state_machine {
            sm.current_definition()
                .map(|def| {
                    def.transitions
                        .iter()
                        .filter_map(|t| t.intent.clone())
                        .collect()
                })
                .unwrap_or_default()
        } else {
            Vec::new()
        };

        Ok(DisambiguationContext::from_agent_state(
            recent_messages,
            current_state,
            state_prompt,
            available_tools,
            available_skills,
            available_intents,
            user_context,
        ))
    }

    fn get_available_skills(&self) -> Vec<&SkillDefinition> {
        if let Some(ref sm) = self.state_machine {
            if let Some(state_def) = sm.current_definition() {
                let parent_def = sm.get_parent_definition();
                let effective_skills = state_def.get_effective_skills(parent_def.as_ref());
                if !effective_skills.is_empty() {
                    return self
                        .skills
                        .iter()
                        .filter(|s| effective_skills.contains(&&s.id))
                        .collect();
                }
            }
        }
        self.skills.iter().collect()
    }

    async fn build_messages(&self) -> Result<Vec<ChatMessage>> {
        let system_prompt = self.get_effective_system_prompt().await?;
        let mut messages = vec![ChatMessage::system(&system_prompt)];

        // Use get_context() instead of get_messages() so that the conversation
        // summary produced by CompactingMemory is included in the prompt.
        // Token budget controls what's stored in memory (via handle_memory_overflow)
        // max_context_tokens + ContextOverflowAction handles LLM context limits
        let context = self.memory.get_context().await?;
        let history = if let Some(ref budget) = self.memory_token_budget {
            // Use per-component allocation so summary and recent messages
            // are each capped to their declared token budgets.
            context.to_llm_messages_with_allocation(&budget.allocation)
        } else {
            context.to_llm_messages()
        };
        messages.extend(history);

        let total_tokens = self.estimate_total_tokens(&messages);

        if total_tokens > self.max_context_tokens {
            debug!(
                total = total_tokens,
                limit = self.max_context_tokens,
                "Context overflow"
            );

            match &self.recovery_manager.config().llm.on_context_overflow {
                ContextOverflowAction::Error => {
                    return Err(AgentError::LLM(format!(
                        "Context overflow: {} tokens > {} limit",
                        total_tokens, self.max_context_tokens
                    )));
                }
                ContextOverflowAction::Truncate { keep_recent } => {
                    self.truncate_context(&mut messages, *keep_recent);
                }
                ContextOverflowAction::Summarize {
                    summarizer_llm,
                    max_summary_tokens,
                    custom_prompt,
                    keep_recent,
                    filter,
                } => {
                    self.summarize_context(
                        &mut messages,
                        summarizer_llm.as_deref(),
                        *max_summary_tokens,
                        custom_prompt.as_deref(),
                        *keep_recent,
                        filter.as_ref(),
                    )
                    .await?;
                }
            }
        }

        Ok(messages)
    }

    fn parse_tool_calls(&self, content: &str) -> Option<Vec<ToolCall>> {
        // Try direct JSON parse first
        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
            // Handle JSON array of tool calls (parallel tool calling)
            if let Some(arr) = parsed.as_array() {
                let calls: Vec<ToolCall> = arr
                    .iter()
                    .filter_map(|v| self.extract_tool_call_from_value(v))
                    .collect();
                if !calls.is_empty() {
                    return Some(calls);
                }
            }
            // Handle single JSON object
            if let Some(tool_call) = self.extract_tool_call_from_value(&parsed) {
                return Some(vec![tool_call]);
            }
        }

        // Try to extract JSON from content (handles extra text/braces from LLM)
        if let Some(json_str) = self.extract_json_from_content(content) {
            if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&json_str) {
                // Handle JSON array of tool calls (parallel tool calling)
                if let Some(arr) = parsed.as_array() {
                    let calls: Vec<ToolCall> = arr
                        .iter()
                        .filter_map(|v| self.extract_tool_call_from_value(v))
                        .collect();
                    if !calls.is_empty() {
                        return Some(calls);
                    }
                }
                // Handle single JSON object
                if let Some(tool_call) = self.extract_tool_call_from_value(&parsed) {
                    return Some(vec![tool_call]);
                }
            }
        }

        None
    }

    fn extract_tool_call_from_value(&self, parsed: &serde_json::Value) -> Option<ToolCall> {
        if let Some(tool_name) = parsed.get("tool").and_then(|v| v.as_str()) {
            let arguments = parsed
                .get("arguments")
                .cloned()
                .unwrap_or(serde_json::json!({}));
            return Some(ToolCall {
                id: uuid::Uuid::new_v4().to_string(),
                name: tool_name.to_string(),
                arguments,
            });
        }
        None
    }

    // Lite models could generate unmatched braces: this function handles such cases
    fn extract_json_from_content(&self, content: &str) -> Option<String> {
        // Try array first (for parallel tool calls), then single object
        if let Some(result) = self.extract_json_array_from_content(content) {
            return Some(result);
        }
        self.extract_json_object_from_content(content)
    }

    /// Extract a JSON array `[...]` containing tool calls from mixed content.
    fn extract_json_array_from_content(&self, content: &str) -> Option<String> {
        let start = content.find('[')?;
        let content_from_start = &content[start..];

        let mut depth = 0;
        let mut end = 0;
        for (i, ch) in content_from_start.char_indices() {
            match ch {
                '[' => depth += 1,
                ']' => {
                    depth -= 1;
                    if depth == 0 {
                        end = i + 1;
                        break;
                    }
                }
                _ => {}
            }
        }

        if end > 0 {
            let json_str = &content_from_start[..end];
            // Verify it looks like an array of tool calls
            if json_str.contains("\"tool\"") {
                return Some(json_str.to_string());
            }
        }

        None
    }

    /// Extract a JSON object `{...}` containing a tool call from mixed content.
    fn extract_json_object_from_content(&self, content: &str) -> Option<String> {
        let start = content.find('{')?;
        let content_from_start = &content[start..];

        // Count braces to find the matching closing brace
        let mut depth = 0;
        let mut end = 0;
        for (i, ch) in content_from_start.char_indices() {
            match ch {
                '{' => depth += 1,
                '}' => {
                    depth -= 1;
                    if depth == 0 {
                        end = i + 1;
                        break;
                    }
                }
                _ => {}
            }
        }

        if end > 0 {
            let json_str = &content_from_start[..end];
            // Verify it looks like a tool call
            if json_str.contains("\"tool\"") {
                return Some(json_str.to_string());
            }
        }

        None
    }

    async fn execute_tool(&self, tool_call: &ToolCall) -> Result<String> {
        let tool = self
            .tools
            .get(&tool_call.name)
            .ok_or_else(|| AgentError::Tool(format!("Tool not found: {}", tool_call.name)))?;

        let result = tool.execute(tool_call.arguments.clone()).await;

        if result.success {
            Ok(result.output)
        } else {
            Err(AgentError::Tool(result.output))
        }
    }

    #[instrument(skip(self, tool_call), fields(tool = %tool_call.name))]
    async fn execute_tool_smart(&self, tool_call: &ToolCall) -> Result<String> {
        let mut tool_call = tool_call.clone();
        info!(args = %tool_call.arguments, "Executing tool");

        self.hooks
            .on_tool_start(&tool_call.name, &tool_call.arguments)
            .await;
        let tool_start = Instant::now();

        let available_tool_ids = self.get_available_tool_ids().await?;
        // Resolve the tool call name to its canonical ID via the registry.
        // The LLM may use the display name (e.g., "HTTP Client") but
        // available_tool_ids contains IDs (e.g., "http"). Registry.get()
        // now matches by ID, display name, and alias.
        let resolved_id = self
            .tools
            .get(&tool_call.name)
            .map(|t| t.id().to_lowercase());
        let tool_name_lower = tool_call.name.to_lowercase();
        if !available_tool_ids.is_empty() {
            let is_available = available_tool_ids.iter().any(|id| {
                let id_lower = id.to_lowercase();
                id_lower == tool_name_lower || resolved_id.as_deref() == Some(&id_lower)
            });
            if !is_available {
                warn!(tool = %tool_call.name, "Tool not available in current context");
                return Err(AgentError::Tool(format!(
                    "Tool '{}' is not available. Available tools: {}",
                    tool_call.name,
                    available_tool_ids.join(", ")
                )));
            }
        }

        // Build language context for HITL localization.
        let hitl_lang_ctx = self.build_hitl_language_context();

        // Compute canonical tool ID before security/HITL checks so both use it.
        // Use the tool ID (e.g. "http") instead of the name (e.g. "HTTP Client") so it matches the HITL config keys in hitl.tools.<id>.
        let hitl_tool_id = resolved_id.as_deref().unwrap_or(&tool_name_lower);

        // Check security FIRST -- don't bother the human if security blocks
        if self.tool_security.config().enabled {
            let security_result = self
                .tool_security
                .check_tool_execution(&tool_call.name, &tool_call.arguments)
                .await?;

            match security_result {
                SecurityCheckResult::Allow => {}
                SecurityCheckResult::Block { reason } => {
                    warn!(reason = %reason, "Tool blocked by security");
                    return Err(AgentError::Tool(format!("Blocked: {}", reason)));
                }
                SecurityCheckResult::RequireConfirmation { message } => {
                    // Route through HITL if available, otherwise error
                    if let Some(ref hitl_engine) = self.hitl_engine {
                        let check_result = hitl_engine
                            .check_tool_with_localization(
                                hitl_tool_id,
                                &tool_call.arguments,
                                &hitl_lang_ctx,
                                self.approval_handler.as_ref(),
                                Some(&self.llm_registry),
                            )
                            .await?;
                        let result = self.request_hitl_approval(check_result).await?;
                        match result {
                            ApprovalResult::Approved | ApprovalResult::Modified { .. } => {}
                            ApprovalResult::Rejected { reason } => {
                                let reason_str = reason.as_deref().unwrap_or("rejected");
                                warn!(tool = %tool_call.name, reason = %reason_str, "Security confirmation rejected by approver");
                                return Err(AgentError::Tool(format!(
                                    "Confirmation rejected: {}",
                                    message
                                )));
                            }
                            _ => {}
                        }
                    } else {
                        warn!(message = %message, "Tool requires confirmation but no HITL handler");
                        return Err(AgentError::Tool(format!(
                            "Confirmation required: {}",
                            message
                        )));
                    }
                }
                SecurityCheckResult::Warn { message } => {
                    warn!(message = %message, "Tool security warning");
                }
            }
        }

        // Check HITL approval for tool (after security passes)
        if let Some(ref hitl_engine) = self.hitl_engine {
            let check_result = hitl_engine
                .check_tool_with_localization(
                    hitl_tool_id,
                    &tool_call.arguments,
                    &hitl_lang_ctx,
                    self.approval_handler.as_ref(),
                    Some(&self.llm_registry),
                )
                .await?;
            if check_result.is_required() {
                let result = self.request_hitl_approval(check_result).await?;
                match result {
                    ApprovalResult::Approved => {}
                    ApprovalResult::Modified { changes } => {
                        if let Some(obj) = tool_call.arguments.as_object_mut() {
                            for (k, v) in changes {
                                obj.insert(k, v);
                            }
                        }
                        info!(tool = %tool_call.name, "Tool arguments modified by approver");
                    }
                    ApprovalResult::Rejected { reason: _reason } => {
                        warn!(tool = %tool_call.name, "Tool execution rejected by HITL");
                        return Err(AgentError::HITLRejected(format!(
                            "Tool '{}' was rejected by human approver. Do not retry.",
                            tool_call.name
                        )));
                    }
                    _ => {}
                }
            }

            // Check conditions (e.g., amount > 1000)
            let condition_check = hitl_engine
                .check_conditions_with_localization(
                    &tool_call.arguments,
                    &hitl_lang_ctx,
                    self.approval_handler.as_ref(),
                    Some(&self.llm_registry),
                )
                .await?;
            if condition_check.is_required() {
                let result = self.request_hitl_approval(condition_check).await?;
                match result {
                    ApprovalResult::Approved => {}
                    ApprovalResult::Modified { changes } => {
                        if let Some(obj) = tool_call.arguments.as_object_mut() {
                            for (k, v) in changes {
                                obj.insert(k, v);
                            }
                        }
                        info!(tool = %tool_call.name, "Tool arguments modified by approver (condition)");
                    }
                    ApprovalResult::Rejected { reason: _reason } => {
                        warn!(tool = %tool_call.name, "Tool execution rejected by HITL condition");
                        return Err(AgentError::HITLRejected(format!(
                            "Tool '{}' was rejected due to policy condition. Do not retry.",
                            tool_call.name
                        )));
                    }
                    _ => {}
                }
            }
        }

        let tool_config = self.recovery_manager.get_tool_config(&tool_call.name);

        let result = if tool_config.max_retries > 0 {
            let retry_config = RetryConfig {
                max_retries: tool_config.max_retries,
                ..Default::default()
            };

            let tool_call_clone = tool_call.clone();
            self.recovery_manager
                .with_retry(
                    &format!("tool:{}", tool_call.name),
                    Some(&retry_config),
                    || {
                        let tc = tool_call_clone.clone();
                        async move { self.execute_tool(&tc).await.map_err(|e| e.classify()) }
                    },
                )
                .await
                .map_err(|e| AgentError::Tool(e.to_string()))
        } else {
            self.execute_tool(&tool_call).await
        };

        // Apply on_failure policy when the tool fails after all retries.
        let result = match result {
            Ok(output) => Ok(output),
            Err(e) => match &tool_config.on_failure {
                ToolFailureAction::Skip => {
                    warn!(
                        tool = %tool_call.name,
                        error = %e,
                        "Tool failed, skipping per on_failure: skip policy"
                    );
                    Ok(format!(
                        "{{\"skipped\": true, \"reason\": \"Tool '{}' was skipped after failure\"}}",
                        tool_call.name
                    ))
                }
                ToolFailureAction::Fallback { fallback_tool } => {
                    warn!(
                        tool = %tool_call.name,
                        fallback = %fallback_tool,
                        error = %e,
                        "Tool failed, trying fallback tool"
                    );
                    let fallback_call = ToolCall {
                        id: tool_call.id.clone(),
                        name: fallback_tool.clone(),
                        arguments: tool_call.arguments.clone(),
                    };
                    self.execute_tool(&fallback_call).await
                }
                ToolFailureAction::ReportError => Err(e),
            },
        };

        let tool_duration_ms = tool_start.elapsed().as_millis() as u64;

        match &result {
            Ok(output) => {
                info!(output_len = output.len(), "Tool execution successful");
                let result_value: Value =
                    serde_json::from_str(output).unwrap_or(Value::String(output.clone()));
                self.record_tool_call(&tool_call.name, result_value);

                let tool_result = ToolResult {
                    success: true,
                    output: output.clone(),
                    metadata: None,
                };
                self.hooks
                    .on_tool_complete(&tool_call.name, &tool_result, tool_duration_ms)
                    .await;
            }
            Err(e) => {
                error!(error = %e, "Tool execution failed");
                self.record_tool_call(&tool_call.name, serde_json::json!({"error": e.to_string()}));

                let tool_result = ToolResult {
                    success: false,
                    output: e.to_string(),
                    metadata: None,
                };
                self.hooks
                    .on_tool_complete(&tool_call.name, &tool_result, tool_duration_ms)
                    .await;
                self.hooks.on_error(e).await;
            }
        }

        result
    }

    /// Result of skill routing — three possible outcomes.
    /// NoMatch: no skill matched, continue to normal LLM chat.
    /// Response: skill executed successfully, here is the response string.
    /// NeedsClarification: skill matched but needs disambiguation first.
    async fn try_skill_route(&self, input: &str) -> Result<SkillRouteResult> {
        if let Some(ref router) = self.skill_router {
            let available_skills = self.get_available_skills();
            let skill_ids: Vec<&str> = available_skills.iter().map(|s| s.id.as_str()).collect();

            if let Some(skill_id) = router.select_skill_filtered(input, &skill_ids).await? {
                info!(skill_id = %skill_id, "Skill selected");

                let skill = router
                    .get_skill(&skill_id)
                    .ok_or_else(|| AgentError::Skill(format!("Skill not found: {}", skill_id)))?;

                // Skill-level disambiguation check: if the skill has a disambiguation
                // override and it's enabled, run a second disambiguation pass before
                // executing. This lets skills enforce stricter thresholds and
                // required_clarity fields (e.g. transfer_money requiring recipient + amount).
                if let Some(ref skill_disambig) = skill.disambiguation {
                    if skill_disambig.enabled.unwrap_or(false) {
                        if let Some(ref disambiguator) = self.disambiguation_manager {
                            let context = self.build_disambiguation_context().await?;
                            let state_override = self
                                .state_machine
                                .as_ref()
                                .and_then(|sm| sm.current_definition())
                                .and_then(|def| def.disambiguation.clone());

                            match disambiguator
                                .process_input_with_override(
                                    input,
                                    &context,
                                    state_override.as_ref(),
                                    Some(skill_disambig),
                                )
                                .await?
                            {
                                DisambiguationResult::Clear => {
                                    debug!(skill_id = %skill_id, "Skill disambiguation: clear");
                                    // Fall through to execute_skill below
                                }
                                DisambiguationResult::NeedsClarification {
                                    question,
                                    detection,
                                } => {
                                    info!(
                                        skill_id = %skill_id,
                                        ambiguity_type = ?detection.ambiguity_type,
                                        confidence = detection.confidence,
                                        "Skill requires clarification before execution"
                                    );
                                    // Tag the runtime with the skill ID so the next turn
                                    // routes directly to this skill after clarification resolves.
                                    *self.pending_skill_id.write() = Some(skill_id.clone());

                                    return Ok(SkillRouteResult::NeedsClarification(
                                        AgentResponse::new(&question.question).with_metadata(
                                            "disambiguation",
                                            serde_json::json!({
                                                "status": "awaiting_clarification",
                                                "skill_id": skill_id,
                                                "options": question.options,
                                                "clarifying": question.clarifying,
                                                "detection": {
                                                    "type": detection.ambiguity_type,
                                                    "confidence": detection.confidence,
                                                    "what_is_unclear": detection.what_is_unclear,
                                                }
                                            }),
                                        ),
                                    ));
                                }
                                DisambiguationResult::Clarified { enriched_input, .. } => {
                                    info!(
                                        skill_id = %skill_id,
                                        enriched = %enriched_input,
                                        "Skill disambiguation clarified, executing with enriched input"
                                    );
                                    return Ok(SkillRouteResult::Response(
                                        self.execute_skill(skill, &enriched_input).await?,
                                    ));
                                }
                                DisambiguationResult::ProceedWithBestGuess { enriched_input } => {
                                    info!(skill_id = %skill_id, "Skill disambiguation: proceeding with best guess");
                                    return Ok(SkillRouteResult::Response(
                                        self.execute_skill(skill, &enriched_input).await?,
                                    ));
                                }
                                DisambiguationResult::GiveUp { reason } => {
                                    warn!(skill_id = %skill_id, reason = %reason, "Skill disambiguation gave up");
                                    let apology = self
                                        .generate_localized_apology(
                                            "Generate a brief, polite apology saying you couldn't understand the request. Be concise.",
                                            &reason,
                                        )
                                        .await
                                        .unwrap_or_else(|_| {
                                            format!("I'm sorry, I couldn't understand your request: {}", reason)
                                        });
                                    return Ok(SkillRouteResult::NeedsClarification(
                                        AgentResponse::new(&apology),
                                    ));
                                }
                                DisambiguationResult::Escalate { reason } => {
                                    info!(skill_id = %skill_id, reason = %reason, "Skill disambiguation escalating");
                                    let apology = self
                                        .generate_localized_apology(
                                            "Explain briefly that you're transferring the user to a human agent for help.",
                                            &reason,
                                        )
                                        .await
                                        .unwrap_or_else(|_| {
                                            format!("I need human assistance to help with your request: {}", reason)
                                        });
                                    return Ok(SkillRouteResult::NeedsClarification(
                                        AgentResponse::new(&apology),
                                    ));
                                }
                                DisambiguationResult::Abandoned { .. } => {
                                    // User abandoned during skill-level disambiguation.
                                    // Return NoMatch so run_loop falls through to normal processing.
                                    debug!(skill_id = %skill_id, "Skill disambiguation abandoned");
                                    return Ok(SkillRouteResult::NoMatch);
                                }
                            }
                        }
                    }
                }

                return Ok(SkillRouteResult::Response(
                    self.execute_skill(skill, input).await?,
                ));
            }
        }
        Ok(SkillRouteResult::NoMatch)
    }

    /// Execute a skill with reasoning and reflection, returning the response string.
    async fn execute_skill(&self, skill: &SkillDefinition, input: &str) -> Result<String> {
        if let Some(ref executor) = self.skill_executor {
            let skill_reasoning = self.get_skill_reasoning_config(skill);
            let skill_reflection = self.get_skill_reflection_config(skill);

            debug!(
                skill_id = %skill.id,
                reasoning_mode = ?skill_reasoning.mode,
                reflection_enabled = ?skill_reflection.enabled,
                "Skill reasoning/reflection config"
            );

            let response = executor
                .execute(skill, input, serde_json::json!({}))
                .await?;

            if skill_reflection.requires_evaluation() && skill_reflection.is_enabled() {
                let should_reflect = self
                    .should_reflect_with_config(input, &response, &skill_reflection)
                    .await?;
                if should_reflect {
                    let evaluated = self
                        .evaluate_and_retry_with_config(input, response, &skill_reflection)
                        .await?;
                    return Ok(evaluated);
                }
            }

            return Ok(response);
        }
        Err(AgentError::Skill(
            "No skill executor configured".to_string(),
        ))
    }

    /// Execute a skill by ID, bypassing the skill router.
    /// Used after skill-triggered disambiguation resolves to route directly to the matched skill.
    async fn execute_skill_by_id(&self, skill_id: &str, input: &str) -> Result<String> {
        let skill = self
            .skill_router
            .as_ref()
            .and_then(|r| r.get_skill(skill_id).cloned())
            .ok_or_else(|| AgentError::Skill(format!("Skill not found: {}", skill_id)))?;
        self.execute_skill(&skill, input).await
    }

    async fn should_reflect_with_config(
        &self,
        input: &str,
        response: &str,
        config: &ReflectionConfig,
    ) -> Result<bool> {
        if !config.requires_evaluation() {
            return Ok(false);
        }

        if config.is_enabled() {
            return Ok(true);
        }

        let evaluator_llm = config
            .evaluator_llm
            .as_ref()
            .and_then(|alias| self.llm_registry.get(alias).ok())
            .or_else(|| self.llm_registry.router().ok())
            .or_else(|| self.llm_registry.default().ok());

        let Some(llm) = evaluator_llm else {
            return Ok(false);
        };

        let prompt = format!(
            r#"Should this response be evaluated for quality? Consider if it's a complex or important response.

User query: "{}"
Response: "{}"

Answer YES or NO only."#,
            input,
            &response[..response.len().min(500)]
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let result = llm.complete(&messages, None).await;

        match result {
            Ok(resp) => Ok(resp.content.trim().to_uppercase().contains("YES")),
            Err(_) => Ok(false),
        }
    }

    async fn evaluate_and_retry_with_config(
        &self,
        input: &str,
        mut response: String,
        config: &ReflectionConfig,
    ) -> Result<String> {
        let llm = self.get_state_llm()?;
        let mut attempts = 0u32;
        let max_retries = config.max_retries;

        loop {
            let evaluation = self
                .evaluate_response_with_config(input, &response, config)
                .await?;

            if evaluation.passed || attempts >= max_retries {
                info!(
                    passed = evaluation.passed,
                    confidence = evaluation.confidence,
                    attempts = attempts + 1,
                    "Skill reflection evaluation complete"
                );
                return Ok(response);
            }

            debug!(
                attempt = attempts + 1,
                failed_criteria = evaluation.failed_criteria().count(),
                "Skill response did not meet criteria, retrying"
            );

            let feedback: Vec<String> = evaluation
                .failed_criteria()
                .map(|c| format!("- {}", c.criterion))
                .collect();

            let retry_prompt = format!(
                "Your previous response did not meet these criteria:\n{}\n\nPlease provide an improved response to: {}",
                feedback.join("\n"),
                input
            );

            let messages = vec![ChatMessage::user(&retry_prompt)];
            let retry_response = llm
                .complete(&messages, None)
                .await
                .map_err(|e| AgentError::LLM(e.to_string()))?;

            response = retry_response.content.trim().to_string();
            attempts += 1;
        }
    }

    async fn evaluate_response_with_config(
        &self,
        input: &str,
        response: &str,
        config: &ReflectionConfig,
    ) -> Result<EvaluationResult> {
        let evaluator_llm = config
            .evaluator_llm
            .as_ref()
            .and_then(|alias| self.llm_registry.get(alias).ok())
            .or_else(|| self.llm_registry.router().ok())
            .or_else(|| self.llm_registry.default().ok())
            .ok_or_else(|| AgentError::Config("No LLM available for evaluation".into()))?;

        let criteria = &config.criteria;
        let criteria_list = criteria
            .iter()
            .enumerate()
            .map(|(i, c)| format!("{}. {}", i + 1, c))
            .collect::<Vec<_>>()
            .join("\n");

        let prompt = format!(
            r#"Evaluate this response against the criteria.

User query: "{}"

Response to evaluate: "{}"

Criteria:
{}

For each criterion, respond with:
- criterion number
- PASS or FAIL
- brief reason

Then provide overall confidence (0.0 to 1.0) and whether it passes overall.

Format:
1. PASS/FAIL - reason
2. PASS/FAIL - reason
...
CONFIDENCE: 0.X
OVERALL: PASS/FAIL"#,
            input, response, criteria_list
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let eval_response = evaluator_llm
            .complete(&messages, None)
            .await
            .map_err(|e| AgentError::LLM(format!("Evaluation failed: {}", e)))?;

        let content = eval_response.content.to_uppercase();
        let llm_pass = content.contains("OVERALL: PASS");

        let confidence = content
            .lines()
            .find(|l| l.contains("CONFIDENCE:"))
            .and_then(|l| {
                l.split(':')
                    .nth(1)
                    .and_then(|v| v.trim().parse::<f32>().ok())
            })
            .unwrap_or(if llm_pass { 0.8 } else { 0.4 });

        // Gate pass against confidence threshold.
        // LLM may say PASS but with low confidence - the threshold catches this.
        let overall_pass = llm_pass && confidence >= config.pass_threshold;

        let mut criteria_results = Vec::new();
        for (i, criterion) in criteria.iter().enumerate() {
            let line_marker = format!("{}.", i + 1);
            let passed = eval_response
                .content
                .lines()
                .find(|l| l.contains(&line_marker))
                .map(|l| l.to_uppercase().contains("PASS"))
                .unwrap_or(overall_pass);

            if passed {
                criteria_results.push(CriterionResult::pass(criterion));
            } else {
                criteria_results.push(CriterionResult::fail(criterion, "Did not meet criterion"));
            }
        }

        Ok(EvaluationResult::new(overall_pass, confidence).with_criteria(criteria_results))
    }

    /// Process input through the pipeline (state-level override or agent-level).
    async fn process_input(&self, input: &str) -> Result<ProcessData> {
        if let Some(processor) = self.get_state_process_processor() {
            return processor.process_input(input).await;
        }
        if let Some(ref processor) = self.process_processor {
            processor.process_input(input).await
        } else {
            Ok(ProcessData::new(input))
        }
    }

    /// Process output through the pipeline (state-level override or agent-level).
    async fn process_output(
        &self,
        output: &str,
        input_context: &std::collections::HashMap<String, serde_json::Value>,
    ) -> Result<ProcessData> {
        if let Some(processor) = self.get_state_process_processor() {
            return processor.process_output(output, input_context).await;
        }
        if let Some(ref processor) = self.process_processor {
            processor.process_output(output, input_context).await
        } else {
            Ok(ProcessData::new(output))
        }
    }

    /// Build a ProcessProcessor from the current state's process config, if any.
    fn get_state_process_processor(&self) -> Option<ProcessProcessor> {
        let sm = self.state_machine.as_ref()?;
        let def = sm.current_definition()?;
        let config = def.process.as_ref()?;
        let mut processor = ProcessProcessor::new(config.clone());
        if let Some(ref registry) = Some(self.llm_registry.clone()) {
            processor = processor.with_llm_registry(registry.clone());
        }
        Some(processor)
    }

    async fn check_turn_timeout(&self) -> Result<()> {
        if let Some(ref sm) = self.state_machine {
            if let Some(timeout_state) = sm.check_timeout() {
                let from_state = sm.current();
                self.execute_state_exit_actions(&from_state).await;
                sm.transition_to(&timeout_state, "max_turns exceeded")?;
                self.execute_state_enter_actions(&timeout_state).await;
                info!(to = %timeout_state, "Timeout transition");
            }
        }
        Ok(())
    }

    fn increment_turn(&self) {
        if let Some(ref sm) = self.state_machine {
            sm.increment_turn();
        }
    }

    async fn evaluate_transitions(&self, user_message: &str, response: &str) -> Result<bool> {
        let (transitions, evaluator, current_state) =
            match (&self.state_machine, &self.transition_evaluator) {
                (Some(sm), Some(eval)) => {
                    let auto_transitions: Vec<_> = sm
                        .auto_transitions()
                        .into_iter()
                        .filter(|t| {
                            // S7: skip transitions on cooldown
                            match t.cooldown_turns {
                                Some(cd) if cd > 0 => {
                                    let resolved =
                                        sm.config().resolve_full_path(&sm.current(), &t.to);
                                    !sm.is_on_cooldown(&resolved, cd)
                                }
                                _ => true,
                            }
                        })
                        .collect();
                    if auto_transitions.is_empty() {
                        return Ok(false);
                    }
                    (auto_transitions, eval, sm.current())
                }
                _ => return Ok(false),
            };

        let context_map = self.context_manager.get_all();
        let context = TransitionContext::new(user_message, response, &current_state)
            .with_context(context_map);

        if let Some(index) = evaluator.select_transition(&transitions, &context).await? {
            let target = transitions[index].to.clone();
            let reason = if transitions[index].when.is_empty() {
                "guard condition met".to_string()
            } else {
                transitions[index].when.clone()
            };

            if let Some(ref sm) = self.state_machine {
                // Check HITL approval for state transition
                let approved = self
                    .check_state_hitl(Some(&context.current_state), &target)
                    .await?;
                if !approved {
                    info!(to = %target, "State transition rejected by HITL");
                    return Ok(false);
                }

                // Execute on_exit actions for the current state
                self.execute_state_exit_actions(&context.current_state)
                    .await;

                sm.transition_to(&target, &reason)?;
                sm.reset_no_transition();

                // Execute on_enter actions for the new state
                self.execute_state_enter_actions(&target).await;

                self.hooks
                    .on_state_transition(Some(&context.current_state), &target, &reason)
                    .await;
                info!(from = %context.current_state, to = %target, "State transition");
            }
            return Ok(true);
        }

        if let Some(ref sm) = self.state_machine {
            sm.increment_no_transition();
            if let Some(fallback) = sm.check_fallback() {
                let from_state = current_state.clone();

                // Check HITL approval for fallback transition
                let approved = self.check_state_hitl(Some(&from_state), &fallback).await?;
                if !approved {
                    info!(to = %fallback, "Fallback transition rejected by HITL");
                    return Ok(false);
                }

                // Execute on_exit actions for the current state
                self.execute_state_exit_actions(&from_state).await;

                sm.transition_to(&fallback, "fallback after no transitions")?;

                // Execute on_enter actions for the fallback state
                self.execute_state_enter_actions(&fallback).await;

                self.hooks
                    .on_state_transition(
                        Some(&from_state),
                        &fallback,
                        "fallback after no transitions",
                    )
                    .await;
                info!(to = %fallback, "Fallback transition");
                return Ok(true);
            }
        }

        Ok(false)
    }

    /// Execute on_exit actions for a state being left.
    async fn execute_state_exit_actions(&self, state_path: &str) {
        if let Some(ref sm) = self.state_machine {
            if let Some(def) = sm.get_definition(state_path) {
                if !def.on_exit.is_empty() {
                    debug!(state = %state_path, count = def.on_exit.len(), "Executing on_exit actions");
                    self.execute_state_actions(&def.on_exit).await;
                }
            }
        }
    }

    /// Execute on_enter (or on_reenter) actions for a state being entered.
    async fn execute_state_enter_actions(&self, state_path: &str) {
        if let Some(ref sm) = self.state_machine {
            if let Some(def) = sm.get_definition(state_path) {
                // Check if this state was previously visited
                let is_reentry = sm.history().iter().any(|e| e.to == state_path);

                if is_reentry && !def.on_reenter.is_empty() {
                    debug!(state = %state_path, count = def.on_reenter.len(), "Executing on_reenter actions");
                    self.execute_state_actions(&def.on_reenter).await;
                } else if !def.on_enter.is_empty() {
                    debug!(state = %state_path, count = def.on_enter.len(), "Executing on_enter actions");
                    self.execute_state_actions(&def.on_enter).await;
                }
            }
        }
    }

    /// Execute a list of state actions (tool calls, skill invocations, context updates, LLM prompts).
    async fn execute_state_actions(&self, actions: &[StateAction]) {
        for action in actions {
            match action {
                StateAction::Tool { tool, args } => {
                    let raw_args = args.clone().unwrap_or(Value::Object(Default::default()));
                    // Render template variables in tool args (e.g., {{ context.order_id }})
                    let args_value = self.render_action_args(&raw_args);
                    if let Some(t) = self.tools.get(tool) {
                        self.hooks.on_tool_start(tool, &args_value).await;
                        let start = Instant::now();
                        let result = t.execute(args_value).await;
                        let duration_ms = start.elapsed().as_millis() as u64;
                        self.hooks
                            .on_tool_complete(tool, &result, duration_ms)
                            .await;
                        if result.success {
                            debug!(tool = %tool, "State action: tool executed");
                            // Store tool result in context so YAML prompts can reference it
                            let context_key = format!("last_tool_result");
                            let _ = self
                                .context_manager
                                .set(&context_key, serde_json::Value::String(result.output));
                        } else {
                            warn!(tool = %tool, error = %result.output, "State action: tool failed");
                        }
                    } else {
                        warn!(tool = %tool, "State action: tool not found");
                    }
                }
                StateAction::Skill { skill } => {
                    if let Some(ref executor) = self.skill_executor {
                        if let Some(def) = self.skills.iter().find(|s| s.id == *skill) {
                            match executor.execute(def, "", serde_json::json!({})).await {
                                Ok(_) => debug!(skill = %skill, "State action: skill executed"),
                                Err(e) => {
                                    warn!(skill = %skill, error = %e, "State action: skill failed")
                                }
                            }
                        } else {
                            warn!(skill = %skill, "State action: skill not found");
                        }
                    }
                }
                StateAction::SetContext { set_context } => {
                    for (key, value) in set_context {
                        if let Err(e) = self.context_manager.set(key, value.clone()) {
                            warn!(key = %key, error = %e, "State action: set_context failed");
                        } else {
                            debug!(key = %key, "State action: context set");
                        }
                    }
                }
                StateAction::Prompt {
                    prompt,
                    llm,
                    store_as,
                } => {
                    let llm_result = if let Some(alias) = llm {
                        self.llm_registry.get(alias)
                    } else {
                        self.llm_registry.default()
                    };
                    match llm_result {
                        Ok(llm_provider) => {
                            // Render template variables and include conversation context
                            let context = self.context_manager.get_all();
                            let rendered_prompt = self
                                .template_renderer
                                .render(prompt, &context)
                                .unwrap_or_else(|_| prompt.clone());
                            let recent =
                                self.memory.get_messages(Some(5)).await.unwrap_or_default();
                            let mut messages: Vec<ChatMessage> = recent;
                            messages.push(ChatMessage::user(&rendered_prompt));
                            match llm_provider.complete(&messages, None).await {
                                Ok(response) => {
                                    if let Some(key) = store_as {
                                        let _ = self
                                            .context_manager
                                            .set(key, Value::String(response.content));
                                        debug!(key = %key, "State action: prompt result stored");
                                    }
                                }
                                Err(e) => {
                                    warn!(error = %e, "State action: prompt LLM call failed");
                                }
                            }
                        }
                        Err(e) => {
                            warn!(error = %e, "State action: LLM not found for prompt");
                        }
                    }
                }
            }
        }
    }

    /// Run context extractors for the current state on the user's input.
    async fn run_context_extractors(&self, user_message: &str) {
        let extractors = match &self.state_machine {
            Some(sm) => match sm.current_definition() {
                Some(def) if !def.extract.is_empty() => def.extract.clone(),
                _ => return,
            },
            None => return,
        };

        for extractor in &extractors {
            let prompt = if let Some(ref custom) = extractor.llm_extract {
                format!(
                    "User message:\n\"{}\"\n\nInstruction:\n{}",
                    user_message, custom
                )
            } else if let Some(ref desc) = extractor.description {
                format!(
                    "From the following message, extract: {}\n\n\
                     Message: \"{}\"\n\n\
                     If the information is present, return ONLY the extracted value.\n\
                     If NOT present, return exactly: __NONE__",
                    desc, user_message
                )
            } else {
                continue;
            };

            let llm = match self
                .llm_registry
                .get(&extractor.llm)
                .or_else(|_| self.llm_registry.get("router"))
                .or_else(|_| self.llm_registry.get("default"))
            {
                Ok(llm) => llm,
                Err(e) => {
                    warn!(key = %extractor.key, error = %e, "Extractor LLM not found");
                    continue;
                }
            };

            let messages = vec![ChatMessage::user(&prompt)];
            match llm.complete(&messages, None).await {
                Ok(response) => {
                    let value = response.content.trim().to_string();
                    if value != "__NONE__" && !value.is_empty() {
                        let _ = self
                            .context_manager
                            .update(&extractor.key, serde_json::Value::String(value.clone()));
                        debug!(key = %extractor.key, value = %value, "Context extracted");
                    } else if extractor.required {
                        warn!(key = %extractor.key, "Required extraction returned no value");
                    }
                }
                Err(e) => {
                    warn!(key = %extractor.key, error = %e, "Context extraction LLM call failed");
                }
            }
        }
    }

    async fn check_memory_compression(&self) -> Result<()> {
        if self.memory.needs_compression() {
            let result = self.memory.compress(None).await?;
            if let CompressResult::Compressed {
                messages_summarized,
                new_summary_length,
                tokens_saved,
            } = result
            {
                let event = MemoryCompressEvent::new(
                    messages_summarized,
                    tokens_saved,
                    new_summary_length as u32,
                );
                self.hooks.on_memory_compress(&event).await;
                debug!(
                    messages = messages_summarized,
                    tokens_saved = tokens_saved,
                    "Memory compressed"
                );
            }
        }

        // Handle overflow AFTER compression, then check warning threshold
        self.handle_memory_overflow().await?;
        self.check_memory_budget().await;

        Ok(())
    }

    async fn check_memory_budget(&self) {
        let Some(ref budget) = self.memory_token_budget else {
            return;
        };

        let context = match self.memory.get_context().await {
            Ok(ctx) => ctx,
            Err(_) => return,
        };

        // Overall budget warning
        let used_tokens = context.estimated_tokens();
        if budget.is_over_warn_threshold(used_tokens) {
            let event = MemoryBudgetEvent::new("memory", used_tokens, budget.total);
            self.hooks.on_memory_budget_warning(&event).await;
            debug!(
                used = used_tokens,
                total = budget.total,
                percent = event.usage_percent,
                "Memory budget warning"
            );
        }

        // Per-component warning: summary
        if let Some(ref summary) = context.summary {
            let summary_tokens = ai_agents_memory::estimate_tokens(summary);
            let summary_budget = budget.allocation.summary;
            if summary_budget > 0 {
                let warn_threshold =
                    (summary_budget as f64 * budget.warn_at_percent as f64 / 100.0) as u32;
                if summary_tokens >= warn_threshold {
                    let event = MemoryBudgetEvent::new("summary", summary_tokens, summary_budget);
                    self.hooks.on_memory_budget_warning(&event).await;
                }
            }
        }

        // Per-component warning: recent_messages
        let recent_tokens: u32 = context
            .messages
            .iter()
            .map(ai_agents_memory::estimate_message_tokens)
            .sum();
        let recent_budget = budget.allocation.recent_messages;
        if recent_budget > 0 {
            let warn_threshold =
                (recent_budget as f64 * budget.warn_at_percent as f64 / 100.0) as u32;
            if recent_tokens >= warn_threshold {
                let event = MemoryBudgetEvent::new("recent_messages", recent_tokens, recent_budget);
                self.hooks.on_memory_budget_warning(&event).await;
            }
        }
    }

    async fn handle_memory_overflow(&self) -> Result<()> {
        let Some(ref budget) = self.memory_token_budget else {
            return Ok(());
        };

        let context = self.memory.get_context().await?;
        let used_tokens = context.estimated_tokens();

        if used_tokens <= budget.total {
            return Ok(());
        }

        match budget.overflow_strategy {
            OverflowStrategy::TruncateOldest => {
                let tokens_to_free = used_tokens - budget.total;
                let messages_to_evict = self.calculate_eviction_count(tokens_to_free);
                if messages_to_evict > 0 {
                    self.evict_messages(messages_to_evict, EvictionReason::TokenBudgetExceeded)
                        .await?;
                }
            }
            OverflowStrategy::SummarizeMore => {
                self.memory.compress(None).await?;
            }
            OverflowStrategy::Error => {
                return Err(AgentError::MemoryBudgetExceeded {
                    used: used_tokens,
                    budget: budget.total,
                });
            }
        }
        Ok(())
    }

    fn calculate_eviction_count(&self, tokens_to_free: u32) -> usize {
        // Estimate ~50 tokens per message on average
        ((tokens_to_free as f64 / 50.0).ceil() as usize).max(1)
    }

    async fn evict_messages(&self, count: usize, reason: EvictionReason) -> Result<()> {
        let evicted = self.memory.evict_oldest(count).await?;
        if !evicted.is_empty() {
            let event = MemoryEvictEvent {
                reason,
                messages_evicted: evicted.len(),
                importance_scores: vec![],
            };
            self.hooks.on_memory_evict(&event).await;
            debug!(count = evicted.len(), "Messages evicted from memory");
        }
        Ok(())
    }

    #[instrument(skip(self, input), fields(agent = %self.info.name))]
    async fn determine_reasoning_mode(&self, input: &str) -> Result<ReasoningMode> {
        let effective_config = self.get_effective_reasoning_config();

        if !matches!(effective_config.mode, ReasoningMode::Auto) {
            return Ok(effective_config.mode.clone());
        }

        let judge_llm = effective_config
            .judge_llm
            .as_ref()
            .and_then(|alias| self.llm_registry.get(alias).ok())
            .or_else(|| self.llm_registry.router().ok())
            .or_else(|| self.llm_registry.default().ok());

        let Some(llm) = judge_llm else {
            return Ok(ReasoningMode::None);
        };

        let prompt = format!(
            r#"Analyze this user request and determine the appropriate reasoning mode.

User request: "{}"

Choose ONE of these modes:
- none: Simple queries, greetings, direct answers (fastest)
- cot: Complex analysis, multi-step reasoning, math problems
- react: Tasks requiring multiple tool calls with observation
- plan_and_execute: Complex multi-step tasks requiring coordination

Respond with ONLY the mode name (none, cot, react, or plan_and_execute)."#,
            input
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let response = llm.complete(&messages, None).await;

        match response {
            Ok(resp) => {
                let mode_str = resp.content.trim().to_lowercase();
                Ok(match mode_str.as_str() {
                    "cot" => ReasoningMode::CoT,
                    "react" => ReasoningMode::React,
                    "plan_and_execute" => ReasoningMode::PlanAndExecute,
                    _ => ReasoningMode::None,
                })
            }
            Err(_) => Ok(ReasoningMode::None),
        }
    }

    async fn should_reflect(&self, input: &str, response: &str) -> Result<bool> {
        let effective_config = self.get_effective_reflection_config();

        if !effective_config.requires_evaluation() {
            return Ok(false);
        }

        if effective_config.is_enabled() {
            return Ok(true);
        }

        let evaluator_llm = effective_config
            .evaluator_llm
            .as_ref()
            .and_then(|alias| self.llm_registry.get(alias).ok())
            .or_else(|| self.llm_registry.router().ok())
            .or_else(|| self.llm_registry.default().ok());

        let Some(llm) = evaluator_llm else {
            return Ok(false);
        };

        let prompt = format!(
            r#"Should this response be evaluated for quality? Consider if it's a complex or important response.

User query: "{}"
Response: "{}"

Answer YES or NO only."#,
            input,
            &response[..response.len().min(500)]
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let result = llm.complete(&messages, None).await;

        match result {
            Ok(resp) => Ok(resp.content.trim().to_uppercase().contains("YES")),
            Err(_) => Ok(false),
        }
    }

    fn build_cot_system_prompt(&self, base_prompt: &str) -> String {
        format!(
            "{}\n\n<instruction>\nThink through this step by step before answering:\n1. Understand what is being asked\n2. Break down the problem\n3. Work through each part\n4. Provide your final answer\n\nShow your thinking process, then give your final answer.\n</instruction>",
            base_prompt
        )
    }

    fn build_react_system_prompt(&self, base_prompt: &str) -> String {
        format!(
            "{}\n\n<instruction>\nUse the Reason-Act-Observe pattern:\n1. Thought: Think about what to do\n2. Action: Use a tool if needed\n3. Observation: Analyze the result\n4. Repeat until you have the answer\n\nFormat your response showing Thought/Action/Observation steps.\n</instruction>",
            base_prompt
        )
    }

    async fn generate_plan(&self, input: &str) -> Result<Plan> {
        let effective = self.get_effective_reasoning_config();
        let planning_config = effective.get_planning();

        let planner_llm = planning_config
            .and_then(|c| c.planner_llm.as_ref())
            .and_then(|alias| self.llm_registry.get(alias).ok())
            .or_else(|| self.llm_registry.router().ok())
            .or_else(|| self.llm_registry.default().ok())
            .ok_or_else(|| AgentError::Config("No LLM available for planning".into()))?;

        let mut available_tool_ids: Vec<String> = self
            .get_available_tool_ids()
            .await
            .unwrap_or_else(|_| self.tools.list_ids());
        let mut available_skills: Vec<String> = self.skills.iter().map(|s| s.id.clone()).collect();

        // Apply planning-level tool and skill filters.
        if let Some(config) = planning_config {
            if !config.available.tools.is_all() {
                available_tool_ids.retain(|t| config.available.tools.allows(t));
            }
            if !config.available.skills.is_all() {
                available_skills.retain(|s| config.available.skills.allows(s));
            }
        }

        // Build tool descriptions with argument schemas so the planner
        // knows how to construct valid args for each step.
        let tool_descriptions: Vec<String> = available_tool_ids
            .iter()
            .filter_map(|id| {
                self.tools.get(id).map(|tool| {
                    let schema = tool.input_schema();
                    let args_desc = schema
                        .get("properties")
                        .and_then(|p| serde_json::to_string(p).ok())
                        .unwrap_or_else(|| "{}".to_string());
                    format!(
                        "- {} ({}): {}\n  Arguments: {}",
                        id,
                        tool.name(),
                        tool.description(),
                        args_desc
                    )
                })
            })
            .collect();

        let tools_section = if tool_descriptions.is_empty() {
            "Available tools: none".to_string()
        } else {
            format!("Available tools:\n{}", tool_descriptions.join("\n"))
        };

        let skills_section = if available_skills.is_empty() {
            "Available skills: none".to_string()
        } else {
            format!("Available skills: {}", available_skills.join(", "))
        };

        let prompt = format!(
            r#"Create a step-by-step plan to accomplish this goal.

Goal: "{}"

{}

{}

Create a plan with clear steps. For each step, specify:
- description: What this step accomplishes
- action_type: "tool", "skill", "think", or "respond"
- action_target: The tool/skill id (if applicable)
- args: The arguments object matching the tool's schema (if action_type is "tool")
- dependencies: List of step IDs this depends on (empty if none)

Respond in JSON format:
{{
  "steps": [
    {{"id": "step1", "description": "...", "action_type": "tool", "action_target": "tool_id", "args": {{"required_field": "value"}}, "dependencies": []}},
    {{"id": "step2", "description": "...", "action_type": "think", "action_target": "...", "dependencies": ["step1"]}}
  ]
}}"#,
            input, tools_section, skills_section,
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let response = planner_llm
            .complete(&messages, None)
            .await
            .map_err(|e| AgentError::LLM(format!("Planning failed: {}", e)))?;

        let mut plan = Plan::new(input);

        if let Some(json_start) = response.content.find('{') {
            if let Some(json_end) = response.content.rfind('}') {
                let json_str = &response.content[json_start..=json_end];
                if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(json_str) {
                    if let Some(steps) = parsed.get("steps").and_then(|s| s.as_array()) {
                        for step_value in steps {
                            let id = step_value
                                .get("id")
                                .and_then(|v| v.as_str())
                                .unwrap_or("step");
                            let desc = step_value
                                .get("description")
                                .and_then(|v| v.as_str())
                                .unwrap_or("");
                            let action_type = step_value
                                .get("action_type")
                                .and_then(|v| v.as_str())
                                .unwrap_or("think");
                            let action_target = step_value
                                .get("action_target")
                                .and_then(|v| v.as_str())
                                .unwrap_or("");
                            let args = step_value
                                .get("args")
                                .cloned()
                                .unwrap_or(serde_json::json!({}));
                            let deps: Vec<String> = step_value
                                .get("dependencies")
                                .and_then(|v| v.as_array())
                                .map(|arr| {
                                    arr.iter()
                                        .filter_map(|v| v.as_str().map(String::from))
                                        .collect()
                                })
                                .unwrap_or_default();

                            let action = match action_type {
                                "tool" => PlanAction::tool(action_target, args),
                                "skill" => PlanAction::skill(action_target),
                                "respond" => PlanAction::respond(action_target),
                                _ => PlanAction::think(desc),
                            };

                            let step = PlanStep::new(desc, action)
                                .with_id(id)
                                .with_dependencies(deps);
                            plan.add_step(step);
                        }
                    }
                }
            }
        }

        if plan.steps.is_empty() {
            plan.add_step(PlanStep::new(
                "Process the request",
                PlanAction::think(input),
            ));
            plan.add_step(PlanStep::new(
                "Provide response",
                PlanAction::respond("Answer based on analysis"),
            ));
        }

        Ok(plan)
    }

    async fn execute_plan(&self, plan: &mut Plan) -> Result<String> {
        let llm = self.get_state_llm()?;
        let mut results: HashMap<String, serde_json::Value> = HashMap::new();
        let effective = self.get_effective_reasoning_config();
        let max_steps = effective.get_planning().map(|c| c.max_steps).unwrap_or(10);

        plan.status = PlanStatus::InProgress;

        for step_idx in 0..plan.steps.len().min(max_steps as usize) {
            let step = &plan.steps[step_idx];

            let deps_satisfied = step.dependencies.iter().all(|dep| {
                plan.steps
                    .iter()
                    .find(|s| &s.id == dep)
                    .map(|s| s.status.is_completed())
                    .unwrap_or(false)
            });

            if !deps_satisfied {
                continue;
            }

            plan.steps[step_idx].mark_running();

            let result = match &plan.steps[step_idx].action {
                PlanAction::Tool { tool, args } => {
                    // When a tool step has dependency results, ask the LLM to
                    // produce the correct arguments given the context and tool schema.
                    // This avoids brittle {{stepN}} template substitution and lets
                    // the LLM handle type adaptation (e.g. picking the iso field
                    // from a datetime result for a downstream format call).
                    let has_dep_results = plan.steps[step_idx]
                        .dependencies
                        .iter()
                        .any(|dep| results.contains_key(dep));

                    let final_args = if has_dep_results {
                        let dep_context: String = plan.steps[step_idx]
                            .dependencies
                            .iter()
                            .filter_map(|dep| results.get(dep).map(|r| format!("{}: {}", dep, r)))
                            .collect::<Vec<_>>()
                            .join("\n");

                        let tool_schema = self
                            .tools
                            .get(tool)
                            .map(|t| {
                                let schema = t.input_schema();
                                let props = schema
                                    .get("properties")
                                    .and_then(|p| serde_json::to_string(p).ok())
                                    .unwrap_or_else(|| "{}".to_string());
                                format!(
                                    "{}: {}\nArguments schema: {}",
                                    t.id(),
                                    t.description(),
                                    props
                                )
                            })
                            .unwrap_or_default();

                        let step_desc = &plan.steps[step_idx].description;
                        let arg_prompt = format!(
                            "Generate the JSON arguments for a tool call.\n\n\
                             Tool: {}\n\n\
                             Task: {}\n\n\
                             Previous step results:\n{}\n\n\
                             Planner's draft arguments: {}\n\n\
                             Produce ONLY a valid JSON object with the correct argument values.\n\
                             Use actual values from the previous step results, not template references.",
                            tool_schema,
                            step_desc,
                            dep_context,
                            serde_json::to_string(args).unwrap_or_default()
                        );
                        let messages = vec![ChatMessage::user(&arg_prompt)];
                        match llm.complete(&messages, None).await {
                            Ok(resp) => {
                                let content = resp.content.trim();
                                // Parse the LLM's JSON response, fall back to planner args.
                                let json_start = content.find('{');
                                let json_end = content.rfind('}');
                                if let (Some(start), Some(end)) = (json_start, json_end) {
                                    serde_json::from_str(&content[start..=end])
                                        .unwrap_or_else(|_| args.clone())
                                } else {
                                    args.clone()
                                }
                            }
                            Err(_) => args.clone(),
                        }
                    } else {
                        args.clone()
                    };

                    let tool_call = ToolCall {
                        id: uuid::Uuid::new_v4().to_string(),
                        name: tool.clone(),
                        arguments: final_args,
                    };
                    match self.execute_tool_smart(&tool_call).await {
                        Ok(output) => serde_json::json!({ "output": output }),
                        Err(e) => {
                            plan.steps[step_idx].mark_failed(e.to_string());
                            continue;
                        }
                    }
                }
                PlanAction::Skill { skill } => {
                    if let Some(skill_def) = self.skills.iter().find(|s| &s.id == skill) {
                        if let Some(ref executor) = self.skill_executor {
                            match executor.execute(skill_def, "", serde_json::json!({})).await {
                                Ok(output) => serde_json::json!({ "output": output }),
                                Err(e) => {
                                    plan.steps[step_idx].mark_failed(e.to_string());
                                    continue;
                                }
                            }
                        } else {
                            serde_json::json!({ "output": "Skill executor not available" })
                        }
                    } else {
                        plan.steps[step_idx].mark_failed("Skill not found");
                        continue;
                    }
                }
                PlanAction::Think { prompt } => {
                    let context: String = results
                        .iter()
                        .map(|(k, v)| format!("{}: {}", k, v))
                        .collect::<Vec<_>>()
                        .join("\n");

                    let think_prompt = format!("Context:\n{}\n\nTask: {}", context, prompt);
                    let messages = vec![ChatMessage::user(&think_prompt)];

                    match llm.complete(&messages, None).await {
                        Ok(resp) => serde_json::json!({ "output": resp.content }),
                        Err(e) => {
                            plan.steps[step_idx].mark_failed(e.to_string());
                            continue;
                        }
                    }
                }
                PlanAction::Respond { template } => {
                    let context: String = results
                        .iter()
                        .map(|(k, v)| format!("{}: {}", k, v))
                        .collect::<Vec<_>>()
                        .join("\n");

                    let respond_prompt = format!(
                        "Based on this context:\n{}\n\nGenerate a response following this template/instruction: {}",
                        context, template
                    );
                    let messages = vec![ChatMessage::user(&respond_prompt)];

                    match llm.complete(&messages, None).await {
                        Ok(resp) => serde_json::json!({ "output": resp.content }),
                        Err(e) => {
                            plan.steps[step_idx].mark_failed(e.to_string());
                            continue;
                        }
                    }
                }
            };

            results.insert(plan.steps[step_idx].id.clone(), result.clone());
            plan.steps[step_idx].mark_completed(Some(result));
        }

        // Set plan status based on whether any steps actually failed.
        let has_failures = plan.steps.iter().any(|s| s.status.is_failed());
        if has_failures {
            let failed_ids: Vec<String> = plan
                .steps
                .iter()
                .filter(|s| s.status.is_failed())
                .map(|s| s.id.clone())
                .collect();
            plan.status = PlanStatus::Failed {
                error: format!("Steps failed: {}", failed_ids.join(", ")),
            };
        } else {
            plan.status = PlanStatus::Completed;
        }

        // Synthesize final output from all completed step results.
        let all_outputs: Vec<String> = plan
            .steps
            .iter()
            .filter(|s| s.status.is_completed())
            .filter_map(|s| {
                s.result
                    .as_ref()
                    .and_then(|r| r.get("output"))
                    .and_then(|o| o.as_str())
                    .map(|o| format!("{}: {}", s.description, o))
            })
            .collect();

        if all_outputs.is_empty() {
            return Ok("Plan execution completed but produced no results.".to_string());
        }

        if all_outputs.len() == 1 {
            return Ok(all_outputs.into_iter().next().unwrap());
        }

        // Synthesize a coherent summary from multiple step results via LLM.
        let context = all_outputs.join("\n\n");
        let prompt = format!(
            "You completed a multi-step plan for: \"{}\"\n\nStep results:\n{}\n\nProvide a coherent final response that synthesizes these results.",
            plan.goal, context
        );
        let messages = vec![ChatMessage::user(&prompt)];
        match llm.complete(&messages, None).await {
            Ok(resp) => Ok(resp.content.trim().to_string()),
            Err(_) => Ok(context),
        }
    }

    async fn evaluate_response(&self, input: &str, response: &str) -> Result<EvaluationResult> {
        let effective_config = self.get_effective_reflection_config();
        self.evaluate_response_with_config(input, response, &effective_config)
            .await
    }

    fn extract_thinking(&self, content: &str) -> (Option<String>, String) {
        if let Some(start) = content.find("<thinking>") {
            if let Some(end) = content.find("</thinking>") {
                let thinking = content[start + 10..end].trim().to_string();
                let answer = content[end + 11..].trim().to_string();
                return (Some(thinking), answer);
            }
        }
        (None, content.to_string())
    }

    fn format_response_with_thinking(&self, thinking: Option<&str>, answer: &str) -> String {
        match self.get_effective_reasoning_config().output {
            ReasoningOutput::Hidden => answer.to_string(),
            ReasoningOutput::Visible => {
                if let Some(t) = thinking {
                    format!("Thinking:\n{}\n\nAnswer:\n{}", t, answer)
                } else {
                    answer.to_string()
                }
            }
            ReasoningOutput::Tagged => {
                if let Some(t) = thinking {
                    format!("<thinking>{}</thinking>\n{}", t, answer)
                } else {
                    answer.to_string()
                }
            }
        }
    }

    async fn run_loop(&self, input: &str) -> Result<AgentResponse> {
        info!(input_len = input.len(), "Starting chat");

        self.hooks.on_message_received(input).await;

        // One-shot context initialization: load runtime defaults, resolve env vars,
        // populate builtin sources (session, agent), etc.  This must happen before
        // the first template render so that {{ context.* }} variables are available.
        if !self.context_initialized.swap(true, Ordering::SeqCst) {
            self.context_manager.initialize().await?;
            debug!("Context manager initialized (defaults, env, builtins)");
        }

        self.check_turn_timeout().await?;
        self.context_manager.refresh_per_turn().await?;

        // Clear stale disambiguation context from previous turns.
        // This prevents resolved_intent from leaking across turns and causing incorrect deterministic routing on subsequent inputs.
        self.clear_disambiguation_context();

        // Disambiguation check (before input processing)
        if let Some(ref disambiguator) = self.disambiguation_manager {
            let disambiguation_context = self.build_disambiguation_context().await?;

            // Get state-level disambiguation override
            let state_override = self
                .state_machine
                .as_ref()
                .and_then(|sm| sm.current_definition())
                .and_then(|def| def.disambiguation.clone());

            match disambiguator
                .process_input_with_override(
                    input,
                    &disambiguation_context,
                    state_override.as_ref(),
                    None,
                )
                .await?
            {
                DisambiguationResult::Clear => {
                    debug!("Input is clear, proceeding normally");
                }
                DisambiguationResult::NeedsClarification {
                    question,
                    detection,
                } => {
                    info!(
                        ambiguity_type = ?detection.ambiguity_type,
                        confidence = detection.confidence,
                        "Input requires clarification"
                    );

                    self.memory.add_message(ChatMessage::user(input)).await?;
                    self.memory
                        .add_message(ChatMessage::assistant(&question.question))
                        .await?;

                    return Ok(AgentResponse::new(&question.question).with_metadata(
                        "disambiguation",
                        serde_json::json!({
                            "status": "awaiting_clarification",
                            "options": question.options,
                            "clarifying": question.clarifying,
                            "detection": {
                                "type": detection.ambiguity_type,
                                "confidence": detection.confidence,
                                "what_is_unclear": detection.what_is_unclear,
                            }
                        }),
                    ));
                }
                DisambiguationResult::Clarified {
                    enriched_input,
                    resolved,
                    ..
                } => {
                    info!(
                        resolved_count = resolved.len(),
                        enriched = %enriched_input,
                        "Input clarified, injecting resolved intent into context"
                    );

                    // Routing uses `resolved` (structured, deterministic)
                    // This is what makes post-disambiguation routing DETERMINISTIC
                    for (key, value) in &resolved {
                        let context_key = format!("disambiguation.{}", key);
                        let _ = self.context_manager.set(&context_key, value.clone());
                    }

                    if let Some(intent) = resolved.get("intent") {
                        let _ = self.context_manager.set("resolved_intent", intent.clone());
                    }

                    let _ = self
                        .context_manager
                        .set("disambiguation.resolved", serde_json::Value::Bool(true));

                    // Check if this clarification was triggered by a skill-level override.
                    // If so, route directly to the matched skill instead of going through
                    // skill routing again (which might match a different skill).
                    let skill_id = self.pending_skill_id.read().clone();
                    if let Some(skill_id) = skill_id {
                        info!(skill_id = %skill_id, "Re-checking skill disambiguation on clarified input");
                        return self
                            .recheck_skill_disambiguation(&skill_id, &enriched_input)
                            .await;
                    }

                    return self.run_loop_internal(&enriched_input).await;
                }
                DisambiguationResult::ProceedWithBestGuess { enriched_input } => {
                    info!("Proceeding with best guess interpretation");

                    // Same skill-id re-check for best-guess path
                    let skill_id = self.pending_skill_id.read().clone();
                    if let Some(skill_id) = skill_id {
                        info!(skill_id = %skill_id, "Re-checking skill disambiguation on best-guess input");
                        return self
                            .recheck_skill_disambiguation(&skill_id, &enriched_input)
                            .await;
                    }

                    return self.run_loop_internal(&enriched_input).await;
                }
                DisambiguationResult::GiveUp { reason } => {
                    *self.pending_skill_id.write() = None;
                    warn!(reason = %reason, "Disambiguation gave up");
                    let apology = self
                        .generate_localized_apology(
                            "Generate a brief, polite apology saying you couldn't understand the request. Be concise.",
                            &reason,
                        )
                        .await
                        .unwrap_or_else(|_| {
                            format!("I'm sorry, I couldn't understand your request: {}", reason)
                        });
                    return Ok(AgentResponse::new(&apology));
                }
                DisambiguationResult::Escalate { reason } => {
                    *self.pending_skill_id.write() = None;
                    info!(reason = %reason, "Escalating to human");
                    if let Some(ref hitl) = self.hitl_engine {
                        let trigger =
                            ApprovalTrigger::condition("disambiguation_escalation", reason.clone());
                        let mut context_map = HashMap::new();
                        context_map.insert("original_input".to_string(), serde_json::json!(input));
                        context_map.insert("reason".to_string(), serde_json::json!(&reason));
                        let check_result = HITLCheckResult::required(
                            trigger,
                            context_map,
                            format!("User request needs human assistance: {}", reason),
                            Some(hitl.config().default_timeout_seconds),
                        );
                        let result = self.request_hitl_approval(check_result).await?;
                        if matches!(
                            result,
                            ApprovalResult::Approved | ApprovalResult::Modified { .. }
                        ) {
                            return self.run_loop_internal(input).await;
                        }
                    }
                    let apology = self
                        .generate_localized_apology(
                            "Explain briefly that you're transferring the user to a human agent for help.",
                            &reason,
                        )
                        .await
                        .unwrap_or_else(|_| {
                            format!("I need human assistance to help with your request: {}", reason)
                        });
                    return Ok(AgentResponse::new(&apology));
                }
                DisambiguationResult::Abandoned { new_input } => {
                    *self.pending_skill_id.write() = None;

                    info!(
                        has_new_input = new_input.is_some(),
                        "Clarification abandoned by user"
                    );

                    self.memory.add_message(ChatMessage::user(input)).await?;

                    match new_input {
                        Some(fresh_input) => {
                            // Topic switch: process the user's new input from scratch.
                            // The LLM sees full conversation context including the abandoned exchange.
                            return self.run_loop_internal(&fresh_input).await;
                        }
                        None => {
                            // Pure abandonment: generate a brief acknowledgment.
                            let ack = self
                                .generate_localized_apology(
                                    "The user changed their mind about their previous request. \
                                     Generate a brief, friendly acknowledgment (e.g. 'OK, no problem. What else can I help with?'). \
                                     Do NOT apologize excessively. Be concise.",
                                    "User abandoned clarification",
                                )
                                .await
                                .unwrap_or_else(|_| {
                                    "OK, no problem. What else can I help with?".to_string()
                                });

                            self.memory
                                .add_message(ChatMessage::assistant(&ack))
                                .await?;

                            return Ok(AgentResponse::new(&ack));
                        }
                    }
                }
            }
        }

        self.run_loop_internal(input).await
    }

    /// Generate a localized response using the router LLM
    async fn generate_localized_apology(&self, instruction: &str, reason: &str) -> Result<String> {
        let llm = self.llm_registry.router().map_err(|e| {
            AgentError::LLM(format!(
                "Router LLM not available for localized response: {}",
                e
            ))
        })?;

        let recent: Vec<String> = self
            .memory
            .get_messages(Some(3))
            .await?
            .iter()
            .map(|m| m.content.clone())
            .collect();

        let context_hint = if recent.is_empty() {
            String::new()
        } else {
            format!(
                "\nRecent conversation (detect the user's language from this):\n{}\n",
                recent.join("\n")
            )
        };

        let prompt = format!(
            "{}\nReason: {}\n{}Respond in the same language as the user. Output ONLY the message, nothing else.",
            instruction, reason, context_hint
        );

        let messages = vec![ChatMessage::user(&prompt)];
        let response = llm
            .complete(&messages, None)
            .await
            .map_err(|e| AgentError::LLM(format!("Localized response generation failed: {}", e)))?;

        Ok(response.content.trim().to_string())
    }

    /// Clear disambiguation-related keys from the context manager.
    ///
    /// Render template variables in state action args using the context manager.
    fn render_action_args(&self, args: &Value) -> Value {
        let context = self.context_manager.get_all();
        match args {
            Value::Object(map) => {
                let mut rendered = serde_json::Map::new();
                for (k, v) in map {
                    match v {
                        Value::String(s) if s.contains("{{") => {
                            match self.template_renderer.render(s, &context) {
                                Ok(rendered_str) => {
                                    rendered.insert(k.clone(), Value::String(rendered_str));
                                }
                                Err(_) => {
                                    rendered.insert(k.clone(), v.clone());
                                }
                            }
                        }
                        _ => {
                            rendered.insert(k.clone(), v.clone());
                        }
                    }
                }
                Value::Object(rendered)
            }
            _ => args.clone(),
        }
    }

    /// Called at the start of each turn to prevent stale `resolved_intent` from leaking across turns.
    fn clear_disambiguation_context(&self) {
        let _ = self
            .context_manager
            .set("resolved_intent", serde_json::Value::Null);

        let all = self.context_manager.get_all();
        for key in all.keys() {
            if key.starts_with("disambiguation.") {
                let _ = self.context_manager.set(key, serde_json::Value::Null);
            }
        }
    }

    /// Re-run skill disambiguation on enriched input before executing the skill.
    /// After clarification resolves, the enriched input may still be missing required_clarity fields (e.g. "Transfer money to Jane." still lacks amount).
    /// This method re-runs the skill's disambiguation pass.
    /// If fields are still missing, it returns the new clarification question and keeps pending_skill_id set.
    /// If all fields are present (Clear), it executes the skill and returns the response.
    async fn recheck_skill_disambiguation(
        &self,
        skill_id: &str,
        enriched_input: &str,
    ) -> Result<AgentResponse> {
        let skill = self
            .skill_router
            .as_ref()
            .and_then(|r| r.get_skill(skill_id).cloned());

        // If the skill has disambiguation enabled, re-run it on the enriched input.
        if let Some(ref skill) = skill {
            if let Some(ref skill_disambig) = skill.disambiguation {
                if skill_disambig.enabled.unwrap_or(false) {
                    if let Some(ref disambiguator) = self.disambiguation_manager {
                        let context = self.build_disambiguation_context().await?;
                        let state_override = self
                            .state_machine
                            .as_ref()
                            .and_then(|sm| sm.current_definition())
                            .and_then(|def| def.disambiguation.clone());

                        match disambiguator
                            .process_input_with_override(
                                enriched_input,
                                &context,
                                state_override.as_ref(),
                                Some(skill_disambig),
                            )
                            .await?
                        {
                            DisambiguationResult::Clear => {
                                debug!(skill_id = %skill_id, "Skill re-check: all fields present");
                            }
                            DisambiguationResult::NeedsClarification {
                                question,
                                detection,
                            } => {
                                info!(
                                    skill_id = %skill_id,
                                    ambiguity_type = ?detection.ambiguity_type,
                                    what_is_unclear = ?detection.what_is_unclear,
                                    "Skill re-check: still missing fields, asking again"
                                );
                                // Keep pending_skill_id set (do NOT clear it).
                                // The next turn will resolve this new clarification and
                                // re-enter this method until all fields are present.
                                self.memory
                                    .add_message(ChatMessage::user(enriched_input))
                                    .await?;
                                self.memory
                                    .add_message(ChatMessage::assistant(&question.question))
                                    .await?;

                                return Ok(AgentResponse::new(&question.question).with_metadata(
                                    "disambiguation",
                                    serde_json::json!({
                                        "status": "awaiting_clarification",
                                        "skill_id": skill_id,
                                        "options": question.options,
                                        "clarifying": question.clarifying,
                                        "detection": {
                                            "type": detection.ambiguity_type,
                                            "confidence": detection.confidence,
                                            "what_is_unclear": detection.what_is_unclear,
                                        }
                                    }),
                                ));
                            }
                            DisambiguationResult::Clarified {
                                enriched_input: re_enriched,
                                ..
                            } => {
                                debug!(skill_id = %skill_id, "Skill re-check: clarified immediately, executing");
                                // Fall through to execute with the further-enriched input.
                                *self.pending_skill_id.write() = None;
                                let skill_response =
                                    self.execute_skill_by_id(skill_id, &re_enriched).await?;
                                self.memory
                                    .add_message(ChatMessage::user(&re_enriched))
                                    .await?;
                                return self
                                    .handle_skill_response(
                                        &re_enriched,
                                        skill_response,
                                        &HashMap::new(),
                                    )
                                    .await;
                            }
                            DisambiguationResult::ProceedWithBestGuess {
                                enriched_input: re_enriched,
                            } => {
                                debug!(skill_id = %skill_id, "Skill re-check: proceeding with best guess");
                                *self.pending_skill_id.write() = None;
                                let skill_response =
                                    self.execute_skill_by_id(skill_id, &re_enriched).await?;
                                self.memory
                                    .add_message(ChatMessage::user(&re_enriched))
                                    .await?;
                                return self
                                    .handle_skill_response(
                                        &re_enriched,
                                        skill_response,
                                        &HashMap::new(),
                                    )
                                    .await;
                            }
                            DisambiguationResult::GiveUp { reason } => {
                                *self.pending_skill_id.write() = None;
                                let apology = self
                                    .generate_localized_apology(
                                        "Generate a brief, polite apology saying you couldn't understand the request. Be concise.",
                                        &reason,
                                    )
                                    .await
                                    .unwrap_or_else(|_| {
                                        format!("I'm sorry, I couldn't understand your request: {}", reason)
                                    });
                                return Ok(AgentResponse::new(&apology));
                            }
                            DisambiguationResult::Escalate { reason } => {
                                *self.pending_skill_id.write() = None;
                                let apology = self
                                    .generate_localized_apology(
                                        "Explain briefly that you're transferring the user to a human agent for help.",
                                        &reason,
                                    )
                                    .await
                                    .unwrap_or_else(|_| {
                                        format!("I need human assistance to help with your request: {}", reason)
                                    });
                                return Ok(AgentResponse::new(&apology));
                            }
                            DisambiguationResult::Abandoned { new_input } => {
                                // User abandoned during skill re-check.
                                // Clear skill routing state and fall through to normal execution.
                                *self.pending_skill_id.write() = None;
                                debug!(skill_id = %skill_id, "Skill re-check: abandoned by user");
                                if let Some(fresh) = new_input {
                                    return self.run_loop_internal(&fresh).await;
                                }
                                let ack = self
                                    .generate_localized_apology(
                                        "The user changed their mind about their previous request. \
                                         Generate a brief, friendly acknowledgment (e.g. 'OK, no problem. What else can I help with?'). \
                                         Do NOT apologize excessively. Be concise.",
                                        "User abandoned clarification",
                                    )
                                    .await
                                    .unwrap_or_else(|_| {
                                        "OK, no problem. What else can I help with?".to_string()
                                    });
                                self.memory
                                    .add_message(ChatMessage::assistant(&ack))
                                    .await?;
                                return Ok(AgentResponse::new(&ack));
                            }
                        }
                    }
                }
            }
        }

        // Skill has no disambiguation or all fields present: execute.
        *self.pending_skill_id.write() = None;
        let skill_response = self.execute_skill_by_id(skill_id, enriched_input).await?;
        self.memory
            .add_message(ChatMessage::user(enriched_input))
            .await?;
        self.handle_skill_response(enriched_input, skill_response, &HashMap::new())
            .await
    }

    /// Handle skill routing result: output processing, memory, transitions.
    /// Returns a fully formed AgentResponse for skill-routed requests.
    async fn handle_skill_response(
        &self,
        processed_input: &str,
        skill_response: String,
        input_context: &HashMap<String, Value>,
    ) -> Result<AgentResponse> {
        let output_data = self.process_output(&skill_response, input_context).await?;
        let final_response = output_data.content;

        self.memory
            .add_message(ChatMessage::assistant(&final_response))
            .await?;

        self.check_memory_compression().await?;

        self.increment_turn();
        self.evaluate_transitions(processed_input, &final_response)
            .await?;

        let response = AgentResponse::new(final_response);
        self.hooks.on_response(&response).await;
        Ok(response)
    }

    /// Run the Plan-and-Execute flow: generate plan, execute steps, finalize.
    /// Supports plan-level reflection with replan loop when configured.
    async fn handle_plan_and_execute(
        &self,
        processed_input: &str,
        input_context: &HashMap<String, Value>,
        auto_detected: bool,
    ) -> Result<AgentResponse> {
        let effective = self.get_effective_reasoning_config();
        let plan_reflection = effective
            .get_planning()
            .map(|c| c.reflection.clone())
            .unwrap_or_default();

        let max_attempts = if plan_reflection.enabled {
            1 + plan_reflection.max_replans
        } else {
            1
        };

        let mut plan = self.generate_plan(processed_input).await?;
        info!(
            plan_id = %plan.id,
            steps = plan.steps.len(),
            "Plan generated"
        );

        let mut plan_result = String::new();

        for attempt in 0..max_attempts {
            *self.current_plan.write() = Some(plan.clone());
            plan_result = self.execute_plan(&mut plan).await?;

            info!(
                plan_status = ?plan.status,
                completed_steps = plan.completed_steps().count(),
                attempt = attempt + 1,
                "Plan execution completed"
            );

            if !plan_reflection.enabled {
                break;
            }

            let has_failures = plan.steps.iter().any(|s| s.status.is_failed());
            if !has_failures {
                break;
            }

            if attempt + 1 >= max_attempts {
                break;
            }

            match plan_reflection.on_step_failure {
                StepFailureAction::Replan => {
                    info!(attempt = attempt + 1, "Plan had failures, replanning");
                    plan = self.generate_plan(processed_input).await?;
                }
                StepFailureAction::Abort => {
                    warn!("Plan step failed, aborting");
                    break;
                }
                StepFailureAction::Skip | StepFailureAction::Continue => {
                    break;
                }
            }
        }

        *self.current_plan.write() = Some(plan);

        let output_data = self.process_output(&plan_result, input_context).await?;
        let final_content = output_data.content;

        self.memory
            .add_message(ChatMessage::assistant(&final_content))
            .await?;

        self.check_memory_compression().await?;
        self.increment_turn();
        self.evaluate_transitions(processed_input, &final_content)
            .await?;

        let reasoning_metadata =
            ReasoningMetadata::new(ReasoningMode::PlanAndExecute).with_auto_detected(auto_detected);

        let response = AgentResponse::new(&final_content).with_metadata(
            "reasoning",
            serde_json::to_value(&reasoning_metadata).unwrap_or_default(),
        );

        self.hooks.on_response(&response).await;
        Ok(response)
    }

    /// Inject CoT/ReAct reasoning prompt into the system message (first iteration only).
    fn inject_reasoning_prompt(
        &self,
        messages: &mut [ChatMessage],
        reasoning_mode: &ReasoningMode,
        is_first_iteration: bool,
    ) {
        if !is_first_iteration {
            return;
        }
        match reasoning_mode {
            ReasoningMode::CoT => {
                if let Some(msg) = messages.first_mut() {
                    if matches!(msg.role, ai_agents_core::Role::System) {
                        msg.content = self.build_cot_system_prompt(&msg.content);
                        debug!("Applied Chain-of-Thought system prompt");
                    }
                }
            }
            ReasoningMode::React => {
                if let Some(msg) = messages.first_mut() {
                    if matches!(msg.role, ai_agents_core::Role::System) {
                        msg.content = self.build_react_system_prompt(&msg.content);
                        debug!("Applied ReAct system prompt");
                    }
                }
            }
            _ => {}
        }
    }

    /// Handle tool calls: check transitions, execute tools in parallel, handle HITL rejection.
    async fn handle_tool_calls(
        &self,
        processed_input: &str,
        content: &str,
        tool_calls: Vec<ToolCall>,
        all_tool_calls: &mut Vec<ToolCall>,
    ) -> Result<ToolCallOutcome> {
        // Check if a transition should fire before executing the LLM's tool call.
        // If a transition fires, on_enter actions handle the tool call correctly
        // (with proper URLs from YAML), so skip the LLM's tool call.
        let transition_fired = self.evaluate_transitions(processed_input, content).await?;
        if transition_fired {
            self.memory
                .add_message(ChatMessage::assistant(
                    "(Transitioned to new state — tool call handled by workflow)",
                ))
                .await?;
            return Ok(ToolCallOutcome::TransitionFired);
        }

        // Store the assistant's tool-call message so the LLM sees its own decision in conversation history. Without this, the model only sees the tool result and may repeat the same call.
        self.memory
            .add_message(ChatMessage::assistant(content))
            .await?;

        let results = self.execute_tools_parallel(&tool_calls).await;

        for ((_id, result), tool_call) in results.into_iter().zip(tool_calls.iter()) {
            match result {
                Ok(output) => {
                    self.memory
                        .add_message(ChatMessage::function(&tool_call.name, &output))
                        .await?;
                }
                Err(e) => {
                    // Check if this is a HITL rejection - if so, break the loop
                    if matches!(e, AgentError::HITLRejected(_)) {
                        self.memory
                            .add_message(ChatMessage::assistant(&format!(
                                "The operation was rejected by the approver: {}",
                                e
                            )))
                            .await?;
                        // Return the rejection message to user, don't continue loop
                        return Ok(ToolCallOutcome::Rejected(AgentResponse {
                            content: format!("Operation cancelled: {}", e),
                            metadata: None,
                            tool_calls: Some(all_tool_calls.clone()),
                        }));
                    }
                    self.memory
                        .add_message(ChatMessage::function(
                            &tool_call.name,
                            &format!("Error: {}", e),
                        ))
                        .await?;
                }
            }
            all_tool_calls.push(tool_call.clone());
        }
        Ok(ToolCallOutcome::Continue)
    }

    /// Run the reflection loop on a response, returning (improved_content, reflection_metadata).
    async fn run_reflection(
        &self,
        llm: &dyn LLMProvider,
        processed_input: &str,
        mut content: String,
    ) -> Result<(String, Option<ReflectionMetadata>)> {
        let should_reflect = self.should_reflect(processed_input, &content).await?;
        if !should_reflect {
            return Ok((content, None));
        }

        info!("Starting response reflection evaluation");
        let mut attempts = 0u32;
        let max_retries = self.reflection_config.max_retries;
        let mut history: Vec<ReflectionAttempt> = Vec::new();

        loop {
            let evaluation = self.evaluate_response(processed_input, &content).await?;

            if evaluation.passed || attempts >= max_retries {
                info!(
                    passed = evaluation.passed,
                    confidence = evaluation.confidence,
                    attempts = attempts + 1,
                    "Reflection evaluation complete"
                );
                let reflection_metadata = Some(
                    ReflectionMetadata::new(evaluation)
                        .with_attempts(attempts + 1)
                        .with_history(history),
                );
                return Ok((content, reflection_metadata));
            }

            debug!(
                attempt = attempts + 1,
                failed_criteria = evaluation.failed_criteria().count(),
                "Response did not meet criteria, retrying"
            );

            history.push(
                ReflectionAttempt::new(&content, evaluation.clone())
                    .with_feedback("Response did not meet quality criteria"),
            );

            let feedback: Vec<String> = evaluation
                .failed_criteria()
                .map(|c| format!("- {}", c.criterion))
                .collect();

            let retry_prompt = format!(
                "Your previous response did not meet these criteria:\n{}\n\nPlease provide an improved response.",
                feedback.join("\n")
            );

            self.memory
                .add_message(ChatMessage::user(&retry_prompt))
                .await?;

            let retry_messages = self.build_messages().await?;
            let retry_response = llm
                .complete(&retry_messages, None)
                .await
                .map_err(|e| AgentError::LLM(e.to_string()))?;

            content = retry_response.content.trim().to_string();
            attempts += 1;
        }
    }

    /// Record the assistant turn, evaluate transitions, and decide what to do next.
    /// Returns PostLoopResult so callers can apply_post_loop_result for re-dispatch.
    async fn post_loop_processing(
        &self,
        processed_input: &str,
        content: String,
    ) -> Result<PostLoopResult> {
        // Do NOT add the assistant message to memory yet.
        // evaluate_transitions receives content as a direct parameter, so the message does not need to be in memory for transitions to evaluate correctly.
        // For NeedsRedispatch we skip adding the stale old-state response entirely, keeping memory clean for the re-dispatched handler.

        self.increment_turn();

        // Run context extractors so guards can check freshly-extracted values.
        self.run_context_extractors(processed_input).await;

        let transitioned = self.evaluate_transitions(processed_input, &content).await?;

        if !transitioned {
            self.memory
                .add_message(ChatMessage::assistant(&content))
                .await?;
            self.check_memory_compression().await?;
            return Ok(PostLoopResult::NoTransition(content));
        }

        // Check if we should skip re-generation after this transition.
        if !self.should_regenerate_after_transition() {
            self.memory
                .add_message(ChatMessage::assistant(&content))
                .await?;
            self.check_memory_compression().await?;
            return Ok(PostLoopResult::Transitioned(content));
        }

        // Check if the new state needs full dispatch.
        // Orchestration states (concurrent, group_chat, pipeline, handoff, delegate) need their dedicated handlers.
        // Any non-None effective reasoning mode needs CoT/ReAct prompt injection, the plan-and-execute handler, or Auto re-determination - all of which live in run_loop_internal, not here.
        if self.needs_redispatch_for_new_state() {
            info!("Post-transition NeedsRedispatch: new state requires full dispatch");
            // Stale old-state content is NOT added to memory.
            // apply_post_loop_result will increment redispatch_depth and call run_loop_internal, which produces the correct response and adds it.
            return Ok(PostLoopResult::NeedsRedispatch);
        }

        // Normal post-transition re-generation (plain LLM with optional tool calls).
        // Add the stale response to history so the new LLM sees the conversation.
        self.memory
            .add_message(ChatMessage::assistant(&content))
            .await?;
        self.check_memory_compression().await?;

        // If a transition fired, on_enter actions already executed (e.g., HTTP calls).
        // The current content was generated in the OLD state context and is stale.
        // Re-generate in the new state context so the LLM can reference on_enter results.
        // If the LLM responds with a tool call, execute it in a mini-loop so the
        // result is not returned as raw JSON text.
        let new_llm = self.get_state_llm()?;
        let mut final_content;

        for post_iter in 0..self.max_iterations {
            let new_messages = self.build_messages().await?;
            if post_iter == 0 {
                if let Some(system_msg) = new_messages.first() {
                    if system_msg.role == ai_agents_core::Role::System {
                        debug!(
                            prompt_preview =
                                &system_msg.content[system_msg.content.len().saturating_sub(200)..],
                            "Post-transition system prompt (last 200 chars)"
                        );
                    }
                }
            }

            let new_response = new_llm
                .complete(&new_messages, None)
                .await
                .map_err(|e| AgentError::LLM(e.to_string()))?;
            final_content = new_response.content.trim().to_string();

            // Check if the post-transition response contains tool calls.
            // If so, execute them and loop so the LLM can summarize the result.
            if let Some(tool_calls) = self.parse_tool_calls(&final_content) {
                debug!(
                    post_iter = post_iter,
                    tools = tool_calls.len(),
                    "Post-transition tool call detected, executing"
                );

                self.memory
                    .add_message(ChatMessage::assistant(&final_content))
                    .await?;

                let results = self.execute_tools_parallel(&tool_calls).await;
                for ((_id, result), tool_call) in results.into_iter().zip(tool_calls.iter()) {
                    match result {
                        Ok(output) => {
                            self.memory
                                .add_message(ChatMessage::function(&tool_call.name, &output))
                                .await?;
                        }
                        Err(e) => {
                            self.memory
                                .add_message(ChatMessage::function(
                                    &tool_call.name,
                                    &format!("Error: {}", e),
                                ))
                                .await?;
                        }
                    }
                }
                // Loop to let the LLM see the tool result and produce a text response.
                continue;
            }

            // No tool call - this is the final text response.
            self.memory
                .add_message(ChatMessage::assistant(&final_content))
                .await?;
            return Ok(PostLoopResult::Transitioned(final_content));
        }

        // Exhausted post-transition iterations (unlikely). Return last content.
        final_content = "Post-transition processing completed.".to_string();
        self.memory
            .add_message(ChatMessage::assistant(&final_content))
            .await?;

        Ok(PostLoopResult::Transitioned(final_content))
    }

    /// Build the final AgentResponse with all metadata.
    /// Check whether to re-generate a response after a state transition.
    fn should_regenerate_after_transition(&self) -> bool {
        if let Some(ref sm) = self.state_machine {
            // Global setting
            if !sm.config().regenerate_on_transition {
                return false;
            }
            // Per-state override on the new (current) state
            if let Some(def) = sm.current_definition() {
                if let Some(regen) = def.regenerate_on_enter {
                    return regen;
                }
            }
        }
        true
    }

    /// Return true when the new state requires full dispatch via run_loop_internal.
    /// Covers orchestration states and any non-None effective reasoning mode.
    fn needs_redispatch_for_new_state(&self) -> bool {
        if let Some(ref sm) = self.state_machine {
            if let Some(def) = sm.current_definition() {
                if def.concurrent.is_some()
                    || def.group_chat.is_some()
                    || def.pipeline.is_some()
                    || def.handoff.is_some()
                    || def.delegate.is_some()
                {
                    return true;
                }
                // Any non-None effective reasoning mode requires the main dispatch loop.
                let effective = self.get_effective_reasoning_config();
                if !matches!(effective.mode, ReasoningMode::None) {
                    return true;
                }
            }
        }
        false
    }

    /// Consume a PostLoopResult. NeedsRedispatch re-enters run_loop_internal.
    /// The user message is already in memory - redispatch_depth suppresses re-adding it.
    async fn apply_post_loop_result(
        &self,
        processed_input: &str,
        result: PostLoopResult,
    ) -> Result<String> {
        match result {
            PostLoopResult::NoTransition(content) | PostLoopResult::Transitioned(content) => {
                Ok(content)
            }
            PostLoopResult::NeedsRedispatch => {
                const MAX_REDISPATCH_DEPTH: u32 = 3;
                let current_depth = *self.redispatch_depth.read();
                if current_depth >= MAX_REDISPATCH_DEPTH {
                    warn!(
                        depth = current_depth,
                        "Post-transition re-dispatch depth limit reached, returning empty response"
                    );
                    let content = String::new();
                    self.memory
                        .add_message(ChatMessage::assistant(&content))
                        .await?;
                    return Ok(content);
                }
                *self.redispatch_depth.write() += 1;
                info!(
                    depth = current_depth + 1,
                    "Re-dispatching for new state after transition"
                );
                let resp = Box::pin(self.run_loop_internal(processed_input)).await;
                *self.redispatch_depth.write() -= 1;
                resp.map(|r| r.content)
            }
        }
    }

    fn build_agent_response(
        &self,
        content: String,
        all_tool_calls: Vec<ToolCall>,
        reasoning_mode: ReasoningMode,
        auto_detected: bool,
        iterations: u32,
        thinking: Option<String>,
        reflection_metadata: Option<ReflectionMetadata>,
    ) -> AgentResponse {
        let reasoning_metadata = ReasoningMetadata::new(reasoning_mode.clone())
            .with_thinking(thinking.clone().unwrap_or_default())
            .with_iterations(iterations)
            .with_auto_detected(auto_detected);

        let mut response = AgentResponse::new(&content);
        if !all_tool_calls.is_empty() {
            response = response.with_tool_calls(all_tool_calls);
        }

        if let Some(state) = self.current_state() {
            response = response.with_metadata("current_state", serde_json::json!(state));
        }

        response = response.with_metadata(
            "reasoning",
            serde_json::to_value(&reasoning_metadata).unwrap_or_default(),
        );

        if let Some(ref refl_meta) = reflection_metadata {
            response = response.with_metadata(
                "reflection",
                serde_json::to_value(refl_meta).unwrap_or_default(),
            );
        }

        response
    }

    // Handle delegation: forward user input to a registry agent.
    async fn handle_delegated_state(
        &self,
        input: &str,
        delegate_id: &str,
        state_def: &ai_agents_state::StateDefinition,
    ) -> Result<AgentResponse> {
        use std::time::Instant;

        let registry = self.spawner_registry.as_ref().ok_or_else(|| {
            AgentError::Config(format!(
                "State delegates to '{}' but no agent registry is configured. \
                 Add a spawner section with auto_spawn to your YAML.",
                delegate_id
            ))
        })?;

        let state_name = self
            .state_machine
            .as_ref()
            .map(|sm| sm.current())
            .unwrap_or_else(|| "unknown".to_string());

        self.hooks.on_delegate_start(delegate_id, &state_name).await;
        let start = Instant::now();

        let delegate = registry.get(delegate_id).ok_or_else(|| {
            AgentError::Other(format!(
                "State '{}' delegates to '{}' but no agent with that ID exists in the registry.",
                state_name, delegate_id
            ))
        })?;

        // Prepare input based on delegate_context mode.
        let context_mode = state_def.delegate_context.clone().unwrap_or_default();
        let effective_input = crate::orchestration::context::prepare_delegate_input(
            input,
            &context_mode,
            &*self.memory,
            self.llm_registry.get("router").ok().as_deref(),
        )
        .await?;

        let response = delegate.chat(&effective_input).await?;

        let duration_ms = start.elapsed().as_millis() as u64;
        self.hooks
            .on_delegate_complete(delegate_id, &state_name, duration_ms)
            .await;

        // Backward-compatible context key.
        let ctx_key = format!("delegation.{}.last_response", delegate_id);
        let _ = self.context_manager.set(
            &ctx_key,
            serde_json::Value::String(response.content.clone()),
        );

        // Structured orchestration context.
        let _ = self.context_manager.set(
            "orchestration",
            serde_json::json!({
                "type": "delegate",
                "agent": delegate_id,
                "state": state_name,
                "response": response.content,
                "duration_ms": duration_ms,
            }),
        );

        // Add user message to parent memory for tracking (skip on redispatch - already present).
        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ai_agents_llm::ChatMessage::user(input))
                .await?;
        }

        // post_loop_processing records the assistant turn and evaluates transitions.
        // apply_post_loop_result handles NeedsRedispatch by re-entering run_loop_internal.
        let post_result = self
            .post_loop_processing(
                input,
                format!("[Delegated to {}]: {}", delegate_id, response.content),
            )
            .await?;
        let final_content = self.apply_post_loop_result(input, post_result).await?;

        let mut result = AgentResponse::new(final_content);

        let metadata = serde_json::json!({
            "orchestration": {
                "type": "delegate",
                "agent": delegate_id,
                "state": state_name,
                "response": response.content,
                "duration_ms": duration_ms,
            }
        });
        result.metadata = Some(
            serde_json::from_value::<std::collections::HashMap<String, serde_json::Value>>(
                metadata,
            )
            .unwrap_or_default(),
        );

        self.hooks.on_response(&result).await;
        Ok(result)
    }

    // Handle concurrent execution: run multiple registry agents in parallel and aggregate.
    async fn handle_concurrent_state(
        &self,
        input: &str,
        config: &ai_agents_state::ConcurrentStateConfig,
    ) -> Result<AgentResponse> {
        use std::time::Instant;

        let registry = self.spawner_registry.as_ref().ok_or_else(|| {
            AgentError::Config(
                "Concurrent state requires an agent registry. Add a spawner section.".into(),
            )
        })?;

        // Render input template if provided, otherwise use the raw input.
        // Uses direct minijinja rendering (same approach as pipeline) so variables
        // are top-level: {{ user_input }}, not {{ context.user_input }}.
        // Enrich input with parent conversation history when context_mode is set.
        let context_mode = config.context_mode.clone().unwrap_or_default();
        let context_input = crate::orchestration::context::prepare_delegate_input(
            input,
            &context_mode,
            &*self.memory,
            self.llm_registry.get("router").ok().as_deref(),
        )
        .await?;

        let effective_input = if let Some(ref tmpl) = config.input {
            render_concurrent_template(tmpl, &context_input, &self.context_manager.get_all())
                .unwrap_or_else(|_| context_input.clone())
        } else {
            context_input
        };

        let start = Instant::now();

        let llm_name = config
            .aggregation
            .synthesizer_llm
            .as_deref()
            .unwrap_or("router");
        let llm_provider = self.llm_registry.get(llm_name).ok();

        let result = crate::orchestration::concurrent(
            registry,
            &effective_input,
            &config.agents,
            &config.aggregation,
            llm_provider.as_deref(),
            config.min_required,
            config.timeout_ms,
            config.on_partial_failure.clone(),
        )
        .await?;

        let duration_ms = start.elapsed().as_millis() as u64;
        let agent_ids: Vec<String> = config.agents.iter().map(|a| a.id().to_string()).collect();
        let strategy = format!("{:?}", config.aggregation.strategy);
        self.hooks
            .on_concurrent_complete(&agent_ids, &strategy, duration_ms)
            .await;

        // Backward-compatible context key.
        let _ = self.context_manager.set(
            "concurrent.result",
            serde_json::Value::String(result.response.content.clone()),
        );

        // Build per-agent result data for context and metadata.
        let agents_json: Vec<serde_json::Value> = result
            .agent_results
            .iter()
            .map(|ar| {
                serde_json::json!({
                    "id": ar.agent_id,
                    "response": ar.response.as_ref().map(|r| r.content.as_str()),
                    "success": ar.success,
                    "error": ar.error,
                    "duration_ms": ar.duration_ms,
                })
            })
            .collect();

        // Structured orchestration context with per-agent results.
        let _ = self.context_manager.set(
            "orchestration",
            serde_json::json!({
                "type": "concurrent",
                "result": result.response.content,
                "strategy": strategy,
                "agents": agents_json,
                "duration_ms": duration_ms,
            }),
        );

        // Add user message to parent memory for tracking (skip on redispatch - already present).
        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ai_agents_llm::ChatMessage::user(input))
                .await?;
        }

        let post_result = self
            .post_loop_processing(input, result.response.content.clone())
            .await?;
        let final_content = self.apply_post_loop_result(input, post_result).await?;

        let mut response = AgentResponse::new(final_content);
        let metadata = serde_json::json!({
            "orchestration": {
                "type": "concurrent",
                "result": result.response.content,
                "strategy": strategy,
                "agents": agents_json,
                "duration_ms": duration_ms,
            }
        });
        response.metadata = Some(
            serde_json::from_value::<std::collections::HashMap<String, serde_json::Value>>(
                metadata,
            )
            .unwrap_or_default(),
        );

        self.hooks.on_response(&response).await;
        Ok(response)
    }

    // Handle group chat: run a multi-turn multi-agent conversation.
    async fn handle_group_chat_state(
        &self,
        input: &str,
        config: &ai_agents_state::GroupChatStateConfig,
    ) -> Result<AgentResponse> {
        use std::time::Instant;

        let registry = self.spawner_registry.as_ref().ok_or_else(|| {
            AgentError::Config(
                "Group chat state requires an agent registry. Add a spawner section.".into(),
            )
        })?;

        let start = Instant::now();

        let llm_provider = self.llm_registry.get("router").ok();

        // Enrich input with parent conversation history when context_mode is set.
        let context_mode = config.context_mode.clone().unwrap_or_default();
        let context_input = crate::orchestration::context::prepare_delegate_input(
            input,
            &context_mode,
            &*self.memory,
            self.llm_registry.get("router").ok().as_deref(),
        )
        .await?;

        // Render input template if provided, otherwise use the raw user message as topic.
        let effective_topic = if let Some(ref tmpl) = config.input {
            render_concurrent_template(tmpl, &context_input, &self.context_manager.get_all())
                .unwrap_or_else(|_| context_input.clone())
        } else {
            context_input
        };

        let result = crate::orchestration::group_chat(
            registry,
            &effective_topic,
            config,
            llm_provider.as_deref(),
            Some(&*self.hooks),
        )
        .await?;

        let duration_ms = start.elapsed().as_millis() as u64;

        // Backward-compatible context key.
        let _ = self.context_manager.set(
            "group_chat.conclusion",
            serde_json::Value::String(result.response.content.clone()),
        );

        // Build transcript data for context and metadata.
        let transcript_json: Vec<serde_json::Value> = result
            .transcript
            .iter()
            .map(|t| {
                serde_json::json!({
                    "speaker": t.speaker,
                    "round": t.round,
                    "content": t.content,
                })
            })
            .collect();

        // Structured orchestration context with full transcript.
        let _ = self.context_manager.set(
            "orchestration",
            serde_json::json!({
                "type": "group_chat",
                "conclusion": result.response.content,
                "transcript": transcript_json,
                "rounds": result.rounds_completed,
                "termination": result.termination_reason,
                "duration_ms": duration_ms,
            }),
        );

        // Add user message to parent memory for tracking (skip on redispatch - already present).
        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ai_agents_llm::ChatMessage::user(input))
                .await?;
        }

        let post_result = self
            .post_loop_processing(input, result.response.content.clone())
            .await?;
        let final_content = self.apply_post_loop_result(input, post_result).await?;

        let mut response = AgentResponse::new(final_content);
        let metadata = serde_json::json!({
            "orchestration": {
                "type": "group_chat",
                "conclusion": result.response.content,
                "transcript": transcript_json,
                "rounds": result.rounds_completed,
                "termination": result.termination_reason,
                "duration_ms": duration_ms,
            }
        });
        response.metadata = Some(
            serde_json::from_value::<std::collections::HashMap<String, serde_json::Value>>(
                metadata,
            )
            .unwrap_or_default(),
        );

        self.hooks.on_response(&response).await;
        Ok(response)
    }

    // Handle pipeline: run agents sequentially with per-stage input templates.
    async fn handle_pipeline_state(
        &self,
        input: &str,
        config: &ai_agents_state::PipelineStateConfig,
    ) -> Result<AgentResponse> {
        use std::time::Instant;

        let registry = self.spawner_registry.as_ref().ok_or_else(|| {
            AgentError::Config(
                "Pipeline state requires an agent registry. Add a spawner section.".into(),
            )
        })?;

        let start = Instant::now();

        let stages: Vec<crate::orchestration::PipelineStage> = config
            .stages
            .iter()
            .map(|entry| {
                let mut stage = crate::orchestration::PipelineStage::id(entry.id());
                if let Some(tmpl) = entry.input() {
                    stage = stage.with_input(tmpl);
                }
                stage
            })
            .collect();

        // Enrich input with parent conversation history when context_mode is set.
        let context_mode = config.context_mode.clone().unwrap_or_default();
        let context_input = crate::orchestration::context::prepare_delegate_input(
            input,
            &context_mode,
            &*self.memory,
            self.llm_registry.get("router").ok().as_deref(),
        )
        .await?;

        let context_values = self.context_manager.get_all();
        let result = crate::orchestration::pipeline(
            registry,
            &context_input,
            &stages,
            config.timeout_ms,
            Some(&*self.hooks),
            Some(&context_values),
        )
        .await?;

        let duration_ms = start.elapsed().as_millis() as u64;

        // Backward-compatible context key.
        let _ = self.context_manager.set(
            "pipeline.result",
            serde_json::Value::String(result.response.content.clone()),
        );

        // Build per-stage data for context and metadata.
        let stages_json: Vec<serde_json::Value> = result
            .stage_outputs
            .iter()
            .map(|s| {
                serde_json::json!({
                    "agent_id": s.agent_id,
                    "output": s.output,
                    "duration_ms": s.duration_ms,
                    "skipped": s.skipped,
                })
            })
            .collect();

        // Structured orchestration context.
        let _ = self.context_manager.set(
            "orchestration",
            serde_json::json!({
                "type": "pipeline",
                "result": result.response.content,
                "stages": stages_json,
                "duration_ms": duration_ms,
            }),
        );

        // Add user message to parent memory for tracking (skip on redispatch - already present).
        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ai_agents_llm::ChatMessage::user(input))
                .await?;
        }

        let post_result = self
            .post_loop_processing(input, result.response.content.clone())
            .await?;
        let final_content = self.apply_post_loop_result(input, post_result).await?;

        let mut response = AgentResponse::new(final_content);
        let metadata = serde_json::json!({
            "orchestration": {
                "type": "pipeline",
                "result": result.response.content,
                "stages": stages_json,
                "duration_ms": duration_ms,
            }
        });
        response.metadata = Some(
            serde_json::from_value::<std::collections::HashMap<String, serde_json::Value>>(
                metadata,
            )
            .unwrap_or_default(),
        );

        self.hooks.on_response(&response).await;
        Ok(response)
    }

    // Handle handoff: LLM-directed agent-to-agent control transfer.
    async fn handle_handoff_state(
        &self,
        input: &str,
        config: &ai_agents_state::HandoffStateConfig,
    ) -> Result<AgentResponse> {
        use std::time::Instant;

        let registry = self.spawner_registry.as_ref().ok_or_else(|| {
            AgentError::Config(
                "Handoff state requires an agent registry. Add a spawner section.".into(),
            )
        })?;

        let llm = self
            .llm_registry
            .get("router")
            .map_err(|_| AgentError::Config("Handoff state requires a router LLM.".into()))?;

        let start = Instant::now();

        // Enrich input with parent conversation history when context_mode is set.
        let context_mode = config.context_mode.clone().unwrap_or_default();
        let context_input = crate::orchestration::context::prepare_delegate_input(
            input,
            &context_mode,
            &*self.memory,
            self.llm_registry.get("router").ok().as_deref(),
        )
        .await?;

        // Render input template if provided, otherwise forward the raw user message.
        let effective_input = if let Some(ref tmpl) = config.input {
            render_concurrent_template(tmpl, &context_input, &self.context_manager.get_all())
                .unwrap_or_else(|_| context_input.clone())
        } else {
            context_input
        };

        let result = crate::orchestration::handoff(
            registry,
            &effective_input,
            &config.initial_agent,
            &config.available_agents,
            config.max_handoffs,
            llm.as_ref(),
            Some(&*self.hooks),
        )
        .await?;

        let duration_ms = start.elapsed().as_millis() as u64;

        // Backward-compatible context key.
        let _ = self.context_manager.set(
            "handoff.result",
            serde_json::Value::String(result.response.content.clone()),
        );

        // Build handoff chain data for context and metadata.
        let chain_json: Vec<serde_json::Value> = result
            .handoff_chain
            .iter()
            .map(|h| {
                serde_json::json!({
                    "from": h.from_agent,
                    "to": h.to_agent,
                    "reason": h.reason,
                })
            })
            .collect();

        // Structured orchestration context.
        let _ = self.context_manager.set(
            "orchestration",
            serde_json::json!({
                "type": "handoff",
                "result": result.response.content,
                "final_agent": result.final_agent,
                "handoff_chain": chain_json,
                "duration_ms": duration_ms,
            }),
        );

        // Add user message to parent memory for tracking (skip on redispatch - already present).
        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ai_agents_llm::ChatMessage::user(input))
                .await?;
        }

        let post_result = self
            .post_loop_processing(input, result.response.content.clone())
            .await?;
        let final_content = self.apply_post_loop_result(input, post_result).await?;

        let mut response = AgentResponse::new(final_content);
        let metadata = serde_json::json!({
            "orchestration": {
                "type": "handoff",
                "result": result.response.content,
                "final_agent": result.final_agent,
                "handoff_chain": chain_json,
                "duration_ms": duration_ms,
            }
        });
        response.metadata = Some(
            serde_json::from_value::<std::collections::HashMap<String, serde_json::Value>>(
                metadata,
            )
            .unwrap_or_default(),
        );

        self.hooks.on_response(&response).await;
        Ok(response)
    }

    // run_loop_internal: blocking (non-streaming) agent pipeline.
    async fn run_loop_internal(&self, input: &str) -> Result<AgentResponse> {
        let input_data = self.process_input(input).await?;

        // Inject process context (detect/extract results) into agent context
        // so system prompt templates can use {{ context.detected_language }} etc.
        for (key, value) in &input_data.context {
            let _ = self.context_manager.set(key, value.clone());
        }

        if input_data.metadata.rejected {
            let reason = input_data
                .metadata
                .rejection_reason
                .unwrap_or_else(|| "Input rejected".to_string());
            warn!(reason = %reason, "Input rejected");
            return Ok(AgentResponse::new(reason));
        }

        let processed_input = &input_data.content;

        // Handle orchestration states (delegate, concurrent, group_chat, pipeline, handoff).
        if let Some(ref sm) = self.state_machine {
            if let Some(def) = sm.current_definition() {
                if let Some(ref delegate_id) = def.delegate {
                    return self
                        .handle_delegated_state(processed_input, delegate_id, &def)
                        .await;
                }
                if let Some(ref concurrent_config) = def.concurrent {
                    return self
                        .handle_concurrent_state(processed_input, concurrent_config)
                        .await;
                }
                if let Some(ref group_chat_config) = def.group_chat {
                    return self
                        .handle_group_chat_state(processed_input, group_chat_config)
                        .await;
                }
                if let Some(ref pipeline_config) = def.pipeline {
                    return self
                        .handle_pipeline_state(processed_input, pipeline_config)
                        .await;
                }
                if let Some(ref handoff_config) = def.handoff {
                    return self
                        .handle_handoff_state(processed_input, handoff_config)
                        .await;
                }
            }
        }

        match self.try_skill_route(processed_input).await? {
            SkillRouteResult::Response(skill_response) => {
                if *self.redispatch_depth.read() == 0 {
                    self.memory
                        .add_message(ChatMessage::user(processed_input))
                        .await?;
                }
                return self
                    .handle_skill_response(processed_input, skill_response, &input_data.context)
                    .await;
            }
            SkillRouteResult::NeedsClarification(response) => {
                if *self.redispatch_depth.read() == 0 {
                    self.memory
                        .add_message(ChatMessage::user(processed_input))
                        .await?;
                }
                if let Some(q) = response
                    .metadata
                    .as_ref()
                    .and_then(|m| m.get("disambiguation"))
                    .and_then(|d| d.get("status"))
                    .and_then(|s| s.as_str())
                {
                    if q == "awaiting_clarification" {
                        // Store the clarification question in memory so the next turn
                        // can be handled as a clarification response.
                        self.memory
                            .add_message(ChatMessage::assistant(&response.content))
                            .await?;
                    }
                }
                return Ok(response);
            }
            SkillRouteResult::NoMatch => {} // continue to normal LLM chat
        }

        let effective_reasoning = self.get_effective_reasoning_config();
        let reasoning_mode = self.determine_reasoning_mode(processed_input).await?;
        let auto_detected = matches!(effective_reasoning.mode, ReasoningMode::Auto);

        info!(
            reasoning_mode = ?reasoning_mode,
            auto_detected = auto_detected,
            reflection_enabled = ?self.reflection_config.enabled,
            "Reasoning mode determined"
        );

        if matches!(reasoning_mode, ReasoningMode::PlanAndExecute) {
            if *self.redispatch_depth.read() == 0 {
                self.memory
                    .add_message(ChatMessage::user(processed_input))
                    .await?;
            }
            return self
                .handle_plan_and_execute(processed_input, &input_data.context, auto_detected)
                .await;
        }

        if *self.redispatch_depth.read() == 0 {
            self.memory
                .add_message(ChatMessage::user(processed_input))
                .await?;
        }

        let mut iterations = 0u32;
        let mut all_tool_calls: Vec<ToolCall> = Vec::new();
        let mut thinking_content: Option<String> = None;

        let llm = self.get_state_llm()?;

        loop {
            // When reasoning is active, cap iterations at the reasoning-specific limit.
            let effective_max = if reasoning_mode != ReasoningMode::None {
                let rc = self.get_effective_reasoning_config();
                self.max_iterations.min(rc.max_iterations)
            } else {
                self.max_iterations
            };

            if iterations >= effective_max {
                let err = AgentError::Other(format!("Max iterations ({}) exceeded", effective_max));
                self.hooks.on_error(&err).await;
                error!(iterations = iterations, "Max iterations exceeded");
                return Err(err);
            }
            iterations += 1;
            *self.iteration_count.write() = iterations;

            debug!(iteration = iterations, max = effective_max, "LLM call");

            let mut messages = self.build_messages().await?;
            self.inject_reasoning_prompt(&mut messages, &reasoning_mode, iterations == 1);

            self.hooks.on_llm_start(&messages).await;
            let llm_start = Instant::now();

            // Try primary LLM (with retry if configured), then apply on_failure policy.
            let response = {
                let primary_result = if self.recovery_manager.config().default.max_retries > 0 {
                    self.recovery_manager
                        .with_retry("llm_call", None, || async {
                            llm.complete(&messages, None)
                                .await
                                .map_err(|e| e.classify())
                        })
                        .await
                        .map_err(|e| AgentError::LLM(e.to_string()))
                } else {
                    llm.complete(&messages, None)
                        .await
                        .map_err(|e| AgentError::LLM(e.to_string()))
                };

                match primary_result {
                    Ok(resp) => resp,
                    Err(primary_err) => match &self.recovery_manager.config().llm.on_failure {
                        LLMFailureAction::FallbackLlm { fallback_llm } => {
                            warn!(
                                fallback = %fallback_llm,
                                error = %primary_err,
                                "Primary LLM failed, attempting fallback LLM"
                            );
                            let fb = self.llm_registry.get(fallback_llm).map_err(|e| {
                                AgentError::Config(format!(
                                    "Fallback LLM '{}' not found: {}",
                                    fallback_llm, e
                                ))
                            })?;
                            fb.complete(&messages, None)
                                .await
                                .map_err(|e| AgentError::LLM(e.to_string()))?
                        }
                        LLMFailureAction::FallbackResponse { message } => {
                            warn!(
                                error = %primary_err,
                                "Primary LLM failed, using static fallback response"
                            );
                            LLMResponse::new(message.clone(), FinishReason::Stop)
                        }
                        LLMFailureAction::Error => {
                            return Err(primary_err);
                        }
                    },
                }
            };

            let llm_duration_ms = llm_start.elapsed().as_millis() as u64;
            self.hooks.on_llm_complete(&response, llm_duration_ms).await;

            let content = response.content.trim();

            if let Some(tool_calls) = self.parse_tool_calls(content) {
                match self
                    .handle_tool_calls(processed_input, content, tool_calls, &mut all_tool_calls)
                    .await?
                {
                    ToolCallOutcome::Continue | ToolCallOutcome::TransitionFired => continue,
                    ToolCallOutcome::Rejected(resp) => return Ok(resp),
                }
            }

            let (extracted_thinking, answer) = self.extract_thinking(content);
            if extracted_thinking.is_some() {
                thinking_content = extracted_thinking;
            }

            let output_data = self.process_output(&answer, &input_data.context).await?;

            let mut final_content = if output_data.metadata.rejected {
                output_data
                    .metadata
                    .rejection_reason
                    .unwrap_or_else(|| answer.to_string())
            } else {
                output_data.content
            };

            // Run reflection (blocking LLM calls for retries)
            let reflection_metadata;
            (final_content, reflection_metadata) = self
                .run_reflection(&*llm, processed_input, final_content)
                .await?;

            final_content =
                self.format_response_with_thinking(thinking_content.as_deref(), &final_content);

            // Post-loop: memory, transitions, post-transition re-generation.
            // apply_post_loop_result handles NeedsRedispatch by re-entering
            // run_loop_internal so the new state's full dispatch activates.
            let final_content = {
                let result = self
                    .post_loop_processing(processed_input, final_content)
                    .await?;
                self.apply_post_loop_result(processed_input, result).await?
            };

            let reflected = reflection_metadata.is_some();
            let reasoning_mode_debug = format!("{:?}", reasoning_mode);

            let response = self.build_agent_response(
                final_content,
                all_tool_calls,
                reasoning_mode,
                auto_detected,
                iterations,
                thinking_content,
                reflection_metadata,
            );

            self.hooks.on_response(&response).await;

            let tool_call_count = response.tool_calls.as_ref().map(|tc| tc.len()).unwrap_or(0);
            info!(
                tool_calls = tool_call_count,
                response_len = response.content.len(),
                reasoning_mode = %reasoning_mode_debug,
                reflected = reflected,
                "Chat completed"
            );
            return Ok(response);
        }
    }

    /// Streaming agent pipeline
    /// Uses all the same shared helpers as run_loop_internal.
    /// The ONLY difference: LLM calls use complete_stream() + yield deltas.
    fn run_loop_internal_stream<'a>(
        &'a self,
        input: &'a str,
    ) -> Pin<Box<dyn Stream<Item = StreamChunk> + Send + 'a>> {
        let include_tool_events = self.streaming.include_tool_events;
        let include_state_events = self.streaming.include_state_events;

        Box::pin(async_stream::stream! {
            let input_data = match self.process_input(input).await {
                Ok(data) => data,
                Err(e) => {
                    yield StreamChunk::error(e.to_string());
                    return;
                }
            };

            // Inject process context (detect/extract results) into agent context
            for (key, value) in &input_data.context {
                let _ = self.context_manager.set(key, value.clone());
            }

            if input_data.metadata.rejected {
                let reason = input_data
                    .metadata
                    .rejection_reason
                    .unwrap_or_else(|| "Input rejected".to_string());
                warn!(reason = %reason, "Input rejected (stream)");
                yield StreamChunk::error(reason);
                return;
            }

            let processed_input = &input_data.content;

            // Handle orchestration states in streaming mode.
            if let Some(ref sm) = self.state_machine {
                if let Some(def) = sm.current_definition() {
                    let orchestration_result = if let Some(ref delegate_id) = def.delegate {
                        Some(self.handle_delegated_state(processed_input, delegate_id, &def).await)
                    } else if let Some(ref concurrent_config) = def.concurrent {
                        Some(self.handle_concurrent_state(processed_input, concurrent_config).await)
                    } else if let Some(ref group_chat_config) = def.group_chat {
                        Some(self.handle_group_chat_state(processed_input, group_chat_config).await)
                    } else if let Some(ref pipeline_config) = def.pipeline {
                        Some(self.handle_pipeline_state(processed_input, pipeline_config).await)
                    } else if let Some(ref handoff_config) = def.handoff {
                        Some(self.handle_handoff_state(processed_input, handoff_config).await)
                    } else {
                        None
                    };

                    if let Some(result) = orchestration_result {
                        match result {
                            Ok(response) => {
                                yield StreamChunk::content(&response.content);
                                yield StreamChunk::Done {};
                            }
                            Err(e) => {
                                yield StreamChunk::error(e.to_string());
                            }
                        }
                        return;
                    }
                }
            }

            // Skill routing
            match self.try_skill_route(processed_input).await {
                Ok(SkillRouteResult::Response(skill_response)) => {
                    if *self.redispatch_depth.read() == 0 {
                        let _ = self.memory.add_message(ChatMessage::user(processed_input)).await;
                    }
                    match self.handle_skill_response(processed_input, skill_response, &input_data.context).await {
                        Ok(resp) => {
                            yield StreamChunk::content(&resp.content);
                            yield StreamChunk::Done {};
                            return;
                        }
                        Err(e) => {
                            yield StreamChunk::error(e.to_string());
                            return;
                        }
                    }
                }
                Ok(SkillRouteResult::NeedsClarification(response)) => {
                    if *self.redispatch_depth.read() == 0 {
                        let _ = self.memory.add_message(ChatMessage::user(processed_input)).await;
                    }
                    let _ = self.memory.add_message(ChatMessage::assistant(&response.content)).await;
                    yield StreamChunk::content(&response.content);
                    yield StreamChunk::Done {};
                    return;
                }
                Ok(SkillRouteResult::NoMatch) => {} // no skill matched, continue
                Err(e) => {
                    yield StreamChunk::error(e.to_string());
                    return;
                }
            }

            // Reasoning mode determination
            let effective_reasoning = self.get_effective_reasoning_config();
            let reasoning_mode = match self.determine_reasoning_mode(processed_input).await {
                Ok(mode) => mode,
                Err(e) => {
                    yield StreamChunk::error(e.to_string());
                    return;
                }
            };
            let auto_detected = matches!(effective_reasoning.mode, ReasoningMode::Auto);

            info!(
                reasoning_mode = ?reasoning_mode,
                auto_detected = auto_detected,
                "Reasoning mode determined (stream)"
            );

            // Plan-and-Execute: yield final result as single chunk (not token-by-token)
            if matches!(reasoning_mode, ReasoningMode::PlanAndExecute) {
                if *self.redispatch_depth.read() == 0 {
                    let _ = self.memory.add_message(ChatMessage::user(processed_input)).await;
                }
                match self.handle_plan_and_execute(processed_input, &input_data.context, auto_detected).await {
                    Ok(resp) => {
                        yield StreamChunk::content(&resp.content);
                        yield StreamChunk::Done {};
                        return;
                    }
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                }
            }

            if *self.redispatch_depth.read() == 0 {
                let _ = self.memory.add_message(ChatMessage::user(processed_input)).await;
            }

            let llm = match self.get_state_llm() {
                Ok(llm) => llm,
                Err(e) => {
                    yield StreamChunk::error(e.to_string());
                    return;
                }
            };

            let mut iterations = 0u32;
            let mut all_tool_calls: Vec<ToolCall> = Vec::new();
            let mut thinking_content: Option<String> = None;

            loop {
                // When reasoning is active, cap iterations at the reasoning-specific limit.
                let effective_max = if reasoning_mode != ReasoningMode::None {
                    let rc = self.get_effective_reasoning_config();
                    self.max_iterations.min(rc.max_iterations)
                } else {
                    self.max_iterations
                };

                if iterations >= effective_max {
                    let err_msg = format!("Max iterations ({}) exceeded", effective_max);
                    let err = AgentError::Other(err_msg.clone());
                    self.hooks.on_error(&err).await;
                    error!(iterations = iterations, "Max iterations exceeded (stream)");
                    yield StreamChunk::error(err_msg);
                    return;
                }
                iterations += 1;
                *self.iteration_count.write() = iterations;

                debug!(iteration = iterations, max = effective_max, "LLM call (stream)");

                let mut messages = match self.build_messages().await {
                    Ok(m) => m,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };
                self.inject_reasoning_prompt(&mut messages, &reasoning_mode, iterations == 1);

                self.hooks.on_llm_start(&messages).await;
                let llm_start = Instant::now();

                // Check if reflection is active — if so, suppress streaming for this LLM call
                // because we may need to retry and the user would see a stale first attempt.
                let reflection_active = match self.should_reflect(processed_input, "").await {
                    Ok(v) => v,
                    Err(_) => false,
                };

                let content = if reflection_active {
                    // Use blocking call so reflection can retry without partial output
                    let response = match llm.complete(&messages, None).await {
                        Ok(r) => r,
                        Err(e) => {
                            yield StreamChunk::error(e.to_string());
                            return;
                        }
                    };
                    let llm_duration_ms = llm_start.elapsed().as_millis() as u64;
                    self.hooks.on_llm_complete(&response, llm_duration_ms).await;
                    response.content.trim().to_string()
                } else {
                    // Streaming LLM call
                    let llm_stream = match llm.complete_stream(&messages, None).await {
                        Ok(s) => s,
                        Err(e) => {
                            yield StreamChunk::error(e.to_string());
                            return;
                        }
                    };
                    let mut accumulated = String::new();
                    let mut stream_inner = llm_stream;
                    while let Some(chunk_result) = stream_inner.next().await {
                        match chunk_result {
                            Ok(chunk) => {
                                accumulated.push_str(&chunk.delta);
                                yield StreamChunk::content(chunk.delta);
                            }
                            Err(e) => {
                                yield StreamChunk::error(e.to_string());
                                return;
                            }
                        }
                    }
                    let llm_duration_ms = llm_start.elapsed().as_millis() as u64;
                    // Construct LLMResponse for hooks
                    let llm_response = ai_agents_core::LLMResponse::new(
                        accumulated.trim(),
                        ai_agents_core::FinishReason::Stop,
                    );
                    self.hooks.on_llm_complete(&llm_response, llm_duration_ms).await;
                    accumulated.trim().to_string()
                };

                // Tool call handling
                if let Some(tool_calls) = self.parse_tool_calls(&content) {
                    // Emit tool events for streaming
                    // First check transitions (same as blocking path)
                    let transition_fired = match self.evaluate_transitions(processed_input, &content).await {
                        Ok(v) => v,
                        Err(e) => {
                            yield StreamChunk::error(e.to_string());
                            return;
                        }
                    };
                    if transition_fired {
                        let _ = self.memory.add_message(ChatMessage::assistant(
                            "(Transitioned to new state — tool call handled by workflow)",
                        )).await;

                        if include_state_events {
                            if let Some(state) = self.current_state() {
                                yield StreamChunk::state_transition(None, state);
                            }
                        }
                        continue;
                    }

                    // Store the assistant's tool-call message (same as blocking path)
                    let _ = self.memory.add_message(ChatMessage::assistant(&content)).await;

                    // Execute tools with streaming events
                    let results = self.execute_tools_parallel(&tool_calls).await;

                    for ((_id, result), tool_call) in results.into_iter().zip(tool_calls.iter()) {
                        if include_tool_events {
                            yield StreamChunk::tool_start(&tool_call.id, &tool_call.name);
                        }

                        match result {
                            Ok(output) => {
                                if include_tool_events {
                                    yield StreamChunk::tool_result(
                                        &tool_call.id,
                                        &tool_call.name,
                                        &output,
                                        true,
                                    );
                                }
                                let _ = self.memory
                                    .add_message(ChatMessage::function(&tool_call.name, &output))
                                    .await;
                            }
                            Err(e) => {
                                if matches!(e, AgentError::HITLRejected(_)) {
                                    let _ = self.memory.add_message(ChatMessage::assistant(
                                        &format!("The operation was rejected by the approver: {}", e),
                                    )).await;
                                    yield StreamChunk::error(format!("Operation cancelled: {}", e));
                                    yield StreamChunk::Done {};
                                    return;
                                }
                                if include_tool_events {
                                    yield StreamChunk::tool_result(
                                        &tool_call.id,
                                        &tool_call.name,
                                        &e.to_string(),
                                        false,
                                    );
                                }
                                let _ = self.memory
                                    .add_message(ChatMessage::function(
                                        &tool_call.name,
                                        &format!("Error: {}", e),
                                    ))
                                    .await;
                            }
                        }
                        all_tool_calls.push(tool_call.clone());

                        if include_tool_events {
                            yield StreamChunk::tool_end(&tool_call.id);
                        }
                    }
                    continue;
                }

                // Extract thinking, process output
                let (extracted_thinking, answer) = self.extract_thinking(&content);
                if extracted_thinking.is_some() {
                    thinking_content = extracted_thinking;
                }

                let output_data = match self.process_output(&answer, &input_data.context).await {
                    Ok(d) => d,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };

                let final_content = if output_data.metadata.rejected {
                    output_data
                        .metadata
                        .rejection_reason
                        .unwrap_or_else(|| answer.to_string())
                } else {
                    output_data.content
                };

                // Reflection (uses blocking LLM calls for retries)
                let (final_content, _reflection_metadata) = match self
                    .run_reflection(&*llm, processed_input, final_content)
                    .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };

                let final_content = self.format_response_with_thinking(
                    thinking_content.as_deref(),
                    &final_content,
                );

                // If reflection was active (we used blocking call), yield the final content now
                if reflection_active {
                    yield StreamChunk::content(&final_content);
                }

                // Post-loop: memory, transitions, post-transition re-generation.
                // For NeedsRedispatch, run_loop_internal handles the new state's full
                // dispatch and its result is yielded as a single non-streamed chunk.
                let post_result = match self
                    .post_loop_processing(processed_input, final_content)
                    .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };

                let (final_content, transitioned) = match post_result {
                    PostLoopResult::NoTransition(content) => (content, false),
                    PostLoopResult::Transitioned(content) => (content, true),
                    PostLoopResult::NeedsRedispatch => {
                        const MAX_REDISPATCH_DEPTH: u32 = 3;
                        let current_depth = *self.redispatch_depth.read();
                        let content = if current_depth >= MAX_REDISPATCH_DEPTH {
                            warn!(
                                depth = current_depth,
                                "Post-transition re-dispatch depth limit reached (stream)"
                            );
                            let c = String::new();
                            let _ = self.memory.add_message(ChatMessage::assistant(&c)).await;
                            c
                        } else {
                            *self.redispatch_depth.write() += 1;
                            info!(
                                depth = current_depth + 1,
                                "Re-dispatching for new state after transition (stream)"
                            );
                            let result = self.run_loop_internal(processed_input).await;
                            *self.redispatch_depth.write() -= 1;
                            match result {
                                Ok(resp) => resp.content,
                                Err(e) => {
                                    yield StreamChunk::error(e.to_string());
                                    return;
                                }
                            }
                        };
                        (content, true)
                    }
                };

                if transitioned {
                    if include_state_events {
                        if let Some(state) = self.current_state() {
                            yield StreamChunk::state_transition(None, state);
                        }
                    }
                    // Yield the post-transition re-generated or re-dispatched content.
                    yield StreamChunk::content(&final_content);
                }

                yield StreamChunk::Done {};
                return;
            }
        })
    }

    /// Streaming entry point with disambiguation
    /// Mirrors run_loop but yields StreamChunks instead of AgentResponse.
    fn run_loop_stream<'a>(
        &'a self,
        input: &'a str,
    ) -> Pin<Box<dyn Stream<Item = StreamChunk> + Send + 'a>> {
        Box::pin(async_stream::stream! {
            self.hooks.on_message_received(input).await;

            // One-shot context initialization (mirrors run_loop)
            if !self.context_initialized.swap(true, Ordering::SeqCst) {
                if let Err(e) = self.context_manager.initialize().await {
                    yield StreamChunk::error(e.to_string());
                    return;
                }
                debug!("Context manager initialized (defaults, env, builtins)");
            }

            if let Err(e) = self.check_turn_timeout().await {
                yield StreamChunk::error(e.to_string());
                return;
            }
            if let Err(e) = self.context_manager.refresh_per_turn().await {
                yield StreamChunk::error(e.to_string());
                return;
            }

            // Clear stale disambiguation context from previous turns.
            self.clear_disambiguation_context();

            // Disambiguation check (before input processing)
            if let Some(ref disambiguator) = self.disambiguation_manager {
                let disambiguation_context = match self.build_disambiguation_context().await {
                    Ok(ctx) => ctx,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };

                let state_override = self
                    .state_machine
                    .as_ref()
                    .and_then(|sm| sm.current_definition())
                    .and_then(|def| def.disambiguation.clone());

                let result = match disambiguator
                    .process_input_with_override(
                        input,
                        &disambiguation_context,
                        state_override.as_ref(),
                        None,
                    )
                    .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        yield StreamChunk::error(e.to_string());
                        return;
                    }
                };

                match result {
                    DisambiguationResult::Clear => {
                        debug!("Input is clear, proceeding normally (stream)");
                    }
                    DisambiguationResult::NeedsClarification {
                        question,
                        detection,
                    } => {
                        info!(
                            ambiguity_type = ?detection.ambiguity_type,
                            confidence = detection.confidence,
                            "Input requires clarification (stream)"
                        );
                        let _ = self.memory.add_message(ChatMessage::user(input)).await;
                        let _ = self
                            .memory
                            .add_message(ChatMessage::assistant(&question.question))
                            .await;
                        yield StreamChunk::content(&question.question);
                        yield StreamChunk::Done {};
                        return;
                    }
                    DisambiguationResult::Clarified {
                        enriched_input,
                        resolved,
                        ..
                    } => {
                        info!(
                            resolved_count = resolved.len(),
                            enriched = %enriched_input,
                            "Input clarified (stream)"
                        );
                        for (key, value) in &resolved {
                            let context_key = format!("disambiguation.{}", key);
                            let _ = self.context_manager.set(&context_key, value.clone());
                        }
                        if let Some(intent) = resolved.get("intent") {
                            let _ = self.context_manager.set("resolved_intent", intent.clone());
                        }
                        let _ = self
                            .context_manager
                            .set("disambiguation.resolved", serde_json::Value::Bool(true));

                        // Check if this clarification was triggered by a skill-level override.
                        // Re-run skill disambiguation to verify all required_clarity fields
                        // are present before executing.
                        let skill_id = self.pending_skill_id.read().clone();
                        if let Some(skill_id) = skill_id {
                            info!(skill_id = %skill_id, "Re-checking skill disambiguation on clarified input (stream)");
                            match self.recheck_skill_disambiguation(&skill_id, &enriched_input).await {
                                Ok(resp) => {
                                    yield StreamChunk::content(&resp.content);
                                    yield StreamChunk::Done {};
                                    return;
                                }
                                Err(e) => {
                                    yield StreamChunk::error(e.to_string());
                                    return;
                                }
                            }
                        }

                        // Forward to internal stream with enriched input
                        let mut inner = self.run_loop_internal_stream(&enriched_input);
                        while let Some(chunk) = inner.next().await {
                            yield chunk;
                        }
                        return;
                    }
                    DisambiguationResult::ProceedWithBestGuess { enriched_input } => {
                        info!("Proceeding with best guess (stream)");

                        // Same skill-id re-check for best-guess path
                        let skill_id = self.pending_skill_id.read().clone();
                        if let Some(skill_id) = skill_id {
                            info!(skill_id = %skill_id, "Re-checking skill disambiguation on best-guess input (stream)");
                            match self.recheck_skill_disambiguation(&skill_id, &enriched_input).await {
                                Ok(resp) => {
                                    yield StreamChunk::content(&resp.content);
                                    yield StreamChunk::Done {};
                                    return;
                                }
                                Err(e) => {
                                    yield StreamChunk::error(e.to_string());
                                    return;
                                }
                            }
                        }

                        let mut inner = self.run_loop_internal_stream(&enriched_input);
                        while let Some(chunk) = inner.next().await {
                            yield chunk;
                        }
                        return;
                    }
                    DisambiguationResult::GiveUp { reason } => {
                        *self.pending_skill_id.write() = None;
                        warn!(reason = %reason, "Disambiguation gave up (stream)");
                        let apology = self
                            .generate_localized_apology(
                                "Generate a brief, polite apology saying you couldn't understand the request. Be concise.",
                                &reason,
                            )
                            .await
                            .unwrap_or_else(|_| {
                                format!("I'm sorry, I couldn't understand your request: {}", reason)
                            });
                        yield StreamChunk::content(&apology);
                        yield StreamChunk::Done {};
                        return;
                    }
                    DisambiguationResult::Escalate { reason } => {
                        *self.pending_skill_id.write() = None;
                        info!(reason = %reason, "Escalating to human (stream)");
                        if let Some(ref hitl) = self.hitl_engine {
                            let trigger =
                                ApprovalTrigger::condition("disambiguation_escalation", reason.clone());
                            let mut context_map = HashMap::new();
                            context_map.insert("original_input".to_string(), serde_json::json!(input));
                            context_map.insert("reason".to_string(), serde_json::json!(&reason));
                            let check_result = HITLCheckResult::required(
                                trigger,
                                context_map,
                                format!("User request needs human assistance: {}", reason),
                                Some(hitl.config().default_timeout_seconds),
                            );
                            match self.request_hitl_approval(check_result).await {
                                Ok(result) if matches!(result, ApprovalResult::Approved | ApprovalResult::Modified { .. }) => {
                                    let mut inner = self.run_loop_internal_stream(input);
                                    while let Some(chunk) = inner.next().await {
                                        yield chunk;
                                    }
                                    return;
                                }
                                Ok(_) => {}
                                Err(e) => {
                                    yield StreamChunk::error(e.to_string());
                                    return;
                                }
                            }
                        }
                        let apology = self
                            .generate_localized_apology(
                                "Explain briefly that you're transferring the user to a human agent for help.",
                                &reason,
                            )
                            .await
                            .unwrap_or_else(|_| {
                                format!("I need human assistance to help with your request: {}", reason)
                            });
                        yield StreamChunk::content(&apology);
                        yield StreamChunk::Done {};
                        return;
                    }
                    DisambiguationResult::Abandoned { new_input } => {
                        *self.pending_skill_id.write() = None;

                        info!(
                            has_new_input = new_input.is_some(),
                            "Clarification abandoned by user (stream)"
                        );

                        let _ = self.memory.add_message(ChatMessage::user(input)).await;

                        match new_input {
                            Some(fresh_input) => {
                                // Topic switch: forward to internal stream with fresh input.
                                let mut inner = self.run_loop_internal_stream(&fresh_input);
                                while let Some(chunk) = inner.next().await {
                                    yield chunk;
                                }
                                return;
                            }
                            None => {
                                // Pure abandonment: generate a brief acknowledgment.
                                let ack = self
                                    .generate_localized_apology(
                                        "The user changed their mind about their previous request. \
                                         Generate a brief, friendly acknowledgment (e.g. 'OK, no problem. What else can I help with?'). \
                                         Do NOT apologize excessively. Be concise.",
                                        "User abandoned clarification",
                                    )
                                    .await
                                    .unwrap_or_else(|_| {
                                        "OK, no problem. What else can I help with?".to_string()
                                    });

                                let _ = self
                                    .memory
                                    .add_message(ChatMessage::assistant(&ack))
                                    .await;

                                yield StreamChunk::content(&ack);
                                yield StreamChunk::Done {};
                                return;
                            }
                        }
                    }
                }
            }

            // No disambiguation or Clear result — proceed with internal stream
            let mut inner = self.run_loop_internal_stream(input);
            while let Some(chunk) = inner.next().await {
                yield chunk;
            }
        })
    }

    pub fn info(&self) -> AgentInfo {
        self.info.clone()
    }

    pub fn skills(&self) -> &[SkillDefinition] {
        &self.skills
    }

    pub async fn reset(&self) -> Result<()> {
        self.memory.clear().await?;
        *self.iteration_count.write() = 0;
        self.tool_call_history.write().clear();
        *self.pending_skill_id.write() = None;
        if let Some(ref sm) = self.state_machine {
            sm.reset();
        }
        Ok(())
    }

    pub fn max_context_tokens(&self) -> u32 {
        self.max_context_tokens
    }

    pub fn llm_registry(&self) -> &Arc<LLMRegistry> {
        &self.llm_registry
    }

    pub fn state_machine(&self) -> Option<&Arc<StateMachine>> {
        self.state_machine.as_ref()
    }

    pub fn context_manager(&self) -> &Arc<ContextManager> {
        &self.context_manager
    }

    pub fn tool_call_history(&self) -> Vec<ToolCallRecord> {
        self.tool_call_history.read().clone()
    }

    pub fn memory_token_budget(&self) -> Option<&MemoryTokenBudget> {
        self.memory_token_budget.as_ref()
    }

    pub fn parallel_tools_config(&self) -> &ParallelToolsConfig {
        &self.parallel_tools
    }

    pub fn streaming_config(&self) -> &StreamingConfig {
        &self.streaming
    }

    pub fn hooks(&self) -> &Arc<dyn AgentHooks> {
        &self.hooks
    }

    pub fn hitl_engine(&self) -> Option<&HITLEngine> {
        self.hitl_engine.as_ref()
    }

    pub fn approval_handler(&self) -> &Arc<dyn ApprovalHandler> {
        &self.approval_handler
    }

    /// Build a context map with language hints from context_manager for HITL message localization.
    fn build_hitl_language_context(&self) -> HashMap<String, Value> {
        let mut ctx = HashMap::new();
        for key in &["user.language", "input.detected.language", "language"] {
            if let Some(val) = self.context_manager.get(key) {
                ctx.insert(key.to_string(), val);
            }
        }
        ctx
    }

    /// Send a HITL check result through the approval flow and return the full ApprovalResult.
    async fn request_hitl_approval(&self, check_result: HITLCheckResult) -> Result<ApprovalResult> {
        let Some(request) = check_result.into_request() else {
            return Ok(ApprovalResult::Approved);
        };

        self.hooks.on_approval_requested(&request).await;

        let request_id = request.id.clone();
        let timeout = request.timeout;

        let result = if let Some(duration) = timeout {
            match tokio::time::timeout(duration, self.approval_handler.request_approval(request))
                .await
            {
                Ok(result) => result,
                Err(_) => ApprovalResult::timeout(),
            }
        } else {
            self.approval_handler.request_approval(request).await
        };

        self.hooks.on_approval_result(&request_id, &result).await;

        // Resolve timeout action
        let result = match result {
            ApprovalResult::Timeout => {
                if let Some(ref engine) = self.hitl_engine {
                    match engine.config().on_timeout {
                        TimeoutAction::Approve => ApprovalResult::Approved,
                        TimeoutAction::Reject => ApprovalResult::Rejected {
                            reason: Some("Timeout".to_string()),
                        },
                        TimeoutAction::Error => {
                            return Err(AgentError::Other("HITL approval timeout".to_string()));
                        }
                    }
                } else {
                    ApprovalResult::Rejected {
                        reason: Some("Timeout (no engine)".to_string()),
                    }
                }
            }
            other => other,
        };

        Ok(result)
    }

    pub async fn check_state_hitl(&self, from: Option<&str>, to: &str) -> Result<bool> {
        if let Some(ref hitl_engine) = self.hitl_engine {
            let hitl_lang_ctx = self.build_hitl_language_context();
            let check_result = hitl_engine
                .check_state_transition_with_localization(
                    from,
                    to,
                    &hitl_lang_ctx,
                    self.approval_handler.as_ref(),
                    Some(&self.llm_registry),
                )
                .await?;
            if check_result.is_required() {
                let result = self.request_hitl_approval(check_result).await?;
                return Ok(matches!(
                    result,
                    ApprovalResult::Approved | ApprovalResult::Modified { .. }
                ));
            }
        }
        Ok(true)
    }

    /// Execute multiple tools in parallel
    async fn execute_tools_parallel(
        &self,
        tool_calls: &[ToolCall],
    ) -> Vec<(String, Result<String>)> {
        if !self.parallel_tools.enabled || tool_calls.len() <= 1 {
            let mut results = Vec::new();
            for tc in tool_calls {
                let result = self.execute_tool_smart(tc).await;
                results.push((tc.id.clone(), result));
            }
            return results;
        }

        let chunks: Vec<_> = tool_calls
            .chunks(self.parallel_tools.max_parallel)
            .collect();

        let mut all_results = Vec::new();

        for chunk in chunks {
            let futures: Vec<_> = chunk
                .iter()
                .map(|tc| {
                    let tc = tc.clone();
                    async move {
                        let result = self.execute_tool_smart(&tc).await;
                        (tc.id.clone(), result)
                    }
                })
                .collect();

            let results = futures::future::join_all(futures).await;
            all_results.extend(results);
        }

        all_results
    }

    /// Stream a chat response with real-time updates
    pub async fn chat_stream<'a>(
        &'a self,
        input: &'a str,
    ) -> Result<Pin<Box<dyn Stream<Item = StreamChunk> + Send + 'a>>> {
        info!(input_len = input.len(), "Starting streaming chat");
        Ok(self.run_loop_stream(input))
    }
}

#[async_trait]
impl Agent for RuntimeAgent {
    async fn chat(&self, input: &str) -> Result<AgentResponse> {
        self.run_loop(input).await
    }

    fn info(&self) -> AgentInfo {
        self.info.clone()
    }

    async fn reset(&self) -> Result<()> {
        self.memory.clear().await?;
        *self.iteration_count.write() = 0;
        self.tool_call_history.write().clear();
        if let Some(ref sm) = self.state_machine {
            sm.reset();
        }
        Ok(())
    }
}

//
// Render a concurrent input template using direct minijinja.
// Same approach as pipeline's render_stage_template so variables are top-level.
//
// Available variables:
//   {{ user_input }}    - the user's actual message
//   {{ context.<key> }} - values from the context manager
//
fn render_concurrent_template(
    template: &str,
    user_input: &str,
    context_values: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<String> {
    let mut env = minijinja::Environment::new();
    env.add_template("concurrent", template)
        .map_err(|e| AgentError::Other(format!("Concurrent template parse error: {}", e)))?;

    let mut ctx = std::collections::BTreeMap::new();
    ctx.insert("user_input".to_string(), minijinja::Value::from(user_input));

    // Expose context manager values under {{ context.<key> }}.
    let context_obj = minijinja::Value::from_serialize(context_values);
    ctx.insert("context".to_string(), context_obj);

    let tmpl = env
        .get_template("concurrent")
        .map_err(|e| AgentError::Other(format!("Concurrent template error: {}", e)))?;

    tmpl.render(minijinja::Value::from_serialize(&ctx))
        .map_err(|e| AgentError::Other(format!("Concurrent template render error: {}", e)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::AgentBuilder;
    use ai_agents_llm::mock::MockLLMProvider;

    fn mock_with_response(response: &str) -> MockLLMProvider {
        let mut mock = MockLLMProvider::new("test");
        mock.set_response(response);
        mock
    }

    fn mock_with_responses(responses: Vec<&str>) -> MockLLMProvider {
        let mut mock = MockLLMProvider::new("test");
        mock.set_responses(responses.into_iter().map(String::from).collect(), true);
        mock
    }

    // Basic YAML → Build → Chat flow
    #[tokio::test]
    async fn test_integration_yaml_to_chat_basic() {
        let mock = mock_with_response("Hello! How can I help you?");
        let agent = AgentBuilder::new()
            .system_prompt("You are a test assistant.")
            .llm(Arc::new(mock))
            .build()
            .unwrap();

        let response = agent.chat("Hi").await.unwrap();
        assert!(!response.content.is_empty());
        assert_eq!(response.content, "Hello! How can I help you?");
    }

    // Multi-turn conversation
    #[tokio::test]
    async fn test_integration_multi_turn_conversation() {
        let mock = mock_with_responses(vec![
            "Hello! I'm your assistant.",
            "The weather is sunny today.",
            "Goodbye!",
        ]);
        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm(Arc::new(mock))
            .build()
            .unwrap();

        let r1 = agent.chat("Hi").await.unwrap();
        assert_eq!(r1.content, "Hello! I'm your assistant.");

        let r2 = agent.chat("What's the weather?").await.unwrap();
        assert_eq!(r2.content, "The weather is sunny today.");

        let r3 = agent.chat("Bye").await.unwrap();
        assert_eq!(r3.content, "Goodbye!");

        // Verify memory accumulated messages
        let messages = agent.memory.get_messages(None).await.unwrap();
        // 3 user + 3 assistant = 6 messages
        assert_eq!(messages.len(), 6);
    }

    // Tool execution in chat flow
    #[tokio::test]
    async fn test_integration_tool_execution() {
        // Mock LLM that returns a tool call then a final answer
        let mock = mock_with_responses(vec![
            // First response: tool call
            r#"I'll calculate that for you.
[TOOL_CALL: {"name": "calculator", "arguments": {"expression": "2+2"}}]"#,
            // After tool result: final answer
            "The answer is 4.",
        ]);
        let mut tools = ai_agents_tools::ToolRegistry::new();
        tools
            .register(Arc::new(ai_agents_tools::CalculatorTool))
            .unwrap();

        let agent = AgentBuilder::new()
            .system_prompt("You are a calculator assistant.")
            .llm(Arc::new(mock))
            .tools(tools)
            .build()
            .unwrap();

        let response = agent.chat("What is 2+2?").await.unwrap();
        // The agent should eventually produce a response
        assert!(!response.content.is_empty());
    }

    // State machine transitions
    #[tokio::test]
    async fn test_integration_state_machine_basic() {
        let yaml = r#"
name: StateAgent
system_prompt: "You are a support agent."
states:
  initial: greeting
  states:
    greeting:
      prompt: "Welcome the user warmly."
      transitions:
        - to: support
          when: "User needs help"
          auto: true
    support:
      prompt: "Help solve the user's problem."
"#;
        let mock = mock_with_responses(vec![
            "Welcome! How can I help?", // greeting response
            "1",                        // transition evaluator picks first (index 0)
            "I'll help you with that.", // support response
        ]);
        let builder = AgentBuilder::from_yaml(yaml).unwrap();
        let agent = builder.llm(Arc::new(mock)).build().unwrap();

        assert_eq!(agent.current_state(), Some("greeting".to_string()));
        let _ = agent.chat("I need help").await.unwrap();
        // After transition evaluation, state may or may not have changed
        // depending on mock evaluator response - the key is that it doesn't crash
    }

    // State on_enter/on_exit actions
    #[tokio::test]
    async fn test_integration_state_on_enter_set_context() {
        let yaml = r#"
name: ActionAgent
system_prompt: "You are helpful."
states:
  initial: step1
  states:
    step1:
      prompt: "Step 1"
      on_exit:
        - set_context:
            step1_exited: true
      transitions:
        - to: step2
          when: "always"
          auto: true
    step2:
      prompt: "Step 2"
      on_enter:
        - set_context:
            step2_entered: true
"#;
        // The transition evaluator will pick the first transition (index 0)
        let mock = mock_with_responses(vec![
            "Processing step 1.",
            "0", // transition evaluator response: select first transition
        ]);
        let builder = AgentBuilder::from_yaml(yaml).unwrap();
        let agent = builder.llm(Arc::new(mock)).build().unwrap();

        assert_eq!(agent.current_state(), Some("step1".to_string()));

        // Manually transition to test on_enter/on_exit
        agent.transition_to("step2").await.unwrap();

        assert_eq!(agent.current_state(), Some("step2".to_string()));

        // Verify context was set by on_exit and on_enter actions
        let ctx = agent.get_context();
        assert_eq!(ctx.get("step1_exited"), Some(&serde_json::json!(true)));
        assert_eq!(ctx.get("step2_entered"), Some(&serde_json::json!(true)));
    }

    // Process pipeline transforms input
    #[tokio::test]
    async fn test_integration_process_normalize() {
        let yaml = r#"
name: ProcessAgent
system_prompt: "You are helpful."
process:
  input:
    - type: normalize
      config:
        trim: true
        collapse_whitespace: true
"#;
        let mock = mock_with_response("Got your message.");
        let builder = AgentBuilder::from_yaml(yaml).unwrap();
        let agent = builder.llm(Arc::new(mock.clone())).build().unwrap();

        let _ = agent.chat("  hello   world  ").await.unwrap();

        // Verify the LLM received the normalized input (trimmed + collapsed whitespace)
        let history = mock.call_history();
        assert!(!history.is_empty());
        // The user message in LLM call should be normalized
        let last_call = history.last().unwrap();
        let user_msg = last_call
            .messages
            .iter()
            .find(|m| m.role == ai_agents_core::Role::User)
            .unwrap();
        assert_eq!(user_msg.content, "hello world");
    }

    // ═══════════════════════════════════════════════════════════
    // Integration Test 2.1.7: Memory compression triggers
    // ═══════════════════════════════════════════════════════════
    #[tokio::test]
    async fn test_integration_memory_compression() {
        let yaml = r#"
name: MemoryAgent
system_prompt: "You are helpful."
memory:
  type: compacting
  max_messages: 100
  compress_threshold: 5
  max_recent_messages: 3
  summarize_batch_size: 2
"#;
        // Provide enough responses for compression to trigger
        let responses: Vec<&str> = (0..8).map(|_| "Response from assistant.").collect();
        let mock = mock_with_responses(responses);
        let builder = AgentBuilder::from_yaml(yaml).unwrap();
        let agent = builder.llm(Arc::new(mock)).build().unwrap();

        // Send enough messages to trigger compression
        for i in 0..6 {
            let _ = agent.chat(&format!("Message {}", i)).await.unwrap();
        }

        // Memory should have compressed - verify it didn't crash
        // and that messages are bounded
        let messages = agent.memory.get_messages(None).await.unwrap();
        // With compress_threshold=5 and max_recent_messages=3,
        // after 6 turns (12 messages), compression should have run
        assert!(messages.len() <= 12); // At most all messages if no compression, fewer if compressed
    }

    // YAML with multiple LLMs
    #[tokio::test]
    async fn test_integration_multi_llm_registry() {
        let mut mock_default = MockLLMProvider::new("default");
        mock_default.set_response("Default LLM response.");
        let mut mock_router = MockLLMProvider::new("router");
        mock_router.set_response("Router response.");

        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm_alias("default", Arc::new(mock_default))
            .llm_alias("router", Arc::new(mock_router))
            .build()
            .unwrap();

        let response = agent.chat("Hello").await.unwrap();
        assert_eq!(response.content, "Default LLM response.");
    }

    // Agent reset clears state
    #[tokio::test]
    async fn test_integration_agent_reset() {
        let mock = mock_with_responses(vec!["Hello!", "Hello again!"]);
        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm(Arc::new(mock))
            .build()
            .unwrap();

        let _ = agent.chat("Hi").await.unwrap();
        let messages = agent.memory.get_messages(None).await.unwrap();
        assert_eq!(messages.len(), 2); // user + assistant

        agent.reset().await.unwrap();
        let messages = agent.memory.get_messages(None).await.unwrap();
        assert_eq!(messages.len(), 0);
    }

    // Process pipeline rejects input
    #[tokio::test]
    async fn test_integration_process_validate_reject() {
        use ai_agents_process::{ProcessConfig, ProcessProcessor};

        let validate_config = ai_agents_process::ValidateStage {
            id: Some("length_check".to_string()),
            condition: None,
            config: ai_agents_process::ValidateConfig {
                rules: vec![ai_agents_process::ValidationRule::MinLength {
                    min_length: 10,
                    on_fail: ai_agents_process::ValidationAction {
                        action: ai_agents_process::ValidationActionType::Reject,
                        message: None,
                    },
                }],
                ..Default::default()
            },
        };
        let process_config = ProcessConfig {
            input: vec![ai_agents_process::ProcessStage::Validate(validate_config)],
            ..Default::default()
        };
        let processor = ProcessProcessor::new(process_config);

        let mock = mock_with_response("Should not reach here.");
        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm(Arc::new(mock))
            .process_processor(processor)
            .build()
            .unwrap();

        let response = agent.chat("Hi").await.unwrap();
        // Rejected input should produce a rejection response, not call LLM
        assert!(
            response.content.contains("rejected")
                || response.content.contains("Input rejected")
                || response.content.contains("too short")
                || response.content.contains("Too short")
                || response.content.len() < 50, // rejection message is typically short
            "Expected rejection response, got: {}",
            response.content
        );
    }

    // LLM fallback: primary fails, fallback LLM responds
    #[tokio::test]
    async fn test_llm_fallback_on_failure() {
        use ai_agents_recovery::{ErrorRecoveryConfig, LLMFailureAction, LLMRecoveryConfig};

        let mut primary = MockLLMProvider::new("primary");
        primary.set_error("Primary LLM is unavailable");

        let mut fallback = MockLLMProvider::new("fallback");
        fallback.set_response("Fallback response works!");

        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm_alias("default", Arc::new(primary))
            .llm_alias("backup", Arc::new(fallback))
            .recovery_manager(RecoveryManager::new(ErrorRecoveryConfig {
                llm: LLMRecoveryConfig {
                    on_failure: LLMFailureAction::FallbackLlm {
                        fallback_llm: "backup".to_string(),
                    },
                    ..Default::default()
                },
                ..Default::default()
            }))
            .build()
            .unwrap();

        let response = agent.chat("Hello").await.unwrap();
        assert!(
            response.content.contains("Fallback response"),
            "Expected fallback response, got: {}",
            response.content
        );
    }

    // LLM fallback: primary fails, static message returned
    #[tokio::test]
    async fn test_llm_fallback_response_static_message() {
        use ai_agents_recovery::{ErrorRecoveryConfig, LLMFailureAction, LLMRecoveryConfig};

        let mut primary = MockLLMProvider::new("primary");
        primary.set_error("Primary LLM is unavailable");

        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm(Arc::new(primary))
            .recovery_manager(RecoveryManager::new(ErrorRecoveryConfig {
                llm: LLMRecoveryConfig {
                    on_failure: LLMFailureAction::FallbackResponse {
                        message: "I am temporarily unavailable. Please try again later."
                            .to_string(),
                    },
                    ..Default::default()
                },
                ..Default::default()
            }))
            .build()
            .unwrap();

        let response = agent.chat("Hello").await.unwrap();
        assert!(
            response.content.contains("temporarily unavailable"),
            "Expected static fallback message, got: {}",
            response.content
        );
    }

    // Tool skip: tool fails, on_failure: skip absorbs the error
    #[tokio::test]
    async fn test_tool_failure_skip() {
        use ai_agents_recovery::{
            ErrorRecoveryConfig, ToolFailureAction, ToolRecoveryConfig, ToolRetryConfig,
        };

        // LLM requests a nonexistent tool, then responds after seeing the skip result
        let mock = mock_with_responses(vec![
            r#"I'll use the nonexistent tool.
[TOOL_CALL: {"name": "nonexistent_tool", "arguments": {}}]"#,
            "The tool was unavailable, but I can still help you.",
        ]);

        let agent = AgentBuilder::new()
            .system_prompt("You are helpful.")
            .llm(Arc::new(mock))
            .recovery_manager(RecoveryManager::new(ErrorRecoveryConfig {
                tools: ToolRecoveryConfig {
                    default: ToolRetryConfig {
                        max_retries: 0,
                        timeout_ms: None,
                        on_failure: ToolFailureAction::Skip,
                    },
                    ..Default::default()
                },
                ..Default::default()
            }))
            .build()
            .unwrap();

        // The tool will fail (not found), but on_failure: skip absorbs the error
        let response = agent.chat("Use the nonexistent tool").await;
        assert!(
            response.is_ok(),
            "Expected Ok with skip policy, got: {:?}",
            response
        );
    }
}