lc_agents/executor/

engine.rs

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

use super::budget::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::index_tools;
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::types::{AgentAction, AgentOutput, AgentStep, ToolInput};
use lc_callbacks::{semconv::GEN_AI_OPERATION_NAME, 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::path::PathBuf;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::Duration;
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).
pub(super) 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));
            }
        }
        // T10: root run is a GenAI agent invocation, not a generic chain.
        root_run = root_run.with_metadata(GEN_AI_OPERATION_NAME, json!("invoke_agent"));

        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),
            }
        }
        // T10: classify the root run as the 2026 `invoke_agent` GenAI operation
        // before any handler sees on_chain_start, so OtelHandler names the span
        // `invoke_agent` (unstamped chain runs still produce "chain").
        root_run = root_run.with_metadata(GEN_AI_OPERATION_NAME, json!("invoke_agent"));

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

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()
    }
}

#[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);
    }
}