car-engine 0.32.1

//! 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::{ApprovalDecision, ApprovalLedger, ApprovalRecord, PermissionTier, 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,
    /// When true, validator/policy/capability rejections (ActionStatus::Rejected)
    /// also trigger rollback + replan, not just runtime Failed actions.
    /// Default false (conservative: preserves prior abort-only-on-Failed behavior).
    pub replan_on_rejected: bool,
}

/// How the runtime treats a proposal that conflicts with the current
/// shared state on a pre-execution transactional check (survey §4.3/§5.2.4
/// — `car_verify::check_transaction` against the versioned `StateStore`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum TransactionCheckMode {
    /// Don't run the check (default — preserves prior behavior exactly).
    #[default]
    Off,
    /// Run it and emit any conflicts as `TransactionConflict` telemetry, but
    /// execute anyway.
    Warn,
    /// Run it; if any conflict is found, reject the proposal without
    /// executing (the conflicting actions become `Rejected` results).
    Strict,
}

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

/// 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>;

    /// Variant that also carries the originating proposal `Action.id`
    /// and its `timeout_ms` budget.
    ///
    /// `action_id`: WS-based executors (`car-server-core::WsToolExecutor`)
    /// use it so the daemon-initiated `tools.execute` request to the client
    /// carries the same id the host's process-wide handler is keyed on
    /// — without this round-trip the host can't disambiguate concurrent
    /// callbacks for the same tool (Parslee-ai/car-releases#43 follow-up).
    ///
    /// `timeout_ms`: the action's per-call budget. WS executors MUST bound
    /// their callback wait by this (falling back to a default when `None`)
    /// so the daemon→host wait and the executor's own action deadline stay
    /// coordinated — otherwise a hardcoded inner wait reaps a call the outer
    /// deadline still permits (Parslee-ai/car#259). In-process executors
    /// that don't need either can keep the default forward to [`execute`].
    async fn execute_with_action(
        &self,
        tool: &str,
        params: &Value,
        _action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        self.execute(tool, params).await
    }

    /// Streaming entry point for detached invocation modes (C2). Start
    /// the tool and return a channel of [`car_ir::ToolStreamChunk`]s; the
    /// runtime drains it into the per-runtime handle registry while the
    /// DAG proceeds. End the stream with a terminal chunk (`done` /
    /// `error`); dropping the sender without one is reported as failure.
    /// A cooperative executor should stop work when the receiver returned
    /// here is dropped (that's what cancellation looks like from its side).
    ///
    /// Default: unsupported — existing one-shot executors compile and
    /// behave unchanged; a detached action against them is rejected with
    /// this error.
    async fn execute_stream(
        &self,
        tool: &str,
        _params: &Value,
        _action_id: &str,
    ) -> Result<tokio::sync::mpsc::Receiver<car_ir::ToolStreamChunk>, String> {
        Err(format!(
            "tool '{tool}': this executor does not support streaming/long-running invocation"
        ))
    }
}

/// Deterministic key for idempotency deduplication.
/// Deterministic key for idempotency deduplication. Carries the tenant
/// dimension (linus review): without it, tenant A's cached result for
/// an identical idempotent action was served to tenant B — a
/// cross-tenant data leak through the dedup cache. Unscoped executions
/// keep their historical keys (empty tenant segment).
fn idempotency_key(action: &Action, scope: Option<&crate::scope::RuntimeScope>) -> String {
    let sorted: std::collections::BTreeMap<_, _> = action.parameters.iter().collect();
    let params = serde_json::to_string(&sorted).unwrap_or_default();
    let tenant = scope.and_then(|s| s.tenant_id.as_deref()).unwrap_or("");
    format!(
        "{}:{}:{}:{}",
        tenant,
        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>,
    /// Pre-execution transactional conflict check mode (survey §4.3/§5.2.4).
    /// `Off` by default. When `Warn`/`Strict`, each proposal is checked
    /// against the versioned shared state before execution; `Strict`
    /// rejects on conflict. See [`Runtime::set_transaction_check_mode`].
    transaction_check: TokioRwLock<TransactionCheckMode>,
    /// The live harness operating config the Evolution Agent tunes (survey
    /// §3.5). `None` until one is installed via [`Runtime::set_harness_config`]
    /// — so default behavior is byte-identical to before (no cap on
    /// per-action retries, built-in backoff). Once installed, the runtime
    /// *reads* it: `max_retries` caps per-action retry budgets and
    /// `retry_backoff_ms` sets the inter-attempt delay; the setter also maps
    /// `planning_max_replans` onto the replan config. This is what makes an
    /// applied [`car_memgine::HarnessConfigPatch`] take effect.
    harness_config: TokioRwLock<Option<car_memgine::HarnessConfig>>,
    /// Canonical tool registry (optional — new code should use this).
    pub registry: Arc<crate::registry::ToolRegistry>,
    /// The environment the agent's side-effecting built-in tools act within.
    /// Defaults to [`crate::substrate::LocalSubstrate`] (host fs/process), which
    /// reproduces the historic `agent_basics` host behavior byte-for-byte.
    /// Bind a different environment (e.g. a VM via `McpSubstrate`) with
    /// [`Runtime::with_substrate`] / [`Runtime::set_substrate`]. `calculate`
    /// stays pure and never consults the substrate.
    substrate: TokioRwLock<Arc<dyn crate::substrate::Substrate>>,
    /// Proposal-admission gates — the pre-execution safety seam (EPIC A /
    /// task A1). Each registered [`crate::admission::AdmissionGate`] runs
    /// during admission, before any action executes; a proposal that any
    /// gate blocks (or escalates to approval) is refused. Empty by default,
    /// so a runtime that registers no gates behaves exactly as before.
    /// Individual gates (information-flow, concurrency, blocking-policy)
    /// are layered on via [`Runtime::register_admission_gate`].
    admission_gates: TokioRwLock<Vec<Arc<dyn crate::admission::AdmissionGate>>>,
    /// Durable human-in-the-loop approval ledger (EPIC A / A7). When set,
    /// an admission gate's `NeedsApproval` verdict is resolved against this
    /// ledger by fingerprint: a prior Approved decision admits the
    /// proposal, a Rejected decision blocks it, an unseen one stays pending
    /// (fail-closed). `None` by default — escalations fail closed with an
    /// explanatory reason until a ledger is installed.
    approval_ledger: TokioRwLock<Option<ApprovalLedger>>,
    /// Optional JSONL journal backing the idempotency cache (EPIC A / C3).
    /// `None` by default — the cache is in-memory and lost on restart. When
    /// set via [`Runtime::set_idempotency_cache_path`], idempotent results
    /// are persisted and reloaded so a crash-restart doesn't re-execute a
    /// completed idempotent action (avoiding duplicate side effects).
    idempotency_journal: TokioRwLock<Option<std::path::PathBuf>>,
    /// Detached tool invocations (C2): a `ToolCall` with a
    /// `streaming`/`long_running` invocation mode is registered here and
    /// its handle returned as the action's output while the DAG proceeds.
    /// Chunks are drained via [`Runtime::tool_poll`], cancelled via
    /// [`Runtime::tool_cancel`], and fanned out to
    /// [`Runtime::subscribe_tool_events`] subscribers.
    pub tool_handles: Arc<crate::tool_handles::ToolHandleRegistry>,
}

/// A durable idempotency-cache record (C3). `result: None` is a tombstone
/// recorded when a cached entry is invalidated by a rollback.
#[derive(serde::Serialize, serde::Deserialize)]
struct IdempotencyEntry {
    key: String,
    result: Option<ActionResult>,
}

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()),
            transaction_check: TokioRwLock::new(TransactionCheckMode::Off),
            harness_config: TokioRwLock::new(None),
            registry: Arc::new(crate::registry::ToolRegistry::new()),
            substrate: TokioRwLock::new(Arc::new(crate::substrate::LocalSubstrate::new())),
            admission_gates: TokioRwLock::new(Vec::new()),
            approval_ledger: TokioRwLock::new(None),
            idempotency_journal: TokioRwLock::new(None),
            tool_handles: Arc::new(crate::tool_handles::ToolHandleRegistry::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()),
            transaction_check: TokioRwLock::new(TransactionCheckMode::Off),
            harness_config: TokioRwLock::new(None),
            registry: Arc::new(crate::registry::ToolRegistry::new()),
            substrate: TokioRwLock::new(Arc::new(crate::substrate::LocalSubstrate::new())),
            admission_gates: TokioRwLock::new(Vec::new()),
            approval_ledger: TokioRwLock::new(None),
            idempotency_journal: TokioRwLock::new(None),
            tool_handles: Arc::new(crate::tool_handles::ToolHandleRegistry::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(())
    }

    /// Load declarative deny rules from a project's `.car/policies/`
    /// directory and register them on the global policy engine (EPIC A /
    /// task A2).
    ///
    /// `car_dir` is the project's `.car` directory (the engine discovers
    /// it by walking up from cwd); this looks for `car_dir/policies/*.toml`.
    /// A missing directory is not an error (returns 0). A malformed rule
    /// file *is* an error — a dropped security rule must surface loudly.
    /// Returns the number of rules registered.
    ///
    /// These rules are *deny*-only and additive on top of any code-
    /// registered policies. Once A9 makes policy violations blocking at
    /// admission, a matching action refuses the proposal.
    pub async fn load_project_policies(
        &self,
        car_dir: impl AsRef<std::path::Path>,
    ) -> Result<usize, car_policy::PolicyLoadError> {
        let dir = car_dir.as_ref().join("policies");
        let rules = car_policy::load_policy_dir(&dir)?;
        let count =
            rules.deny_tool.len() + rules.deny_keyword.len() + rules.deny_tool_param.len();
        let mut engine = self.policies.write().await;
        rules.apply(&mut engine);
        Ok(count)
    }

    /// Load information-flow tool labels from a project's `.car` directory
    /// and register the information-flow admission gate (EPIC A / A3+A4).
    ///
    /// Reads `car_dir/tool-labels.json` (merged over built-in defaults) and
    /// registers an [`crate::flow::InformationFlowGate`] so every admitted
    /// proposal is checked for data exfiltration (blocked) and forbidden
    /// tool orderings (escalated to approval). A missing labels file is
    /// fine — the built-in defaults still mark the network tools as sinks.
    /// A malformed file is a loud error.
    pub async fn install_information_flow_gate(
        &self,
        car_dir: impl AsRef<std::path::Path>,
    ) -> Result<(), crate::flow::FlowLoadError> {
        let config = crate::flow::load_tool_labels(car_dir)?;
        let gate = Arc::new(crate::flow::InformationFlowGate::new(config));
        self.register_admission_gate(gate).await;
        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);
    }

    /// Bind the execution substrate the side-effecting built-in tools
    /// (`read_file`/`write_file`/`edit_file`/`list_dir`/`find_files`/
    /// `grep_files`) act within (builder). Defaults to
    /// [`crate::substrate::LocalSubstrate`]; bind e.g.
    /// [`crate::substrate::McpSubstrate`] to make those tools hit a VM.
    /// `calculate` stays pure and ignores the substrate.
    pub fn with_substrate(self, substrate: Arc<dyn crate::substrate::Substrate>) -> Self {
        // Use try_write for non-async init context. Safe because we just created the lock.
        if let Ok(mut guard) = self.substrate.try_write() {
            *guard = substrate;
        }
        self
    }

    /// Set the execution substrate at runtime.
    pub async fn set_substrate(&self, substrate: Arc<dyn crate::substrate::Substrate>) {
        *self.substrate.write().await = substrate;
    }

    /// Clone the currently bound substrate.
    pub async fn substrate(&self) -> Arc<dyn crate::substrate::Substrate> {
        self.substrate.read().await.clone()
    }

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

    /// Set the pre-execution transactional conflict-check mode (survey
    /// §4.3/§5.2.4). `Off` (default) preserves prior behavior; `Warn`
    /// records conflicts as telemetry; `Strict` rejects a conflicting
    /// proposal before executing it.
    pub async fn set_transaction_check_mode(&self, mode: TransactionCheckMode) {
        *self.transaction_check.write().await = mode;
    }

    /// Register a proposal-admission gate (EPIC A / task A1).
    ///
    /// Gates run during admission, before any action executes, in the
    /// order they were registered. Each is a verified pre-execution safety
    /// check — information-flow (A4), concurrency (A5), blocking-policy
    /// (A9) — that can block a proposal or escalate it to human approval.
    /// Registering no gates leaves behavior unchanged.
    pub async fn register_admission_gate(
        &self,
        gate: Arc<dyn crate::admission::AdmissionGate>,
    ) {
        self.admission_gates.write().await.push(gate);
    }

    /// Remove all registered admission gates (primarily for tests and
    /// reconfiguration). After this, proposal admission reverts to the
    /// transactional pre-check only.
    pub async fn clear_admission_gates(&self) {
        self.admission_gates.write().await.clear();
    }

    /// The number of currently-registered admission gates.
    pub async fn admission_gate_count(&self) -> usize {
        self.admission_gates.read().await.len()
    }

    /// Register the VIGIL intent gate (arXiv 2601.05755 — the live
    /// verify-before-commit call-site). Installs
    /// [`crate::intent_gate::IntentGate`] as an admission gate: a
    /// forbidden capability or a tool-stream-influenced out-of-intent
    /// action hard-rejects the proposal; untainted drift escalates to
    /// the durable approval flow (A7) by content-bound fingerprint.
    /// Replans are covered automatically (gates re-run on every
    /// replanned proposal).
    pub async fn install_intent_gate(&self, config: crate::intent_gate::IntentGateConfig) {
        self.register_admission_gate(Arc::new(crate::intent_gate::IntentGate::new(config)))
            .await;
    }

    /// Load `.car/intent.json` from `car_dir` and install the intent
    /// gate when present. Absent file → Ok(false) (opt-in, ungated);
    /// malformed file → loud error, never a silently-ungated session.
    pub async fn install_intent_gate_from_project(
        &self,
        car_dir: impl AsRef<std::path::Path>,
    ) -> Result<bool, crate::intent_gate::IntentLoadError> {
        match crate::intent_gate::load_intent_config(car_dir)? {
            Some(cfg) => {
                self.install_intent_gate(cfg).await;
                Ok(true)
            }
            None => Ok(false),
        }
    }

    /// Register the skill deployment-tier ceiling gate (EPIC A / A8).
    ///
    /// Requires a memgine to be attached (skills live there). When a
    /// proposal names its driving skill in `context["skill"]`, the gate
    /// caps the proposal's actions at that skill's persisted
    /// `deployment_tier`, escalating an over-ceiling action to the durable
    /// approval flow (A7). Returns false if no memgine is attached.
    pub async fn install_skill_ceiling_gate(&self) -> bool {
        match &self.memgine {
            Some(mem) => {
                let gate = Arc::new(crate::skill_ceiling::SkillCeilingGate::new(mem.clone()));
                self.register_admission_gate(gate).await;
                true
            }
            None => false,
        }
    }

    /// Drain buffered chunks + status for a detached tool invocation (C2).
    /// `None` for an unknown or fully-consumed handle. See
    /// [`crate::tool_handles::ToolHandleRegistry::poll`] for the
    /// consume-on-terminal contract.
    pub async fn tool_poll(
        &self,
        handle_id: &str,
    ) -> Option<crate::tool_handles::ToolPollResult> {
        self.tool_handles.poll(handle_id).await
    }

    /// Request cancellation of a detached tool invocation (C2): fires the
    /// handle's cancel token (dropping the executor's chunk receiver) and
    /// seals its status as `cancelled` unless already terminal. Returns
    /// false for an unknown handle.
    pub async fn tool_cancel(&self, handle_id: &str) -> bool {
        self.tool_handles.cancel(handle_id).await
    }

    /// Subscribe to the live [`car_ir::ToolStreamEvent`] fanout for all
    /// detached tool invocations on this runtime (C2). The WS layer
    /// forwards these as `tools.stream.event` notifications.
    pub fn subscribe_tool_events(
        &self,
    ) -> tokio::sync::broadcast::Receiver<car_ir::ToolStreamEvent> {
        self.tool_handles.subscribe()
    }

    /// Enable tamper-evident hash chaining on the event log (EPIC A / A9).
    /// Every event appended from now on is linked to its predecessor by a
    /// content hash, so an after-the-fact edit to a chained event — or an
    /// interior deletion/reordering — is detectable via
    /// [`Runtime::verify_event_log_chain`]. Truncation at either end of the
    /// log (dropping a prefix or suffix wholesale) is NOT detectable — the
    /// chain has no anchored head hash and no trusted tail witness; that is
    /// out of scope until the chain head is anchored. Opt-in: existing logs
    /// stay byte-identical until enabled.
    pub async fn enable_event_log_hash_chaining(&self) {
        self.log.lock().await.enable_hash_chaining();
    }

    /// Verify the event log's tamper-evidence chain (EPIC A / A9). Returns
    /// `Ok(n)` for `n` verified chained events, or `Err(index)` naming the
    /// first event whose hash/linkage doesn't match — the point of an
    /// interior edit, deletion, or reordering. Head/tail truncation is not
    /// detectable (no anchored head hash; the first chained event's
    /// `prev_hash` is taken on trust) — see `EventLog::verify_chain`.
    pub async fn verify_event_log_chain(&self) -> Result<usize, usize> {
        self.log.lock().await.verify_chain()
    }

    /// Install a durable HITL approval ledger backed by a JSONL journal
    /// (EPIC A / A7). Loads any existing decisions so approvals survive a
    /// restart, then resolves future admission-gate `NeedsApproval`
    /// verdicts against it. The canonical daemon path is
    /// `~/.car/approvals.jsonl`.
    pub async fn set_approval_ledger_path(
        &self,
        path: impl Into<std::path::PathBuf>,
    ) -> std::io::Result<()> {
        let ledger = ApprovalLedger::with_journal(path.into())?;
        // Surface journal corruption (or a torn concurrent write) instead of
        // silently trusting a partial ledger (review A7 — the doc on
        // `skipped_on_load` promises callers surface it).
        if ledger.skipped_on_load() > 0 {
            tracing::warn!(
                skipped = ledger.skipped_on_load(),
                "approval ledger journal had unparseable lines skipped on load — \
                 the ledger may be missing decisions"
            );
        }
        *self.approval_ledger.write().await = Some(ledger);
        Ok(())
    }

    /// Use an in-memory approval ledger (no persistence) — primarily for
    /// tests and ephemeral runtimes.
    pub async fn set_approval_ledger_in_memory(&self) {
        *self.approval_ledger.write().await = Some(ApprovalLedger::new());
    }

    /// Record a human approval for an admission fingerprint (EPIC A / A7).
    /// A subsequently re-submitted proposal whose escalation matches this
    /// fingerprint is admitted without asking again.
    pub async fn approve_admission(
        &self,
        fingerprint: &str,
        reviewer: &str,
        reason: &str,
    ) -> Result<(), String> {
        self.record_admission_decision(fingerprint, ApprovalDecision::Approved, reviewer, reason)
            .await
    }

    /// Record a human rejection for an admission fingerprint (EPIC A / A7).
    /// A proposal whose escalation matches a rejected fingerprint is
    /// blocked outright.
    pub async fn reject_admission(
        &self,
        fingerprint: &str,
        reviewer: &str,
        reason: &str,
    ) -> Result<(), String> {
        self.record_admission_decision(fingerprint, ApprovalDecision::Rejected, reviewer, reason)
            .await
    }

    async fn record_admission_decision(
        &self,
        fingerprint: &str,
        decision: ApprovalDecision,
        reviewer: &str,
        reason: &str,
    ) -> Result<(), String> {
        {
            let mut guard = self.approval_ledger.write().await;
            let ledger = guard
                .as_mut()
                .ok_or_else(|| "no approval ledger installed".to_string())?;
            ledger
                .record(ApprovalRecord {
                    fingerprint: fingerprint.to_string(),
                    // Admission escalations aren't tier-classified; record the
                    // most restrictive tier so the decision reads as "elevated".
                    required_tier: PermissionTier::FullAccess,
                    decision,
                    reviewer: reviewer.to_string(),
                    reason: reason.to_string(),
                    evidence: None,
                    decided_at: chrono::Utc::now().to_rfc3339(),
                })
                // A journal write failure means the decision is NOT durable —
                // surface it instead of emitting a false ApprovalRecorded
                // audit event (review A7).
                .map_err(|e| format!("failed to persist approval decision: {e}"))?;
        }
        let approval = match decision {
            ApprovalDecision::Approved => "approved",
            ApprovalDecision::Rejected => "rejected",
        };
        let mut log = self.log.lock().await;
        log.append(
            EventKind::ApprovalRecorded,
            None,
            None,
            [
                ("fingerprint".to_string(), Value::from(fingerprint)),
                ("approval".to_string(), Value::from(approval)),
                ("reviewer".to_string(), Value::from(reviewer)),
                ("reason".to_string(), Value::from(reason)),
            ]
            .into(),
        );
        Ok(())
    }

    /// Back the idempotency cache with a durable JSONL journal (EPIC A / C3).
    ///
    /// Loads any existing entries into the in-memory cache, then persists
    /// future idempotent results (and rollback invalidations, as
    /// tombstones) to the journal. After a crash-restart, re-submitting a
    /// completed idempotent action returns the cached result instead of
    /// re-executing it — preventing duplicate external side effects. The
    /// canonical daemon path is `~/.car/idempotency.jsonl`. Returns the
    /// number of live entries loaded.
    pub async fn set_idempotency_cache_path(
        &self,
        path: impl Into<std::path::PathBuf>,
    ) -> std::io::Result<usize> {
        let path = path.into();
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        // Replay the journal (last entry per key wins; a tombstone removes).
        let mut loaded: HashMap<String, ActionResult> = HashMap::new();
        if path.exists() {
            let content = std::fs::read_to_string(&path)?;
            for line in content.lines() {
                if line.trim().is_empty() {
                    continue;
                }
                if let Ok(entry) = serde_json::from_str::<IdempotencyEntry>(line) {
                    match entry.result {
                        Some(r) => {
                            loaded.insert(entry.key, r);
                        }
                        None => {
                            loaded.remove(&entry.key);
                        }
                    }
                }
            }
        }
        {
            let mut cache = self.idempotency_cache.lock().await;
            for (k, v) in loaded.iter() {
                cache.entry(k.clone()).or_insert_with(|| v.clone());
            }
        }
        let count = loaded.len();
        *self.idempotency_journal.write().await = Some(path);
        Ok(count)
    }

    /// Append an idempotency record to the journal if one is configured.
    /// `result = None` writes a tombstone (invalidation).
    async fn journal_idempotency(&self, key: &str, result: Option<&ActionResult>) {
        let path = {
            let guard = self.idempotency_journal.read().await;
            guard.clone()
        };
        let Some(path) = path else {
            return;
        };
        let entry = IdempotencyEntry {
            key: key.to_string(),
            result: result.cloned(),
        };
        if let Ok(mut line) = serde_json::to_string(&entry) {
            line.push('\n');
            use std::io::Write;
            // Persistence failure must be VISIBLE (linus review): a
            // silently-dropped journal write means an idempotent action
            // re-executes its side effects after a restart while the
            // operator believes it's covered. Fail-open (the in-memory
            // cache still dedups this process) but loudly.
            let write = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(&path)
                .and_then(|mut f| f.write_all(line.as_bytes()));
            if let Err(e) = write {
                tracing::warn!(
                    path = %path.display(),
                    error = %e,
                    "idempotency journal write failed — durable dedup is NOT covering this result"
                );
            }
        }
    }

    /// Look up the current decision for an admission fingerprint, if any.
    pub async fn admission_decision(&self, fingerprint: &str) -> Option<ApprovalDecision> {
        let guard = self.approval_ledger.read().await;
        guard
            .as_ref()
            .and_then(|l| l.lookup(fingerprint).map(|r| r.decision))
    }

    /// Verify a model's tool-use claims against the runtime's own execution
    /// receipts (EPIC A / A6 — arXiv 2603.10060). The runtime owns tool
    /// execution and logs it, so it holds unforgeable ground truth: this
    /// projects [`car_eventlog::tool_receipts::ToolReceipt`]s from the event
    /// log and cross-checks the supplied claims, catching a fabricated tool
    /// reference, a misstated result count, or a false "found nothing".
    ///
    /// `proposal_id` scopes the cross-check window to a single proposal's
    /// events (pass the proposal whose response the claims came from) — a
    /// claim is never judged against another run's receipts. The check is
    /// retention-coherent (review A6): when the log has trimmed events and
    /// the window can't be proven complete (unscoped, or the proposal's
    /// `ProposalReceived` marker — which precedes every receipt of that
    /// proposal — was itself evicted), a claim without a receipt comes back
    /// in `ReceiptReport::ungroundable` ("window evicted") instead of being
    /// mis-flagged `fabricated_tool_reference`.
    ///
    /// Returns the [`ReceiptReport`]; when it is not grounded, a
    /// `ToolReceiptHallucination` event is emitted so the caller's
    /// verdict→action loop (reject/flag the response) is auditable.
    /// Deterministic, zero-inference. Claims arrive structured — CAR's
    /// thesis is that intent is structured IR, so a caller extracts claims
    /// from the model's tool_calls / IR rather than regexing prose.
    pub async fn verify_tool_receipts(
        &self,
        claims: &[car_eventlog::tool_receipts::ToolClaim],
        proposal_id: Option<&str>,
    ) -> car_eventlog::tool_receipts::ReceiptReport {
        let (receipts, window_complete) = {
            let log = self.log.lock().await;
            let complete = if log.trimmed_events() == 0 {
                // Nothing was ever evicted — the window is complete whether
                // or not it is scoped.
                true
            } else {
                match proposal_id {
                    // A proposal's window is complete iff its ProposalReceived
                    // marker survived retention: every receipt of the proposal
                    // was appended after it, so if the marker is retained, so
                    // are the receipts.
                    Some(pid) => log.events().iter().any(|e| {
                        e.kind == EventKind::ProposalReceived
                            && e.proposal_id.as_deref() == Some(pid)
                    }),
                    // Unscoped check over a trimmed log: unknowable.
                    None => false,
                }
            };
            (
                car_eventlog::tool_receipts::receipts_from_events_scoped(
                    log.events(),
                    proposal_id,
                ),
                complete,
            )
        };
        let report = car_eventlog::tool_receipts::verify_tool_claims_windowed(
            claims,
            &receipts,
            window_complete,
        );
        if !report.grounded {
            let mut log = self.log.lock().await;
            log.append(
                EventKind::ToolReceiptHallucination,
                None,
                proposal_id,
                [
                    (
                        "count".to_string(),
                        Value::from(report.hallucinations.len()),
                    ),
                    (
                        "hallucinations".to_string(),
                        serde_json::to_value(&report.hallucinations).unwrap_or_default(),
                    ),
                ]
                .into(),
            );
        }
        report
    }

    /// Run every registered admission gate against a proposal and fold
    /// their verdicts into a single [`crate::admission::AdmissionDecision`].
    ///
    /// Each gate's outcome is recorded as an `AdmissionGateDecision` event
    /// so a denial is attributable. Aggregation is conjunctive and
    /// fail-closed: the proposal is admitted only if every gate allowed it.
    /// Returns an admit decision immediately when no gates are registered
    /// (zero overhead on the common path).
    async fn run_admission_gates(
        &self,
        proposal: &ActionProposal,
        session_id: Option<&str>,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> crate::admission::AdmissionDecision {
        use crate::admission::{AdmissionDecision, GateContext};

        let gates = self.admission_gates.read().await;
        if gates.is_empty() {
            return AdmissionDecision::admit();
        }

        // One consistent snapshot for every gate this pass.
        let (state, versions) = self.state.versioned_snapshot();
        let ctx = GateContext {
            session_id,
            scope,
            state: &state,
            versions: &versions,
        };

        let mut decision = AdmissionDecision::admit();
        for gate in gates.iter() {
            let outcome = gate.check(proposal, &ctx).await;
            // Audit every gate decision (allow included) so the trail shows
            // which gates ran, not just which objected.
            let mut props: HashMap<String, Value> = HashMap::new();
            props.insert("gate".to_string(), Value::from(gate.name()));
            // Pre-execution proposal admission (vs the multi-agent
            // commit barrier, which emits the same event kind with
            // phase:"commit_barrier").
            props.insert("phase".to_string(), Value::from("admission"));
            props.insert("decision".to_string(), Value::from(outcome.label()));
            match &outcome {
                crate::admission::GateOutcome::Allow => {}
                crate::admission::GateOutcome::Reject { blocked, reason } => {
                    props.insert("reason".to_string(), Value::from(reason.clone()));
                    props.insert(
                        "blocked".to_string(),
                        serde_json::to_value(blocked).unwrap_or_default(),
                    );
                }
                crate::admission::GateOutcome::NeedsApproval {
                    actions,
                    fingerprint,
                    reason,
                } => {
                    props.insert("reason".to_string(), Value::from(reason.clone()));
                    props.insert(
                        "blocked".to_string(),
                        serde_json::to_value(actions).unwrap_or_default(),
                    );
                    props.insert("fingerprint".to_string(), Value::from(fingerprint.clone()));
                }
            }
            {
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::AdmissionGateDecision,
                    None,
                    Some(&proposal.id),
                    props,
                );
            }
            decision.absorb(gate.name(), outcome);
        }
        decision
    }

    /// Install a harness operating config — the live end of the Evolution
    /// Agent loop (survey §3.5). After the meta-agent's `HarnessConfig::apply`
    /// produces a governed, regression-gated config, hand it here to take
    /// effect: `max_retries`/`retry_backoff_ms` drive the per-action retry
    /// loop, and `planning_max_replans` is mapped onto the replan budget.
    /// Every `HarnessConfig` knob is consumed here — none is aspirational.
    pub async fn set_harness_config(&self, cfg: car_memgine::HarnessConfig) {
        self.replan_config.write().await.max_replans = cfg.planning_max_replans;
        *self.harness_config.write().await = Some(cfg);
    }

    /// The current harness operating config, if one has been installed.
    pub async fn harness_config(&self) -> Option<car_memgine::HarnessConfig> {
        self.harness_config.read().await.clone()
    }

    /// Atomically read-modify-write the harness operating config under ONE
    /// write lock (installing the default first when none is set). The
    /// get→mutate→set alternative is a lost-update race when two requests
    /// mutate concurrently — the daemon's `evolution.run` harness-apply path
    /// uses this instead (kernel review S3). Keeps the same replan-budget
    /// propagation as [`Self::set_harness_config`].
    pub async fn update_harness_config<R>(
        &self,
        f: impl FnOnce(&mut car_memgine::HarnessConfig) -> R,
    ) -> R {
        let mut guard = self.harness_config.write().await;
        let cfg = guard.get_or_insert_with(car_memgine::HarnessConfig::default);
        let out = f(cfg);
        let max_replans = cfg.planning_max_replans;
        drop(guard);
        self.replan_config.write().await.max_replans = max_replans;
        out
    }

    /// Run the pre-execution transactional check against the current
    /// versioned shared state. Returns the conflicting action ids when the
    /// mode is `Strict` and conflicts exist (so the caller can reject those
    /// actions); always emits `TransactionConflict` telemetry for each
    /// conflict found. Empty/`None` means "proceed".
    async fn transaction_precheck(
        &self,
        proposal: &ActionProposal,
    ) -> std::collections::HashSet<String> {
        let mode = *self.transaction_check.read().await;
        if mode == TransactionCheckMode::Off {
            return std::collections::HashSet::new();
        }
        let (state, versions) = self.state.versioned_snapshot();
        let report = car_verify::check_transaction(proposal, &versions, Some(&state));
        if report.consistent {
            return std::collections::HashSet::new();
        }
        let mut blocked = std::collections::HashSet::new();
        let mut log = self.log.lock().await;
        for c in &report.conflicts {
            for aid in &c.actions {
                blocked.insert(aid.clone());
            }
            log.append(
                EventKind::TransactionConflict,
                c.actions.first().map(|s| s.as_str()),
                Some(&proposal.id),
                [
                    // Use the serde representation (snake_case: write_write
                    // / read_write / stale_assumption) the .d.ts/.pyi/doc
                    // contract is written against — NOT Debug, which would
                    // emit "writewrite" (neo review #4).
                    (
                        "kind".to_string(),
                        serde_json::to_value(c.kind).unwrap_or_default(),
                    ),
                    ("key".to_string(), Value::from(c.key.clone())),
                    (
                        "actions".to_string(),
                        serde_json::to_value(&c.actions).unwrap_or_default(),
                    ),
                    (
                        "explanation".to_string(),
                        Value::from(c.explanation.clone()),
                    ),
                    ("resolution".to_string(), Value::from(c.resolution.clone())),
                ]
                .into(),
            );
        }
        drop(log);
        // Only Strict blocks execution; Warn records and proceeds.
        match mode {
            TransactionCheckMode::Strict => blocked,
            _ => std::collections::HashSet::new(),
        }
    }

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

    /// Remove a tool from both the canonical registry and the legacy
    /// `tools` schema map, so the model no longer sees it and the
    /// validator no longer accepts it. Used when a remote MCP connector
    /// tool is disabled or its connector is removed. Returns true if the
    /// tool was present in either store.
    pub async fn unregister_tool(&self, name: &str) -> bool {
        let removed_entry = self.registry.remove(name).await.is_some();
        let removed_schema = self.tools.write().await.remove(name).is_some();
        removed_entry || removed_schema
    }

    /// 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), None, 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, None, cancel)
            .await
    }

    /// Execute a proposal with an attached [`RuntimeScope`]
    /// (Parslee-ai/car#187 phase 3).
    ///
    /// Same contract as [`Self::execute`] plus a per-execution
    /// identity surface — typically built by the car-a2a dispatcher
    /// from the verified `Identity` and cooperative `a2a_caller`
    /// metadata on the inbound `ActionProposal`. The scope is
    /// recorded on the event log so downstream audit / log analysis
    /// can see which caller / tenant issued each action.
    ///
    /// **What this enforces today**: scope is captured + logged.
    /// Memgine queries and state-store ops still hit global
    /// namespaces — those follow-ups are tracked under #187.
    /// Tool / policy code that needs per-tenant behaviour right now
    /// should keep reading `proposal.context["a2a_caller_verified"]`
    /// directly (the phase 1 / 2 surface).
    pub async fn execute_scoped(
        &self,
        proposal: &ActionProposal,
        scope: &crate::scope::RuntimeScope,
    ) -> ProposalResult {
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_scoped_with_cancel(proposal, scope, &token)
            .await
    }

    /// Combined scoped + cancellable execute. Mirrors the shape of
    /// [`Self::execute_with_session_and_cancel`] for symmetry — both
    /// add a side-channel (session id / scope) on top of the
    /// cancellable form.
    pub async fn execute_scoped_with_cancel(
        &self,
        proposal: &ActionProposal,
        scope: &crate::scope::RuntimeScope,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        self.execute_with_optional_session(proposal, None, Some(scope), 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>,
        scope: Option<&crate::scope::RuntimeScope>,
        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;

        // Phase 3 foundation (Parslee-ai/car#187): record the scope
        // on the event log so audit / log analysis can correlate
        // actions to the caller / tenant that triggered them. Only
        // logged when at least one identity field is set — keeps
        // the existing in-process call sites free of noise.
        if let Some(s) = scope {
            if !s.is_unscoped() {
                let mut props: HashMap<String, Value> = HashMap::new();
                if let Some(cid) = &s.caller_id {
                    props.insert("caller_id".to_string(), Value::from(cid.as_str()));
                }
                if let Some(tid) = &s.tenant_id {
                    props.insert("tenant_id".to_string(), Value::from(tid.as_str()));
                }
                if !s.claims.is_empty() {
                    if let Ok(claims_json) = serde_json::to_value(&s.claims) {
                        props.insert("claims".to_string(), claims_json);
                    }
                }
                let mut log = self.log.lock().await;
                log.append(EventKind::SessionScope, None, Some(&proposal.id), props);
            }
        }

        // Pre-execution transactional conflict check (survey §4.3/§5.2.4).
        // Off by default; Warn records conflicts; Strict rejects the whole
        // proposal before any action runs, since a transactional conflict is
        // a property of the action *set* against current state, not an
        // isolated action. This is an advisory planning-time gate, not a
        // substitute for per-action execution-time validation. Replanned
        // proposals are re-checked inside the loop (at the replan quality
        // gate below). A pre-execution rejection here deliberately produces
        // no execution trajectory — nothing ran; the emitted
        // `TransactionConflict` events are the audit record.
        let blocked = self.transaction_precheck(&current_proposal).await;
        if !blocked.is_empty() {
            let results = current_proposal
                .actions
                .iter()
                .map(|a| {
                    rejected_result(
                        &a.id,
                        "transactional conflict with current shared state (strict mode); \
                         see TransactionConflict events for details and resolution"
                            .to_string(),
                    )
                })
                .collect();
            return ProposalResult {
                proposal_id: proposal.id.clone(),
                results,
                cost: Default::default(),
            };
        }

        // Pre-execution admission gates (EPIC A / task A1 — the safety
        // seam). Runs every registered AdmissionGate against the proposal
        // before any action executes. Aggregation is conjunctive and
        // fail-closed: a proposal any gate blocks (or escalates to
        // approval) does not run. Like the transactional pre-check above, a
        // rejection here produces no execution trajectory — the emitted
        // AdmissionGateDecision events are the audit record. No gates
        // registered → zero overhead, identical behavior.
        let admission = self
            .run_admission_gates(&current_proposal, session_id, scope)
            .await;
        if !admission.admitted {
            let gate = admission.deciding_gate.as_deref().unwrap_or("admission");
            let base_reason = admission
                .reason
                .clone()
                .unwrap_or_else(|| "blocked by admission gate".to_string());
            // Resolve approval escalations against the durable ledger (A7).
            // A hard `Reject` from ANY gate is never overridable — the
            // ledger is not consulted at all in that case (an old approval
            // for one gate's escalation must not steamroll another gate's
            // deny). Otherwise EVERY escalation must resolve to Approved,
            // each by its own fingerprint: one Rejected fingerprint blocks,
            // one unseen fingerprint stays pending (fail-closed).
            let mut approved = false;
            let reason = if admission.hard_rejected {
                format!("{base_reason} (gate: {gate})")
            } else if admission.needs_approval() {
                let mut blocking_reason = None;
                for esc in &admission.escalations {
                    match self.admission_decision(&esc.fingerprint).await {
                        Some(ApprovalDecision::Approved) => continue,
                        Some(ApprovalDecision::Rejected) => {
                            blocking_reason = Some(format!(
                                "rejected by operator (gate: {}; fingerprint: {})",
                                esc.gate, esc.fingerprint
                            ));
                            break;
                        }
                        None => {
                            blocking_reason = Some(format!(
                                "requires human approval (gate: {}; {}); \
                                 approve fingerprint '{}' then re-run",
                                esc.gate, esc.reason, esc.fingerprint
                            ));
                            break;
                        }
                    }
                }
                match blocking_reason {
                    None => {
                        approved = true;
                        // Audit every durable approval taking effect.
                        let mut log = self.log.lock().await;
                        for esc in &admission.escalations {
                            log.append(
                                EventKind::ApprovalRecorded,
                                None,
                                Some(&proposal.id),
                                [
                                    (
                                        "fingerprint".to_string(),
                                        Value::from(esc.fingerprint.as_str()),
                                    ),
                                    ("gate".to_string(), Value::from(esc.gate.as_str())),
                                    ("approval".to_string(), Value::from("approved")),
                                    ("applied".to_string(), Value::from(true)),
                                ]
                                .into(),
                            );
                        }
                        String::new()
                    }
                    Some(r) => r,
                }
            } else {
                format!("{base_reason} (gate: {gate})")
            };
            if approved {
                // Escalation cleared by a durable approval — proceed to
                // execution as if admitted.
            } else {
            let results = current_proposal
                .actions
                .iter()
                .map(|a| {
                    // Name the specific reason on the offending actions; a
                    // generic note on the rest (the whole proposal is held,
                    // since a safety hazard is a property of the set).
                    if admission.blocked.is_empty() || admission.blocked.contains(&a.id) {
                        rejected_result(&a.id, reason.clone())
                    } else {
                        rejected_result(
                            &a.id,
                            format!("proposal blocked by admission gate: {gate}"),
                        )
                    }
                })
                .collect();
            return ProposalResult {
                proposal_id: proposal.id.clone(),
                results,
                cost: Default::default(),
            };
            }
        }

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

            // Statuses that trigger rollback + replan: always runtime Failed,
            // and (opt-in) validator/policy/capability Rejected.
            let replan_triggers = |s: &ActionStatus| {
                *s == ActionStatus::Failed
                    || (config.replan_on_rejected && *s == ActionStatus::Rejected)
            };

            // Check if we aborted
            let aborted = result.results.iter().any(|r| replan_triggers(&r.status));
            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| replan_triggers(&r.status))
                .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(),
                );
                // Deep-telemetry breadcrumbs (§3.5.1): record the fork the
                // harness took (replan vs accept vs abandon) and the
                // approach it discarded, so failure-mode diagnosis can see
                // the path *not* taken, not just the path taken.
                let rejected_ids: Vec<String> = ctx
                    .failed_actions
                    .iter()
                    .map(|f| f.action_id.clone())
                    .collect();
                log.append(
                    EventKind::BranchDecision,
                    None,
                    Some(&proposal.id),
                    [
                        ("branch".to_string(), Value::from("replan")),
                        (
                            "reason".to_string(),
                            Value::from("actions failed; requesting a revised plan"),
                        ),
                        ("attempt".to_string(), Value::from(attempt + 1)),
                    ]
                    .into(),
                );
                log.append(
                    EventKind::AlternativeRejected,
                    None,
                    Some(&proposal.id),
                    [
                        (
                            "alternative".to_string(),
                            serde_json::to_value(&rejected_ids).unwrap_or_default(),
                        ),
                        (
                            "reason".to_string(),
                            Value::from("plan superseded by replan after action failure"),
                        ),
                    ]
                    .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 current_state = self.state.snapshot();
                        // Verify against the full registered schemas
                        // (not just names) so a replan with a bad
                        // parameter type / missing required field is
                        // rejected here, per register_tool_schema's
                        // contract (car-releases#56). The read guard is
                        // held across the synchronous verify call.
                        let tools_guard = self.tools.read().await;
                        let vr = car_verify::verify_with_schemas(
                            &new_proposal,
                            Some(&current_state),
                            Some(&tools_guard),
                            100,
                        );
                        drop(tools_guard);
                        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;
                        }
                    }

                    // Transactional re-check of the replan against the now-
                    // mutated shared state (neo review #1): a replan runs
                    // after earlier actions changed state, so it's exactly
                    // where a fresh stale-assumption/write conflict appears.
                    // A Strict conflict rejects this replan attempt via the
                    // same bounded ReplanRejected path (capped by
                    // max_replans), never an infinite loop.
                    let replan_blocked = self.transaction_precheck(&new_proposal).await;
                    if !replan_blocked.is_empty() {
                        let mut log = self.log.lock().await;
                        log.append(
                            EventKind::ReplanRejected,
                            None,
                            Some(&proposal.id),
                            [
                                (
                                    "errors".to_string(),
                                    Value::from("transactional conflict with current state"),
                                ),
                                ("attempt".to_string(), Value::from(attempt + 1)),
                            ]
                            .into(),
                        );
                        drop(log);
                        attempt += 1;
                        continue;
                    }

                    // Admission gates re-run on EVERY replanned proposal
                    // (linus review C-2): the replan callback is exactly
                    // where injected/adversarial content reshapes a plan,
                    // so a proposal that was clean at first admission must
                    // not smuggle a hazardous replan past the gates. Same
                    // fail-closed contract as first admission — a hard
                    // reject or any unapproved escalation rejects this
                    // replan attempt (bounded by max_replans).
                    let replan_admission = self
                        .run_admission_gates(&new_proposal, session_id, scope)
                        .await;
                    if !replan_admission.admitted {
                        let mut cleared = !replan_admission.hard_rejected
                            && replan_admission.needs_approval();
                        if cleared {
                            for esc in &replan_admission.escalations {
                                if self.admission_decision(&esc.fingerprint).await
                                    != Some(ApprovalDecision::Approved)
                                {
                                    cleared = false;
                                    break;
                                }
                            }
                        }
                        if !cleared {
                            let gate = replan_admission
                                .deciding_gate
                                .as_deref()
                                .unwrap_or("admission");
                            let why = replan_admission
                                .reason
                                .clone()
                                .unwrap_or_else(|| "blocked by admission gate".to_string());
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    (
                                        "errors".to_string(),
                                        Value::from(format!(
                                            "admission gate blocked replan (gate: {gate}; {why})"
                                        )),
                                    ),
                                    ("attempt".to_string(), Value::from(attempt + 1)),
                                ]
                                .into(),
                            );
                            drop(log);
                            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>,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> (ProposalResult, HashMap<String, HashMap<String, Value>>) {
        // Whether validator/policy/capability rejections should count toward
        // the abort-and-replan path (default false). Read once up front so the
        // per-action loop doesn't take the replan_config lock repeatedly (and
        // never inside join_all). Independent lock from log/tools/policies/
        // capabilities, so no lock-ordering conflict.
        let replan_on_rejected = self.replan_config.read().await.replan_on_rejected;

        // 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. When the proposal carries a tenant scope,
        // snapshot only that tenant's namespace so a rollback can't clobber
        // concurrent tenants' state (EPIC E / E2). No tenant → unchanged
        // full snapshot.
        let rollback_tenant: Option<&str> = scope.and_then(|s| s.tenant_id.as_deref());
        let snapshot = match rollback_tenant {
            Some(_) => self.state.snapshot_scoped(rollback_tenant),
            None => 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,
                            scope,
                        )
                        .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
                        || (replan_on_rejected && ar.status == ActionStatus::Rejected))
                        && 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,
                            scope,
                        )
                    })
                    .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 {
            // Tenant-scoped rollback when the proposal is tenant-scoped
            // (EPIC E / E2), else the full restore.
            match rollback_tenant {
                Some(_) => {
                    self.state
                        .restore_scoped(rollback_tenant, snapshot.clone(), transition_count)
                }
                None => 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 invalidated: Vec<String> = Vec::new();
            {
                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 {
                                let key = idempotency_key(action, scope);
                                cache.remove(&key);
                                invalidated.push(key);
                            }
                        }
                    }
                }
            }
            // Record tombstones so the durable journal (C3) doesn't resurrect
            // a rolled-back result on reload.
            for key in &invalidated {
                self.journal_idempotency(key, None).await;
            }
        }

        // 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();
                    // Validation-gated ingest (SkillOpt-inspired). Distilled
                    // skills are NOT trusted straight into the active pool; each
                    // enters as a PROVISIONAL candidate on trial and must prove
                    // itself (or beat its incumbent) before the promotion gate in
                    // consolidate() makes it Active. Tenant is threaded through so
                    // candidates are stamped for the calling tenant's namespace.
                    let tenant = scope.and_then(|s| s.tenant_id.as_deref());
                    let provisional = engine.ingest_provisional_candidates(&skills, tenant);

                    // 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)),
                            ("provisional_ingested".to_string(), Value::from(provisional)),
                            (
                                "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: capability → validate → policy → idempotency → execute.
    /// (Idempotency dedup runs AFTER the deny checks — review C3: a durable
    /// cached result must never outlive a tool's revocation.)
    /// 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>,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> (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, scope)
            .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>,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> (ActionResult, u32) {
        // 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);
            }
        }

        // Idempotency check — deliberately AFTER capability + policy
        // (linus review C3): with the durable journal, a cached result
        // consulted first would let a tool that is denied TODAY serve
        // yesterday's cached result after every restart, forever. Deny
        // rules must win over dedup.
        if action.idempotent && !action.invocation_mode.is_detached() {
            let key = idempotency_key(action, scope);
            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,
                );
            }
        }

        // 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, scope).await;

        // Cache idempotent results. Detached invocations are exempt
        // (linus review D4): their Succeeded output is a live tool
        // HANDLE, exactly as uncacheable here as in the result cache —
        // deduping would return a stale handle instead of starting the
        // tool, and journaling would replay a handle id that doesn't
        // exist in a fresh registry after restart.
        if action.idempotent
            && !action.invocation_mode.is_detached()
            && result.status == ActionStatus::Succeeded
        {
            let key = idempotency_key(action, scope);
            {
                let mut cache = self.idempotency_cache.lock().await;
                cache.insert(key.clone(), result.clone());
            }
            // Persist to the durable journal (C3) so the result survives a
            // restart and the action isn't re-executed.
            self.journal_idempotency(&key, Some(&result)).await;
        }

        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,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> (ActionResult, u32) {
        // The harness config (Evolution Agent, §3.5), when installed, caps
        // the per-action retry budget and sets the inter-attempt backoff —
        // this is where an applied retry-config mutation actually takes
        // effect. `None` (the default) preserves prior behavior exactly.
        let hc = self.harness_config.read().await.clone();
        let retry_cap = hc.as_ref().map(|c| c.max_retries).unwrap_or(u32::MAX);
        let backoff_override = hc.as_ref().map(|c| c.retry_backoff_ms).filter(|&v| v > 0);
        let max_attempts = if action.failure_behavior == FailureBehavior::Retry {
            action.max_retries.min(retry_cap) + 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;
                // Base delay from the harness config when installed (an
                // applied retry-config mutation), else the built-in default.
                let base = backoff_override.unwrap_or(RETRY_BASE_DELAY_MS);
                let delay = base * 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, scope),
                )
                .await
                {
                    Ok(r) => r,
                    Err(_) => Err(format!("action timed out after {}ms", timeout_ms)),
                }
            } else {
                self.dispatch(action, scope).await
            };

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

            match exec_result {
                Ok(output) => {
                    // Commit post-effects. Tenant-scoped when the
                    // proposal carries a scope (Parslee-ai/car#187
                    // phase 3) — distinct tenants write to disjoint
                    // namespaces so concurrent multi-tenant proposals
                    // don't trample each other's keys.
                    let tenant_for_effects = scope.and_then(|s| s.tenant_id.as_deref());
                    let scoped_state = self.state.scoped(tenant_for_effects);
                    let mut state_changes: HashMap<String, Value> = HashMap::new();
                    for (key, value) in &action.expected_effects {
                        scoped_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;
                    // Record the tool name + result count so the tool-receipt
                    // verifier (A6) can project ground-truth receipts from the
                    // log and catch tool-use hallucinations.
                    let mut succ_data: HashMap<String, Value> = HashMap::new();
                    if let Some(tool) = &action.tool {
                        succ_data.insert("tool".to_string(), Value::from(tool.as_str()));
                        succ_data.insert("ok".to_string(), Value::from(true));
                        if let Value::Array(arr) = &output {
                            succ_data
                                .insert("result_count".to_string(), Value::from(arr.len() as u64));
                        }
                    }
                    // Record duration through the standardized metric path
                    // (§3.5.1) so it feeds `EventLog::metrics_totals` via the
                    // same contract as inference token metrics — one path,
                    // not a coincidentally-matching `"duration_ms"` string.
                    log.append_metered(
                        EventKind::ActionSucceeded,
                        Some(&action.id),
                        Some(proposal_id),
                        succ_data,
                        car_eventlog::Metrics::latency(duration_ms),
                    );

                    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;
                    let mut fail_data: HashMap<String, Value> = [
                        ("error".to_string(), Value::from(e.as_str())),
                        ("attempt".to_string(), Value::from(attempt + 1)),
                    ]
                    .into();
                    // Tool name + ok=false so the receipt verifier (A6) records
                    // that the tool *executed* (a failed call still ran).
                    if let Some(tool) = &action.tool {
                        fail_data.insert("tool".to_string(), Value::from(tool.as_str()));
                        fail_data.insert("ok".to_string(), Value::from(false));
                    }
                    log.append(
                        EventKind::ActionFailed,
                        Some(&action.id),
                        Some(proposal_id),
                        fail_data,
                    );
                }
            }
        }

        // 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.
    ///
    /// `scope` is the per-execution caller / tenant surface
    /// (Parslee-ai/car#187 phase 3). When the scope carries a
    /// tenant id, state R/W operations (`StateWrite`, `StateRead`,
    /// `Assertion`) route through `StateStore::scoped(tenant_id)`
    /// so distinct tenants can't see each other's keys. Unscoped
    /// proposals get the legacy flat-namespace behaviour
    /// automatically.
    async fn dispatch(
        &self,
        action: &Action,
        scope: Option<&crate::scope::RuntimeScope>,
    ) -> 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(),
                );

                // Detached invocation modes (C2): start the tool via the
                // configured executor's streaming entry point, register a
                // handle, and return immediately — the DAG must not block
                // on a streaming/long-running tool. Deliberately BEFORE
                // the result cache (a handle is a live invocation, never a
                // cacheable value) but still behind the rate limiter.
                // Built-ins don't stream; a detached call requires a
                // configured executor that implements execute_stream.
                if action.invocation_mode.is_detached() {
                    self.rate_limiter.acquire(tool_name).await;
                    let configured = {
                        let guard = self.tool_executor.lock().await;
                        guard.as_ref().cloned()
                    };
                    let executor = configured.ok_or_else(|| {
                        format!(
                            "tool '{tool_name}': detached invocation requires a tool executor"
                        )
                    })?;
                    let rx = executor
                        .execute_stream(tool_name, &params, &action.id)
                        .await?;
                    let (handle, cancel) =
                        self.tool_handles.register(tool_name, &action.id).await;
                    crate::tool_handles::spawn_drain(
                        self.tool_handles.clone(),
                        handle.id.clone(),
                        rx,
                        cancel,
                    );
                    return Ok(serde_json::json!({
                        "tool_handle": handle.id,
                        "status": "running",
                    }));
                }

                // 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(),
                            );
                            // Per-candidate gate telemetry (SkillOpt-inspired):
                            // one event per promotion/rejection so the skill
                            // bank's evolution is auditable.
                            for key in &report.candidates_promoted {
                                log.append(
                                    EventKind::CandidatePromoted,
                                    None,
                                    None,
                                    [("candidate".to_string(), Value::from(key.as_str()))].into(),
                                );
                            }
                            for key in &report.candidates_rejected {
                                log.append(
                                    EventKind::CandidateRejected,
                                    None,
                                    None,
                                    [("candidate".to_string(), Value::from(key.as_str()))].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_with_action(tool_name, &params, &action.id, action.timeout_ms)
                        .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;
                    }
                }

                // Built-in commodity tools resolve against the bound substrate
                // (default: LocalSubstrate → historic host behavior). `calculate`
                // stays pure inside agent_basics and ignores the substrate.
                let substrate = self.substrate.read().await.clone();
                if let Some(result) =
                    crate::agent_basics::execute(&substrate, 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);
                let tenant = scope.and_then(|s| s.tenant_id.as_deref());
                self.state.scoped(tenant).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")?;
                let tenant = scope.and_then(|s| s.tenant_id.as_deref());
                Ok(self.state.scoped(tenant).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 tenant = scope.and_then(|s| s.tenant_id.as_deref());
                let actual = self.state.scoped(tenant).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()
    }
}

#[cfg(test)]
mod timeout_integration_tests {
    //! Integration coverage for the #259/#262 timeout coordination the unit
    //! tests in `car-server-core` / `car-ffi-common` only exercise as pure
    //! selection helpers (#266 item 4): that the executor's per-action deadline
    //! reaps a slow dispatch first, and that `action.timeout_ms` flows
    //! end-to-end onto `execute_with_action`.
    use super::*;
    use car_ir::ActionProposal;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// A tool executor that (a) records the `timeout_ms` it was handed and
    /// (b) sleeps `delay_ms` before returning, so a test can prove the
    /// executor's own `timeout(action.timeout_ms, dispatch)` reaps a call that
    /// outlives its budget.
    struct RecordingExecutor {
        seen_timeout_ms: Arc<AtomicU64>,
        delay_ms: u64,
    }

    #[async_trait::async_trait]
    impl ToolExecutor for RecordingExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            Ok(Value::Null)
        }
        async fn execute_with_action(
            &self,
            _tool: &str,
            _params: &Value,
            _action_id: &str,
            timeout_ms: Option<u64>,
        ) -> Result<Value, String> {
            // `u64::MAX` sentinel = "None was passed".
            self.seen_timeout_ms
                .store(timeout_ms.unwrap_or(u64::MAX), Ordering::SeqCst);
            tokio::time::sleep(Duration::from_millis(self.delay_ms)).await;
            Ok(serde_json::json!({ "ok": true }))
        }
    }

    fn one_tool_proposal(timeout_ms: Option<u64>) -> ActionProposal {
        let mut action = serde_json::json!({
            "id": "a0",
            "type": "tool_call",
            "tool": "slow",
            "parameters": {},
            "dependencies": [],
        });
        if let Some(ms) = timeout_ms {
            action["timeout_ms"] = serde_json::json!(ms);
        }
        serde_json::from_value(serde_json::json!({
            "source": "test",
            "actions": [action],
        }))
        .expect("proposal deserializes")
    }

    #[tokio::test]
    async fn action_timeout_ms_reaches_executor_and_reaps_slow_dispatch() {
        let seen = Arc::new(AtomicU64::new(0));
        let rt = Runtime::new();
        rt.register_tool("slow").await;
        rt.set_executor(Arc::new(RecordingExecutor {
            seen_timeout_ms: seen.clone(),
            delay_ms: 2_000, // outlives the 100ms budget below
        }))
        .await;

        // 100ms budget against a 2s dispatch: the executor's own
        // `timeout(action.timeout_ms, dispatch)` must reap it (#262 — the
        // executor is the authority), and the budget must have reached
        // `execute_with_action` (the #262 harness→action propagation, proven
        // end-to-end here rather than only via the pure helper).
        let result = rt.execute(&one_tool_proposal(Some(100))).await;
        assert_eq!(
            seen.load(Ordering::SeqCst),
            100,
            "budget must reach the executor"
        );
        let action_result = &result.results[0];
        assert!(
            action_result
                .error
                .as_deref()
                .unwrap_or("")
                .contains("timed out"),
            "slow dispatch must be reaped by the action deadline: {:?}",
            action_result.error
        );
    }

    #[tokio::test]
    async fn no_budget_applies_no_executor_deadline() {
        // The `None` path: the executor applies NO deadline, so a dispatch that
        // outlives any per-attempt budget still completes (the WS callback wait
        // is the sole bound on that path, exercised in car-server-core). Here a
        // 300ms dispatch with no budget must succeed, not be reaped.
        let seen = Arc::new(AtomicU64::new(0));
        let rt = Runtime::new();
        rt.register_tool("slow").await;
        rt.set_executor(Arc::new(RecordingExecutor {
            seen_timeout_ms: seen.clone(),
            delay_ms: 300,
        }))
        .await;

        let result = rt.execute(&one_tool_proposal(None)).await;
        assert_eq!(
            seen.load(Ordering::SeqCst),
            u64::MAX,
            "None budget must be forwarded as None"
        );
        assert_eq!(result.results[0].status, ActionStatus::Succeeded);
    }
}