car_engine/

executor.rs

//! Core execution engine — propose → validate → execute → commit.

use crate::cache::ResultCache;
use crate::capabilities::CapabilitySet;
use crate::checkpoint::Checkpoint;
use crate::rate_limit::{RateLimit, RateLimiter};
use car_eventlog::{EventKind, EventLog, SpanStatus};
use car_ir::{
    build_dag, Action, ActionProposal, ActionResult, ActionStatus, ActionType, CostSummary,
    FailureBehavior, ProposalResult, ToolSchema,
};
use car_policy::PolicyEngine;
use car_state::StateStore;
use car_validator::validate_action;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex as TokioMutex, RwLock as TokioRwLock};
use tokio::time::timeout;
use tracing::instrument;
use uuid::Uuid;

/// Retry backoff constants.
const RETRY_BASE_DELAY_MS: u64 = 100;
const RETRY_BACKOFF_FACTOR: u64 = 2;

// ---------------------------------------------------------------------------
// Replan types — failure recovery via model callback
// ---------------------------------------------------------------------------

/// Callback trait for replanning failed proposals.
/// Implement this to let the runtime ask the model for an alternative plan
/// when a proposal aborts.
#[async_trait::async_trait]
pub trait ReplanCallback: Send + Sync {
    async fn replan(&self, ctx: &ReplanContext) -> Result<ActionProposal, String>;
}

/// Context provided to the replan callback so the model can generate an alternative.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ReplanContext {
    /// Original proposal ID.
    pub proposal_id: String,
    /// Which replan attempt this is (1-indexed; attempt 1 = first replan after initial failure).
    pub attempt: u32,
    /// Actions that failed and caused the abort.
    pub failed_actions: Vec<FailedActionSummary>,
    /// Action IDs that succeeded before the abort (now rolled back).
    pub completed_action_ids: Vec<String>,
    /// State snapshot after rollback.
    pub state_snapshot: HashMap<String, Value>,
    /// How many replans remain.
    pub replans_remaining: u32,
    /// Original proposal source (model name, agent, etc.).
    pub original_source: String,
    /// Total actions in the original proposal.
    pub original_action_count: usize,
    /// The original goal/context from the proposal (for generating alternatives).
    pub original_context: HashMap<String, Value>,
}

/// Summary of a failed action, included in ReplanContext.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct FailedActionSummary {
    pub action_id: String,
    pub tool: Option<String>,
    pub error: String,
    pub parameters: HashMap<String, Value>,
}

/// Configuration for the replan loop.
#[derive(Debug, Clone)]
pub struct ReplanConfig {
    /// Maximum number of replan attempts. 0 = disabled (default).
    pub max_replans: u32,
    /// Delay in milliseconds between replan attempts. Prevents burning through
    /// attempts instantly if the model returns garbage fast. 0 = no delay.
    pub delay_ms: u64,
    /// If true, replan proposals are scored via car-planner's verify() before
    /// execution. Proposals with errors are rejected without executing.
    /// Prevents the engine from running a worse plan than the one that failed.
    pub verify_before_execute: bool,
}

impl Default for ReplanConfig {
    fn default() -> Self {
        Self {
            max_replans: 0,
            delay_ms: 0,
            verify_before_execute: true,
        }
    }
}

/// Trait for tool execution. Implement this to provide tools to the runtime.
///
/// In-process: implement directly with function calls.
/// Daemon mode: implement by sending JSON-RPC to the client.
#[async_trait::async_trait]
pub trait ToolExecutor: Send + Sync {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String>;
}

/// Deterministic key for idempotency deduplication.
fn idempotency_key(action: &Action) -> String {
    let sorted: std::collections::BTreeMap<_, _> = action.parameters.iter().collect();
    let params = serde_json::to_string(&sorted).unwrap_or_default();
    format!(
        "{}:{}:{}",
        serde_json::to_string(&action.action_type).unwrap_or_default(),
        action.tool.as_deref().unwrap_or(""),
        params
    )
}

fn rejected_result(action_id: &str, error: String) -> ActionResult {
    ActionResult {
        action_id: action_id.to_string(),
        status: ActionStatus::Rejected,
        output: None,
        error: Some(error),
        state_changes: HashMap::new(),
        duration_ms: None,
        timestamp: chrono::Utc::now(),
    }
}

/// Capture only the state keys relevant to an action (state_dependencies + expected_effects).
/// Returns an empty map if the action declares no relevant keys (e.g., tool calls
/// that don't interact with state).
fn snapshot_relevant_keys(
    state: &car_state::StateStore,
    action: &Action,
) -> HashMap<String, Value> {
    let mut keys: std::collections::HashSet<&str> = std::collections::HashSet::new();
    for dep in &action.state_dependencies {
        keys.insert(dep.as_str());
    }
    for key in action.expected_effects.keys() {
        keys.insert(key.as_str());
    }
    // For state_write actions, capture the key being written
    if action.action_type == ActionType::StateWrite {
        if let Some(key) = action.parameters.get("key").and_then(|v| v.as_str()) {
            keys.insert(key);
        }
    }

    if keys.is_empty() {
        // No declared state interaction — skip snapshot (empty map)
        return HashMap::new();
    }

    keys.iter()
        .filter_map(|&k| state.get(k).map(|v| (k.to_string(), v)))
        .collect()
}

fn skipped_result(action_id: &str, reason: &str) -> ActionResult {
    ActionResult {
        action_id: action_id.to_string(),
        status: ActionStatus::Skipped,
        output: None,
        error: Some(reason.to_string()),
        state_changes: HashMap::new(),
        duration_ms: None,
        timestamp: chrono::Utc::now(),
    }
}

/// Prefix on the `error` field of an `ActionResult` that distinguishes
/// "the user pulled the plug" from "earlier abort cascaded." The
/// `ActionStatus` itself is `Skipped` in both cases (introducing a
/// new variant ripples through every IR consumer + FFI binding); the
/// prefix lets callers like the A2A bridge tell the cases apart
/// without string-matching a magic literal.
pub const CANCELED_PREFIX: &str = "canceled: ";

/// Result for an action that didn't run because the proposal was
/// cancelled mid-flight. Distinct from `Skipped` so callers can tell
/// "didn't run because of an earlier abort" from "didn't run because
/// the user pulled the plug."
fn canceled_result(action_id: &str, reason: &str) -> ActionResult {
    ActionResult {
        action_id: action_id.to_string(),
        status: ActionStatus::Skipped,
        output: None,
        error: Some(format!("{}{}", CANCELED_PREFIX, reason)),
        state_changes: HashMap::new(),
        duration_ms: None,
        timestamp: chrono::Utc::now(),
    }
}

/// Format a tool result for feeding back to a model.
pub fn format_tool_result(result: &ActionResult) -> String {
    match result.status {
        ActionStatus::Succeeded => match &result.output {
            Some(v) => serde_json::to_string(v).unwrap_or_else(|_| v.to_string()),
            None => String::new(),
        },
        ActionStatus::Rejected => format!("[REJECTED] {}", result.error.as_deref().unwrap_or("")),
        ActionStatus::Failed => format!("[FAILED] {}", result.error.as_deref().unwrap_or("")),
        _ => format!(
            "[{:?}] {}",
            result.status,
            result.error.as_deref().unwrap_or("")
        ),
    }
}

/// Budget constraints for proposal execution.
#[derive(Debug, Clone)]
pub struct CostBudget {
    pub max_tool_calls: Option<u32>,
    pub max_duration_ms: Option<f64>,
    pub max_actions: Option<u32>,
}

/// Common Agent Runtime — deterministic execution layer.
///
/// Lock ordering discipline (never hold multiple simultaneously, never hold sync locks across .await):
/// 1. capabilities (RwLock, read-only during execution)
/// 2. tools (RwLock, read-only during execution)
/// 3. policies (RwLock, read-only during execution)
/// 4. session_policies (RwLock to find the per-session engine, then read inner Arc)
/// 5. cost_budget (RwLock, read-only during execution)
/// 6. log (TokioMutex, acquired/released per event)
/// 7. tool_executor (TokioMutex, clone Arc and drop before await)
/// 8. idempotency_cache (TokioMutex, acquired/released per check)
///
/// StateStore uses parking_lot::Mutex (sync) — NEVER hold across .await points.
pub struct Runtime {
    pub state: Arc<StateStore>,
    pub tools: Arc<TokioRwLock<HashMap<String, ToolSchema>>>,
    pub policies: Arc<TokioRwLock<PolicyEngine>>,
    /// Per-session policy registries. Hosts that multiplex multiple
    /// concurrent agent sessions over a single `Runtime` (IDE-style
    /// frontends with per-project rules, multi-tenant servers) call
    /// [`Runtime::open_session`] to mint an id, then
    /// [`Runtime::register_policy_in_session`] to attach session-scoped
    /// rules. Validation under that session walks the global registry
    /// AND the session's — both must pass. Sessions can deny what
    /// global allows; sessions cannot allow what global denies.
    /// Closing a session drops its registry and any closures it holds.
    /// See `docs/proposals/per-session-policy-scoping.md`.
    pub session_policies: Arc<TokioRwLock<HashMap<String, Arc<TokioRwLock<PolicyEngine>>>>>,
    pub log: Arc<TokioMutex<EventLog>>,
    pub rate_limiter: Arc<RateLimiter>,
    pub result_cache: Arc<ResultCache>,
    tool_executor: TokioMutex<Option<Arc<dyn ToolExecutor>>>,
    idempotency_cache: TokioMutex<HashMap<String, ActionResult>>,
    cost_budget: TokioRwLock<Option<CostBudget>>,
    capabilities: TokioRwLock<Option<CapabilitySet>>,
    inference_engine: Option<Arc<car_inference::InferenceEngine>>,
    /// Optional memgine for skill learning (auto-distillation after execution).
    memgine: Option<Arc<TokioMutex<car_memgine::MemgineEngine>>>,
    /// Whether to auto-distill skills after each proposal execution.
    auto_distill: bool,
    /// Optional trajectory store for persisting execution traces.
    trajectory_store: Option<Arc<car_memgine::TrajectoryStore>>,
    /// Optional replan callback for failure recovery.
    replan_callback: TokioMutex<Option<Arc<dyn ReplanCallback>>>,
    /// Replan configuration.
    replan_config: TokioRwLock<ReplanConfig>,
    /// Canonical tool registry (optional — new code should use this).
    pub registry: Arc<crate::registry::ToolRegistry>,
}

impl Runtime {
    pub fn new() -> Self {
        Self {
            state: Arc::new(StateStore::new()),
            tools: Arc::new(TokioRwLock::new(HashMap::new())),
            policies: Arc::new(TokioRwLock::new(PolicyEngine::new())),
            session_policies: Arc::new(TokioRwLock::new(HashMap::new())),
            log: Arc::new(TokioMutex::new(EventLog::new())),
            rate_limiter: Arc::new(RateLimiter::new()),
            result_cache: Arc::new(ResultCache::new()),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            memgine: None,
            auto_distill: false,
            trajectory_store: None,
            replan_callback: TokioMutex::new(None),
            replan_config: TokioRwLock::new(ReplanConfig::default()),
            registry: Arc::new(crate::registry::ToolRegistry::new()),
        }
    }

    /// Create a runtime with shared state, event log, and policies.
    /// Each runtime gets its own tool set, executor, and idempotency cache.
    pub fn with_shared(
        state: Arc<StateStore>,
        log: Arc<TokioMutex<EventLog>>,
        policies: Arc<TokioRwLock<PolicyEngine>>,
    ) -> Self {
        Self {
            state,
            tools: Arc::new(TokioRwLock::new(HashMap::new())),
            policies,
            // Session-policy registries are per-runtime — sharing them
            // across embedders that share global policies would defeat
            // the isolation point. Hosts that genuinely want shared
            // sessions should drive them through one shared Runtime.
            session_policies: Arc::new(TokioRwLock::new(HashMap::new())),
            log,
            rate_limiter: Arc::new(RateLimiter::new()),
            result_cache: Arc::new(ResultCache::new()),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            memgine: None,
            auto_distill: false,
            trajectory_store: None,
            replan_callback: TokioMutex::new(None),
            replan_config: TokioRwLock::new(ReplanConfig::default()),
            registry: Arc::new(crate::registry::ToolRegistry::new()),
        }
    }

    // ─── Session policy lifecycle ───────────────────────────────────
    //
    // Per-session policy scoping. Policies registered against a
    // session apply to proposals executed under that session id (via
    // [`Self::execute_with_session`] / [`Self::execute_with_session_and_cancel`]).
    // Policies registered globally always apply, on top of any
    // session-scoped layer. See `docs/proposals/per-session-policy-scoping.md`.

    /// Mint a new session id and pre-register an empty policy engine
    /// under it. Hosts call this once per concurrent agent context
    /// (an IDE project window, a multi-tenant client, etc.) and pair
    /// it with [`Self::close_session`] when the context ends.
    ///
    /// Returns the opaque id to pass to subsequent
    /// [`Self::register_policy_in_session`] / [`Self::execute_with_session`]
    /// calls. Ids are UUIDs so collisions across concurrent calls
    /// don't matter.
    pub async fn open_session(&self) -> String {
        let id = Uuid::new_v4().to_string();
        let mut sessions = self.session_policies.write().await;
        sessions.insert(id.clone(), Arc::new(TokioRwLock::new(PolicyEngine::new())));
        id
    }

    /// Drop the session and every policy scoped to it. Returns true
    /// if a session by that id existed; false if it didn't (already
    /// closed, never opened, etc.). Idempotent in effect — closing a
    /// missing session is a no-op the caller is free to ignore.
    pub async fn close_session(&self, session_id: &str) -> bool {
        let mut sessions = self.session_policies.write().await;
        sessions.remove(session_id).is_some()
    }

    /// Register a policy under a specific session id. The policy
    /// applies only when a proposal is executed under that session;
    /// proposals executed without a session (the default) only see
    /// global policies.
    ///
    /// Returns `Err(...)` if the session is unknown — callers either
    /// forgot to call [`Self::open_session`] or are using a
    /// stale/closed id.
    pub async fn register_policy_in_session(
        &self,
        session_id: &str,
        name: &str,
        check: car_policy::PolicyCheck,
        description: &str,
    ) -> Result<(), String> {
        let engine = {
            let sessions = self.session_policies.read().await;
            sessions
                .get(session_id)
                .cloned()
                .ok_or_else(|| format!("unknown session id '{session_id}'"))?
        };
        let mut engine = engine.write().await;
        engine.register(name, check, description);
        Ok(())
    }

    /// True if a session with this id is currently open. Mostly for
    /// tests and FFI surface validation — production code should
    /// trust the id it just opened.
    pub async fn session_exists(&self, session_id: &str) -> bool {
        self.session_policies.read().await.contains_key(session_id)
    }

    /// Attach a local inference engine. Registers `infer`, `embed`, `classify`
    /// as built-in tools with real implementations.
    pub fn with_inference(mut self, engine: Arc<car_inference::InferenceEngine>) -> Self {
        self.inference_engine = Some(engine);
        // Register inference tool schemas (non-async init, use try_lock)
        if let Ok(mut tools) = self.tools.try_write() {
            for schema in car_inference::service::all_schemas() {
                tools.insert(schema.name.clone(), schema);
            }
        }
        self
    }

    /// Attach a memgine for automatic skill learning after execution.
    /// When `auto_distill` is true, execution traces are automatically distilled
    /// into skills and domains are evolved when underperforming.
    pub fn with_learning(
        mut self,
        memgine: Arc<TokioMutex<car_memgine::MemgineEngine>>,
        auto_distill: bool,
    ) -> Self {
        self.memgine = Some(memgine);
        self.auto_distill = auto_distill;
        self
    }

    /// Attach a memgine with auto-distillation enabled (recommended default).
    pub fn with_memgine(self, memgine: Arc<TokioMutex<car_memgine::MemgineEngine>>) -> Self {
        self.with_learning(memgine, true)
    }

    /// Attach a trajectory store for persisting execution traces.
    pub fn with_trajectory_store(mut self, store: Arc<car_memgine::TrajectoryStore>) -> Self {
        self.trajectory_store = Some(store);
        self
    }

    pub fn with_executor(self, executor: Arc<dyn ToolExecutor>) -> Self {
        // Use try_lock for non-async init context. Safe because we just created the mutex.
        if let Ok(mut guard) = self.tool_executor.try_lock() {
            *guard = Some(executor);
        }
        self
    }

    /// Set a tool executor for the next execute() call.
    /// Used by NAPI bindings where executor varies per call.
    pub async fn set_executor(&self, executor: Arc<dyn ToolExecutor>) {
        *self.tool_executor.lock().await = Some(executor);
    }

    pub fn with_event_log(mut self, log: EventLog) -> Self {
        self.log = Arc::new(TokioMutex::new(log));
        self
    }

    /// Attach a replan callback for failure recovery (builder).
    pub fn with_replan(self, callback: Arc<dyn ReplanCallback>, config: ReplanConfig) -> Self {
        if let Ok(mut guard) = self.replan_callback.try_lock() {
            *guard = Some(callback);
        }
        if let Ok(mut guard) = self.replan_config.try_write() {
            *guard = config;
        }
        self
    }

    /// Set a replan callback at runtime.
    pub async fn set_replan_callback(&self, callback: Arc<dyn ReplanCallback>) {
        *self.replan_callback.lock().await = Some(callback);
    }

    /// Set replan configuration at runtime.
    pub async fn set_replan_config(&self, config: ReplanConfig) {
        *self.replan_config.write().await = config;
    }

    /// Register a tool with just a name (backward compatible).
    pub async fn register_tool(&self, name: &str) {
        let schema = ToolSchema {
            name: name.to_string(),
            description: String::new(),
            parameters: serde_json::Value::Object(Default::default()),
            returns: None,
            idempotent: false,
            cache_ttl_secs: None,
            rate_limit: None,
        };
        self.register_tool_schema(schema).await;
    }

    /// Register a tool with full schema.
    pub async fn register_tool_schema(&self, schema: ToolSchema) {
        // Auto-configure cache if schema specifies it
        if let Some(ttl) = schema.cache_ttl_secs {
            self.result_cache.enable_caching(&schema.name, ttl).await;
        }
        // Auto-configure rate limit if schema specifies it
        if let Some(ref rl) = schema.rate_limit {
            self.rate_limiter
                .set_limit(
                    &schema.name,
                    RateLimit {
                        max_calls: rl.max_calls,
                        interval_secs: rl.interval_secs,
                    },
                )
                .await;
        }
        self.tools.write().await.insert(schema.name.clone(), schema);
    }

    /// Register a tool via the canonical registry.
    /// This is the preferred way to register tools — it updates both the
    /// registry and the legacy tools HashMap for backward compatibility.
    pub async fn register_tool_entry(&self, entry: crate::registry::ToolEntry) {
        let schema = entry.schema.clone();
        self.registry.register(entry).await;
        self.register_tool_schema(schema).await;
    }

    /// Register CAR's built-in agent utility stdlib.
    ///
    /// This is an opt-in convenience layer for common local-file and text tools.
    /// Existing runtimes remain unchanged until this is called.
    pub async fn register_agent_basics(&self) {
        for entry in crate::agent_basics::entries() {
            self.register_tool_entry(entry).await;
        }
    }

    /// Get all registered tool schemas (for model prompt generation).
    pub async fn tool_schemas(&self) -> Vec<ToolSchema> {
        self.tools.read().await.values().cloned().collect()
    }

    /// Set a cost budget that limits proposal execution.
    pub async fn set_cost_budget(&self, budget: CostBudget) {
        *self.cost_budget.write().await = Some(budget);
    }

    /// Set per-agent capability permissions that restrict tools, state keys, and action count.
    pub async fn set_capabilities(&self, caps: CapabilitySet) {
        *self.capabilities.write().await = Some(caps);
    }

    /// Set a per-tool rate limit (token bucket).
    ///
    /// `max_calls` tokens are available per `interval_secs` window.
    /// When the bucket is empty, `dispatch()` applies backpressure by
    /// waiting until a token refills.
    pub async fn set_rate_limit(&self, tool: &str, max_calls: u32, interval_secs: f64) {
        self.rate_limiter
            .set_limit(
                tool,
                RateLimit {
                    max_calls,
                    interval_secs,
                },
            )
            .await;
    }

    /// Enable cross-proposal result caching for a tool with a TTL in seconds.
    pub async fn enable_tool_cache(&self, tool: &str, ttl_secs: u64) {
        self.result_cache.enable_caching(tool, ttl_secs).await;
    }

    /// Execute a proposal with automatic replanning on failure.
    ///
    /// If a `ReplanCallback` is registered and `max_replans > 0`, the runtime
    /// will catch abort failures, roll back state, ask the model for an
    /// alternative proposal via the callback, and re-execute. This transforms
    /// "execute-and-hope" into "execute-and-recover."
    ///
    /// If no callback is registered or `max_replans == 0`, behaves identically
    /// to a single `execute_inner()` call (zero overhead, fully backward compatible).
    #[instrument(
        name = "proposal.execute",
        skip_all,
        fields(
            proposal_id = %proposal.id,
            action_count = proposal.actions.len(),
        )
    )]
    pub async fn execute(&self, proposal: &ActionProposal) -> ProposalResult {
        // Forward to the cancel-aware variant with a never-cancelled
        // token. Existing callers see no behaviour change.
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_with_cancel(proposal, &token).await
    }

    /// Execute a proposal scoped to a specific session id.
    ///
    /// Validation walks the global policy registry plus the session's
    /// own registry — both must pass for an action to run. Session
    /// policies can deny what global allows; they cannot allow what
    /// global denies (validation is conjunctive).
    ///
    /// Returns the same [`ProposalResult`] shape as [`Self::execute`].
    /// Errors with an action-level rejection if the session id is
    /// unknown — callers should check via [`Self::session_exists`] or
    /// trust an id they minted via [`Self::open_session`].
    pub async fn execute_with_session(
        &self,
        proposal: &ActionProposal,
        session_id: &str,
    ) -> ProposalResult {
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_with_session_and_cancel(proposal, session_id, &token).await
    }

    /// Combined session-scoped + cancellable execute. The session id
    /// is passed verbatim to the per-action policy check; the cancel
    /// token behaves identically to [`Self::execute_with_cancel`].
    pub async fn execute_with_session_and_cancel(
        &self,
        proposal: &ActionProposal,
        session_id: &str,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        self.execute_with_optional_session(proposal, Some(session_id), cancel).await
    }

    /// Execute a proposal with cooperative cancellation.
    ///
    /// The runtime checks `token.is_cancelled()` at each DAG level
    /// boundary. When set, every action that hadn't yet started runs
    /// is reported as `Skipped` with `error = "canceled: ..."` so
    /// callers can distinguish "user pulled the plug" from "earlier
    /// abort cascaded." Actions already in flight continue to
    /// completion — tool calls dispatched to user-provided executors
    /// can't be safely interrupted from the engine.
    ///
    /// The CAR A2A bridge uses this so `tasks/cancel` produces a
    /// `ProposalResult` with clean partial state rather than relying
    /// on `JoinHandle::abort` to interrupt mid-await (which leaves
    /// no record of which actions actually ran).
    ///
    /// **FFI exposure:** this method is intentionally not surfaced
    /// through the NAPI / PyO3 / `car-server-core` JSON-RPC bindings.
    /// Those consumers (Node, Python, WebSocket) don't currently
    /// expose long-running async-task surfaces that need
    /// cancellation; the bridge is the lone consumer. When a binding
    /// gains a long-running task surface, the path is clear: add a
    /// per-binding token registry keyed by some caller-provided id,
    /// expose `cancelExecution(id)` / `cancel_execution(id)` /
    /// `proposal.cancel { id }`, and have the runtime call
    /// `execute_with_cancel` with the matching token. Skipping that
    /// today avoids speculative API surface that bloats bindings
    /// without a consumer.
    pub async fn execute_with_cancel(
        &self,
        proposal: &ActionProposal,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        self.execute_with_optional_session(proposal, None, cancel).await
    }

    /// Internal entry point that backs both
    /// [`Self::execute_with_cancel`] (no session) and
    /// [`Self::execute_with_session_and_cancel`]. Holds the replan
    /// loop and threads `session_id` into the per-action validation
    /// path so session-scoped policies stack on top of global ones.
    async fn execute_with_optional_session(
        &self,
        proposal: &ActionProposal,
        session_id: Option<&str>,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        let config = self.replan_config.read().await.clone();
        let mut current_proposal = proposal.clone();
        let mut attempt: u32 = 0;

        loop {
            let (result, state_before_map) = self
                .execute_inner_with_cancel(&current_proposal, Some(cancel), session_id)
                .await;

            // Check if we aborted
            let aborted = result
                .results
                .iter()
                .any(|r| r.status == ActionStatus::Failed);
            if !aborted || attempt >= config.max_replans {
                if aborted && attempt > 0 {
                    // Exhausted all replan attempts
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ReplanExhausted,
                        None,
                        Some(&proposal.id),
                        [("attempts".to_string(), Value::from(attempt))].into(),
                    );
                }

                // Persist trajectory
                let outcome = if !aborted {
                    if attempt > 0 {
                        car_memgine::TrajectoryOutcome::ReplanSuccess
                    } else {
                        car_memgine::TrajectoryOutcome::Success
                    }
                } else if attempt > 0 {
                    car_memgine::TrajectoryOutcome::ReplanExhausted
                } else {
                    car_memgine::TrajectoryOutcome::Failed
                };
                if let Some(err) = self.persist_trajectory(
                    proposal,
                    &current_proposal,
                    &result,
                    outcome,
                    attempt,
                    &state_before_map,
                ) {
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ActionFailed,
                        None,
                        Some(&proposal.id),
                        [(
                            "trajectory_persist_error".to_string(),
                            Value::from(err.as_str()),
                        )]
                        .into(),
                    );
                }

                return result;
            }

            // Get replan callback (clone Arc, drop lock immediately)
            let callback = {
                let guard = self.replan_callback.lock().await;
                guard.clone()
            };
            let Some(callback) = callback else {
                // No callback registered — persist trajectory and return
                if let Some(err) = self.persist_trajectory(
                    proposal,
                    &current_proposal,
                    &result,
                    car_memgine::TrajectoryOutcome::Failed,
                    attempt,
                    &state_before_map,
                ) {
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ActionFailed,
                        None,
                        Some(&proposal.id),
                        [(
                            "trajectory_persist_error".to_string(),
                            Value::from(err.as_str()),
                        )]
                        .into(),
                    );
                }
                return result;
            };

            // Build replan context
            let failed_actions: Vec<FailedActionSummary> = result
                .results
                .iter()
                .filter(|r| r.status == ActionStatus::Failed)
                .map(|r| {
                    let action = current_proposal
                        .actions
                        .iter()
                        .find(|a| a.id == r.action_id);
                    FailedActionSummary {
                        action_id: r.action_id.clone(),
                        tool: action.and_then(|a| a.tool.clone()),
                        error: r.error.clone().unwrap_or_default(),
                        parameters: action.map(|a| a.parameters.clone()).unwrap_or_default(),
                    }
                })
                .collect();

            let completed_action_ids: Vec<String> = result
                .results
                .iter()
                .filter(|r| r.status == ActionStatus::Succeeded)
                .map(|r| r.action_id.clone())
                .collect();

            let ctx = ReplanContext {
                proposal_id: proposal.id.clone(),
                attempt: attempt + 1,
                failed_actions,
                completed_action_ids,
                state_snapshot: self.state.snapshot(),
                replans_remaining: config.max_replans.saturating_sub(attempt + 1),
                original_source: proposal.source.clone(),
                original_action_count: proposal.actions.len(),
                original_context: proposal.context.clone(),
            };

            // Backoff delay between replan attempts
            if config.delay_ms > 0 {
                tokio::time::sleep(Duration::from_millis(config.delay_ms)).await;
            }

            // Log replan attempt
            {
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ReplanAttempted,
                    None,
                    Some(&proposal.id),
                    [
                        ("attempt".to_string(), Value::from(attempt + 1)),
                        (
                            "failed_count".to_string(),
                            Value::from(ctx.failed_actions.len()),
                        ),
                    ]
                    .into(),
                );
            }

            // Call the model for a new plan
            match callback.replan(&ctx).await {
                Ok(new_proposal) => {
                    // Quality gate: verify replan proposal before executing
                    if config.verify_before_execute {
                        let tools_guard = self.tools.read().await;
                        let tool_names: std::collections::HashSet<String> =
                            tools_guard.keys().cloned().collect();
                        drop(tools_guard);

                        let current_state = self.state.snapshot();
                        let vr = car_verify::verify(
                            &new_proposal,
                            Some(&current_state),
                            Some(&tool_names),
                            100,
                        );
                        if !vr.valid {
                            let error_msgs: Vec<String> = vr
                                .issues
                                .iter()
                                .filter(|i| i.severity == "error")
                                .map(|i| i.message.clone())
                                .collect();
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    ("errors".to_string(), Value::from(error_msgs.join("; "))),
                                    ("attempt".to_string(), Value::from(attempt + 1)),
                                ]
                                .into(),
                            );
                            // Don't execute a broken replan — count as failed attempt
                            attempt += 1;
                            continue;
                        }
                    }

                    // Log accepted proposal
                    {
                        let mut log = self.log.lock().await;
                        log.append(
                            EventKind::ReplanProposalReceived,
                            None,
                            Some(&proposal.id),
                            [
                                ("attempt".to_string(), Value::from(attempt + 1)),
                                (
                                    "new_action_count".to_string(),
                                    Value::from(new_proposal.actions.len()),
                                ),
                            ]
                            .into(),
                        );
                    }
                    current_proposal = new_proposal;
                    attempt += 1;
                }
                Err(e) => {
                    // Replan callback itself failed — log and return original failure
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ReplanExhausted,
                        None,
                        Some(&proposal.id),
                        [
                            ("reason".to_string(), Value::from("callback_error")),
                            ("error".to_string(), Value::from(e.as_str())),
                            ("attempt".to_string(), Value::from(attempt + 1)),
                        ]
                        .into(),
                    );
                    if let Some(err) = self.persist_trajectory(
                        proposal,
                        &current_proposal,
                        &result,
                        car_memgine::TrajectoryOutcome::Failed,
                        attempt,
                        &state_before_map,
                    ) {
                        log.append(
                            EventKind::ActionFailed,
                            None,
                            Some(&proposal.id),
                            [(
                                "trajectory_persist_error".to_string(),
                                Value::from(err.as_str()),
                            )]
                            .into(),
                        );
                    }
                    return result;
                }
            }
        }
    }

    /// Persist a trajectory to the store if configured.
    fn persist_trajectory(
        &self,
        proposal: &ActionProposal,
        current_proposal: &ActionProposal,
        result: &ProposalResult,
        outcome: car_memgine::TrajectoryOutcome,
        attempt: u32,
        state_before_map: &HashMap<String, HashMap<String, Value>>,
    ) -> Option<String> {
        let store = self.trajectory_store.as_ref()?;

        let trace_events: Vec<car_memgine::TraceEvent> = result
            .results
            .iter()
            .map(|r| {
                let kind = match r.status {
                    ActionStatus::Succeeded => "action_succeeded",
                    ActionStatus::Failed => "action_failed",
                    ActionStatus::Rejected => "action_rejected",
                    ActionStatus::Skipped => "action_skipped",
                    _ => "unknown",
                };
                let tool = current_proposal
                    .actions
                    .iter()
                    .find(|a| a.id == r.action_id)
                    .and_then(|a| a.tool.clone());
                let reward = match r.status {
                    ActionStatus::Succeeded => Some(1.0),
                    ActionStatus::Failed => Some(0.0),
                    ActionStatus::Rejected => Some(0.0),
                    ActionStatus::Skipped => None,
                    _ => None,
                };
                car_memgine::TraceEvent {
                    kind: kind.to_string(),
                    action_id: Some(r.action_id.clone()),
                    tool,
                    data: r
                        .error
                        .as_ref()
                        .map(|e| serde_json::json!({"error": e}))
                        .unwrap_or(serde_json::json!({})),
                    duration_ms: r.duration_ms,
                    state_before: state_before_map.get(&r.action_id).cloned(),
                    state_after: if !r.state_changes.is_empty() {
                        Some(r.state_changes.clone())
                    } else {
                        None
                    },
                    reward,
                }
            })
            .collect();

        let trajectory = car_memgine::Trajectory {
            proposal_id: proposal.id.clone(),
            source: proposal.source.clone(),
            action_count: current_proposal.actions.len(),
            events: trace_events,
            outcome,
            timestamp: chrono::Utc::now(),
            duration_ms: result.cost.total_duration_ms,
            replan_attempts: attempt,
        };

        match store.append(&trajectory) {
            Ok(()) => None,
            Err(e) => Some(e.to_string()),
        }
    }

    /// Score N candidate proposals, execute the best valid one, fall back to
    /// next-best on failure. Combines car-planner scoring with engine execution.
    ///
    /// Returns the result from whichever proposal was executed (best or fallback).
    /// If all candidates fail verification, returns an error result for the first.
    pub async fn plan_and_execute(
        &self,
        candidates: &[ActionProposal],
        planner_config: Option<car_planner::PlannerConfig>,
        feedback: Option<&car_planner::ToolFeedback>,
    ) -> ProposalResult {
        if candidates.is_empty() {
            return ProposalResult {
                proposal_id: "empty".to_string(),
                results: vec![],
                cost: car_ir::CostSummary::default(),
            };
        }

        // Score all candidates
        let planner = car_planner::Planner::new(planner_config.unwrap_or_default());
        let tools_guard = self.tools.read().await;
        let tool_names: std::collections::HashSet<String> = tools_guard.keys().cloned().collect();
        drop(tools_guard);

        let pre_plan_snapshot = self.state.snapshot();
        let pre_plan_transitions = self.state.transition_count();
        let ranked = planner.rank_with_feedback(
            candidates,
            Some(&pre_plan_snapshot),
            Some(&tool_names),
            feedback,
        );

        // Try each valid candidate in score order
        let mut first_failure: Option<ProposalResult> = None;
        for scored in &ranked {
            if !scored.valid {
                continue;
            }

            // Restore clean state before each candidate (don't rely on execute's
            // internal rollback — it only fires on Abort, not Skip/Retry failures)
            self.state
                .restore(pre_plan_snapshot.clone(), pre_plan_transitions);

            let proposal = &candidates[scored.index];
            let result = self.execute(proposal).await;

            if result.all_succeeded() {
                return result;
            }

            tracing::info!(
                proposal_id = %proposal.id,
                score = scored.score,
                "plan_and_execute: proposal failed, trying next candidate"
            );

            if first_failure.is_none() {
                first_failure = Some(result);
            }
        }

        // Return the first failure result (don't re-execute — avoids duplicate side effects)
        first_failure.unwrap_or_else(|| ProposalResult {
            proposal_id: candidates[0].id.clone(),
            results: vec![],
            cost: car_ir::CostSummary::default(),
        })
    }

    /// Execute a single proposal through the runtime loop (no replanning).
    /// Returns (result, state_before_map) where state_before_map has per-action snapshots.
    ///
    /// `session_id`, when `Some`, scopes per-action policy validation
    /// to the named session in addition to global policies. Both
    /// layers must pass for an action to run; the session layer cannot
    /// loosen what global denies.
    async fn execute_inner_with_cancel(
        &self,
        proposal: &ActionProposal,
        cancel: Option<&tokio_util::sync::CancellationToken>,
        session_id: Option<&str>,
    ) -> (ProposalResult, HashMap<String, HashMap<String, Value>>) {
        // Generate trace_id for this proposal execution
        let trace_id = Uuid::new_v4().to_string();

        // Begin root span for proposal execution
        let root_span_id = {
            let mut log = self.log.lock().await;
            log.begin_span(
                "proposal.execute",
                &trace_id,
                None,
                [("proposal_id".to_string(), Value::from(proposal.id.as_str()))].into(),
            )
        };

        // Log proposal received
        {
            let mut log = self.log.lock().await;
            log.append(
                EventKind::ProposalReceived,
                None,
                Some(&proposal.id),
                [
                    ("source".to_string(), Value::from(proposal.source.as_str())),
                    (
                        "action_count".to_string(),
                        Value::from(proposal.actions.len()),
                    ),
                ]
                .into(),
            );
        }

        // Capability check: max_actions budget for entire proposal
        {
            let caps = self.capabilities.read().await;
            if let Some(ref cap) = *caps {
                if !cap.actions_within_budget(proposal.actions.len() as u32) {
                    let mut action_results = Vec::new();
                    for action in &proposal.actions {
                        action_results.push(rejected_result(
                            &action.id,
                            format!(
                                "capability denied: proposal has {} actions, max allowed is {:?}",
                                proposal.actions.len(),
                                cap.max_actions
                            ),
                        ));
                    }
                    return (
                        ProposalResult {
                            proposal_id: proposal.id.clone(),
                            results: action_results,
                            cost: CostSummary::default(),
                        },
                        HashMap::new(),
                    );
                }
            }
        }

        // Snapshot for rollback
        let snapshot = self.state.snapshot();
        let transition_count = self.state.transition_count();

        let mut results: Vec<ActionResult> = Vec::new();
        // Per-action state snapshots captured before execution (for TraceEvent.state_before).
        let mut state_before_map: HashMap<String, HashMap<String, Value>> = HashMap::new();
        let mut aborted = false;
        let mut budget_exceeded = false;
        let mut total_retries: u32 = 0;

        // Running cost counters for budget enforcement
        let mut running_tool_calls: u32 = 0;
        let mut running_actions: u32 = 0;
        let mut running_duration_ms: f64 = 0.0;

        // Snapshot the budget once
        let budget = self.cost_budget.read().await.clone();

        // Build DAG
        let levels = build_dag(&proposal.actions);

        let mut canceled = false;
        for level in &levels {
            // Cooperative cancellation check at the level boundary.
            // Actions already in flight aren't interrupted (we can't
            // safely cancel a tool call dispatched to a user-provided
            // executor), but every action that hadn't started runs
            // is recorded as canceled with a clear reason.
            if !canceled {
                if let Some(token) = cancel {
                    if token.is_cancelled() {
                        canceled = true;
                    }
                }
            }
            if canceled {
                for &idx in level {
                    results.push(canceled_result(
                        &proposal.actions[idx].id,
                        "cancellation requested by caller",
                    ));
                }
                continue;
            }
            if aborted || budget_exceeded {
                let skip_reason = if budget_exceeded {
                    "cost budget exceeded"
                } else {
                    "skipped due to earlier abort"
                };
                for &idx in level {
                    results.push(skipped_result(&proposal.actions[idx].id, skip_reason));
                }
                continue;
            }

            // Check if any action in this level has ABORT behavior
            let has_abort = level
                .iter()
                .any(|&i| proposal.actions[i].failure_behavior == FailureBehavior::Abort);

            if level.len() == 1 || has_abort {
                // Sequential execution
                for &idx in level {
                    if aborted || budget_exceeded {
                        let skip_reason = if budget_exceeded {
                            "cost budget exceeded"
                        } else {
                            "skipped due to abort"
                        };
                        results.push(skipped_result(&proposal.actions[idx].id, skip_reason));
                        continue;
                    }

                    // Budget check before execution
                    if let Some(ref b) = budget {
                        if let Some(max) = b.max_actions {
                            if running_actions >= max {
                                budget_exceeded = true;
                                results.push(skipped_result(
                                    &proposal.actions[idx].id,
                                    "cost budget exceeded",
                                ));
                                continue;
                            }
                        }
                        if let Some(max) = b.max_tool_calls {
                            if proposal.actions[idx].action_type == ActionType::ToolCall
                                && running_tool_calls >= max
                            {
                                budget_exceeded = true;
                                results.push(skipped_result(
                                    &proposal.actions[idx].id,
                                    "cost budget exceeded",
                                ));
                                continue;
                            }
                        }
                        if let Some(max) = b.max_duration_ms {
                            if running_duration_ms >= max {
                                budget_exceeded = true;
                                results.push(skipped_result(
                                    &proposal.actions[idx].id,
                                    "cost budget exceeded",
                                ));
                                continue;
                            }
                        }
                    }

                    state_before_map.insert(
                        proposal.actions[idx].id.clone(),
                        snapshot_relevant_keys(&self.state, &proposal.actions[idx]),
                    );
                    let (ar, action_retries) = self
                        .process_action(
                            &proposal.actions[idx],
                            &proposal.id,
                            &trace_id,
                            &root_span_id,
                            session_id,
                        )
                        .await;
                    total_retries += action_retries;

                    // Update running counters
                    if ar.status == ActionStatus::Succeeded
                        && proposal.actions[idx].action_type == ActionType::ToolCall
                    {
                        running_tool_calls += 1;
                    }
                    if ar.status != ActionStatus::Skipped {
                        running_actions += 1;
                    }
                    if let Some(d) = ar.duration_ms {
                        running_duration_ms += d;
                    }

                    if ar.status == ActionStatus::Failed
                        && proposal.actions[idx].failure_behavior == FailureBehavior::Abort
                    {
                        aborted = true;
                    }
                    results.push(ar);
                }
            } else {
                // Concurrent execution via futures::join_all
                // Snapshot only relevant keys per action (all see same pre-level state)
                for &idx in level {
                    state_before_map.insert(
                        proposal.actions[idx].id.clone(),
                        snapshot_relevant_keys(&self.state, &proposal.actions[idx]),
                    );
                }
                let futs: Vec<_> = level
                    .iter()
                    .map(|&idx| {
                        self.process_action(
                            &proposal.actions[idx],
                            &proposal.id,
                            &trace_id,
                            &root_span_id,
                            session_id,
                        )
                    })
                    .collect();
                let level_results = futures::future::join_all(futs).await;

                for (i, (ar, action_retries)) in level_results.into_iter().enumerate() {
                    let idx = level[i];
                    total_retries += action_retries;
                    if ar.status == ActionStatus::Succeeded
                        && proposal.actions[idx].action_type == ActionType::ToolCall
                    {
                        running_tool_calls += 1;
                    }
                    if ar.status != ActionStatus::Skipped {
                        running_actions += 1;
                    }
                    if let Some(d) = ar.duration_ms {
                        running_duration_ms += d;
                    }
                    results.push(ar);
                }
            }
        }

        // Handle rollback
        if aborted {
            self.state.restore(snapshot.clone(), transition_count);

            let mut log = self.log.lock().await;
            log.append(
                EventKind::StateSnapshot,
                None,
                Some(&proposal.id),
                [(
                    "state".to_string(),
                    serde_json::to_value(&snapshot).unwrap_or_default(),
                )]
                .into(),
            );
            log.append(
                EventKind::StateRollback,
                None,
                Some(&proposal.id),
                [(
                    "rolled_back_to".to_string(),
                    Value::from("pre-proposal snapshot"),
                )]
                .into(),
            );

            // Clear idempotency cache for rolled-back actions
            let mut cache = self.idempotency_cache.lock().await;
            for r in &results {
                if r.status == ActionStatus::Succeeded {
                    for action in &proposal.actions {
                        if action.id == r.action_id && action.idempotent {
                            cache.remove(&idempotency_key(action));
                        }
                    }
                }
            }
        }

        // Sort results to match original action order
        let action_order: HashMap<String, usize> = proposal
            .actions
            .iter()
            .enumerate()
            .map(|(i, a)| (a.id.clone(), i))
            .collect();
        results.sort_by_key(|r| {
            action_order
                .get(&r.action_id)
                .copied()
                .unwrap_or(usize::MAX)
        });

        // Compute cost summary from results
        let mut cost = CostSummary::default();
        for r in &results {
            let action = action_order
                .get(&r.action_id)
                .and_then(|&i| proposal.actions.get(i));
            match r.status {
                ActionStatus::Succeeded => {
                    cost.actions_executed += 1;
                    if let Some(a) = action {
                        if a.action_type == ActionType::ToolCall {
                            cost.tool_calls += 1;
                        }
                    }
                }
                ActionStatus::Failed | ActionStatus::Rejected => {
                    cost.actions_executed += 1;
                }
                ActionStatus::Skipped => {
                    cost.actions_skipped += 1;
                }
                _ => {}
            }
            if let Some(d) = r.duration_ms {
                cost.total_duration_ms += d;
            }
        }

        // Set retries from inline counter
        cost.retries = total_retries;

        // End root span — Ok if no abort, Error if aborted
        {
            let span_status = if aborted {
                SpanStatus::Error
            } else {
                SpanStatus::Ok
            };
            let mut log = self.log.lock().await;
            log.end_span(&root_span_id, span_status);
        }

        let proposal_result = ProposalResult {
            proposal_id: proposal.id.clone(),
            results,
            cost,
        };

        // Post-execution: auto-distill skills from this execution trace
        if self.auto_distill {
            if let Some(ref memgine) = self.memgine {
                // Convert results to TraceEvents for distillation
                let trace_events: Vec<car_memgine::TraceEvent> = proposal_result
                    .results
                    .iter()
                    .map(|r| {
                        let kind = match r.status {
                            ActionStatus::Succeeded => "action_succeeded",
                            ActionStatus::Failed => "action_failed",
                            ActionStatus::Rejected => "action_rejected",
                            ActionStatus::Skipped => "action_skipped",
                            _ => "unknown",
                        };
                        // Find the matching action to get the tool name
                        let tool = proposal
                            .actions
                            .iter()
                            .find(|a| a.id == r.action_id)
                            .and_then(|a| a.tool.clone());
                        let mut data = serde_json::Map::new();
                        if let Some(ref e) = r.error {
                            data.insert("error".into(), Value::from(e.as_str()));
                        }
                        if let Some(ref o) = r.output {
                            data.insert("output".into(), o.clone());
                        }
                        car_memgine::TraceEvent {
                            kind: kind.to_string(),
                            action_id: Some(r.action_id.clone()),
                            tool,
                            data: Value::Object(data),
                            duration_ms: r.duration_ms,
                            reward: match r.status {
                                ActionStatus::Succeeded => Some(1.0),
                                ActionStatus::Failed | ActionStatus::Rejected => Some(0.0),
                                _ => None,
                            },
                            ..Default::default()
                        }
                    })
                    .collect();

                let mut engine = memgine.lock().await;
                let skills = engine.distill_skills(&trace_events).await;
                if !skills.is_empty() {
                    let count = skills.len();
                    engine.ingest_distilled_skills(&skills);

                    // Log the distillation event
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::SkillDistilled,
                        None,
                        Some(&proposal_result.proposal_id),
                        [
                            ("skills_count".to_string(), Value::from(count)),
                            (
                                "skill_names".to_string(),
                                Value::from(
                                    skills.iter().map(|s| s.name.as_str()).collect::<Vec<_>>(),
                                ),
                            ),
                        ]
                        .into(),
                    );

                    // Check if any domains need evolution
                    let threshold = engine.evolution_threshold();
                    let domains = engine.domains_needing_evolution(threshold);
                    for domain in &domains {
                        // Collect failed events for this domain
                        let failed: Vec<car_memgine::TraceEvent> = trace_events
                            .iter()
                            .filter(|e| {
                                matches!(e.kind.as_str(), "action_failed" | "action_rejected")
                            })
                            .cloned()
                            .collect();
                        if !failed.is_empty() {
                            let evolved = engine.evolve_skills(&failed, domain).await;
                            if !evolved.is_empty() {
                                log.append(
                                    EventKind::EvolutionTriggered,
                                    None,
                                    Some(&proposal_result.proposal_id),
                                    [
                                        ("domain".to_string(), Value::from(domain.as_str())),
                                        ("new_skills".to_string(), Value::from(evolved.len())),
                                    ]
                                    .into(),
                                );
                            }
                        }
                    }
                }
            }
        }

        (proposal_result, state_before_map)
    }

    /// Process a single action: idempotency → validate → policy → execute.
    /// Returns (ActionResult, retries_count).
    async fn process_action(
        &self,
        action: &Action,
        proposal_id: &str,
        trace_id: &str,
        parent_span_id: &str,
        session_id: Option<&str>,
    ) -> (ActionResult, u32) {
        // Derive action type name for span naming
        let action_type_name = serde_json::to_string(&action.action_type)
            .unwrap_or_default()
            .trim_matches('"')
            .to_string();
        let span_name = format!("action.{}", action_type_name);

        // Begin child span for this action
        let action_span_id = {
            let mut attrs: HashMap<String, Value> = HashMap::new();
            attrs.insert("action_id".to_string(), Value::from(action.id.as_str()));
            if let Some(ref tool) = action.tool {
                attrs.insert("tool".to_string(), Value::from(tool.as_str()));
            }
            let mut log = self.log.lock().await;
            log.begin_span(&span_name, trace_id, Some(parent_span_id), attrs)
        };

        // Execute the action pipeline and capture result
        let (result, retries) = self
            .process_action_inner(action, proposal_id, session_id)
            .await;

        // End action span based on result status
        let span_status = match result.status {
            ActionStatus::Succeeded => SpanStatus::Ok,
            ActionStatus::Failed | ActionStatus::Rejected => SpanStatus::Error,
            _ => SpanStatus::Unset,
        };
        {
            let mut log = self.log.lock().await;
            log.end_span(&action_span_id, span_status);
        }

        (result, retries)
    }

    /// Inner action processing: idempotency -> validate -> policy -> execute.
    /// Returns (ActionResult, retries_count).
    #[instrument(
        name = "action.process",
        skip_all,
        fields(
            action_id = %action.id,
            action_type = ?action.action_type,
            tool = action.tool.as_deref().unwrap_or("none"),
        )
    )]
    async fn process_action_inner(
        &self,
        action: &Action,
        proposal_id: &str,
        session_id: Option<&str>,
    ) -> (ActionResult, u32) {
        // Idempotency check
        if action.idempotent {
            let key = idempotency_key(action);
            let cache = self.idempotency_cache.lock().await;
            if let Some(cached) = cache.get(&key) {
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ActionDeduplicated,
                    Some(&action.id),
                    Some(proposal_id),
                    [(
                        "cached_action_id".to_string(),
                        Value::from(cached.action_id.as_str()),
                    )]
                    .into(),
                );
                return (
                    ActionResult {
                        action_id: action.id.clone(),
                        status: cached.status.clone(),
                        output: cached.output.clone(),
                        error: cached.error.clone(),
                        state_changes: cached.state_changes.clone(),
                        duration_ms: Some(0.0),
                        timestamp: chrono::Utc::now(),
                    },
                    0,
                );
            }
        }

        // Capability check
        {
            let caps = self.capabilities.read().await;
            if let Some(ref cap) = *caps {
                // Check tool capability for ToolCall actions
                if action.action_type == ActionType::ToolCall {
                    if let Some(ref tool_name) = action.tool {
                        if !cap.tool_allowed(tool_name) {
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ActionRejected,
                                Some(&action.id),
                                Some(proposal_id),
                                HashMap::new(),
                            );
                            return (
                                rejected_result(
                                    &action.id,
                                    format!("capability denied: tool '{}' not allowed", tool_name),
                                ),
                                0,
                            );
                        }
                    }
                }

                // Check state key capability for StateWrite/StateRead actions
                if action.action_type == ActionType::StateWrite
                    || action.action_type == ActionType::StateRead
                {
                    if let Some(key) = action.parameters.get("key").and_then(|v| v.as_str()) {
                        if !cap.state_key_allowed(key) {
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ActionRejected,
                                Some(&action.id),
                                Some(proposal_id),
                                HashMap::new(),
                            );
                            return (
                                rejected_result(
                                    &action.id,
                                    format!("capability denied: state key '{}' not allowed", key),
                                ),
                                0,
                            );
                        }
                    }
                }
            }
        }

        // Validate
        let tools = self.tools.read().await;
        let validation = validate_action(action, &self.state, &tools);
        drop(tools);

        if !validation.valid() {
            let error = validation
                .errors
                .iter()
                .map(|e| e.reason.as_str())
                .collect::<Vec<_>>()
                .join("; ");
            let mut log = self.log.lock().await;
            log.append(
                EventKind::ActionRejected,
                Some(&action.id),
                Some(proposal_id),
                HashMap::new(),
            );
            return (rejected_result(&action.id, error), 0);
        }

        // Policy check — global registry plus, when the proposal is
        // executed under a session, that session's registry. Both
        // layers must pass for the action to proceed; session
        // policies are additive deny rules — they can deny what
        // global allows but cannot allow what global denies.
        {
            let mut violations = {
                let policies = self.policies.read().await;
                policies.check(action, &self.state)
            };
            if let Some(sid) = session_id {
                // Snapshot the per-session engine handle out from under
                // the outer registry lock so the inner check holds
                // only the engine's own RwLock — preserves the
                // documented lock-ordering discipline.
                let session_engine = {
                    let sessions = self.session_policies.read().await;
                    sessions.get(sid).cloned()
                };
                if let Some(engine) = session_engine {
                    let engine = engine.read().await;
                    violations.extend(engine.check(action, &self.state));
                } else {
                    // Unknown session id — refuse the action rather
                    // than silently fall back to global-only. A
                    // proposal submitted under a closed session
                    // shouldn't run with looser rules than the caller
                    // intended.
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::PolicyViolation,
                        Some(&action.id),
                        Some(proposal_id),
                        HashMap::new(),
                    );
                    return (
                        rejected_result(
                            &action.id,
                            format!(
                                "unknown session id '{sid}' — open one via Runtime::open_session before executing under a session"
                            ),
                        ),
                        0,
                    );
                }
            }
            if !violations.is_empty() {
                let error = violations
                    .iter()
                    .map(|v| format!("policy '{}': {}", v.policy_name, v.reason))
                    .collect::<Vec<_>>()
                    .join("; ");
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::PolicyViolation,
                    Some(&action.id),
                    Some(proposal_id),
                    HashMap::new(),
                );
                return (rejected_result(&action.id, error), 0);
            }
        }

        // Validated
        {
            let mut log = self.log.lock().await;
            log.append(
                EventKind::ActionValidated,
                Some(&action.id),
                Some(proposal_id),
                HashMap::new(),
            );
        }

        // Execute with retry
        let (result, retries) = self.execute_with_retry(action, proposal_id).await;

        // Cache idempotent results
        if action.idempotent && result.status == ActionStatus::Succeeded {
            let mut cache = self.idempotency_cache.lock().await;
            cache.insert(idempotency_key(action), result.clone());
        }

        tracing::info!(
            status = ?result.status,
            duration_ms = result.duration_ms,
            "action completed"
        );

        (result, retries)
    }

    /// Execute with retry logic and timeout.
    /// Returns (ActionResult, retries_count).
    async fn execute_with_retry(&self, action: &Action, proposal_id: &str) -> (ActionResult, u32) {
        let max_attempts = if action.failure_behavior == FailureBehavior::Retry {
            action.max_retries + 1
        } else {
            1
        };

        let mut last_error: Option<String> = None;
        let mut retries: u32 = 0;

        for attempt in 0..max_attempts {
            if attempt > 0 {
                retries += 1;
                let delay = RETRY_BASE_DELAY_MS * RETRY_BACKOFF_FACTOR.pow(attempt as u32 - 1);
                tokio::time::sleep(Duration::from_millis(delay)).await;
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ActionRetrying,
                    Some(&action.id),
                    Some(proposal_id),
                    [("attempt".to_string(), Value::from(attempt + 1))].into(),
                );
            }

            {
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ActionExecuting,
                    Some(&action.id),
                    Some(proposal_id),
                    HashMap::new(),
                );
            }

            let start = std::time::Instant::now();
            let transitions_before = self.state.transition_count();

            // Execute with optional timeout
            let exec_result = if let Some(timeout_ms) = action.timeout_ms {
                match timeout(Duration::from_millis(timeout_ms), self.dispatch(action)).await {
                    Ok(r) => r,
                    Err(_) => Err(format!("action timed out after {}ms", timeout_ms)),
                }
            } else {
                self.dispatch(action).await
            };

            let duration_ms = start.elapsed().as_secs_f64() * 1000.0;

            match exec_result {
                Ok(output) => {
                    // Commit post-effects
                    let mut state_changes: HashMap<String, Value> = HashMap::new();
                    for (key, value) in &action.expected_effects {
                        self.state.set(key, value.clone(), &action.id);
                        state_changes.insert(key.clone(), value.clone());
                    }

                    // Capture state changes from dispatch
                    for t in self.state.transitions_since(transitions_before) {
                        if !state_changes.contains_key(&t.key) {
                            if let Some(v) = t.new_value {
                                state_changes.insert(t.key.clone(), v);
                            }
                        }
                    }

                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ActionSucceeded,
                        Some(&action.id),
                        Some(proposal_id),
                        [("duration_ms".to_string(), Value::from(duration_ms))].into(),
                    );

                    if !state_changes.is_empty() {
                        log.append(
                            EventKind::StateChanged,
                            Some(&action.id),
                            Some(proposal_id),
                            [(
                                "changes".to_string(),
                                serde_json::to_value(&state_changes).unwrap_or_default(),
                            )]
                            .into(),
                        );
                    }

                    return (
                        ActionResult {
                            action_id: action.id.clone(),
                            status: ActionStatus::Succeeded,
                            output: Some(output),
                            error: None,
                            state_changes,
                            duration_ms: Some(duration_ms),
                            timestamp: chrono::Utc::now(),
                        },
                        retries,
                    );
                }
                Err(e) => {
                    last_error = Some(e.clone());
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ActionFailed,
                        Some(&action.id),
                        Some(proposal_id),
                        [
                            ("error".to_string(), Value::from(e.as_str())),
                            ("attempt".to_string(), Value::from(attempt + 1)),
                        ]
                        .into(),
                    );
                }
            }
        }

        // All attempts exhausted
        if action.failure_behavior == FailureBehavior::Skip {
            return (
                skipped_result(
                    &action.id,
                    last_error.as_deref().unwrap_or("all attempts exhausted"),
                ),
                retries,
            );
        }

        (
            ActionResult {
                action_id: action.id.clone(),
                status: ActionStatus::Failed,
                output: None,
                error: last_error,
                state_changes: HashMap::new(),
                duration_ms: None,
                timestamp: chrono::Utc::now(),
            },
            retries,
        )
    }

    /// Dispatch an action to the appropriate handler.
    async fn dispatch(&self, action: &Action) -> Result<Value, String> {
        match action.action_type {
            ActionType::ToolCall => {
                let tool_name = action.tool.as_deref().ok_or("tool_call has no tool")?;
                let params = Value::Object(
                    action
                        .parameters
                        .iter()
                        .map(|(k, v)| (k.clone(), v.clone()))
                        .collect(),
                );

                // Check cross-proposal result cache.
                if let Some(cached) = self.result_cache.get(tool_name, &params).await {
                    return Ok(cached);
                }

                // Apply rate limit backpressure before executing.
                self.rate_limiter.acquire(tool_name).await;

                // Try built-in inference tools first when the inference engine is available.
                if matches!(
                    tool_name,
                    "infer" | "infer.grounded" | "embed" | "classify" | "transcribe" | "synthesize"
                ) {
                    if let Some(ref engine) = self.inference_engine {
                        // For "infer.grounded" or "infer" with memgine available,
                        // build context from memory and attach it to the request.
                        let params = {
                            let should_ground =
                                tool_name == "infer.grounded" || tool_name == "infer";
                            if should_ground {
                                if let Some(ref memgine) = self.memgine {
                                    if let Some(prompt) =
                                        params.get("prompt").and_then(|v| v.as_str())
                                    {
                                        let ctx = {
                                            let mut m = memgine.lock().await;
                                            m.build_context(prompt)
                                        };
                                        if !ctx.is_empty() {
                                            let mut p = params.clone();
                                            if let Some(obj) = p.as_object_mut() {
                                                obj.insert("context".to_string(), Value::from(ctx));
                                            }
                                            p
                                        } else {
                                            params
                                        }
                                    } else {
                                        params
                                    }
                                } else {
                                    params
                                }
                            } else {
                                params
                            }
                        };

                        // Route "infer.grounded" to "infer" for the service layer
                        let effective_tool = if tool_name == "infer.grounded" {
                            "infer"
                        } else {
                            tool_name
                        };
                        let result =
                            car_inference::service::execute_tool(engine, effective_tool, &params)
                                .await
                                .map_err(|e| e.to_string());

                        if let Ok(ref value) = result {
                            self.result_cache
                                .put(tool_name, &params, value.clone())
                                .await;
                        }

                        return result;
                    }
                }

                // Built-in memory consolidation tool.
                if tool_name == "memory.consolidate" {
                    if let Some(ref memgine) = self.memgine {
                        let report = {
                            let mut m = memgine.lock().await;
                            m.consolidate().await
                        };
                        // Log the consolidation event
                        {
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::Consolidated,
                                None,
                                None,
                                [
                                    (
                                        "expired_pruned".to_string(),
                                        Value::from(report.expired_pruned),
                                    ),
                                    (
                                        "superseded_gc".to_string(),
                                        Value::from(report.superseded_gc),
                                    ),
                                    (
                                        "stale_embeddings_removed".to_string(),
                                        Value::from(report.stale_embeddings_removed),
                                    ),
                                    (
                                        "nodes_embedded".to_string(),
                                        Value::from(report.nodes_embedded),
                                    ),
                                    (
                                        "domains_evolved".to_string(),
                                        Value::from(report.domains_evolved.clone()),
                                    ),
                                    ("total_nodes".to_string(), Value::from(report.total_nodes)),
                                    ("total_edges".to_string(), Value::from(report.total_edges)),
                                ]
                                .into(),
                            );
                        }
                        return Ok(serde_json::to_value(&report).unwrap_or(Value::Null));
                    } else {
                        return Err(
                            "memory.consolidate requires memgine (attach with with_learning)"
                                .into(),
                        );
                    }
                }

                // Prefer a configured tool_executor for any tool it claims to handle.
                // Fall through to agent_basics only when the configured executor is absent
                // or explicitly returns "unknown tool" — this prevents agent_basics' built-in
                // read_file/write_file (which resolve paths via std::env::current_dir) from
                // silently overriding an executor that carries its own working_dir.
                let configured = {
                    let guard = self.tool_executor.lock().await;
                    guard.as_ref().cloned()
                };

                if let Some(ref executor) = configured {
                    let result = executor.execute(tool_name, &params).await;
                    let fall_through = matches!(&result, Err(e) if e.starts_with("unknown tool"));
                    if !fall_through {
                        if let Ok(ref value) = result {
                            self.result_cache
                                .put(tool_name, &params, value.clone())
                                .await;
                        }
                        return result;
                    }
                }

                if let Some(result) = crate::agent_basics::execute(tool_name, &params).await {
                    if let Ok(ref value) = result {
                        self.result_cache
                            .put(tool_name, &params, value.clone())
                            .await;
                    }
                    return result;
                }

                Err(format!("no handler for tool '{}'", tool_name))
            }
            ActionType::StateWrite => {
                let key = action
                    .parameters
                    .get("key")
                    .and_then(|v| v.as_str())
                    .ok_or("state_write requires 'key' parameter")?;
                let value = action
                    .parameters
                    .get("value")
                    .cloned()
                    .unwrap_or(Value::Null);
                self.state.set(key, value, &action.id);
                Ok(Value::from(format!("written: {}", key)))
            }
            ActionType::StateRead => {
                let key = action
                    .parameters
                    .get("key")
                    .and_then(|v| v.as_str())
                    .ok_or("state_read requires 'key' parameter")?;
                Ok(self.state.get(key).unwrap_or(Value::Null))
            }
            ActionType::Assertion => {
                let key = action
                    .parameters
                    .get("key")
                    .and_then(|v| v.as_str())
                    .ok_or("assertion requires 'key' parameter")?;
                let expected = action
                    .parameters
                    .get("expected")
                    .cloned()
                    .unwrap_or(Value::Null);
                let actual = self.state.get(key).unwrap_or(Value::Null);
                if actual != expected {
                    Err(format!(
                        "assertion failed: state['{}'] = {:?}, expected {:?}",
                        key, actual, expected
                    ))
                } else {
                    Ok(serde_json::json!({"asserted": key, "value": actual}))
                }
            }
        }
    }

    // --- Checkpoint and resume ---

    /// Save a checkpoint of the current runtime state.
    pub async fn save_checkpoint(&self) -> Checkpoint {
        let state = self.state.snapshot();
        let tools: Vec<String> = self.tools.read().await.keys().cloned().collect();
        let log = self.log.lock().await;
        let events: Vec<Value> = log
            .events()
            .iter()
            .map(|e| serde_json::to_value(e).unwrap_or_default())
            .collect();

        Checkpoint {
            checkpoint_id: Uuid::new_v4().to_string(),
            created_at: chrono::Utc::now(),
            state,
            events,
            tools,
            metadata: HashMap::new(),
        }
    }

    /// Save checkpoint to a JSON file.
    pub async fn save_checkpoint_to_file(&self, path: &str) -> Result<(), String> {
        let checkpoint = self.save_checkpoint().await;
        let json = serde_json::to_string_pretty(&checkpoint)
            .map_err(|e| format!("serialize error: {}", e))?;
        tokio::fs::write(path, json)
            .await
            .map_err(|e| format!("write error: {}", e))?;
        Ok(())
    }

    /// Load a checkpoint from a JSON file and restore state.
    pub async fn load_checkpoint_from_file(&self, path: &str) -> Result<Checkpoint, String> {
        let json = tokio::fs::read_to_string(path)
            .await
            .map_err(|e| format!("read error: {}", e))?;
        let checkpoint: Checkpoint =
            serde_json::from_str(&json).map_err(|e| format!("deserialize error: {}", e))?;
        self.restore_checkpoint(&checkpoint).await;
        Ok(checkpoint)
    }

    /// Restore runtime state from a checkpoint.
    pub async fn restore_checkpoint(&self, checkpoint: &Checkpoint) {
        // Replace state completely — don't merge, don't create synthetic transitions
        self.state.replace_all(checkpoint.state.clone());
        // Clear idempotency cache — stale results from pre-checkpoint execution
        // must not bypass validation/policy on the restored state
        self.idempotency_cache.lock().await.clear();
        // Restore tools (as name-only schemas; full schemas are not persisted in checkpoint)
        let mut tools = self.tools.write().await;
        tools.clear();
        for tool_name in &checkpoint.tools {
            let schema = ToolSchema {
                name: tool_name.clone(),
                description: String::new(),
                parameters: serde_json::Value::Object(Default::default()),
                returns: None,
                idempotent: false,
                cache_ttl_secs: None,
                rate_limit: None,
            };
            tools.insert(tool_name.clone(), schema);
        }
    }

    /// Register a subprocess tool and set up the subprocess executor.
    /// If no executor exists, creates a new SubprocessToolExecutor.
    /// If one already exists, creates a new SubprocessToolExecutor with the
    /// existing executor as fallback.
    pub async fn register_subprocess_tool(
        &self,
        name: &str,
        tool: crate::subprocess::SubprocessTool,
    ) {
        use crate::subprocess::SubprocessToolExecutor;

        let schema = ToolSchema {
            name: name.to_string(),
            description: format!("Subprocess tool: {}", tool.command),
            parameters: serde_json::Value::Object(Default::default()),
            returns: None,
            idempotent: false,
            cache_ttl_secs: None,
            rate_limit: None,
        };
        self.register_tool_schema(schema).await;

        let mut guard = self.tool_executor.lock().await;
        let mut executor = match guard.take() {
            Some(existing) => {
                let mut sub = SubprocessToolExecutor::new();
                sub = sub.with_fallback(existing);
                sub
            }
            None => SubprocessToolExecutor::new(),
        };
        executor.register(name, tool);
        *guard = Some(std::sync::Arc::new(executor));
    }
}

impl Default for Runtime {
    fn default() -> Self {
        Self::new()
    }
}