lc_agents/executor/

engine.rs

// lc-agents/src/executor/engine.rs
//! `AgentExecutor` — the execution loop (plan -> act -> observe).

use super::budget::{
    budget_cost_gate, budget_iteration_gate, budget_token_gate, budget_tool_gate, BudgetConfig,
};
use super::compaction::CompactionConfig;
use super::hooks::{run_after_completion_hooks, run_before_completion_hooks};
use super::semantic_memory::{SemanticMemoryHook, SEMANTIC_MEMORY_INPUT_KEY};
use super::tools::{
    execute_tool_for_stream, execute_tools_parallel_for_stream, index_tools, tool_error_observation,
};
use super::{
    AgentError, BaseAgent, CACHE_NS, DEFAULT_MAX_CONCURRENCY, MAX_MAX_ITERATIONS,
    MIN_MAX_ITERATIONS,
};
use crate::approval::{ApprovalDecision, ApprovalHandler};
use crate::cache::ResponseCache;
use crate::hooks::{AgentHook, HookError};
use crate::metrics::AgentMetrics;
use crate::policy::ToolPolicy;
use crate::resume::{PendingApproval, ResumeStore};
use crate::streaming::state::AgentStreamEvent;
use crate::types::{AgentAction, AgentOutput, AgentStep, ToolInput};
use futures_util::Stream;
use lc_callbacks::{CallbackManager, RunTree, RunType};
use lc_core::cost::CostTracker;
use lc_core::observability::{MetricsSink, ObsEvent};
use lc_core::runnables::{RunnableConfig, RUN_META_PARENT_RUN_ID, RUN_META_TRACE_ID};
use lc_core::tools::BaseTool;
use lc_memory::{BaseMemory, MemoryExtractor, TwoTierMemory};
use serde_json::json;
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::Semaphore;

/// A18: builds the per-round [`RunnableConfig`] handed to
/// [`BaseAgent::plan`]/[`BaseAgent::plan_stream`] for one agent run.
///
/// The config carries the effective callback manager plus reserved trace-linkage
/// metadata (parent = this run's chain root, trace = root trace id), so the
/// provider-built LLM [`RunTree`] is dispatched under the agent chain tree
/// instead of becoming a trace root. Returns `None` when no callbacks are
/// configured, preserving the pre-A18 behavior where providers fire nothing for
/// observers-less executors (and zero per-round allocation happens).
fn build_plan_config(
    callbacks: &Option<Arc<CallbackManager>>,
    root_run: &RunTree,
) -> Option<RunnableConfig> {
    let manager = callbacks.as_ref()?;
    let trace_id = root_run.trace_id.unwrap_or(root_run.id);
    Some(
        RunnableConfig::new()
            .with_callbacks(manager.clone())
            .with_metadata(RUN_META_PARENT_RUN_ID, json!(root_run.id.to_string()))
            .with_metadata(RUN_META_TRACE_ID, json!(trace_id.to_string())),
    )
}

/// Agent executor.
///
/// Responsible for executing the agent's decision loop: Plan -> Act -> Observe.
pub struct AgentExecutor {
    /// Agent instance.
    pub(crate) agent: Arc<dyn BaseAgent>,

    /// Available tools.
    pub(crate) tools: Vec<Arc<dyn BaseTool>>,

    /// A11: prebuilt name → tool index for O(1) lookups. Kept in sync with `tools`
    /// (built in `new`, extended in `with_memory_tool`, cloned in the merged-executor
    /// copy); `index_tools` preserves first-match-wins on name collisions.
    pub(crate) tools_by_name: HashMap<String, Arc<dyn BaseTool>>,

    /// Max iterations.
    pub(crate) max_iterations: usize,

    /// Verbose output.
    pub(crate) verbose: bool,

    /// Memory (optional).
    pub(crate) memory: Option<Arc<tokio::sync::Mutex<dyn BaseMemory>>>,

    /// Callback manager (optional).
    pub(crate) callbacks: Option<Arc<CallbackManager>>,

    /// Agent hooks (optional).
    pub(crate) hooks: Vec<Arc<dyn AgentHook>>,

    /// Tool execution timeout (None = no timeout).
    pub(crate) tool_timeout: Option<Duration>,

    /// Maximum number of tools executed concurrently.
    pub(crate) max_concurrency: usize,

    /// Semaphore guarding concurrent tool execution.
    pub(crate) concurrency_sem: Arc<Semaphore>,

    /// Most recent execution metrics (P1-5). Arc-shared so merged executors
    /// created by `invoke_with_config` write back to the original executor.
    pub(crate) metrics_store: Arc<Mutex<Option<AgentMetrics>>>,

    /// LLM result cache (P2-1): `plan()` results hit on `(namespace, inputs, steps)`;
    /// deterministic prompts are reused directly, skipping the LLM round-trip.
    /// `None` = no caching.
    pub(crate) response_cache: Option<Arc<dyn ResponseCache>>,
    /// This instance's cache namespace (isolates executors sharing the same cache).
    pub(crate) cache_namespace: String,

    /// Tool permission policy (permission tiering + sandbox gate, P2-9).
    /// `None` = no checks.
    pub(crate) tool_policy: Option<ToolPolicy>,

    /// Approval gate (§4.2): async approval before each tool execution. `None` = no
    /// interception (default off).
    pub(crate) approval: Option<Arc<dyn ApprovalHandler>>,
    /// Budget gate (§4.2): hard limits. `None` = unlimited (default off).
    pub(crate) budget: Option<BudgetConfig>,

    /// Context compaction (0.21.0 S6.1): drops the oldest intermediate steps at
    /// whole-step boundaries when the trigger fires. `None` = off (default).
    pub(crate) compaction: Option<CompactionConfig>,

    /// Cross-process resume (§4.2): checkpoint store. When `Some`, `execute_tool`
    /// persists the pending approval before awaiting approval and clears it once the
    /// decision lands; a new process can inspect it via `pending_approval()` and
    /// continue via `resume(decision)`. `None` = off (default).
    pub(crate) resume_store: Option<Arc<dyn ResumeStore>>,

    /// Observability sink (v0.20.2): exports one `AgentMetrics` event per run
    /// (invoke / stream / resume). `None` = off (default). Failures are `warn` only.
    pub(crate) metrics_sink: Option<Arc<dyn MetricsSink>>,

    /// B3 (0.22.4): shared USD spend tracker. Attach the same `Arc<CostTracker>`
    /// that the tracking LLM (`TokenTrackingLLM::with_cost_tracker`) records into
    /// and the `max_cost_usd` budget gate reads cumulative spend after each LLM
    /// call. `None` = off (default); a cost limit without a tracker never trips.
    pub(crate) cost_tracker: Option<Arc<CostTracker>>,

    /// B4 (0.22.4): two-tier semantic memory hook. When present, each run recalls
    /// relevant facts into `inputs["semantic_memory"]` before planning, and after a
    /// successful answer spawns a **detached** extraction task (never blocking the
    /// answer). `None` = off (default). Distinct from conversation `memory`
    /// (history injected into the next prompt).
    pub(crate) semantic_memory: Option<SemanticMemoryHook>,

    /// 0.22.0 C4 fix: what to do when the loop exhausts `max_iterations`
    /// without a final answer. Default **`Error`** — the previous placeholder
    /// string was indistinguishable from a real answer and downstream
    /// consumers (PlanExecute) recorded failed steps as completed.
    pub(crate) on_max_iterations: MaxIterationsPolicy,

    /// A2 Rule-of-Two (v0.22.1 §S8): when enabled, a tool whose declared risk profile
    /// arms all three properties (`count_armed() >= 3`) is **blocked before execution**
    /// and the loop gets a rejection observation. Default `false` (off, zero change) —
    /// an undeclared tool has an all-false profile and is never intercepted.
    pub(crate) rule_of_two: bool,

    /// A1 Spotlighting (v0.22.1 §S8): when enabled, tool output observations are wrapped in
    /// `<untrusted_data>…</untrusted_data>` before entering the intermediate steps, so the
    /// model reads untrusted tool results as delimited data. Default `false` (off, zero change).
    pub(crate) spotlight_tool_output: bool,
}

/// 0.22.0 C4 fix: behavior when the agent loop exhausts its iteration budget.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum MaxIterationsPolicy {
    /// Fail the run with [`AgentError::MaxIterationsReached`] (default —
    /// "failure disguised as success" is no longer possible).
    #[default]
    Error,
    /// Legacy behavior (≤ 0.21.x): return the stopped-response placeholder
    /// string. Opt in explicitly when a caller cannot handle errors.
    Placeholder,
}

impl AgentExecutor {
    /// Creates a new AgentExecutor.
    pub fn new(agent: Arc<dyn BaseAgent>, tools: Vec<Arc<dyn BaseTool>>) -> Self {
        // A11: build the name index before `tools` is moved into the struct.
        let tools_by_name = index_tools(&tools);
        Self {
            agent,
            tools,
            tools_by_name,
            max_iterations: 10,
            verbose: false,
            memory: None,
            callbacks: None,
            hooks: Vec::new(),
            tool_timeout: None,
            max_concurrency: DEFAULT_MAX_CONCURRENCY,
            concurrency_sem: Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENCY)),
            metrics_store: Arc::new(Mutex::new(None)),
            response_cache: None,
            cache_namespace: format!("exec-{}", CACHE_NS.fetch_add(1, Ordering::SeqCst)),
            tool_policy: None,
            approval: None,
            budget: None,
            compaction: None,
            resume_store: None,
            metrics_sink: None,
            cost_tracker: None,
            semantic_memory: None,
            on_max_iterations: MaxIterationsPolicy::default(),
            rule_of_two: false,
            spotlight_tool_output: false,
        }
    }

    /// Sets max iterations, clamped to `[1, 100]`.
    pub fn with_max_iterations(mut self, max_iterations: usize) -> Self {
        self.max_iterations = max_iterations.clamp(MIN_MAX_ITERATIONS, MAX_MAX_ITERATIONS);
        if max_iterations > MAX_MAX_ITERATIONS {
            log::warn!(
                "max_iterations {} clamped to {}",
                max_iterations,
                MAX_MAX_ITERATIONS
            );
        }
        self
    }

    /// 0.22.0 C4 fix: chooses what happens when the loop exhausts
    /// `max_iterations` without a final answer.
    ///
    /// - `MaxIterationsPolicy::Error` (default): the run fails with
    ///   [`AgentError::MaxIterationsReached`] — the caller can distinguish
    ///   "did not converge" from a real answer, and PlanExecute treats the
    ///   step as failed instead of completed-with-garbage.
    /// - `MaxIterationsPolicy::Placeholder`: legacy ≤ 0.21.x behavior —
    ///   return the stopped-response placeholder string.
    pub fn with_on_max_iterations(mut self, policy: MaxIterationsPolicy) -> Self {
        self.on_max_iterations = policy;
        self
    }

    /// A2 Rule of Two (v0.22.1 §S8): when enabled, a tool whose declared risk profile arms
    /// all three properties (untrusted-input + sensitive-access + state-changing) is blocked
    /// before execution and the loop receives a rejection observation. Default off — an
    /// undeclared tool has an all-false profile and is never intercepted (zero behavior change).
    pub fn with_rule_of_two(mut self, on: bool) -> Self {
        self.rule_of_two = on;
        self
    }

    /// A1 tool-output spotlighting (v0.22.1 §S8): when enabled, tool observations are wrapped
    /// in `<untrusted_data>…</untrusted_data>` before entering intermediate steps. Default off —
    /// tool output passes through unchanged.
    pub fn with_tool_spotlight(mut self, on: bool) -> Self {
        self.spotlight_tool_output = on;
        self
    }

    /// C1: mounts a file-memory tool (v0.22.1 §S8).
    ///
    /// Pushes the [`crate::executor::FileMemoryTool`] adapter over a fresh [`lc_memory::file_memory::FileMemoryStore`]
    /// rooted at `root` into the executor's tool set, letting the agent explicitly `view` /
    /// `create` / `write` / `append` / `delete` / `list` named memories during a run. Default
    /// off — this is explicit opt-in; the tools are only registered when this builder is used.
    /// Err is returned when the root directory cannot be set up.
    pub fn with_memory_tool(
        mut self,
        root: impl Into<PathBuf>,
    ) -> Result<Self, lc_memory::file_memory::FileMemoryError> {
        let tool: Arc<dyn BaseTool> = crate::executor::mount(root)?;
        // A11: keep the name index consistent with the pushed tool (first-wins).
        self.tools_by_name
            .entry(tool.name().to_string())
            .or_insert_with(|| tool.clone());
        self.tools.push(tool);
        Ok(self)
    }

    /// Sets the tool execution timeout.
    ///
    /// A tool that exceeds the timeout returns an error instead of hanging the
    /// whole agent loop. `None` (the default) disables the timeout.
    pub fn with_tool_timeout(mut self, timeout: Duration) -> Self {
        self.tool_timeout = Some(timeout);
        self
    }

    /// Sets the maximum number of tools executed concurrently.
    ///
    /// Clamped to at least 1. The default is 8.
    pub fn with_max_concurrency(mut self, max_concurrency: usize) -> Self {
        let max_concurrency = max_concurrency.max(1);
        self.max_concurrency = max_concurrency;
        self.concurrency_sem = Arc::new(Semaphore::new(max_concurrency));
        self
    }

    /// Enables the LLM result cache (P2-1).
    ///
    /// For deterministic prompts, `plan()` results with the same `(inputs,
    /// intermediate_steps)` are reused directly, skipping the LLM round-trip — suited to
    /// cost-sensitive / repeatedly-evaluated deterministic tasks. Tool execution results
    /// enter the cache key; tools themselves are not cached; the cache applies to the
    /// non-streaming `invoke` path.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let cache = Arc::new(MemoryCache::with_capacity(256));
    /// let executor = AgentExecutor::new(agent, tools).with_response_cache(cache);
    /// ```
    pub fn with_response_cache(mut self, cache: Arc<dyn ResponseCache>) -> Self {
        self.response_cache = Some(cache);
        self
    }

    /// Tool permission policy (permission tiering + sandbox gate, P2-9).
    ///
    /// Checked before every tool execution: tools whose risk exceeds `max_permitted`
    /// are rejected; high-risk tools that are not declared sandboxed
    /// ([`ToolPolicy::sandboxed`]) are also rejected. Unconfigured = everything allowed.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let policy = ToolPolicy::new()
    ///     .risk("code_interpreter", ToolRisk::Dangerous)
    ///     .sandboxed("code_interpreter"); // moved into a restricted environment, allowed to run
    /// let executor = AgentExecutor::new(agent, tools).with_tool_policy(policy);
    /// ```
    pub fn with_tool_policy(mut self, policy: ToolPolicy) -> Self {
        self.tool_policy = Some(policy);
        self
    }

    /// Approval gate (§4.2): async approval before each tool execution.
    ///
    /// Default `None` = no interception; existing behavior unchanged. Approval
    /// decisions (implemented by the caller via [`ApprovalHandler`]):
    /// - [`ApprovalDecision::Allow`](crate::approval::ApprovalDecision::Allow): run as-is;
    /// - [`ApprovalDecision::Deny`](crate::approval::ApprovalDecision::Deny): skip the tool,
    ///   feed the reason back as an observation, and re-plan next round;
    /// - [`ApprovalDecision::Modify`](crate::approval::ApprovalDecision::Modify): run with the
    ///   new arguments substituted.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let executor = AgentExecutor::new(agent, tools)
    ///     .with_approval(Arc::new(AllowAll));
    /// ```
    pub fn with_approval(mut self, handler: Arc<dyn ApprovalHandler>) -> Self {
        self.approval = Some(handler);
        self
    }

    /// Budget gate (§4.2): hard limits, effective on both the `invoke` and `stream`
    /// paths.
    ///
    /// - `invoke`: any limit hit returns [`AgentError::BudgetExceeded`] and stops
    ///   immediately;
    /// - `stream`: any limit hit sends `Err(AgentError::BudgetExceeded)` on the channel
    ///   and stops.
    ///
    /// The caller can catch this error to distinguish a "budget stop" from "the model did
    /// not converge". Default `None` = unlimited.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let budget = BudgetConfig {
    ///     max_tool_calls: Some(3),
    ///     max_tokens: Some(10_000),
    ///     max_duration: Some(Duration::from_secs(60)),
    ///     max_iterations: Some(5),
    ///     max_cost_usd: Some(1.0),
    /// };
    /// let executor = AgentExecutor::new(agent, tools).with_budget(budget);
    /// ```
    pub fn with_budget(mut self, budget: BudgetConfig) -> Self {
        self.budget = Some(budget);
        self
    }

    /// B3 (0.22.4): attaches a shared `CostTracker` used by the
    /// `max_cost_usd` budget gate.
    ///
    /// Pass the **same `Arc`** that records the agent's LLM calls — typically
    /// via `TokenTrackingLLM::with_cost_tracker` (or the equivalent tracked
    /// model wrapper). After every planning call the executor reads
    /// [`CostTracker::total_cost_usd`] and hard-stops with
    /// [`AgentError::BudgetExceeded`] /
    /// [`super::budget::BudgetExceeded::Cost`] once the configured spend is
    /// reached. Attaching a tracker without a `max_cost_usd` limit only
    /// measures; setting a limit without a tracker never trips.
    pub fn with_cost_tracker(mut self, tracker: Arc<CostTracker>) -> Self {
        self.cost_tracker = Some(tracker);
        self
    }

    /// Context compaction (0.21.0 S6.1): drop the oldest intermediate steps
    /// (whole steps — action + observation stay paired) when the trigger fires.
    ///
    /// Checked before every `plan()` round in both the invoke and stream paths.
    /// Off by default (`None` = unlimited history, zero behavior change).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let config = CompactionConfig::new(
    ///     CompactionTrigger::TurnCount(20),
    ///     CompactionStrategy::SlidingWindow { keep_recent_turns: 8 },
    /// );
    /// let executor = AgentExecutor::new(agent, tools).with_compaction(config);
    /// ```
    pub fn with_compaction(mut self, compaction: CompactionConfig) -> Self {
        self.compaction = Some(compaction);
        self
    }

    /// Cross-process resume (§4.2): checkpoint store.
    ///
    /// When enabled, before each tool call enters the approval gate to await approval,
    /// the framework writes the pending tool + the context needed to resume the agent
    /// loop ([`PendingApproval`]) into the store; it is cleared once the approval
    /// decision **lands**. If the process crashes, the checkpoint stays on disk; a new
    /// process rebuilding an executor with the same configuration calls
    /// [`pending_approval`](Self::pending_approval) / [`resume`](Self::resume) to
    /// continue instead of replaying the whole conversation from scratch.
    ///
    /// Applies only to the non-streaming `invoke` path (the streaming path has no
    /// approval gate); only meaningful together with
    /// [`with_approval`](Self::with_approval). Parallel tool execution (multiple tools
    /// approved concurrently) does not participate in cross-process persistence — the
    /// in-process approval still works.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let store = Arc::new(FileResumeStore::new("/var/checkpoints/app")?);
    /// let executor = AgentExecutor::new(agent, tools)
    ///     .with_resume_store(store)
    ///     .with_approval(Arc::new(MyHandler));
    /// ```
    pub fn with_resume_store(mut self, store: Arc<dyn ResumeStore>) -> Self {
        self.resume_store = Some(store);
        self
    }

    /// Tool registration validation (P2-2).
    ///
    /// The Agent declares the tool names it may call via `get_allowed_tools()`; when it
    /// declares some, every one must be present in this executor's `tools`, otherwise an
    /// error is returned listing all missing tools. When the Agent declares nothing
    /// (returns `None`, e.g. a base Agent with no tools), validation is skipped.
    ///
    /// Called before each `invoke` / `stream`: startup fail-fast, turning a mid-loop
    /// `ToolNotFound` into a one-shot, all-configuration-errors-at-once report before
    /// first execution.
    pub fn validate_tool_registration(&self) -> Result<(), AgentError> {
        let Some(allowed) = self.agent.get_allowed_tools() else {
            return Ok(());
        };
        let registered: HashSet<&str> = self.tools.iter().map(|t| t.name()).collect();
        let missing: Vec<&str> = allowed
            .into_iter()
            .filter(|name| !registered.contains(name))
            .collect();
        if missing.is_empty() {
            return Ok(());
        }
        Err(AgentError::ToolNotFound(format!(
            "tools not registered on executor: {}",
            missing.join(", ")
        )))
    }

    /// Sets verbose output.
    pub fn with_verbose(mut self, verbose: bool) -> Self {
        self.verbose = verbose;
        self
    }

    /// Sets memory.
    pub fn with_memory(mut self, memory: Arc<tokio::sync::Mutex<dyn BaseMemory>>) -> Self {
        self.memory = Some(memory);
        self
    }

    /// B4 (v0.22.4): mounts two-tier semantic memory.
    ///
    /// - `store` — the shared [`TwoTierMemory`] (safe to share across executors;
    ///   the `namespace` isolates this executor's facts).
    /// - `namespace` — e.g. a user/session id; recall and extraction never cross it.
    /// - `extractor` — turn-to-facts extractor, typically
    ///   [`crate::LlmMemoryExtractor`].
    ///
    /// On every run the executor recalls relevant facts and injects them into the
    /// prompt inputs under `semantic_memory`; after a successful answer it extracts
    /// durable facts in a **detached background task** and promotes hot/important
    /// facts from the short tier to the weighted-decay long tier. Off by default.
    pub fn with_semantic_memory(
        mut self,
        store: Arc<TwoTierMemory>,
        namespace: impl Into<String>,
        extractor: Arc<dyn MemoryExtractor + Send + Sync>,
    ) -> Self {
        self.semantic_memory = Some(SemanticMemoryHook::new(store, namespace, extractor));
        self
    }

    /// Sets callback manager.
    pub fn with_callbacks(mut self, callbacks: Arc<CallbackManager>) -> Self {
        self.callbacks = Some(callbacks);
        self
    }

    /// Sets an observability sink: each run (invoke / stream / resume) exports one
    /// [`AgentMetrics`] event at the end (v0.20.2). Failures are logged, never
    /// propagated.
    pub fn with_metrics_sink(mut self, sink: Arc<dyn MetricsSink>) -> Self {
        self.metrics_sink = Some(sink);
        self
    }

    /// Adds an agent hook.
    pub fn hook(mut self, hook: impl AgentHook + 'static) -> Self {
        self.hooks.push(Arc::new(hook));
        self
    }

    /// Returns metrics from the most recent invocation, if any.
    pub fn last_metrics(&self) -> Option<AgentMetrics> {
        self.metrics_store.lock().ok().and_then(|g| g.clone())
    }

    /// Exports one run's metrics to the attached sink (v0.20.2). Called at every
    /// run tail (invoke, resume, stream). Failures are logged as `warn` and never
    /// propagate into the agent flow.
    async fn export_metrics(&self, metrics: &AgentMetrics) {
        if let Some(sink) = &self.metrics_sink {
            let evt = ObsEvent::AgentMetrics(metrics.clone());
            if let Err(e) = sink.export(&evt).await {
                log::warn!(target: "lc_agents::metrics", "agent metrics export failed: {e}");
            }
        }
    }

    /// Reads the currently pending approval checkpoint (cross-process resume).
    ///
    /// Returns `Ok(None)` when no [`ResumeStore`] is configured or the store is empty.
    /// After getting a [`PendingApproval`], the caller shows `tool_name` / `arguments`
    /// to an operator, collects the approval decision, then calls
    /// [`resume`](Self::resume) to continue.
    pub async fn pending_approval(&self) -> Result<Option<PendingApproval>, AgentError> {
        let Some(store) = &self.resume_store else {
            return Ok(None);
        };
        store
            .load_pending()
            .await
            .map_err(|e| AgentError::Resume(e.to_string()))
    }

    /// Resumes from a checkpoint (cross-process resume): processes the pending tool with
    /// the given decision, then continues the agent loop from the suspended iteration and
    /// returns the final answer.
    ///
    /// - No [`ResumeStore`] configured or no checkpoint → `Ok(None)` (no-op).
    /// - A checkpoint exists → first **claims** it (clears it) to prevent duplicate
    ///   approval, executes the pending tool, then continues the loop from
    ///   `iteration + 1`; budgets (tool / token / iteration) keep counting from the
    ///   checkpoint's accumulated amounts, and `max_duration` restarts its timer at the
    ///   resume moment (a cross-process monotonic clock is not portable — an honest
    ///   approximation).
    ///
    /// The resuming executor must be constructed identically to the one before the crash
    /// (same agent / tools / store directory) to resume correctly; the approval decision
    /// is injected by the caller and [`ApprovalHandler`] is not re-run.
    pub async fn resume(&self, decision: ApprovalDecision) -> Result<Option<String>, AgentError> {
        let Some(store) = &self.resume_store else {
            return Ok(None);
        };
        let Some(pending) = store
            .load_pending()
            .await
            .map_err(|e| AgentError::Resume(e.to_string()))?
        else {
            return Ok(None);
        };
        // Claim the checkpoint: clear it first. If resume crashes midway, approval is
        // not repeated (at most once).
        store
            .clear_pending()
            .await
            .map_err(|e| AgentError::Resume(e.to_string()))?;

        let action = AgentAction {
            tool: pending.tool_name.clone(),
            tool_input: ToolInput::Object {
                value: pending.arguments.clone(),
            },
            log: String::new(),
        };

        let mut root_run = RunTree::new(
            "AgentExecutor",
            RunType::Chain,
            json!({"input": pending.inputs.get("input").cloned().unwrap_or_default()}),
        );
        // Reuse the original trace_id so the resumed tool child runs keep trace
        // continuity.
        if let Some(tid) = &pending.trace_id {
            if let Ok(id) = uuid::Uuid::parse_str(tid) {
                root_run.trace_id = Some(id);
                root_run = root_run.with_metadata("trace_id", json!(tid));
            }
        }

        let started = std::time::Instant::now();
        let mut metrics = AgentMetrics {
            trace_id: root_run.trace_id.map(|id| id.to_string()),
            tool_calls: pending.tool_calls_consumed,
            total_tokens: pending.tokens_consumed,
            ..Default::default()
        };

        // Execute the pending tool (inject the given decision; do not re-run the
        // approval handler).
        let observation = self
            .execute_tool_inner(&action, &root_run, None, Some(decision))
            .await?;
        let mut steps = pending.steps;
        steps.push(AgentStep::new(action, observation));

        // A18: resumed rounds observe the same callbacks / trace linkage.
        let plan_config = build_plan_config(&self.callbacks, &root_run);

        let result = self
            .run_agent_loop_from(
                pending.inputs,
                steps,
                pending.iteration + 1,
                &mut root_run,
                &mut metrics,
                plan_config.as_ref(),
            )
            .await;

        metrics.duration = started.elapsed();
        metrics.log_summary();
        if let Ok(mut guard) = self.metrics_store.lock() {
            *guard = Some(metrics.clone());
        }
        self.export_metrics(&metrics).await;
        result.map(Some)
    }

    /// Builds the `plan()` cache key: namespace + inputs + intermediate steps (including
    /// tool observations).
    ///
    /// A deterministic Agent always produces the same `AgentOutput` for the same
    /// `(inputs, steps)`, so this hash is the "LLM result" fingerprint; observations are
    /// part of the key, so the cache cannot wrongly hit across different tool results.
    ///
    /// The key must be reproducible across runs. `HashMap` iterates in an order seeded
    /// per-instance (RandomState), so a bare hash of the map's JSON serialization differs
    /// between two `invoke`s even with identical content — breaking cross-run hits. We
    /// serialize inputs as *sorted* key/value pairs to make it canonical. (A10)
    fn cache_key(namespace: &str, inputs: &HashMap<String, String>, steps: &[AgentStep]) -> String {
        use std::hash::{Hash, Hasher};

        let mut input_keys: Vec<&String> = inputs.keys().collect();
        input_keys.sort_unstable();
        let inputs_repr: Vec<String> = input_keys
            .iter()
            .map(|k| {
                // \x1f = unit separator, \x1e = record separator; both are
                // forbidden in JSON keys, so they can't collide with content.
                format!("{}\x1f{}\x1e", *k, inputs[*k])
            })
            .collect();

        let payload = json!({ "ns": namespace, "inputs": inputs_repr, "steps": steps });
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        payload.to_string().hash(&mut hasher);
        format!("{:016x}", hasher.finish())
    }

    /// One-stop lookup / `plan()` / write-back.
    ///
    /// On a hit, returns the previous `AgentOutput` directly and records
    /// `metrics.cache_hits`, without calling the LLM; on a miss, calls `plan()` and
    /// serializes the result back. Corrupt cache content degrades to re-planning.
    pub(crate) async fn plan_cached(
        &self,
        intermediate_steps: &[AgentStep],
        inputs: &HashMap<String, String>,
        metrics: &mut AgentMetrics,
        config: Option<&RunnableConfig>,
    ) -> Result<AgentOutput, AgentError> {
        let cache = self.response_cache.as_ref();
        let key = cache.map(|_| Self::cache_key(&self.cache_namespace, inputs, intermediate_steps));

        if let (Some(cache), Some(key)) = (cache, &key) {
            if let Some(cached) = cache.get(key) {
                match serde_json::from_str::<AgentOutput>(&cached) {
                    Ok(output) => {
                        metrics.cache_hits += 1;
                        log::debug!(target: "lc_agents::cache", "plan cache hit: {}", key);
                        return Ok(output);
                    }
                    Err(_) => {
                        log::warn!(target: "lc_agents::cache", "plan cache corrupt, re-plan");
                    }
                }
            }
        }

        metrics.llm_calls += 1;
        // P2-9: rate-limit / quota check before the LLM call (Reject → abort this round).
        run_before_completion_hooks(&self.hooks, inputs)?;
        let output = self.agent.plan(intermediate_steps, inputs, config).await?;
        let usage = self.agent.last_token_usage();
        if let Some(usage) = &usage {
            metrics.add_token_usage(usage);
        }
        // P2-9: accumulate the real token usage after the LLM call (for the rate-limit
        // hook's accounting).
        run_after_completion_hooks(&self.hooks, &output, usage.as_ref());

        if let (Some(cache), Some(key)) = (cache, &key) {
            if let Ok(serialized) = serde_json::to_string(&output) {
                cache.put(key.clone(), serialized);
            }
        }
        Ok(output)
    }

    /// Executes the agent.
    pub async fn invoke(&self, input: String) -> Result<String, AgentError> {
        self.invoke_inner(input, None).await
    }

    /// Internal execution with optional trace propagation (P1-4).
    ///
    /// `trace_id` comes from `RunnableConfig.metadata["trace_id"]` when present;
    /// `None` for plain `invoke`. The trace_id is stamped onto the root
    /// `RunTree` so every tool child run inherits it via `create_child`.
    async fn invoke_inner(
        &self,
        input: String,
        trace_id: Option<String>,
    ) -> Result<String, AgentError> {
        // P2-2: startup fail-fast — error out on unregistered tools before any LLM call.
        self.validate_tool_registration()?;

        let started = std::time::Instant::now();

        // Hooks: on_agent_start (P1-6)
        for hook in &self.hooks {
            if let Err(e) = hook.on_agent_start(&input) {
                log::warn!("Hook on_agent_start error: {}", e);
            }
        }

        let mut root_run = RunTree::new(
            "AgentExecutor",
            RunType::Chain,
            json!({"input": input.clone()}),
        );

        // P1-4: stamp trace_id onto the root run so tool child runs inherit it.
        if let Some(tid) = trace_id {
            match uuid::Uuid::parse_str(&tid) {
                Ok(id) => {
                    root_run.trace_id = Some(id);
                    root_run = root_run.with_metadata("trace_id", json!(tid));
                }
                Err(_) => log::warn!(target: "lc_agents", "invalid trace_id '{}' ignored", tid),
            }
        }

        if let Some(ref callbacks) = self.callbacks {
            for handler in callbacks.handlers() {
                handler.on_chain_start(&root_run, &root_run.inputs).await;
            }
        }

        let mut inputs = HashMap::new();
        inputs.insert("input".to_string(), input.clone());

        if let Some(memory) = &self.memory {
            let memory_guard = memory.lock().await;
            // P1-7: inject every key from memory_variables() into the prompt rather than
            // hardcoding "history" — so the Agent can also read it when the memory
            // component uses a different key (e.g. VectorStore's "memory").
            let variable_keys: Vec<String> = memory_guard
                .memory_variables()
                .into_iter()
                .map(|k| k.to_string())
                .collect();
            let memory_vars = memory_guard
                .load_memory_variables(&inputs)
                .await
                .map_err(|e| AgentError::Other(format!("Failed to load memory: {}", e)))?;
            drop(memory_guard);

            for key in variable_keys {
                if let Some(value) = memory_vars.get(&key) {
                    if let Some(s) = value.as_str() {
                        inputs.insert(key, s.to_string());
                    }
                }
            }
        }

        // B4: semantic recall — best-effort, isolated by namespace. Injected as a
        // delimited facts block under `semantic_memory`; prompt templates surface
        // it with a {semantic_memory} placeholder.
        if let Some(hook) = &self.semantic_memory {
            if let Some(block) = hook.recall(&input).await {
                inputs.insert(SEMANTIC_MEMORY_INPUT_KEY.to_string(), block);
            }
        }

        let intermediate_steps: Vec<AgentStep> = Vec::new();

        let mut metrics = AgentMetrics {
            trace_id: root_run.trace_id.map(|id| id.to_string()),
            ..Default::default()
        };

        // A18: build after trace stamping so the LLM runs inherit the stamped
        // trace id and become children of this chain root.
        let plan_config = build_plan_config(&self.callbacks, &root_run);

        let result = self
            .run_agent_loop(
                inputs.clone(),
                intermediate_steps,
                &mut root_run,
                &mut metrics,
                plan_config.as_ref(),
            )
            .await;

        // 0.22.0 audit fix (H-A9) reconciled with the F7 error-save contract:
        // the errored round is still part of the conversation — write the user's
        // input back to memory so the next round does not lose it. But the raw
        // error text is never stored as assistant output (H-A9): the assistant
        // slot is left empty, so framework noise can never be mistaken for
        // assistant reasoning. A save failure only warns; it never masks the
        // original agent error.
        if let Some(memory) = &self.memory {
            match &result {
                Ok(output) => {
                    let mut outputs = HashMap::new();
                    outputs.insert("output".to_string(), output.clone());

                    memory
                        .lock()
                        .await
                        .save_context(&inputs, &outputs)
                        .await
                        .map_err(|e| AgentError::Other(format!("Failed to save memory: {}", e)))?;
                }
                Err(e) => {
                    let mut outputs = HashMap::new();
                    outputs.insert("output".to_string(), String::new());
                    if let Err(save_err) = memory.lock().await.save_context(&inputs, &outputs).await
                    {
                        log::warn!(
                            "failed to save errored round to memory: {}, original error: {}",
                            save_err,
                            e
                        );
                    }
                }
            }
        }

        // B4: successful answer only — detached extraction + consolidation. The
        // JoinHandle is deliberately dropped: semantic memory never blocks or
        // fails the run, and errored rounds produce no facts (matching the
        // conversation-memory contract of leaving the assistant slot empty).
        if let Some(hook) = &self.semantic_memory {
            if let Ok(output) = &result {
                hook.spawn_extraction(input.clone(), output.clone());
            }
        }

        match &result {
            Ok(output) => {
                root_run.end(json!({"output": output}));
                if let Some(ref callbacks) = self.callbacks {
                    if let Some(ref outputs) = root_run.outputs {
                        for handler in callbacks.handlers() {
                            handler.on_chain_end(&root_run, outputs).await;
                        }
                    }
                }

                // Hooks: on_agent_end (P1-6)
                for hook in &self.hooks {
                    if let Err(e) = hook.on_agent_end(output) {
                        log::warn!("Hook on_agent_end error: {}", e);
                    }
                }
            }
            Err(e) => {
                root_run.end_with_error(e.to_string());
                if let Some(ref callbacks) = self.callbacks {
                    for handler in callbacks.handlers() {
                        handler.on_chain_error(&root_run, &e.to_string()).await;
                    }
                }

                // Hooks: on_error (P1-6)
                for hook in &self.hooks {
                    hook.on_error(&HookError::Other(e.to_string()));
                }
            }
        }

        // P1-5: finalize and publish metrics.
        metrics.duration = started.elapsed();
        metrics.log_summary();
        if let Ok(mut store) = self.metrics_store.lock() {
            *store = Some(metrics.clone());
        }
        self.export_metrics(&metrics).await;

        result
    }

    /// Execute the agent with a RunnableConfig, merging config callbacks
    /// with the executor's own callbacks.
    ///
    /// This is the entry point used by `AgentRunnable` (LCEL adapter).
    /// Config callbacks take precedence over the executor's callbacks.
    pub async fn invoke_with_config(
        &self,
        input: String,
        config: Option<RunnableConfig>,
    ) -> Result<String, AgentError> {
        // If config has callbacks, temporarily use them; otherwise use executor's own
        let effective_callbacks = config
            .as_ref()
            .and_then(|c| c.callbacks.clone())
            .or_else(|| self.callbacks.clone());

        // P1-4: thread trace_id from config metadata so child runs inherit it.
        let trace_id = config
            .as_ref()
            .and_then(|c| c.metadata.get("trace_id"))
            .and_then(|v| v.as_str())
            .map(|s| s.to_string());

        // Create a temporary executor with merged callbacks; metrics_store is
        // Arc-shared so metrics written here propagate back to this executor.
        let merged_executor = AgentExecutor {
            agent: self.agent.clone(),
            tools_by_name: self.tools_by_name.clone(),
            tools: self.tools.clone(),
            max_iterations: self.max_iterations,
            verbose: self.verbose,
            memory: self.memory.clone(),
            callbacks: effective_callbacks,
            hooks: self.hooks.clone(),
            tool_timeout: self.tool_timeout,
            max_concurrency: self.max_concurrency,
            concurrency_sem: self.concurrency_sem.clone(),
            metrics_store: self.metrics_store.clone(),
            response_cache: self.response_cache.clone(),
            cache_namespace: self.cache_namespace.clone(),
            tool_policy: self.tool_policy.clone(),
            approval: self.approval.clone(),
            budget: self.budget.clone(),
            compaction: self.compaction.clone(),
            resume_store: self.resume_store.clone(),
            metrics_sink: self.metrics_sink.clone(),
            cost_tracker: self.cost_tracker.clone(),
            semantic_memory: self.semantic_memory.clone(),
            on_max_iterations: self.on_max_iterations,
            rule_of_two: self.rule_of_two,
            spotlight_tool_output: self.spotlight_tool_output,
        };

        merged_executor.invoke_inner(input, trace_id).await
    }

    /// Stream agent execution as a true async stream of events.
    ///
    /// Each step of the agent loop (tool calls, observations, final answer)
    /// is emitted as an `AgentStreamEvent` as soon as it occurs.
    ///
    /// # Error semantics (A9, unified)
    /// The stream item is `Result<AgentStreamEvent, AgentError>`. A terminal
    /// failure — a permission-policy rejection, a tool timeout, a guarded-tool
    /// abort, or budget exhaustion (A-S2 / A-H1) — is delivered as an
    /// `Err(AgentError)`, which terminates the stream. There is no successful
    /// `Ok(AgentStreamEvent::Error { .. })`; that variant exists for infallible
    /// streams (e.g. [`crate::StreamingFunctionCallingAgent`]) and in-band errors.
    ///
    /// # `Text` event granularity (F3, honest)
    ///
    /// `Text` events carry model text, but their granularity depends on the
    /// agent's [`BaseAgent::plan_stream`] implementation:
    ///
    /// * **ReAct and FunctionCalling agents** stream from the model's chat API,
    ///   so `Text` events arrive **per token** — concat them as they come for a
    ///   live word-stream. A function-calling step that calls a tool streams
    ///   back empty model text (tool calls aren't carried in stream chunks);
    ///   such steps fall back to the non-streaming path internally, so no
    ///   phantom empty `Text` is emitted.
    /// * **Other agents** (plan-and-execute without a streaming inner agent, …)
    ///   use the non-streaming default, so the whole final answer arrives as a
    ///   single `Text` event immediately before `FinalAnswer`.
    ///
    /// `ToolStart`/`ToolEnd` events are always emitted per tool call.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let mut stream = executor.stream("What is Rust?".to_string());
    /// while let Some(event) = stream.next().await {
    ///     match event {
    ///         Ok(AgentStreamEvent::ToolStart { name, input }) => { /* show tool call */ }
    ///         Ok(AgentStreamEvent::ToolEnd { name, output }) => { /* show result */ }
    ///         Ok(AgentStreamEvent::Text { content }) => { print!("{}", content); } /* model text */
    ///         Ok(AgentStreamEvent::FinalAnswer { content }) => { /* show answer */ }
    ///         Err(e) => { /* terminal failure — the loop has ended */ }
    ///         _ => {}
    ///     }
    /// }
    /// ```
    pub fn stream(
        &self,
        input: String,
    ) -> Pin<Box<dyn Stream<Item = Result<AgentStreamEvent, AgentError>> + Send>> {
        let (tx, rx) = tokio::sync::mpsc::channel(32);

        // 0.20.0 A-H2: dropping the returned stream must stop the background agent
        // loop. Without this, a consumer that stops reading (a client disconnect, an
        // early UI cancel) left the loop running — consuming tool calls and LLM tokens
        // for a listener that is gone. The watch channel is the cancel signal: the loop
        // checks it at iteration / tool boundaries, and the wrapper (`AgentEventStream`)
        // sends `true` on drop.
        let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);

        // P2-2: the streaming path also fails fast — unregistered tools emit one error
        // event before ending.
        if let Err(e) = self.validate_tool_registration() {
            tokio::spawn(async move {
                let _ = tx.send(Err(e)).await;
            });
            return Box::pin(AgentEventStream {
                inner: tokio_stream::wrappers::ReceiverStream::new(rx),
                cancel: cancel_tx,
            });
        }

        let agent = self.agent.clone();
        // A11: the stream loop looks tools up by name via the prebuilt index.
        let tools_by_name = self.tools_by_name.clone();
        let max_iterations = self.max_iterations;
        let verbose = self.verbose;
        let tool_timeout = self.tool_timeout;
        let max_concurrency = self.max_concurrency;
        let hooks = self.hooks.clone();
        let tool_policy = self.tool_policy.clone();
        let budget = self.budget.clone();
        let compaction = self.compaction.clone();
        let metrics_store = self.metrics_store.clone();
        let metrics_sink = self.metrics_sink.clone();
        let cost_tracker = self.cost_tracker.clone();
        let on_max_iterations = self.on_max_iterations;
        // 0.22.0 audit fix (H-A1): the stream path previously dropped the
        // cross-cutting capabilities invoke has. Clone callbacks + memory into
        // the spawned task so chain-level callbacks are dispatched, memory
        // history is loaded before the loop and the final answer is saved.
        let callbacks = self.callbacks.clone();
        let memory = self.memory.clone();
        // B4: cloned into the 'static stream task like `memory`; recall runs
        // before the loop, extraction spawns detached from the final answer.
        let semantic_memory = self.semantic_memory.clone();
        // v0.22.1 §S8: copy the A1/A2 toggles so the spawned stream loop reads locals,
        // not `&self` (disjoint capture holds here; referencing `self.` would borrow the
        // whole executor into the `'static` task because `Mutex<dyn BaseMemory>` is invariant).
        let rule_of_two = self.rule_of_two;
        let spotlight_tool_output = self.spotlight_tool_output;

        tokio::spawn(async move {
            let mut intermediate_steps: Vec<AgentStep> = Vec::new();
            let mut inputs = HashMap::new();
            inputs.insert("input".to_string(), input.clone());

            // H-A1: chain callbacks / trace parity with invoke — build the root
            // RunTree, dispatch on_chain_start and on_agent_start hooks, and load
            // memory variables into the inputs before the loop.
            //
            // A18: planning-round `on_llm_*` callbacks are now dispatched on this
            // path too — `plan_config` carries the same callbacks + trace linkage
            // (`__lc_parent_run_id` / `__lc_trace_id`) the invoke path stamps, so
            // the provider-built LLM runs are children of this chain root.
            //
            // Remaining known gaps (honest): tool-level `on_tool_*` callbacks and
            // RunTree trace_id stamping from RunnableConfig metadata (stream()
            // takes no config) are still not dispatched on this path — invoke's
            // tool child-run tracing has no equivalent here because tool execution
            // goes through `execute_tool_for_stream` without a RunTree.
            let mut root_run = RunTree::new(
                "AgentExecutor",
                RunType::Chain,
                json!({"input": inputs.get("input").cloned().unwrap_or_default()}),
            );
            if let Some(ref callbacks) = callbacks {
                for handler in callbacks.handlers() {
                    handler.on_chain_start(&root_run, &root_run.inputs).await;
                }
            }
            let plan_config = build_plan_config(&callbacks, &root_run);
            for hook in &hooks {
                if let Err(e) = hook.on_agent_start(&input) {
                    log::warn!("Hook on_agent_start error: {}", e);
                }
            }

            if let Some(memory) = &memory {
                let memory_guard = memory.lock().await;
                let variable_keys: Vec<String> = memory_guard
                    .memory_variables()
                    .into_iter()
                    .map(|k| k.to_string())
                    .collect();
                let loaded = match memory_guard.load_memory_variables(&inputs).await {
                    Ok(vars) => vars,
                    Err(e) => {
                        let msg = format!("Failed to load memory: {e}");
                        stream_chain_error(&callbacks, &mut root_run, &msg).await;
                        for hook in &hooks {
                            hook.on_error(&HookError::Other(msg.clone()));
                        }
                        let _ = tx.send(Err(AgentError::Other(msg))).await;
                        return;
                    }
                };
                drop(memory_guard);
                for key in variable_keys {
                    if let Some(value) = loaded.get(&key) {
                        if let Some(s) = value.as_str() {
                            inputs.insert(key, s.to_string());
                        }
                    }
                }
            }

            // B4: semantic recall, same best-effort semantics as invoke.
            if let Some(hook) = &semantic_memory {
                if let Some(block) = hook.recall(&input).await {
                    inputs.insert(SEMANTIC_MEMORY_INPUT_KEY.to_string(), block);
                }
            }

            // Budget gate (§4.2): start the stream timer + accumulate metrics (same
            // semantics as the invoke path).
            let loop_start = Instant::now();
            let mut metrics = AgentMetrics::default();

            for iteration in 0..max_iterations {
                if verbose {
                    log::info!("=== Stream Iteration {} ===", iteration + 1);
                }

                // 0.20.0 A-H2: the consumer dropped the stream → stop before the next
                // plan. Any tool already in flight is allowed to finish (cooperative
                // cancellation), but no new plan / tool starts.
                if *cancel_rx.borrow() {
                    return;
                }

                // Budget gate: iteration-level (iteration count + wall-clock). Over the
                // limit → send Err and stop.
                if let Some(err) =
                    budget_iteration_gate(budget.as_ref(), max_iterations, iteration, loop_start)
                {
                    stream_chain_error(&callbacks, &mut root_run, &err.to_string()).await;
                    publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                    let _ = tx.send(Err(err)).await;
                    return;
                }

                // P2-9: rate-limit / quota check before the LLM call (also applies on
                // the streaming path).
                if let Err(e) = run_before_completion_hooks(&hooks, &inputs) {
                    let msg = e.to_string();
                    stream_chain_error(&callbacks, &mut root_run, &msg).await;
                    publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                    let _ = tx.send(Err(AgentError::Other(msg))).await;
                    return;
                }
                // F3: streaming planning — the agent forwards model text token by token
                // through on_token as Text events. ReAct / FunctionCalling override
                // plan_stream to go through `stream_chat` for a real word-by-word stream;
                // other agents use the default implementation (the whole answer as a
                // single Text event), matching the old path's behavior.
                // 0.21.0 S6.1: context compaction before planning — same semantics as
                // the invoke path (`run_agent_loop_from`), so the two paths cannot diverge.
                if let Some(config) = &compaction {
                    let tokens = metrics.total_tokens.unwrap_or(0);
                    let (kept, dropped) = config.compact(&intermediate_steps, tokens);
                    if dropped > 0 {
                        log::info!(
                            target: "lc_agents::compaction",
                            "compacted {} of {} steps ({} remain) [stream]",
                            dropped,
                            dropped + kept.len(),
                            kept.len()
                        );
                        intermediate_steps = kept;
                        metrics.compactions += 1;
                    }
                }
                let output = {
                    // Must not shadow the outer tx: the closure's `move` would carry it
                    // away, and the ToolStart/FinalAnswer below would no longer be able
                    // to use the outer tx.
                    let send_tx = tx.clone();
                    // The callback receives its own String (F3): the async block owns
                    // the token directly instead of borrowing the argument, so the future
                    // is 'static and can be cast to a trait object with `as`.
                    let mut on_token = move |token: String| {
                        let tx = send_tx.clone();
                        Box::pin(async move {
                            let _ = tx.send(Ok(AgentStreamEvent::Text { content: token })).await;
                        }) as Pin<Box<dyn Future<Output = ()> + Send>>
                    };
                    match agent
                        .plan_stream(
                            &intermediate_steps,
                            &inputs,
                            &mut on_token,
                            plan_config.as_ref(),
                        )
                        .await
                    {
                        Ok(o) => o,
                        Err(e) => {
                            let msg = e.to_string();
                            stream_chain_error(&callbacks, &mut root_run, &msg).await;
                            for hook in &hooks {
                                hook.on_error(&HookError::Other(msg.clone()));
                            }
                            publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start)
                                .await;
                            let _ = tx.send(Err(AgentError::Other(msg))).await;
                            return;
                        }
                    }
                };
                let usage = agent.last_token_usage();
                // P2-9: accumulate the real token usage after the LLM call (same semantics
                // as plan_cached on the invoke path).
                metrics.llm_calls += 1;
                if let Some(u) = &usage {
                    metrics.add_token_usage(u);
                }
                run_after_completion_hooks(&hooks, &output, usage.as_ref());
                // Budget gate: cumulative tokens after the LLM call. Over the limit →
                // send Err and stop.
                if let Some(err) = budget_token_gate(budget.as_ref(), &metrics) {
                    stream_chain_error(&callbacks, &mut root_run, &err.to_string()).await;
                    publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                    let _ = tx.send(Err(err)).await;
                    return;
                }
                // B3 (0.22.4): cumulative USD spend gate, same semantics as invoke.
                if let Some(tracker) = &cost_tracker {
                    let spent = tracker.total_cost_usd().await;
                    if let Some(err) = budget_cost_gate(budget.as_ref(), spent) {
                        stream_chain_error(&callbacks, &mut root_run, &err.to_string()).await;
                        publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                        let _ = tx.send(Err(err)).await;
                        return;
                    }
                }

                match output {
                    AgentOutput::Finish(finish) => {
                        let content = finish.output().unwrap_or("").to_string();
                        // H-A1: memory save on the final answer, same as invoke's
                        // post-answer save_context. A save failure only warns — it
                        // must not mask a successfully finished stream.
                        if let Some(memory) = &memory {
                            let mut outputs = HashMap::new();
                            outputs.insert("output".to_string(), content.clone());
                            if let Err(e) =
                                memory.lock().await.save_context(&inputs, &outputs).await
                            {
                                log::warn!("failed to save final answer to memory [stream]: {e}");
                            }
                        }
                        // B4: detached fact extraction, same as invoke.
                        if let Some(hook) = &semantic_memory {
                            hook.spawn_extraction(input.clone(), content.clone());
                        }

                        root_run.end(json!({"output": content.clone()}));
                        if let Some(ref callbacks) = callbacks {
                            if let Some(ref outputs) = root_run.outputs {
                                for handler in callbacks.handlers() {
                                    handler.on_chain_end(&root_run, outputs).await;
                                }
                            }
                        }
                        for hook in &hooks {
                            if let Err(e) = hook.on_agent_end(&content) {
                                log::warn!("Hook on_agent_end error: {}", e);
                            }
                        }
                        // P1-8 streaming fusion: the model text was already emitted piece
                        // by piece by plan_stream through on_token (Text events); here
                        // only the FinalAnswer terminal event is sent — the full answer is
                        // not repeated.
                        publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                        let _ = tx.send(Ok(AgentStreamEvent::FinalAnswer { content })).await;
                        return;
                    }

                    AgentOutput::Action(action) => {
                        // 0.22.0 audit fix (H-A5): the ReAct parse-repair pseudo-tool
                        // is not a real tool — feed its message back as the
                        // observation so the model can retry (standard ReAct repair
                        // loop), matching the invoke path.
                        if action.tool == crate::react::agent::PARSE_ERROR_TOOL {
                            let observation = match &action.tool_input {
                                ToolInput::String { value } => value.clone(),
                                ToolInput::Object { value } => value.to_string(),
                            };
                            if verbose {
                                log::info!("Parse repair observation: {}", observation);
                            }
                            intermediate_steps.push(AgentStep::new(action, observation));
                            continue;
                        }
                        // P2-9: the streaming path also enforces the tool permission
                        // policy.
                        if let Some(policy) = &tool_policy {
                            if let Err(e) = policy.check(&action.tool) {
                                let msg = e.to_string();
                                stream_chain_error(&callbacks, &mut root_run, &msg).await;
                                publish_metrics(
                                    &metrics,
                                    &metrics_store,
                                    &metrics_sink,
                                    loop_start,
                                )
                                .await;
                                let _ = tx.send(Err(AgentError::Other(msg))).await;
                                return;
                            }
                        }
                        let tool_name = action.tool.clone();
                        let tool_input_str = match &action.tool_input {
                            ToolInput::String { value: s } => s.clone(),
                            ToolInput::Object { value: v } => {
                                serde_json::to_string(v).unwrap_or_default()
                            }
                        };

                        // A11: the budget gate runs **before** `ToolStart` is emitted.
                        // Previously the gate ran after, so a rejection left an orphan
                        // `ToolStart` with no matching `ToolEnd`/error. Order now mirrors
                        // the invoke path: reject first, emit the start event only when
                        // the call is actually allowed.
                        metrics.tool_calls += 1;
                        if let Some(err) = budget_tool_gate(budget.as_ref(), &metrics, loop_start) {
                            stream_chain_error(&callbacks, &mut root_run, &err.to_string()).await;
                            publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start)
                                .await;
                            let _ = tx.send(Err(err)).await;
                            return;
                        }

                        let _ = tx
                            .send(Ok(AgentStreamEvent::ToolStart {
                                name: tool_name.clone(),
                                input: tool_input_str.clone(),
                            }))
                            .await;

                        // 0.20.0 A-H2: dropped mid-iteration → do not start a new tool.
                        if *cancel_rx.borrow() {
                            return;
                        }

                        // Execute the tool. A tool **execution** failure becomes an
                        // observation fed back to the loop (S3.1) so the agent can
                        // recover. Framework guardrails (A-H1, 0.20.0) —
                        // `ControlAbort` handoff/depth guard, `ToolNotFound`, input
                        // serialization — reject the call *before* execution; the agent
                        // cannot recover from them by re-planning, so they end the
                        // stream hard, matching the non-streaming invoke path.
                        let observation = match execute_tool_for_stream(
                            &tools_by_name,
                            &action,
                            tool_timeout,
                            spotlight_tool_output,
                            rule_of_two,
                        )
                        .await
                        {
                            Ok(obs) => obs,
                            Err(e @ AgentError::ToolExecutionError(_)) => {
                                tool_error_observation(&e)
                            }
                            Err(e) => {
                                let msg = e.to_string();
                                stream_chain_error(&callbacks, &mut root_run, &msg).await;
                                publish_metrics(
                                    &metrics,
                                    &metrics_store,
                                    &metrics_sink,
                                    loop_start,
                                )
                                .await;
                                let _ = tx.send(Err(AgentError::Other(msg))).await;
                                return;
                            }
                        };

                        let _ = tx
                            .send(Ok(AgentStreamEvent::ToolEnd {
                                name: tool_name,
                                output: observation.clone(),
                            }))
                            .await;

                        intermediate_steps.push(AgentStep::new(action, observation));
                    }

                    AgentOutput::Actions(actions) => {
                        // P2-9: parallel tools also pass the permission policy first.
                        if let Some(policy) = &tool_policy {
                            for action in &actions {
                                if let Err(e) = policy.check(&action.tool) {
                                    let msg = e.to_string();
                                    stream_chain_error(&callbacks, &mut root_run, &msg).await;
                                    publish_metrics(
                                        &metrics,
                                        &metrics_store,
                                        &metrics_sink,
                                        loop_start,
                                    )
                                    .await;
                                    let _ = tx.send(Err(AgentError::Other(msg))).await;
                                    return;
                                }
                            }
                        }
                        // A11: budget gate runs **before** any `ToolStart` is emitted for the
                        // batch, so a rejection leaves no orphan start events.
                        metrics.tool_calls += actions.len();
                        if let Some(err) = budget_tool_gate(budget.as_ref(), &metrics, loop_start) {
                            stream_chain_error(&callbacks, &mut root_run, &err.to_string()).await;
                            publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start)
                                .await;
                            let _ = tx.send(Err(err)).await;
                            return;
                        }

                        for action in &actions {
                            let tool_name = action.tool.clone();
                            let tool_input_str = match &action.tool_input {
                                ToolInput::String { value: s } => s.clone(),
                                ToolInput::Object { value: v } => {
                                    serde_json::to_string(v).unwrap_or_default()
                                }
                            };

                            let _ = tx
                                .send(Ok(AgentStreamEvent::ToolStart {
                                    name: tool_name.clone(),
                                    input: tool_input_str,
                                }))
                                .await;
                        }

                        // 0.20.0 A-H2: dropped mid-iteration → do not start a new batch.
                        if *cancel_rx.borrow() {
                            return;
                        }

                        let observations = match execute_tools_parallel_for_stream(
                            &tools_by_name,
                            &actions,
                            tool_timeout,
                            max_concurrency,
                            spotlight_tool_output,
                            rule_of_two,
                        )
                        .await
                        {
                            Ok(obs) => obs,
                            // A-H1 (0.20.0): a framework guardrail in any one tool
                            // of the batch ends the stream hard, matching the
                            // invoke-parallel path. Execution errors were already
                            // converted to observations inside the helper.
                            Err(e) => {
                                let msg = e.to_string();
                                stream_chain_error(&callbacks, &mut root_run, &msg).await;
                                publish_metrics(
                                    &metrics,
                                    &metrics_store,
                                    &metrics_sink,
                                    loop_start,
                                )
                                .await;
                                let _ = tx.send(Err(AgentError::Other(msg))).await;
                                return;
                            }
                        };

                        for (action, observation) in
                            actions.into_iter().zip(observations.into_iter())
                        {
                            let _ = tx
                                .send(Ok(AgentStreamEvent::ToolEnd {
                                    name: action.tool.clone(),
                                    output: observation.clone(),
                                }))
                                .await;

                            intermediate_steps.push(AgentStep::new(action, observation));
                        }
                    }
                }
            }

            // 0.22.0 C4 fix: the iteration cap is a failure by default —
            // surface `MaxIterationsReached` on the stream instead of streaming
            // a placeholder that looks like a real answer.
            log::warn!(
                "agent reached max iterations; policy: {:?}",
                on_max_iterations
            );
            if on_max_iterations == MaxIterationsPolicy::Error {
                stream_chain_error(&callbacks, &mut root_run, "max iterations reached").await;
                publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
                let _ = tx.send(Err(AgentError::MaxIterationsReached)).await;
                return;
            }
            // Legacy placeholder policy: the stopped response is treated as a final
            // answer — run the H-A1 terminal path (memory save + chain end) too.
            let finish = agent.return_stopped_response(&intermediate_steps);
            let content = finish.output().unwrap_or("").to_string();
            if let Some(memory) = &memory {
                let mut outputs = HashMap::new();
                outputs.insert("output".to_string(), content.clone());
                if let Err(e) = memory.lock().await.save_context(&inputs, &outputs).await {
                    log::warn!("failed to save final answer to memory [stream]: {e}");
                }
            }
            // B4: detached fact extraction, same as invoke.
            if let Some(hook) = &semantic_memory {
                hook.spawn_extraction(input.clone(), content.clone());
            }
            root_run.end(json!({"output": content.clone()}));
            if let Some(ref callbacks) = callbacks {
                if let Some(ref outputs) = root_run.outputs {
                    for handler in callbacks.handlers() {
                        handler.on_chain_end(&root_run, outputs).await;
                    }
                }
            }
            for hook in &hooks {
                if let Err(e) = hook.on_agent_end(&content) {
                    log::warn!("Hook on_agent_end error: {}", e);
                }
            }
            publish_metrics(&metrics, &metrics_store, &metrics_sink, loop_start).await;
            let _ = tx.send(Ok(AgentStreamEvent::FinalAnswer { content })).await;
        });

        Box::pin(AgentEventStream {
            inner: tokio_stream::wrappers::ReceiverStream::new(rx),
            cancel: cancel_tx,
        })
    }
}

impl std::fmt::Debug for AgentExecutor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AgentExecutor")
            .field("max_iterations", &self.max_iterations)
            .field("verbose", &self.verbose)
            .field("tools_count", &self.tools.len())
            .field("has_memory", &self.memory.is_some())
            .field("tool_timeout", &self.tool_timeout)
            .field("max_concurrency", &self.max_concurrency)
            .field("has_response_cache", &self.response_cache.is_some())
            .field("has_tool_policy", &self.tool_policy.is_some())
            .field("has_resume_store", &self.resume_store.is_some())
            .field("has_metrics_sink", &self.metrics_sink.is_some())
            .field("has_cost_tracker", &self.cost_tracker.is_some())
            .field("has_semantic_memory", &self.semantic_memory.is_some())
            .field(
                "has_metrics",
                &self
                    .metrics_store
                    .lock()
                    .ok()
                    .map(|guard| guard.is_some())
                    .unwrap_or(false),
            )
            .finish()
    }
}

/// Stream wrapper returned by [`AgentExecutor::stream`]: cancels the background agent
/// loop when the consumer drops the stream (0.20.0 A-H2). A dropped stream means the
/// listener is gone — the loop must stop burning tool calls and LLM tokens instead of
/// running the remaining iterations invisibly.
struct AgentEventStream {
    /// The live event channel.
    inner: tokio_stream::wrappers::ReceiverStream<Result<AgentStreamEvent, AgentError>>,
    /// Set to `true` on drop; the loop observes it via `cancel_rx` at iteration / tool
    /// boundaries and stops cooperatively (letting any in-flight tool finish).
    cancel: tokio::sync::watch::Sender<bool>,
}

impl Stream for AgentEventStream {
    type Item = Result<AgentStreamEvent, AgentError>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        Pin::new(&mut self.inner).poll_next(cx)
    }
}

impl Drop for AgentEventStream {
    fn drop(&mut self) {
        let _ = self.cancel.send(true);
    }
}

/// 0.22.0 audit fix (H-A1): dispatches the chain-error callbacks on the stream
/// path, mirroring invoke's `on_chain_error` handling. Best-effort — never
/// fails, only marks the root run as errored first.
async fn stream_chain_error(
    callbacks: &Option<Arc<CallbackManager>>,
    root_run: &mut RunTree,
    message: &str,
) {
    root_run.end_with_error(message.to_string());
    if let Some(callbacks) = callbacks {
        for handler in callbacks.handlers() {
            handler.on_chain_error(root_run, message).await;
        }
    }
}

/// Publishes `AgentMetrics` at the end of a stream (aligned with the invoke path):
/// clone → fill duration → audit log → write `metrics_store`.
///
/// **Ordering constraint (race)**: the stream closure runs in `tokio::spawn`, so every
/// termination path must **`publish_metrics` before `tx.send(terminal event)`** —
/// otherwise a consumer that checks `last_metrics()` immediately after draining the
/// stream may read `None` (the event arrived but the write has not happened yet).
async fn publish_metrics(
    metrics: &AgentMetrics,
    metrics_store: &Arc<Mutex<Option<AgentMetrics>>>,
    metrics_sink: &Option<Arc<dyn MetricsSink>>,
    started: Instant,
) {
    let mut m = metrics.clone();
    m.duration = started.elapsed();
    m.log_summary();
    if let Ok(mut guard) = metrics_store.lock() {
        *guard = Some(m.clone());
    }
    if let Some(sink) = metrics_sink {
        let evt = ObsEvent::AgentMetrics(m);
        if let Err(e) = sink.export(&evt).await {
            log::warn!(target: "lc_agents::metrics", "agent metrics export failed: {e}");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn cache_key_is_deterministic_across_runs() {
        // Two runs with identical content must produce the same key, even
        // though a freshly allocated HashMap may iterate in a different order.
        let a: HashMap<String, String> = HashMap::from([
            ("question".into(), "what is rust".into()),
            ("user_id".into(), "42".into()),
            ("session".into(), "abc".into()),
        ]);
        let b: HashMap<String, String> = HashMap::from([
            ("session".into(), "abc".into()),
            ("question".into(), "what is rust".into()),
            ("user_id".into(), "42".into()),
        ]);

        let key_a = AgentExecutor::cache_key("ns", &a, &[]);
        let key_b = AgentExecutor::cache_key("ns", &b, &[]);

        assert_eq!(key_a, key_b, "identical inputs must hash to the same key");
    }

    #[test]
    fn cache_key_differs_when_inputs_differ() {
        let base: HashMap<String, String> = HashMap::from([("k".into(), "v".into())]);
        let changed: HashMap<String, String> = HashMap::from([("k".into(), "other".into())]);

        let key_base = AgentExecutor::cache_key("ns", &base, &[]);
        let key_changed = AgentExecutor::cache_key("ns", &changed, &[]);

        assert_ne!(key_base, key_changed);
    }
}