car_engine/

executor.rs

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

use crate::cache::ResultCache;
use crate::capabilities::CapabilitySet;
use crate::checkpoint::Checkpoint;
use crate::rate_limit::{RateLimit, RateLimiter};
use car_eventlog::{EventKind, EventLog, SpanStatus};
use car_ir::{
    build_dag, AcceptedProposalPreimage, Action, ActionProposal, ActionResult, ActionStatus,
    ActionType, CostSummary, FailureBehavior, ProposalLineageEntry, ProposalLineageStatus,
    ProposalResult, StateMutation, ToolFailure, ToolFailureClassification, ToolSchema,
};
use car_policy::{ApprovalDecision, ApprovalLedger, ApprovalRecord, PermissionTier, PolicyEngine};
use car_state::{RestoreDurability, StateStore};
use car_validator::validate_action;
use serde_json::Value;
use sha2::{Digest, Sha256};
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;
const ROLLBACK_WARNING: &str =
    "proposal aborted; state effects were rolled back; external effects may remain and were not undone";
const PLAN_FALLBACK_ROLLBACK_WARNING: &str =
    "planning candidate rejected; state effects were rolled back; external effects may remain and were not undone";
const ROLLBACK_DURABILITY_ERROR: &str =
    "durable state rollback failed before publication; in-memory state and idempotency entries were preserved";
const ROLLBACK_DURABILITY_UNKNOWN: &str =
    "state rollback was published but parent-directory durability is unknown; idempotency entries were invalidated for exact retry";

// ---------------------------------------------------------------------------
// 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`](crate::executor::ToolExecutor::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
    }

    /// Variant that also carries the Runtime execution session and the retry
    /// attempt. Executors that retain safety-relevant state across calls (such
    /// as a read-before-edit ledger) override this to keep that state isolated
    /// by session. Existing executors retain their current behavior through the
    /// default delegation.
    ///
    /// `attempt` is **1-based**: `1` on the first try, `2` on the first retry.
    /// WS executors surface it on the `tools.execute` payload so a host can
    /// tell which retry it is serving — `action_id` cannot, being neither
    /// unique across attempts nor varying between them.
    ///
    /// It is threaded from `execute_with_retry`'s own counter rather than
    /// synthesized here. It was previously hardcoded to `1` at the one place
    /// that put it on the wire, which made the field a constant and any join
    /// built on it silently degenerate (Parslee-ai/car#928).
    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        _session_id: Option<&str>,
        _attempt: u32,
    ) -> Result<Value, String> {
        self.execute_with_action(tool, params, action_id, timeout_ms)
            .await
    }

    /// Execute one action while allowing a transport to return runtime-observed
    /// state mutations alongside its ordinary tool output.
    ///
    /// The default preserves every direct/internal executor: its existing
    /// output is wrapped with no mutations, and declared `expected_effects`
    /// remain assertions rather than writes. A transport that overrides this
    /// method must validate its mutation envelope against `expected_effects`
    /// before returning. The Runtime repeats the key-set and I-JSON checks and
    /// applies accepted changes through its own [`StateStore`].
    async fn execute_with_action_state_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        _expected_effects: &HashMap<String, Value>,
        _return_schema: Option<&Value>,
    ) -> Result<ToolExecution, String> {
        self.execute_with_action_in_session(
            tool, params, action_id, timeout_ms, session_id, attempt,
        )
        .await
        .map(ToolExecution::output_only)
    }

    /// Execute one action with typed failure evidence.
    ///
    /// This is a new default-delegating layer rather than a signature change
    /// to the existing execution methods. Existing executors therefore retain
    /// their exact behavior: every legacy string error becomes an ordinary,
    /// non-terminal failure. Executors opt into fail-stop behavior only by
    /// overriding this method and returning [`ToolFailure::terminal`].
    async fn execute_classified(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &HashMap<String, Value>,
        return_schema: Option<&Value>,
    ) -> Result<ToolExecution, ToolFailure> {
        self.execute_with_action_state_in_session(
            tool,
            params,
            action_id,
            timeout_ms,
            session_id,
            attempt,
            expected_effects,
            return_schema,
        )
        .await
        .map_err(ToolFailure::from)
    }

    /// 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"
        ))
    }
}

/// Result of a tool dispatch before the Runtime commits observed state.
#[derive(Debug, Clone, PartialEq)]
pub struct ToolExecution {
    pub output: Value,
    pub state_changes: HashMap<String, Value>,
}

impl ToolExecution {
    pub fn output_only(output: Value) -> Self {
        Self {
            output,
            state_changes: HashMap::new(),
        }
    }
}

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

/// Exact normal-serde proposal preimage plus its lowercase RFC 8785/JCS
/// SHA-256 identity. The shared inference helper is the repository's settled
/// canonicalizer; journal identity must not drift onto a second JSON sorter.
fn proposal_journal_identity(proposal: &ActionProposal) -> Result<(Value, String), String> {
    let preimage = serde_json::to_value(proposal)
        .map_err(|error| format!("proposal serialization failed: {error}"))?;
    let canonical = car_inference::catalog_identity::canonical_json(&preimage)?;
    let digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
    Ok((preimage, digest))
}

fn proposal_digest(proposal: &ActionProposal) -> Result<String, String> {
    proposal_journal_identity(proposal).map(|(_, digest)| digest)
}

fn proposal_rejection_boundary_data(
    proposal: &ActionProposal,
    proposal_digest: Option<&str>,
    reason: &str,
) -> HashMap<String, Value> {
    let mut data: HashMap<String, Value> = [
        ("affected_actions".to_string(), Value::Array(Vec::new())),
        (
            "rolled_back_changes".to_string(),
            Value::Object(serde_json::Map::new()),
        ),
        (
            "changes_semantics".to_string(),
            Value::from("proposal_rejected_no_state_commit"),
        ),
        ("attempted".to_string(), Value::from(false)),
        ("stage".to_string(), Value::from("proposal_rejection")),
        ("rejection_reason".to_string(), Value::from(reason)),
    ]
    .into();
    if let Ok(preimage) = serde_json::to_value(proposal) {
        data.insert("proposal".to_string(), preimage);
    }
    if let Some(digest) = proposal_digest {
        data.insert("proposal_digest".to_string(), Value::from(digest));
    }
    data
}

fn proposal_lineage_entry(
    proposal: &ActionProposal,
    generation: u32,
    status: ProposalLineageStatus,
    rejection_reason: Option<String>,
) -> ProposalLineageEntry {
    ProposalLineageEntry {
        generation,
        proposal_id: proposal.id.clone(),
        proposal_digest: proposal_digest(proposal).ok(),
        status,
        rejection_reason,
    }
}

fn finalize_proposal_result(
    mut result: ProposalResult,
    original_proposal_id: &str,
    lineage: &[ProposalLineageEntry],
    accepted_proposal_preimages: &[AcceptedProposalPreimage],
) -> ProposalResult {
    result.original_proposal_id = original_proposal_id.to_string();
    result.replan_lineage = lineage.to_vec();
    result.accepted_proposal_preimages = accepted_proposal_preimages.to_vec();
    result
}

/// Validate the proposal-local action identity boundary used by DAG execution,
/// action receipts, results, and state-transition attribution. A retry reuses
/// one admitted action id across attempts; two declared actions may not share
/// an id because every downstream join would become ambiguous.
pub fn validate_proposal_action_ids(proposal: &ActionProposal) -> Result<(), String> {
    let mut seen = std::collections::HashSet::with_capacity(proposal.actions.len());
    for action in &proposal.actions {
        if !seen.insert(action.id.as_str()) {
            return Err(format!("duplicate action id '{}'", action.id));
        }
    }
    Ok(())
}

fn validate_proposal_retry_limits(proposal: &ActionProposal) -> Result<(), String> {
    for action in &proposal.actions {
        if action.failure_behavior == FailureBehavior::Retry
            && action.max_retries.checked_add(1).is_none()
        {
            return Err(format!(
                "action '{}' retry attempt count overflows u32",
                action.id
            ));
        }
    }
    Ok(())
}

fn requires_integrity_rollback(result: &ActionResult) -> bool {
    result.status == ActionStatus::Failed
        && result.error.as_deref().is_some_and(|error| {
            error.starts_with("tool output failed JCS/I-JSON validation after dispatch")
        })
}

fn record_rollback_durability_error(
    results: &mut [ActionResult],
    prefix: &str,
    error: &str,
) -> String {
    let detail = format!("{prefix}: {error}");
    if let Some(result) = results
        .iter_mut()
        .find(|result| result.status != ActionStatus::Succeeded)
    {
        result.error = Some(match result.error.take() {
            Some(existing) => format!("{existing}; {detail}"),
            None => detail.clone(),
        });
    }
    detail
}

fn rejected_result(action_id: &str, error: String) -> ActionResult {
    ActionResult {
        action_id: action_id.to_string(),
        status: ActionStatus::Rejected,
        output: None,
        error: Some(error),
        terminal: false,
        state_changes: HashMap::new(),
        rolled_back: false,
        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()),
        terminal: false,
        state_changes: HashMap::new(),
        rolled_back: false,
        duration_ms: None,
        timestamp: chrono::Utc::now(),
    }
}

fn insert_tool_event_provenance(
    data: &mut HashMap<String, Value>,
    action: &Action,
    source: Option<car_ir::ToolSourceKind>,
) {
    if let Some(tool) = action.tool.as_deref() {
        data.insert("tool".to_string(), Value::from(tool));
        data.insert(
            "tool_source".to_string(),
            Value::from(
                source
                    .unwrap_or(car_ir::ToolSourceKind::UserDefined)
                    .as_str(),
            ),
        );
    }
}

/// SHA-256 over the RFC 8785/JCS rendering of only the action parameters.
///
/// The accepted proposal has already crossed `proposal_journal_identity`, so
/// canonicalization cannot fail here without violating that admission
/// invariant. Keeping only the digest on outcome events lets a later join find
/// the raw parameters in `ProposalReceived` without copying them into every
/// failure record.
fn action_params_digest(action: &Action) -> String {
    // Serialize the map directly: RFC 8785/JCS sorts object keys
    // recursively at render time (see `canonical_json` and the
    // `canonical_json_sorts_object_keys_recursively` test), so the map's
    // iteration order does not affect the digest and cloning into a
    // `Value::Object` would be pure per-action allocation overhead.
    let canonical = car_inference::catalog_identity::canonical_json(&action.parameters)
        .expect("accepted action parameters must remain RFC 8785/JCS canonicalizable");
    format!("{:x}", Sha256::digest(canonical.as_bytes()))
}

fn insert_action_outcome_signal(
    data: &mut HashMap<String, Value>,
    action: &Action,
    params_digest: &str,
    error_class: Option<&str>,
) {
    data.insert(
        "params_digest".to_string(),
        Value::from(params_digest.to_string()),
    );
    data.insert(
        "expected_effects".to_string(),
        serde_json::to_value(&action.expected_effects)
            .expect("action expected_effects must serialize as JSON"),
    );
    if let Some(error_class) = error_class {
        data.insert("error_class".to_string(), Value::from(error_class));
    }
}

/// Normalize execution failures into a stable, deliberately low-cardinality
/// event-log vocabulary. Validation and policy checks that happen before
/// dispatch remain `ActionRejected`; this maps failures reached after an
/// action began executing.
pub(crate) fn action_error_class(
    action: &Action,
    error: &str,
    engine_timeout: bool,
) -> &'static str {
    // Classification is by substring over prose and is a published wire
    // contract (docs/agent-ir-spec.md `error_class`). Each literal is pinned
    // to its producer below; changing a producer's wording must update the
    // matching literal (and `action_failure_error_class_mapping_is_stable`).
    let error = error.to_ascii_lowercase();
    // "callback timed out" — the engine deadline itself is typed via
    // `engine_timeout`, so this literal only covers the detached-dispatch
    // producer `tool '{tool}' callback timed out ({s}s)`
    // (car-server-core/src/session.rs, `reap_detached_calls`/wait path).
    // A tool faithfully relaying a *downstream* timeout in its own message
    // also lands here; accepted drift, low-cardinality vocabulary.
    if engine_timeout || error.contains("callback timed out") {
        "timeout"
    } else if error.starts_with("denied by policy:") || error.starts_with("rejected by policy:") {
        // Exact producers return this prefix unwrapped from the tool's
        // guard check: car-server-core/src/assistant/executor.rs and
        // car-server-core/src/coder/shell_tool.rs
        // (`format!("denied by policy: {reason}")`). An executor that
        // *wraps* the denial (e.g. `tool 'x' failed: denied by policy: …`)
        // deliberately falls through to `tool_error` — see the wrapped
        // case in `action_failure_error_class_mapping_is_stable`.
        "rejected_by_policy"
    } else if error.contains("jcs/i-json validation")
        || error.contains("output validation:")
        || error.contains("invalid return json schema")
        || error.contains("must return exact envelope {output,state_changes}")
        || error.contains("state_changes must be an object")
        || error.contains("state_changes keys do not match")
        || error.contains("state_changes serialization failed")
    {
        "validation"
    } else if action.tool.is_some() {
        "tool_error"
    } else {
        "unknown"
    }
}

fn action_outcome_data(
    stage: &str,
    reason: &str,
    attempted: bool,
    attempt: Option<u32>,
) -> HashMap<String, Value> {
    let mut data: HashMap<String, Value> = [
        ("stage".to_string(), Value::from(stage)),
        ("reason".to_string(), Value::from(reason)),
        ("attempted".to_string(), Value::from(attempted)),
    ]
    .into();
    if let Some(attempt) = attempt {
        data.insert("attempt".to_string(), Value::from(attempt));
    }
    data
}

/// 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)),
        terminal: false,
        state_changes: HashMap::new(),
        rolled_back: false,
        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>,
    /// Per-execution-session read ledgers backing the read-before-edit / staleness guard on
    /// the built-in file tools (H1/F4-remainder, audit 2026-07-06). Threaded into
    /// `agent_basics::execute_with_ledger` on the built-in fallback path so a raw
    /// Runtime session (no configured executor) still requires reading a file
    /// before editing or overwriting it. Configured executors
    /// (coder/assistant/bench) own their own ledger, so this one only governs the
    /// built-in fallback path.
    read_ledgers: crate::agent_basics::SessionReadLedgers,
    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 outbound transport backing the `messaging.send` built-in.
    /// `None` by default — a runtime with no sink refuses `messaging.send`
    /// outright rather than falling through to the host executor, and never
    /// advertises the tool. Attach one with
    /// [`Runtime::with_message_sink`]; see [`crate::messaging`] for why
    /// human-directed messaging belongs inside the governed chain at all.
    message_sink: Option<Arc<dyn crate::messaging::MessageSink>>,
    /// 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>>>,
    /// Runtime taint provenance for the VIGIL intent gate
    /// ([`crate::taint::TaintLedger`]). Installed by
    /// [`Runtime::install_intent_gate`] and by nothing else: with no intent
    /// gate there is no ledger, so a runtime that never configures VIGIL
    /// pays no per-action cost and behaves exactly as before. When present,
    /// each successful action records which state keys its result wrote and
    /// whether that result was tainted, and the gate reads it back at
    /// admission to mark actions that READ a tainted key — the cross-
    /// proposal (replan) and trusted-tool-laundering cases the proposal-
    /// local dependency DAG cannot see.
    taint_ledger: TokioRwLock<Option<Arc<crate::taint::TaintLedger>>>,
    /// 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()),
            read_ledgers: crate::agent_basics::SessionReadLedgers::new(),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            message_sink: 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()),
            taint_ledger: TokioRwLock::new(None),
            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()),
            read_ledgers: crate::agent_basics::SessionReadLedgers::new(),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            message_sink: 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()),
            taint_ledger: TokioRwLock::new(None),
            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())));
        self.read_ledgers.ledger_for(Some(&id));
        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;
        let removed = sessions.remove(session_id).is_some();
        if removed {
            self.read_ledgers.remove(session_id);
        }
        removed
    }

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

    /// [`Self::register_policy_in_session`] for a check that forbids `tool`
    /// outright, so `PolicyEngine::blanket_denied_tools` can read it back.
    /// Separate method rather than an extra parameter: the existing signature
    /// is public and crosses the bindings, and this is purely additive.
    pub async fn register_tool_deny_in_session(
        &self,
        session_id: &str,
        name: &str,
        tool: &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_tool_deny(name, tool, check, description);
        Ok(())
    }

    /// Remove a policy by name. `session_id` targets a session's policy set;
    /// `None` targets the global one.
    ///
    /// Returns how many policies were dropped, or `Err` if `session_id` names a
    /// session that doesn't exist. Session-scoped policies could always be
    /// dropped wholesale by [`Self::close_session`], but a *global* policy had
    /// no removal path at all — once registered it lived until the process
    /// exited, so a mistyped or over-broad rule could only be cleared by
    /// restarting the daemon (Parslee-ai/car#623).
    pub async fn unregister_policy(
        &self,
        name: &str,
        session_id: Option<&str>,
    ) -> Result<usize, String> {
        match session_id {
            Some(sid) => {
                let engine = {
                    let sessions = self.session_policies.read().await;
                    sessions
                        .get(sid)
                        .cloned()
                        .ok_or_else(|| format!("unknown session id '{sid}'"))?
                };
                let mut engine = engine.write().await;
                Ok(engine.unregister(name))
            }
            None => {
                let mut engine = self.policies.write().await;
                Ok(engine.unregister(name))
            }
        }
    }

    /// Registered policies as `(name, description)`. `session_id` lists a
    /// session's set; `None` lists the global one. Without this a client could
    /// register a policy but never ask what was in force, so a rejection could
    /// not be explained beyond its single message (Parslee-ai/car#623).
    pub async fn list_policies(
        &self,
        session_id: Option<&str>,
    ) -> Result<Vec<(String, String)>, String> {
        match session_id {
            Some(sid) => {
                let engine = {
                    let sessions = self.session_policies.read().await;
                    sessions
                        .get(sid)
                        .cloned()
                        .ok_or_else(|| format!("unknown session id '{sid}'"))?
                };
                let engine = engine.read().await;
                Ok(engine.policy_details())
            }
            None => {
                let engine = self.policies.read().await;
                Ok(engine.policy_details())
            }
        }
    }

    /// 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 a `.car` directory; this looks for
    /// `car_dir/policies/*.toml`. **The caller chooses the directory and
    /// there is no walk-up here** — this joins the path it is given, once.
    /// The two production callers pick differently: the daemon passes
    /// `$HOME/.car`, and the assistant passes its working directory's
    /// `.car` (`--dir`, else cwd). So a rule file at a repository root does
    /// not govern a `car do` run started from a subdirectory. Say which
    /// directory you mean at the call site rather than assuming discovery.
    ///
    /// 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 additive on top of any code-registered policies, and
    /// every one of them is a prohibition — though `allow_tool_param` states
    /// its prohibition as an allowlist, denying everything about its tool
    /// that it does not name. 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)?;
        // `PolicyRules::len` counts every kind. Summing a hand-picked subset
        // here is what made this under-report once already — it covered the
        // three kinds that existed at the time and was not revisited when
        // more landed, so a file full of allowlists reported zero rules
        // loaded while `apply` registered all of them.
        let count = rules.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).
        // Keep the canonical registry populated too: execution-event provenance
        // is read from ToolEntry.source rather than inferred from a tool name.
        //
        // Contention contract: this runs at construction time on a private
        // runtime, so both try_locks are expected to succeed. If either fails
        // the two stores would disagree (validator reads `tools`, provenance
        // reads the registry first) — loud in debug, logged in release.
        if let Ok(mut tools) = self.tools.try_write() {
            for schema in car_inference::service::all_schemas() {
                let name = schema.name.clone();
                let entry = crate::registry::ToolEntry::builtin(schema);
                tools.insert(name.clone(), entry.schema.clone());
                if !self.registry.try_register(entry) {
                    tracing::warn!(
                        tool = %name,
                        "with_inference: canonical registry contention; \
                         schema map registered the inference tool but the \
                         registry did not — provenance falls back to the map"
                    );
                }
            }
        } else {
            debug_assert!(
                false,
                "with_inference must populate both stores; schema-map lock was contended \
                 at construction time — inference tools stay unregistered"
            );
            tracing::warn!(
                "with_inference: schema map lock contended at construction; \
                 inference tools were NOT registered"
            );
        }
        self
    }

    /// Attach an outbound message sink, making `messaging.send` a real tool on
    /// this runtime.
    ///
    /// Registering the schema here — and ONLY here — is deliberate. A tool the
    /// runtime cannot execute must not be advertised to the model: without a
    /// sink the dispatch arm can only return an error, and a model that keeps
    /// seeing "message the human" in its tool list will keep trying to use it.
    /// So sink and schema arrive together, and neither exists alone.
    ///
    /// The entry is `AskUser` with side effects: reaching a human is
    /// irreversible, so the conservative default is the right one, and hosts
    /// that have their own consent model can re-register with a different
    /// permission. `category = "messaging"` groups it for the capability
    /// surfaces that filter by category.
    ///
    /// Follows [`Self::with_inference`]'s non-async registration pattern
    /// (`try_write` during construction, while the locks are uncontended) —
    /// making the builders async would break every `Runtime::new().with_…()`
    /// chain in the workspace for no gain.
    pub fn with_message_sink(mut self, sink: Arc<dyn crate::messaging::MessageSink>) -> Self {
        self.message_sink = Some(sink);

        let entry = crate::registry::ToolEntry::builtin(car_ir::builtins::messaging_send())
            .with_permission(crate::registry::ToolPermission::AskUser)
            .with_side_effects(true)
            .with_category("messaging");
        let schema = entry.schema.clone();

        // Honour the schema's declared rate limit, the way the async
        // `register_tool_schema` path does — otherwise the backstop on
        // outbound human messaging would silently not exist.
        if let Some(ref rl) = schema.rate_limit {
            self.rate_limiter.try_set_limit(
                &schema.name,
                RateLimit {
                    max_calls: rl.max_calls,
                    interval_secs: rl.interval_secs,
                },
            );
        }
        self.registry.try_register(entry);
        // The legacy schema map is what `validate_action` reads, so the
        // validator can only check `messaging.send` parameters once it lands
        // here.
        if let Ok(mut tools) = self.tools.try_write() {
            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
    }

    /// The attached trajectory store, if any.
    ///
    /// Writing traces was always the point of the store; reading them back is
    /// what makes them a feedback signal rather than an audit log.
    pub fn trajectory_store(&self) -> Option<&Arc<car_memgine::TrajectoryStore>> {
        self.trajectory_store.as_ref()
    }

    /// Per-tool success rates observed over the last `window_days`, or `None`
    /// when no trajectory store is attached.
    ///
    /// Dispatch-conditional (see
    /// [`ToolFeedback::dispatched_from_trajectories`](car_planner::ToolFeedback::dispatched_from_trajectories))
    /// — rejected and skipped actions are excluded, because the tool never ran
    /// and a consumer that models rejection separately would otherwise count
    /// it twice.
    ///
    /// The window exists because a success rate is a claim about how a tool
    /// behaves *now*. An API that was broken for a week six months ago and has
    /// been fixed since should not still be dragging its own rate down, and an
    /// unbounded history makes the rate progressively less responsive to
    /// exactly the recent change an operator is trying to see. 30 days is the
    /// suggested default at the call sites.
    ///
    /// Reads from disk on every call — parsing is bounded by the window (day
    /// files outside it are skipped by filename, never opened), and the callers
    /// are interactive rather than hot-path. If that stops being true, cache
    /// here rather than at each call site.
    pub fn tool_feedback(&self, window_days: u32) -> Option<car_planner::ToolFeedback> {
        let store = self.trajectory_store.as_ref()?;
        let cutoff = chrono::Utc::now() - chrono::Duration::days(i64::from(window_days));
        let trajectories = store.load_since(cutoff);
        Some(car_planner::ToolFeedback::dispatched_from_trajectories(
            &trajectories,
        ))
    }

    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;
        }
        self.read_ledgers.clear();
    }

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

    /// Bind an event log the caller already holds a handle to.
    ///
    /// [`Self::with_event_log`] takes ownership, which leaves no way for
    /// anything outside the runtime to read what was recorded. That was fine
    /// while the log was observability-only, but a model-callable events
    /// surface has to run inside a `ToolExecutor` — and the executor is
    /// constructed *before* the runtime that would own it, so the two can only
    /// meet through a handle created first and shared into both
    /// (Parslee-ai/car#815).
    pub fn with_shared_event_log(mut self, log: Arc<TokioMutex<EventLog>>) -> Self {
        self.log = log;
        self
    }

    /// The runtime's event log handle, for callers that need to read the record
    /// the runtime is writing.
    pub fn event_log_handle(&self) -> Arc<TokioMutex<EventLog>> {
        Arc::clone(&self.log)
    }

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

    /// Register `gate`, **replacing** any already-registered gate answering
    /// to the same [`crate::admission::AdmissionGate::name`] instead of
    /// appending a second one. For gates that are *reconfigured* rather than
    /// layered — the VIGIL intent gate is reloaded whenever an operator
    /// re-reads `.car/intent.json` — appending is a bug, not a no-op: gate
    /// aggregation is fail-closed, so the stale gate's verdict still wins
    /// while it holds state (a [`crate::taint::TaintLedger`]) nothing writes
    /// to any more, and two gates with the same name emit two
    /// `AdmissionGateDecision` events under that name.
    ///
    /// The replaced gate keeps its **position**, so a reconfigure never
    /// silently reorders admission relative to the other gates. Any further
    /// gate carrying the same name is a stale duplicate from an earlier
    /// registration and is dropped, so exactly one gate answers to the name.
    async fn replace_admission_gate(&self, gate: Arc<dyn crate::admission::AdmissionGate>) {
        let name = gate.name().to_string();
        let mut gates = self.admission_gates.write().await;
        match gates.iter().position(|g| g.name() == name) {
            Some(i) => {
                gates[i] = gate;
                let mut idx = 0usize;
                gates.retain(|g| {
                    let keep = idx == i || g.name() != name;
                    idx += 1;
                    keep
                });
            }
            None => gates.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()
    }

    /// The `name()` of every registered admission gate, in registration order.
    ///
    /// Prefer this over [`Self::admission_gate_count`] when asserting that a
    /// *particular* gate is installed: a bare count couples the assertion to
    /// every other gate the runtime happens to register, so adding one breaks
    /// unrelated tests that only cared about their own.
    pub async fn admission_gate_names(&self) -> Vec<String> {
        self.admission_gates
            .read()
            .await
            .iter()
            .map(|g| g.name().to_string())
            .collect()
    }

    /// 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).
    ///
    /// This is also where the runtime's [`crate::taint::TaintLedger`] is
    /// installed. From here on, every successful action records whether its
    /// result was tainted and which state keys it wrote, and the gate marks
    /// any incoming action that READS a tainted key as tool-stream-
    /// influenced. That closes the two holes the static `untrusted_tools`
    /// list leaves open: a trusted tool laundering attacker-controlled
    /// content out of state, and a replanned proposal that carries no
    /// dependency edge back to the poisoned action. Installing no intent
    /// gate installs no ledger, so nothing changes for a runtime that
    /// doesn't configure VIGIL.
    ///
    /// **Calling this again REPLACES the installed gate; it does not add a
    /// second one.** Reinstalling is a real operation — an operator editing
    /// `.car/intent.json` and re-running
    /// [`Runtime::install_intent_gate_from_project`] lands here — and each
    /// call builds a *fresh* ledger that becomes the only one the executor
    /// writes to. Appending would leave the previous gate registered while
    /// holding an orphaned ledger, and because gate aggregation is
    /// fail-closed that frozen gate's verdict would still win: every key
    /// tainted before the reload would stay tainted forever, no trusted
    /// overwrite could clear it, and legitimate out-of-intent work would
    /// hard-refuse with no approval path. So the runtime holds exactly one
    /// intent gate, and it is always the one bound to the live ledger. The
    /// replacement keeps the previous gate's position in the admission
    /// order, so a reload never silently reorders the other gates.
    pub async fn install_intent_gate(&self, config: crate::intent_gate::IntentGateConfig) {
        let ledger = Arc::new(crate::taint::TaintLedger::new(
            config.untrusted_tools.iter().cloned().collect(),
        ));
        *self.taint_ledger.write().await = Some(Arc::clone(&ledger));
        self.replace_admission_gate(Arc::new(crate::intent_gate::IntentGate::with_taint(
            config, ledger,
        )))
        .await;
    }

    /// The runtime's taint provenance ledger, when an intent gate is
    /// installed (see [`Runtime::install_intent_gate`]). `None` otherwise —
    /// the ledger is strictly opt-in, alongside VIGIL.
    pub async fn taint_ledger(&self) -> Option<Arc<crate::taint::TaintLedger>> {
        self.taint_ledger.read().await.clone()
    }

    /// 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 [`car_eventlog::tool_receipts::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(),
            source: car_ir::ToolSourceKind::UserDefined,
            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, mut entry: crate::registry::ToolEntry) {
        // The registry's source is authoritative for both event provenance and
        // the public tools.list/schema view, including literal-built entries.
        entry.schema.source = entry.source.kind();
        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, None, cancel)
            .await
    }

    /// Execute an active-run proposal while requiring every accepted replan
    /// to retain the authenticated proposal id already claimed by the server.
    pub async fn execute_with_session_and_stable_replan_id(
        &self,
        proposal: &ActionProposal,
        session_id: &str,
    ) -> ProposalResult {
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_with_optional_session(
            proposal,
            Some(session_id),
            None,
            Some(&proposal.id),
            &token,
        )
        .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, None, cancel)
            .await
    }

    pub async fn execute_with_stable_replan_id(&self, proposal: &ActionProposal) -> ProposalResult {
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_with_optional_session(proposal, None, None, Some(&proposal.id), &token)
            .await
    }

    /// Execute a proposal with an attached [`RuntimeScope`](crate::scope::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), None, cancel)
            .await
    }

    pub async fn execute_scoped_with_stable_replan_id(
        &self,
        proposal: &ActionProposal,
        scope: &crate::scope::RuntimeScope,
    ) -> ProposalResult {
        let token = tokio_util::sync::CancellationToken::new();
        self.execute_with_optional_session(proposal, None, Some(scope), Some(&proposal.id), &token)
            .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>,
        required_replan_proposal_id: Option<&str>,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        // The shared StateStore has one proposal transaction at a time. The
        // guard spans validation, replanning, and rollback so distinct Runtime
        // facades over the same store cannot interleave state snapshots.
        let _proposal_execution = self.state.lock_proposal_execution().await;
        self.execute_with_optional_session_already_guarded(
            proposal,
            session_id,
            scope,
            required_replan_proposal_id,
            cancel,
        )
        .await
    }

    /// Execute while the caller retains this StateStore's proposal-execution
    /// guard. `plan_and_execute` uses this path so one guard covers its initial
    /// snapshot, ranking, every candidate, and the final commit/rollback.
    async fn execute_with_optional_session_already_guarded(
        &self,
        proposal: &ActionProposal,
        session_id: Option<&str>,
        scope: Option<&crate::scope::RuntimeScope>,
        required_replan_proposal_id: Option<&str>,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> ProposalResult {
        // Establish canonical identity before any other rejection so an
        // undigested lineage entry can only mean the proposal was genuinely
        // impossible to represent as JCS/I-JSON. No journal row is emitted.
        let original_proposal_digest = match proposal_digest(proposal) {
            Ok(digest) => digest,
            Err(error) => {
                self.log.lock().await.append(
                    EventKind::StateRollback,
                    None,
                    Some(&proposal.id),
                    proposal_rejection_boundary_data(proposal, None, &error),
                );
                let result = ProposalResult::for_proposal(
                    proposal,
                    proposal
                        .actions
                        .iter()
                        .map(|action| rejected_result(&action.id, error.clone()))
                        .collect(),
                    CostSummary::default(),
                );
                return finalize_proposal_result(
                    result,
                    &proposal.id,
                    &[ProposalLineageEntry {
                        generation: 0,
                        proposal_id: proposal.id.clone(),
                        proposal_digest: None,
                        status: ProposalLineageStatus::Rejected,
                        rejection_reason: Some(error),
                    }],
                    &[],
                );
            }
        };

        // Proposal-local action ids are the join key for DAG scheduling,
        // result rows, receipts, and StateTransition attribution. Reject a
        // duplicate before scope/admission/transaction logging or any state
        // mutation. Retried attempts remain valid reuse of one admitted id.
        if let Err(error) = validate_proposal_action_ids(proposal) {
            self.log.lock().await.append(
                EventKind::StateRollback,
                None,
                Some(&proposal.id),
                proposal_rejection_boundary_data(proposal, Some(&original_proposal_digest), &error),
            );
            let result = ProposalResult::for_proposal(
                proposal,
                proposal
                    .actions
                    .iter()
                    .map(|action| rejected_result(&action.id, error.clone()))
                    .collect(),
                CostSummary::default(),
            );
            return finalize_proposal_result(
                result,
                &proposal.id,
                &[proposal_lineage_entry(
                    proposal,
                    0,
                    ProposalLineageStatus::Rejected,
                    Some(error),
                )],
                &[],
            );
        }
        if let Err(error) = validate_proposal_retry_limits(proposal) {
            self.log.lock().await.append(
                EventKind::StateRollback,
                None,
                Some(&proposal.id),
                proposal_rejection_boundary_data(proposal, Some(&original_proposal_digest), &error),
            );
            let result = ProposalResult::for_proposal(
                proposal,
                proposal
                    .actions
                    .iter()
                    .map(|action| rejected_result(&action.id, error.clone()))
                    .collect(),
                CostSummary::default(),
            );
            return finalize_proposal_result(
                result,
                &proposal.id,
                &[proposal_lineage_entry(
                    proposal,
                    0,
                    ProposalLineageStatus::Rejected,
                    Some(error),
                )],
                &[],
            );
        }
        let config = self.replan_config.read().await.clone();
        let mut current_proposal = proposal.clone();
        let mut attempt: u32 = 0;
        let mut lineage = vec![proposal_lineage_entry(
            proposal,
            0,
            ProposalLineageStatus::Accepted,
            None,
        )];

        // 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 reason = "transactional conflict with current shared state (strict mode); \
                          see TransactionConflict events for details and resolution"
                .to_string();
            let results = current_proposal
                .actions
                .iter()
                .map(|a| rejected_result(&a.id, reason.clone()))
                .collect();
            self.log.lock().await.append(
                EventKind::StateRollback,
                None,
                Some(&current_proposal.id),
                proposal_rejection_boundary_data(
                    &current_proposal,
                    Some(&original_proposal_digest),
                    &reason,
                ),
            );
            lineage[0].status = ProposalLineageStatus::Rejected;
            lineage[0].rejection_reason = Some(reason);
            return finalize_proposal_result(
                ProposalResult::for_proposal(proposal, results, Default::default()),
                &proposal.id,
                &lineage,
                &[],
            );
        }

        // 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();
                self.log.lock().await.append(
                    EventKind::StateRollback,
                    None,
                    Some(&current_proposal.id),
                    proposal_rejection_boundary_data(
                        &current_proposal,
                        Some(&original_proposal_digest),
                        &reason,
                    ),
                );
                lineage[0].status = ProposalLineageStatus::Rejected;
                lineage[0].rejection_reason = Some(reason.clone());
                let result = finalize_proposal_result(
                    ProposalResult::for_proposal(proposal, results, Default::default()),
                    &proposal.id,
                    &lineage,
                    &[],
                );
                // Record the trajectory even though nothing dispatched. This
                // return is *before* the execution loop's persist calls, so
                // without this an admission rejection is invisible to the
                // trajectory store — and the per-tool success rates
                // `verify.monte_carlo` derives from it would silently skew
                // optimistic, counting only calls that got far enough to run.
                // The state map is empty because no action mutated anything.
                if let Some(err) = self.persist_trajectory(
                    proposal,
                    &current_proposal,
                    &result,
                    car_memgine::TrajectoryOutcome::Failed,
                    0,
                    &HashMap::new(),
                ) {
                    tracing::warn!(
                        error = %err,
                        proposal_id = %proposal.id,
                        "failed to persist trajectory for admission-rejected proposal"
                    );
                }
                return finalize_proposal_result(result, &proposal.id, &lineage, &[]);
            }
        }

        let mut accepted_proposal_preimages = vec![AcceptedProposalPreimage {
            generation: 0,
            proposal_digest: original_proposal_digest,
            proposal: proposal.clone(),
        }];

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

            // A terminal tool failure ends this engine execution. It may not
            // enter the retry loop above or the proposal-replan loop here;
            // daemon-owned session halting is layered on top by the caller.
            let terminal_failure = result.results.iter().any(|result| result.terminal);
            // Check if we aborted
            let aborted = result.results.iter().any(|r| replan_triggers(&r.status));
            let rollback_durability_failed = result.results.iter().any(|result| {
                result.error.as_deref().is_some_and(|error| {
                    error.contains(ROLLBACK_DURABILITY_ERROR)
                        || error.contains(ROLLBACK_DURABILITY_UNKNOWN)
                })
            });
            if !aborted
                || terminal_failure
                || attempt >= config.max_replans
                || rollback_durability_failed
            {
                if aborted && !terminal_failure && attempt > 0 && !rollback_durability_failed {
                    // 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 && !terminal_failure {
                    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 finalize_proposal_result(
                    result,
                    &proposal.id,
                    &lineage,
                    &accepted_proposal_preimages,
                );
            }

            // 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 finalize_proposal_result(
                    result,
                    &proposal.id,
                    &lineage,
                    &accepted_proposal_preimages,
                );
            };

            // Build the stable failure context once. Candidate rejection is a
            // planning-only loop below: it consumes a replan generation but
            // never returns to `execute_inner_with_cancel`, so a failed
            // proposal with external effects cannot be dispatched twice.
            let failed_actions: Vec<FailedActionSummary> = result
                .results
                .iter()
                .filter(|r| replan_triggers(&r.status) && !r.rolled_back)
                .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.rolled_back)
                .map(|r| r.action_id.clone())
                .collect();

            'replan_candidates: loop {
                if attempt >= config.max_replans {
                    let mut log = self.log.lock().await;
                    log.append(
                        EventKind::ReplanExhausted,
                        None,
                        Some(&proposal.id),
                        [("attempts".to_string(), Value::from(attempt))].into(),
                    );
                    drop(log);
                    if let Some(err) = self.persist_trajectory(
                        proposal,
                        &current_proposal,
                        &result,
                        car_memgine::TrajectoryOutcome::ReplanExhausted,
                        attempt,
                        &state_before_map,
                    ) {
                        self.log.lock().await.append(
                            EventKind::ActionFailed,
                            None,
                            Some(&proposal.id),
                            [("trajectory_persist_error".to_string(), Value::from(err))].into(),
                        );
                    }
                    return finalize_proposal_result(
                        result,
                        &proposal.id,
                        &lineage,
                        &accepted_proposal_preimages,
                    );
                }

                let generation = attempt + 1;
                let ctx = ReplanContext {
                    proposal_id: proposal.id.clone(),
                    attempt: generation,
                    failed_actions: failed_actions.clone(),
                    completed_action_ids: completed_action_ids.clone(),
                    state_snapshot: self.state.snapshot(),
                    replans_remaining: config.max_replans.saturating_sub(generation),
                    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(generation)),
                            (
                                "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(generation)),
                        ]
                        .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) => {
                        attempt = generation;
                        let (new_proposal_preimage, new_proposal_digest) =
                            match proposal_journal_identity(&new_proposal) {
                                Ok(identity) => identity,
                                Err(error) => {
                                    lineage.push(ProposalLineageEntry {
                                        generation,
                                        proposal_id: new_proposal.id.clone(),
                                        proposal_digest: None,
                                        status: ProposalLineageStatus::Rejected,
                                        rejection_reason: Some(error.clone()),
                                    });
                                    self.log.lock().await.append(
                                        EventKind::ReplanRejected,
                                        None,
                                        Some(&proposal.id),
                                        [
                                            ("errors".to_string(), Value::from(error)),
                                            ("attempt".to_string(), Value::from(generation)),
                                        ]
                                        .into(),
                                    );
                                    continue 'replan_candidates;
                                }
                            };
                        if required_replan_proposal_id
                            .is_some_and(|required| new_proposal.id != required)
                        {
                            let reason = format!(
                                "replan proposal id `{}` does not retain authenticated proposal id `{}`",
                                new_proposal.id,
                                required_replan_proposal_id.expect("checked above")
                            );
                            lineage.push(ProposalLineageEntry {
                                generation,
                                proposal_id: new_proposal.id.clone(),
                                proposal_digest: Some(new_proposal_digest),
                                status: ProposalLineageStatus::Rejected,
                                rejection_reason: Some(reason.clone()),
                            });
                            self.log.lock().await.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    ("errors".to_string(), Value::from(reason)),
                                    ("attempt".to_string(), Value::from(generation)),
                                ]
                                .into(),
                            );
                            continue 'replan_candidates;
                        }
                        if let Err(error) = validate_proposal_action_ids(&new_proposal) {
                            lineage.push(ProposalLineageEntry {
                                generation,
                                proposal_id: new_proposal.id.clone(),
                                proposal_digest: Some(new_proposal_digest),
                                status: ProposalLineageStatus::Rejected,
                                rejection_reason: Some(error.clone()),
                            });
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    (
                                        "errors".to_string(),
                                        Value::from(format!("invalid proposal: {error}")),
                                    ),
                                    ("attempt".to_string(), Value::from(generation)),
                                ]
                                .into(),
                            );
                            continue 'replan_candidates;
                        }

                        if let Err(error) = validate_proposal_retry_limits(&new_proposal) {
                            lineage.push(ProposalLineageEntry {
                                generation,
                                proposal_id: new_proposal.id.clone(),
                                proposal_digest: Some(new_proposal_digest),
                                status: ProposalLineageStatus::Rejected,
                                rejection_reason: Some(error.clone()),
                            });
                            self.log.lock().await.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    ("errors".to_string(), Value::from(error)),
                                    ("attempt".to_string(), Value::from(generation)),
                                ]
                                .into(),
                            );
                            continue 'replan_candidates;
                        }

                        // 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.
                            // Same helper the admission gate uses, so the two
                            // verification points cannot disagree about what is
                            // fatal. They previously did: this path blocked on the
                            // loop heuristic and on state-dependent findings that
                            // the gate treats as advisory, and it passed the tool
                            // map through unconditionally — so an embedder with no
                            // registered schemas burned its whole replan budget on
                            // "unregistered tool" for every call.
                            let blocking = {
                                let tools_guard = self.tools.read().await;
                                crate::verify_gate::blocking_errors(
                                    &new_proposal,
                                    Some(&current_state),
                                    &tools_guard,
                                    100,
                                )
                            };
                            if !blocking.is_empty() {
                                let error_msgs: Vec<String> =
                                    blocking.iter().map(|i| i.message.clone()).collect();
                                let reason = error_msgs.join("; ");
                                lineage.push(ProposalLineageEntry {
                                    generation,
                                    proposal_id: new_proposal.id.clone(),
                                    proposal_digest: Some(new_proposal_digest),
                                    status: ProposalLineageStatus::Rejected,
                                    rejection_reason: Some(reason.clone()),
                                });
                                let mut log = self.log.lock().await;
                                log.append(
                                    EventKind::ReplanRejected,
                                    None,
                                    Some(&proposal.id),
                                    [
                                        ("errors".to_string(), Value::from(reason)),
                                        ("attempt".to_string(), Value::from(generation)),
                                    ]
                                    .into(),
                                );
                                // Don't execute a broken replan — count as failed attempt
                                continue 'replan_candidates;
                            }
                        }

                        // 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 reason = "transactional conflict with current state".to_string();
                            lineage.push(ProposalLineageEntry {
                                generation,
                                proposal_id: new_proposal.id.clone(),
                                proposal_digest: Some(new_proposal_digest),
                                status: ProposalLineageStatus::Rejected,
                                rejection_reason: Some(reason.clone()),
                            });
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ReplanRejected,
                                None,
                                Some(&proposal.id),
                                [
                                    ("errors".to_string(), Value::from(reason)),
                                    ("attempt".to_string(), Value::from(generation)),
                                ]
                                .into(),
                            );
                            drop(log);
                            continue 'replan_candidates;
                        }

                        // 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 reason =
                                    format!("admission gate blocked replan (gate: {gate}; {why})");
                                lineage.push(ProposalLineageEntry {
                                    generation,
                                    proposal_id: new_proposal.id.clone(),
                                    proposal_digest: Some(new_proposal_digest),
                                    status: ProposalLineageStatus::Rejected,
                                    rejection_reason: Some(reason.clone()),
                                });
                                let mut log = self.log.lock().await;
                                log.append(
                                    EventKind::ReplanRejected,
                                    None,
                                    Some(&proposal.id),
                                    [
                                        ("errors".to_string(), Value::from(reason)),
                                        ("attempt".to_string(), Value::from(generation)),
                                    ]
                                    .into(),
                                );
                                drop(log);
                                continue 'replan_candidates;
                            }
                        }

                        lineage.push(ProposalLineageEntry {
                            generation,
                            proposal_id: new_proposal.id.clone(),
                            proposal_digest: Some(new_proposal_digest.clone()),
                            status: ProposalLineageStatus::Accepted,
                            rejection_reason: None,
                        });
                        accepted_proposal_preimages.push(AcceptedProposalPreimage {
                            generation,
                            proposal_digest: new_proposal_digest.clone(),
                            proposal: new_proposal.clone(),
                        });

                        // Log accepted proposal
                        {
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ReplanProposalReceived,
                                None,
                                Some(&proposal.id),
                                [
                                    ("attempt".to_string(), Value::from(generation)),
                                    (
                                        "proposal_id".to_string(),
                                        Value::from(new_proposal.id.as_str()),
                                    ),
                                    (
                                        "proposal_digest".to_string(),
                                        Value::from(new_proposal_digest),
                                    ),
                                    ("proposal".to_string(), new_proposal_preimage),
                                    (
                                        "new_action_count".to_string(),
                                        Value::from(new_proposal.actions.len()),
                                    ),
                                ]
                                .into(),
                            );
                        }
                        current_proposal = new_proposal;
                        break 'replan_candidates;
                    }
                    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(generation)),
                            ]
                            .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 finalize_proposal_result(
                            result,
                            &proposal.id,
                            &lineage,
                            &accepted_proposal_preimages,
                        );
                    }
                }
            }
        }
    }

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

    async fn rollback_failed_plan_candidate(
        &self,
        proposal: &ActionProposal,
        result: &mut ProposalResult,
        pre_plan_snapshot: &HashMap<String, Value>,
        pre_plan_transitions: usize,
        pre_plan_idempotency_keys: &std::collections::HashSet<String>,
    ) -> Result<(), String> {
        let executed_proposal = result.final_proposal.as_ref().unwrap_or(proposal);
        let executed_proposal_digest = result
            .accepted_proposal_preimages
            .last()
            .map(|accepted| accepted.proposal_digest.clone())
            .unwrap_or_else(|| {
                proposal_digest(executed_proposal)
                    .expect("an executed planning candidate already passed proposal identity")
            });
        let observed_changes_by_action: std::collections::BTreeMap<String, HashMap<String, Value>> =
            result
                .results
                .iter()
                .filter(|action_result| {
                    action_result.status == ActionStatus::Succeeded
                        && !action_result.state_changes.is_empty()
                })
                .map(|action_result| {
                    (
                        action_result.action_id.clone(),
                        action_result.state_changes.clone(),
                    )
                })
                .collect();
        let affected_actions: Vec<String> = observed_changes_by_action.keys().cloned().collect();

        // The planning transaction owns the StateStore proposal guard, so this
        // exact snapshot/count restore cannot erase another proposal's commit.
        let durability = match self
            .state
            .restore(pre_plan_snapshot.clone(), pre_plan_transitions)
        {
            Ok(durability) => durability,
            Err(error) => {
                let detail = record_rollback_durability_error(
                    &mut result.results,
                    ROLLBACK_DURABILITY_ERROR,
                    &error.to_string(),
                );
                self.log.lock().await.append(
                    EventKind::ActionFailed,
                    None,
                    Some(&proposal.id),
                    [
                        ("error".to_string(), Value::from(detail.clone())),
                        ("attempted".to_string(), Value::from(true)),
                        ("publication_succeeded".to_string(), Value::from(false)),
                        ("rollback_succeeded".to_string(), Value::from(false)),
                        ("stage".to_string(), Value::from("plan_fallback_rollback")),
                    ]
                    .into(),
                );
                return Err(detail);
            }
        };
        let durability_error = match &durability {
            RestoreDurability::Durable => None,
            RestoreDurability::DurabilityUnknown { error } => Some(error.clone()),
        };

        {
            let mut log = self.log.lock().await;
            log.append(
                EventKind::StateSnapshot,
                None,
                Some(&proposal.id),
                [(
                    "state".to_string(),
                    serde_json::to_value(pre_plan_snapshot).unwrap_or_default(),
                )]
                .into(),
            );
            log.append(
                EventKind::StateRollback,
                None,
                Some(&proposal.id),
                [
                    (
                        "rolled_back_to".to_string(),
                        Value::from("pre-plan snapshot"),
                    ),
                    (
                        "affected_actions".to_string(),
                        serde_json::to_value(&affected_actions).unwrap_or_default(),
                    ),
                    (
                        "rolled_back_changes".to_string(),
                        serde_json::to_value(&observed_changes_by_action).unwrap_or_default(),
                    ),
                    (
                        "changes_semantics".to_string(),
                        Value::from("rolled_back_failed_planning_candidate"),
                    ),
                    (
                        "proposal_digest".to_string(),
                        Value::from(executed_proposal_digest),
                    ),
                    ("attempted".to_string(), Value::from(true)),
                    (
                        "durability_unknown".to_string(),
                        Value::from(durability_error.is_some()),
                    ),
                    ("publication_succeeded".to_string(), Value::from(true)),
                    ("rollback_succeeded".to_string(), Value::from(true)),
                    ("stage".to_string(), Value::from("plan_fallback_rollback")),
                ]
                .into(),
            );
        }

        // Only invalidate entries introduced by this planning transaction.
        // Pre-existing keys may have been read by a candidate, but their
        // committed result predates the pre-plan snapshot and must survive.
        let mut invalidated = std::collections::BTreeSet::new();
        {
            let mut cache = self.idempotency_cache.lock().await;
            for action_result in &result.results {
                if action_result.status != ActionStatus::Succeeded {
                    continue;
                }
                let Some(action) = executed_proposal
                    .actions
                    .iter()
                    .find(|action| action.id == action_result.action_id)
                else {
                    continue;
                };
                if !action.idempotent || action.invocation_mode.is_detached() {
                    continue;
                }
                let key = idempotency_key(action, None);
                if !pre_plan_idempotency_keys.contains(&key) && cache.remove(&key).is_some() {
                    invalidated.insert(key);
                }
            }
        }
        for key in invalidated {
            self.journal_idempotency(&key, None).await;
        }

        // Same reasoning as the abort path above: a rejected planning
        // candidate loses its state effects, not the record of what ran. This
        // result is observable — `plan_and_execute` returns it directly on a
        // rollback durability error, and otherwise keeps it as
        // `first_failure` (car#1157). `rolled_back` is the machine-readable
        // fact; the warning remains descriptive and must never drive logic.
        for action_result in &mut result.results {
            if action_result.status == ActionStatus::Succeeded {
                action_result.rolled_back = true;
                action_result.error = Some(PLAN_FALLBACK_ROLLBACK_WARNING.to_string());
                action_result.state_changes.clear();
            }
        }
        if let Some(error) = durability_error {
            return Err(record_rollback_durability_error(
                &mut result.results,
                ROLLBACK_DURABILITY_UNKNOWN,
                &error,
            ));
        }
        Ok(())
    }

    /// 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::new("empty", vec![], car_ir::CostSummary::default());
        }

        // One transaction guard covers ranking's state snapshot, every
        // candidate attempt and rollback, and the accepted candidate's commit.
        // Candidate execution calls the already-guarded path to avoid recursive
        // acquisition of the non-reentrant mutex.
        let _proposal_execution = self.state.lock_proposal_execution().await;

        // 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 pre_plan_idempotency_keys: std::collections::HashSet<String> = self
            .idempotency_cache
            .lock()
            .await
            .keys()
            .cloned()
            .collect();
        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;
            }

            let proposal = &candidates[scored.index];
            let cancel = tokio_util::sync::CancellationToken::new();
            let mut result = self
                .execute_with_optional_session_already_guarded(proposal, None, None, None, &cancel)
                .await;

            if result.all_succeeded() {
                return result;
            }

            if self
                .rollback_failed_plan_candidate(
                    proposal,
                    &mut result,
                    &pre_plan_snapshot,
                    pre_plan_transitions,
                    &pre_plan_idempotency_keys,
                )
                .await
                .is_err()
            {
                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::for_proposal(&candidates[0], vec![], 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>>) {
        // Identity is established before CAR admits the proposal or emits any
        // action transition. In particular, an I-JSON-incompatible number may
        // not produce a journal row whose mandatory JCS digest is missing.
        let (proposal_preimage, proposal_digest) = match proposal_journal_identity(proposal) {
            Ok(identity) => identity,
            Err(error) => {
                let results = proposal
                    .actions
                    .iter()
                    .map(|action| rejected_result(&action.id, error.clone()))
                    .collect();
                self.log.lock().await.append(
                    EventKind::StateRollback,
                    None,
                    Some(&proposal.id),
                    proposal_rejection_boundary_data(proposal, None, &error),
                );
                return (
                    ProposalResult::for_proposal(proposal, results, CostSummary::default()),
                    HashMap::new(),
                );
            }
        };

        // 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()),
                    ),
                    ("proposal".to_string(), proposal_preimage.clone()),
                    (
                        "proposal_digest".to_string(),
                        Value::from(proposal_digest.clone()),
                    ),
                ]
                .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 reason = format!(
                        "capability denied: proposal has {} actions, max allowed is {:?}",
                        proposal.actions.len(),
                        cap.max_actions
                    );
                    let mut action_results = Vec::new();
                    let mut log = self.log.lock().await;
                    for action in &proposal.actions {
                        log.append(
                            EventKind::ActionRejected,
                            Some(&action.id),
                            Some(&proposal.id),
                            action_outcome_data("proposal_capability", &reason, false, None),
                        );
                        action_results.push(rejected_result(&action.id, reason.clone()));
                    }
                    log.append(
                        EventKind::StateRollback,
                        None,
                        Some(&proposal.id),
                        proposal_rejection_boundary_data(proposal, Some(&proposal_digest), &reason),
                    );
                    drop(log);
                    return (
                        ProposalResult::for_proposal(
                            proposal,
                            action_results,
                            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 {
                    let action = &proposal.actions[idx];
                    let reason = format!("{CANCELED_PREFIX}cancellation requested by caller");
                    self.log.lock().await.append(
                        EventKind::ActionSkipped,
                        Some(&action.id),
                        Some(&proposal.id),
                        action_outcome_data("cancellation", &reason, false, None),
                    );
                    results.push(canceled_result(
                        &action.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 {
                    let action = &proposal.actions[idx];
                    let stage = if budget_exceeded {
                        "cost_budget"
                    } else {
                        "dependency_abort"
                    };
                    self.log.lock().await.append(
                        EventKind::ActionSkipped,
                        Some(&action.id),
                        Some(&proposal.id),
                        action_outcome_data(stage, skip_reason, false, None),
                    );
                    results.push(skipped_result(&action.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"
                        };
                        let action = &proposal.actions[idx];
                        let stage = if budget_exceeded {
                            "cost_budget"
                        } else {
                            "dependency_abort"
                        };
                        self.log.lock().await.append(
                            EventKind::ActionSkipped,
                            Some(&action.id),
                            Some(&proposal.id),
                            action_outcome_data(stage, skip_reason, false, None),
                        );
                        results.push(skipped_result(&action.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;
                                self.log.lock().await.append(
                                    EventKind::ActionSkipped,
                                    Some(&proposal.actions[idx].id),
                                    Some(&proposal.id),
                                    action_outcome_data(
                                        "cost_budget",
                                        "cost budget exceeded",
                                        false,
                                        None,
                                    ),
                                );
                                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;
                                self.log.lock().await.append(
                                    EventKind::ActionSkipped,
                                    Some(&proposal.actions[idx].id),
                                    Some(&proposal.id),
                                    action_outcome_data(
                                        "cost_budget",
                                        "cost budget exceeded",
                                        false,
                                        None,
                                    ),
                                );
                                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;
                                self.log.lock().await.append(
                                    EventKind::ActionSkipped,
                                    Some(&proposal.actions[idx].id),
                                    Some(&proposal.id),
                                    action_outcome_data(
                                        "cost_budget",
                                        "cost budget exceeded",
                                        false,
                                        None,
                                    ),
                                );
                                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.terminal
                        || requires_integrity_rollback(&ar)
                        || ((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;
                    }
                    if ar.terminal || requires_integrity_rollback(&ar) {
                        aborted = true;
                    }
                    results.push(ar);
                }
            }
        }

        let observed_changes_by_action: std::collections::BTreeMap<String, HashMap<String, Value>> =
            proposal
                .actions
                .iter()
                .filter_map(|action| {
                    results
                        .iter()
                        .find(|result| {
                            result.action_id == action.id
                                && result.status == ActionStatus::Succeeded
                                && !result.state_changes.is_empty()
                        })
                        .map(|result| (action.id.clone(), result.state_changes.clone()))
                })
                .collect();
        let affected_actions: Vec<String> = observed_changes_by_action.keys().cloned().collect();

        // Handle rollback. Per-action StateChanged rows are provisional until
        // this proposal-level transaction boundary is durably recorded.
        if aborted {
            // Tenant-scoped rollback when the proposal is tenant-scoped
            // (EPIC E / E2), else the full restore.
            let rollback = match rollback_tenant {
                Some(_) => {
                    self.state
                        .restore_scoped(rollback_tenant, snapshot.clone(), transition_count)
                }
                None => self.state.restore(snapshot.clone(), transition_count),
            };

            let rollback_durability = match rollback {
                Ok(durability) => Some(durability),
                Err(error) => {
                    let detail = record_rollback_durability_error(
                        &mut results,
                        ROLLBACK_DURABILITY_ERROR,
                        &error.to_string(),
                    );
                    self.log.lock().await.append(
                        EventKind::ActionFailed,
                        None,
                        Some(&proposal.id),
                        [
                            ("error".to_string(), Value::from(detail)),
                            ("attempted".to_string(), Value::from(true)),
                            ("publication_succeeded".to_string(), Value::from(false)),
                            ("in_memory_state_preserved".to_string(), Value::from(true)),
                            (
                                "idempotency_entries_preserved".to_string(),
                                Value::from(true),
                            ),
                            ("rollback_succeeded".to_string(), Value::from(false)),
                            ("stage".to_string(), Value::from("proposal_rollback")),
                        ]
                        .into(),
                    );
                    None
                }
            };

            if let Some(durability) = rollback_durability {
                let durability_error = match &durability {
                    RestoreDurability::Durable => None,
                    RestoreDurability::DurabilityUnknown { error } => Some(error.clone()),
                };
                if let Some(error) = &durability_error {
                    record_rollback_durability_error(
                        &mut results,
                        ROLLBACK_DURABILITY_UNKNOWN,
                        error,
                    );
                }
                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"),
                        ),
                        (
                            "affected_actions".to_string(),
                            serde_json::to_value(&affected_actions).unwrap_or_default(),
                        ),
                        (
                            "rolled_back_changes".to_string(),
                            serde_json::to_value(&observed_changes_by_action).unwrap_or_default(),
                        ),
                        (
                            "changes_semantics".to_string(),
                            Value::from("rolled_back_provisional_state_mutations"),
                        ),
                        (
                            "proposal_digest".to_string(),
                            Value::from(proposal_digest.clone()),
                        ),
                        ("attempted".to_string(), Value::from(true)),
                        (
                            "durability_unknown".to_string(),
                            Value::from(durability_error.is_some()),
                        ),
                        ("publication_succeeded".to_string(), Value::from(true)),
                        ("rollback_succeeded".to_string(), Value::from(true)),
                        ("stage".to_string(), Value::from("proposal_rollback")),
                    ]
                    .into(),
                );
                drop(log);

                // Clear idempotency cache for rolled-back actions only after
                // the state journal has durably accepted the rollback.
                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;
                }
                // The proposal did not commit, so no action's state changes
                // survive. Each action's STATUS still records whether that
                // action executed; `rolled_back` independently records whether
                // that successful execution preceded this rollback. Folding
                // the second fact into ActionStatus erased which action
                // aborted the proposal and marked `failed` precisely the
                // actions whose external effects DID happen and may remain.
                // The warning is human-readable context, never control flow
                // (car#1157).
                for result in &mut results {
                    if result.status == ActionStatus::Succeeded {
                        result.rolled_back = true;
                        result.error = Some(ROLLBACK_WARNING.to_string());
                        result.state_changes.clear();
                    }
                }
            }
        } else {
            self.log.lock().await.append(
                EventKind::StateCommitted,
                None,
                Some(&proposal.id),
                [
                    (
                        "affected_actions".to_string(),
                        serde_json::to_value(&affected_actions).unwrap_or_default(),
                    ),
                    (
                        "committed_changes".to_string(),
                        serde_json::to_value(&observed_changes_by_action).unwrap_or_default(),
                    ),
                    (
                        "changes_semantics".to_string(),
                        Value::from("committed_provisional_state_mutations"),
                    ),
                    ("proposal_digest".to_string(), Value::from(proposal_digest)),
                    ("attempted".to_string(), Value::from(true)),
                    ("stage".to_string(), Value::from("proposal_commit")),
                ]
                .into(),
            );
        }

        // 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;
                        }
                    }
                }
                // A failed action ran — the tool was invoked and errored — so it
                // belongs in `actions_executed`. A rejected one was blocked by
                // the validator or a policy and never started; counting it as
                // executed reported work that never happened (car#624).
                ActionStatus::Failed => {
                    cost.actions_executed += 1;
                }
                ActionStatus::Rejected => {
                    cost.actions_rejected += 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::for_proposal(proposal, 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 reason =
                                format!("capability denied: tool '{}' not allowed", tool_name);
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ActionRejected,
                                Some(&action.id),
                                Some(proposal_id),
                                action_outcome_data("capability", &reason, false, None),
                            );
                            return (rejected_result(&action.id, reason), 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 reason =
                                format!("capability denied: state key '{}' not allowed", key);
                            let mut log = self.log.lock().await;
                            log.append(
                                EventKind::ActionRejected,
                                Some(&action.id),
                                Some(proposal_id),
                                action_outcome_data("capability", &reason, false, None),
                            );
                            return (rejected_result(&action.id, reason), 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),
                action_outcome_data("validation", &error, false, None),
            );
            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 {
                    let error = format!(
                        "unknown session id '{sid}' — open one via Runtime::open_session before executing under a session"
                    );
                    // 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),
                        action_outcome_data("policy", &error, false, None),
                    );
                    return (rejected_result(&action.id, error), 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),
                    action_outcome_data("policy", &error, false, None),
                );
                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()),
                        ),
                        ("attempted".to_string(), Value::from(false)),
                        ("stage".to_string(), Value::from("idempotency_cache")),
                    ]
                    .into(),
                );
                return (
                    ActionResult {
                        action_id: action.id.clone(),
                        status: cached.status.clone(),
                        output: cached.output.clone(),
                        error: cached.error.clone(),
                        terminal: cached.terminal,
                        // The cached action committed these mutations in an
                        // earlier proposal. This proposal did not observe or
                        // apply them, so its actual-change evidence is empty.
                        state_changes: HashMap::new(),
                        rolled_back: false,
                        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, session_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,
        session_id: Option<&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 {
            match action.max_retries.min(retry_cap).checked_add(1) {
                Some(max_attempts) => max_attempts,
                None => {
                    let reason =
                        format!("action '{}' retry attempt count overflows u32", action.id);
                    self.log.lock().await.append(
                        EventKind::ActionRejected,
                        Some(&action.id),
                        Some(proposal_id),
                        action_outcome_data("retry_admission", &reason, false, None),
                    );
                    return (rejected_result(&action.id, reason), 0);
                }
            }
        } else {
            1
        };

        let mut last_error: Option<String> = None;
        let mut retries: u32 = 0;
        let tool_source = self.tool_source_for_action(action).await;
        let params_digest = action_params_digest(action);

        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 - 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 data: HashMap<String, Value> =
                    [("attempt".to_string(), Value::from(attempt + 1))].into();
                insert_tool_event_provenance(&mut data, action, tool_source);
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ActionExecuting,
                    Some(&action.id),
                    Some(proposal_id),
                    data,
                );
            }

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

            // Execute with optional timeout. Keep the timeout bit typed until
            // event classification rather than trying to recover it from the
            // human-readable error later.
            let (exec_result, engine_timeout) = if let Some(timeout_ms) = action.timeout_ms {
                match timeout(
                    Duration::from_millis(timeout_ms),
                    self.dispatch(action, session_id, scope, attempt + 1),
                )
                .await
                {
                    Ok(result) => (result, false),
                    Err(_) => (
                        Err(ToolFailure::ordinary(format!(
                            "action timed out after {}ms",
                            timeout_ms
                        ))),
                        true,
                    ),
                }
            } else {
                (
                    self.dispatch(action, session_id, scope, attempt + 1).await,
                    false,
                )
            };

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

            match exec_result {
                Ok(output) => {
                    if let Err(error) = car_inference::catalog_identity::canonical_json(&output) {
                        let reason = format!(
                            "tool output failed JCS/I-JSON validation after dispatch; external effect may have occurred and was not undone: {error}"
                        );
                        let mut data = action_outcome_data(
                            "output_validation",
                            &reason,
                            true,
                            Some(attempt + 1),
                        );
                        data.insert(
                            "external_effect_status".to_string(),
                            Value::from("may_have_occurred_not_undone"),
                        );
                        insert_tool_event_provenance(&mut data, action, tool_source);
                        insert_action_outcome_signal(
                            &mut data,
                            action,
                            &params_digest,
                            Some("validation"),
                        );
                        if action.tool.is_some() {
                            data.insert("ok".to_string(), Value::from(false));
                        }
                        self.log.lock().await.append(
                            EventKind::ActionFailed,
                            Some(&action.id),
                            Some(proposal_id),
                            data,
                        );
                        return (
                            ActionResult {
                                action_id: action.id.clone(),
                                status: ActionStatus::Failed,
                                output: None,
                                error: Some(reason),
                                terminal: false,
                                state_changes: HashMap::new(),
                                rolled_back: false,
                                duration_ms: Some(duration_ms),
                                timestamp: chrono::Utc::now(),
                            },
                            retries,
                        );
                    }

                    // Snapshot the changes made by dispatch before applying
                    // any result accounting. Deletion and set-to-null are
                    // intentionally tagged differently on the authenticated
                    // wire; a bare JSON null cannot represent both.
                    let runtime_state_changes: HashMap<String, Value> = self
                        .state
                        .transitions_since(transitions_before)
                        .into_iter()
                        // Actions at the same DAG level execute concurrently,
                        // so the shared transition tail can contain a sibling's
                        // writes. StateTransition.action_id is assigned at the
                        // mutation site and is the authoritative attribution
                        // boundary for this per-action journal field.
                        .filter(|transition| transition.action_id == action.id)
                        .map(|transition| {
                            (
                                transition.key,
                                StateMutation::from_new_value(transition.new_value).encode(),
                            )
                        })
                        .collect();

                    // Declared expected effects are plan assertions only.
                    // They must never be applied as if the runtime observed
                    // them, nor merged into committed result evidence.
                    let tenant_for_effects = scope.and_then(|s| s.tenant_id.as_deref());
                    let state_changes = runtime_state_changes.clone();

                    // Record this result's provenance for the VIGIL intent
                    // gate (`crate::taint`) — the only point in the system
                    // that knows which SPECIFIC results were tainted. Only
                    // on success: a failed action commits no effects, so it
                    // changes no taint. No intent gate ⇒ no ledger ⇒ one
                    // `Option` read and nothing else.
                    let ledger = self.taint_ledger.read().await.clone();
                    if let Some(ledger) = ledger {
                        ledger
                            .record_result(
                                tenant_for_effects,
                                action,
                                state_changes.keys().cloned(),
                            )
                            .await;
                    }

                    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();
                    succ_data.insert("attempt".to_string(), Value::from(attempt + 1));
                    succ_data.insert("attempted".to_string(), Value::from(true));
                    succ_data.insert("stage".to_string(), Value::from("dispatch"));
                    insert_tool_event_provenance(&mut succ_data, action, tool_source);
                    insert_action_outcome_signal(&mut succ_data, action, &params_digest, None);
                    if action.tool.is_some() {
                        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 !action.expected_effects.is_empty() || !runtime_state_changes.is_empty() {
                        log.append(
                            EventKind::StateChanged,
                            Some(&action.id),
                            Some(proposal_id),
                            [
                                (
                                    "declared_expected_effects".to_string(),
                                    serde_json::to_value(&action.expected_effects)
                                        .unwrap_or_default(),
                                ),
                                (
                                    "runtime_state_mutations".to_string(),
                                    serde_json::to_value(&runtime_state_changes)
                                        .unwrap_or_default(),
                                ),
                                (
                                    "changes".to_string(),
                                    serde_json::to_value(&state_changes).unwrap_or_default(),
                                ),
                                (
                                    "changes_semantics".to_string(),
                                    Value::from("provisional_state_mutations"),
                                ),
                                ("attempt".to_string(), Value::from(attempt + 1)),
                                ("attempted".to_string(), Value::from(true)),
                                ("stage".to_string(), Value::from("state_effect")),
                            ]
                            .into(),
                        );
                    }

                    return (
                        ActionResult {
                            action_id: action.id.clone(),
                            status: ActionStatus::Succeeded,
                            output: Some(output),
                            error: None,
                            terminal: false,
                            state_changes,
                            rolled_back: false,
                            duration_ms: Some(duration_ms),
                            timestamp: chrono::Utc::now(),
                        },
                        retries,
                    );
                }
                Err(failure) => {
                    let terminal = match failure.classification {
                        ToolFailureClassification::Ordinary => false,
                        ToolFailureClassification::Terminal => true,
                    };
                    let error = failure.message;
                    last_error = Some(error.clone());
                    let mut log = self.log.lock().await;
                    let mut fail_data: HashMap<String, Value> = [
                        ("error".to_string(), Value::from(error.as_str())),
                        ("attempt".to_string(), Value::from(attempt + 1)),
                        ("attempted".to_string(), Value::from(true)),
                        ("stage".to_string(), Value::from("dispatch")),
                    ]
                    .into();
                    // Tool name + ok=false so the receipt verifier (A6) records
                    // that the tool *executed* (a failed call still ran).
                    insert_tool_event_provenance(&mut fail_data, action, tool_source);
                    insert_action_outcome_signal(
                        &mut fail_data,
                        action,
                        &params_digest,
                        Some(action_error_class(action, &error, engine_timeout)),
                    );
                    if action.tool.is_some() {
                        fail_data.insert("ok".to_string(), Value::from(false));
                    }
                    if terminal {
                        fail_data.insert("terminal".to_string(), Value::from(true));
                    }
                    log.append(
                        EventKind::ActionFailed,
                        Some(&action.id),
                        Some(proposal_id),
                        fail_data,
                    );
                    drop(log);

                    if terminal {
                        return (
                            ActionResult {
                                action_id: action.id.clone(),
                                status: ActionStatus::Failed,
                                output: None,
                                error: Some(error),
                                terminal: true,
                                rolled_back: false,
                                state_changes: HashMap::new(),
                                duration_ms: Some(duration_ms),
                                timestamp: chrono::Utc::now(),
                            },
                            retries,
                        );
                    }
                }
            }
        }

        // All attempts exhausted
        if action.failure_behavior == FailureBehavior::Skip {
            let reason = last_error.as_deref().unwrap_or("all attempts exhausted");
            self.log.lock().await.append(
                EventKind::ActionSkipped,
                Some(&action.id),
                Some(proposal_id),
                action_outcome_data("failure_behavior", reason, true, Some(max_attempts)),
            );
            return (skipped_result(&action.id, reason), retries);
        }

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

    /// Resolve the stable source category for an action's tool. Canonical
    /// ToolEntry provenance wins; the ToolSchema fallback keeps legacy direct
    /// registrations observable, and the final default covers old callers that
    /// supplied a tool field on a non-tool action.
    async fn tool_source_for_action(&self, action: &Action) -> Option<car_ir::ToolSourceKind> {
        let tool = action.tool.as_deref()?;
        if let Some(entry) = self.registry.get(tool).await {
            return Some(entry.source.kind());
        }
        Some(
            self.tools
                .read()
                .await
                .get(tool)
                .map(|schema| schema.source)
                .unwrap_or(car_ir::ToolSourceKind::UserDefined),
        )
    }

    /// 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,
        session_id: Option<&str>,
        scope: Option<&crate::scope::RuntimeScope>,
        attempt: u32,
    ) -> Result<Value, ToolFailure> {
        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.map_err(ToolFailure::from);
                    }
                }

                // 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(),
                        );
                    }
                }

                // Built-in outbound human messaging. Sits here, alongside the
                // other built-ins — downstream of `validate_action` and the
                // policy check in the caller, and downstream of the result
                // cache read and `rate_limiter.acquire` above — so a message
                // to a human traverses exactly the same governed chain
                // (validator → policy → rate limit → eventlog) as any other
                // side effect, instead of the hand-rolled transport each agent
                // used to carry.
                if tool_name == "messaging.send" {
                    // No fall-through to the host executor when unconfigured.
                    // Falling through would let an ungoverned host transport
                    // answer the call — the exact path this built-in exists to
                    // close — so an unconfigured runtime refuses instead.
                    let Some(ref sink) = self.message_sink else {
                        return Err("messaging.send: messaging is not configured on this \
                                    runtime — attach a message sink via \
                                    Runtime::with_message_sink"
                            .into());
                    };
                    let msg = crate::messaging::OutboundMessage::from_tool_params(&params)?;
                    let receipt = sink.send(&msg).await?;
                    // Not cached: the schema declares no cache TTL, and a
                    // cached send would swallow a second, genuinely wanted
                    // message with identical text.
                    return serde_json::to_value(&receipt).map_err(|error| {
                        ToolFailure::ordinary(format!("messaging.send: serialize receipt: {error}"))
                    });
                }

                // 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 return_schema = self
                        .tools
                        .read()
                        .await
                        .get(tool_name)
                        .and_then(|schema| schema.returns.clone());
                    let result = executor
                        .execute_classified(
                            tool_name,
                            &params,
                            &action.id,
                            action.timeout_ms,
                            session_id,
                            attempt,
                            &action.expected_effects,
                            return_schema.as_ref(),
                        )
                        .await;
                    let fall_through = matches!(
                        &result,
                        Err(error) if error.classification == ToolFailureClassification::Ordinary
                            && error.message.starts_with("unknown tool")
                    );
                    if !fall_through {
                        match result {
                            Ok(execution) => {
                                if !execution.state_changes.is_empty() {
                                    let expected_keys: std::collections::BTreeSet<&str> = action
                                        .expected_effects
                                        .keys()
                                        .map(String::as_str)
                                        .collect();
                                    let actual_keys: std::collections::BTreeSet<&str> = execution
                                        .state_changes
                                        .keys()
                                        .map(String::as_str)
                                        .collect();
                                    if expected_keys != actual_keys {
                                        return Err(format!(
                                        "tool '{}' callback state_changes keys do not match action '{}': expected {:?}, got {:?}",
                                        tool_name, action.id, expected_keys, actual_keys
                                    )
                                    .into());
                                    }
                                    car_inference::catalog_identity::canonical_json(&execution.output)
                                    .map_err(|error| {
                                        format!(
                                            "tool output failed JCS/I-JSON validation after dispatch; external effect may have occurred and was not undone: {error}"
                                        )
                                    })?;
                                    let changes_value = serde_json::to_value(&execution.state_changes)
                                    .map_err(|error| {
                                        format!(
                                            "tool '{}' callback state_changes serialization failed: {error}",
                                            tool_name
                                        )
                                    })?;
                                    car_inference::catalog_identity::canonical_json(&changes_value)
                                    .map_err(|error| {
                                        format!(
                                            "tool '{}' callback state_changes failed JCS/I-JSON validation: {error}",
                                            tool_name
                                        )
                                    })?;

                                    // Every rejection above happens before the
                                    // first write. Apply in deterministic key order,
                                    // attributed to the active action, through ONE
                                    // batch mutation boundary: a concurrent reader,
                                    // journal replay, and restart recovery observe
                                    // the complete old state or the complete new
                                    // state — never a prefix of this callback's
                                    // keys (Parslee-ai/car#1140).
                                    let mut changes: Vec<(String, Value)> = execution
                                        .state_changes
                                        .iter()
                                        .map(|(key, value)| (key.clone(), value.clone()))
                                        .collect();
                                    changes.sort_by(|(left, _), (right, _)| left.cmp(right));
                                    let tenant = scope.and_then(|scope| scope.tenant_id.as_deref());
                                    self.state.scoped(tenant).set_batch(changes, &action.id);
                                }
                                self.result_cache
                                    .put(tool_name, &params, execution.output.clone())
                                    .await;
                                return Ok(execution.output);
                            }
                            Err(error) => return Err(error),
                        }
                    }
                }

                // 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();
                let read_ledger = self.read_ledgers.ledger_for(session_id);
                if let Some(result) = crate::agent_basics::execute_with_ledger(
                    &substrate,
                    &read_ledger,
                    tool_name,
                    &params,
                )
                .await
                {
                    if let Ok(ref value) = result {
                        self.result_cache
                            .put(tool_name, &params, value.clone())
                            .await;
                    }
                    return result.map_err(ToolFailure::from);
                }

                Err(format!("no handler for tool '{}'", tool_name).into())
            }
            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
                    )
                    .into())
                } 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(),
                source: car_ir::ToolSourceKind::UserDefined,
                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(),
            source: car_ir::ToolSourceKind::Subprocess,
            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_entry(
            crate::registry::ToolEntry::new(schema)
                .with_source(crate::registry::ToolSource::Subprocess),
        )
        .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(start_paused = true)]
    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
        );
    }

    /// An executor that records every `attempt` it is handed and fails until
    /// the Nth, so a test can watch the counter advance across real retries.
    struct AttemptRecordingExecutor {
        seen: Arc<tokio::sync::Mutex<Vec<u32>>>,
        succeed_on: u32,
    }

    #[async_trait::async_trait]
    impl ToolExecutor for AttemptRecordingExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            Ok(Value::Null)
        }
        async fn execute_with_action_in_session(
            &self,
            _tool: &str,
            _params: &Value,
            _action_id: &str,
            _timeout_ms: Option<u64>,
            _session_id: Option<&str>,
            attempt: u32,
        ) -> Result<Value, String> {
            self.seen.lock().await.push(attempt);
            if attempt >= self.succeed_on {
                Ok(serde_json::json!({ "ok": true }))
            } else {
                Err("transient".to_string())
            }
        }
    }

    /// Parslee-ai/car#928 — the retry counter the executor receives must be the
    /// engine's real one, and it must advance.
    ///
    /// The WS executor hardcoded `attempt: 1` onto the `tools.execute` payload,
    /// so the field a host would build a retry-disambiguating join on never
    /// varied. Asserting on the sequence the executor actually observes is the
    /// property that matters: a frame test can only prove the payload carries
    /// whatever it was handed, not that the engine hands it the truth.
    #[tokio::test]
    async fn the_executor_sees_the_engines_real_attempt_sequence() {
        let seen = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let rt = Runtime::new();
        rt.register_tool("flaky").await;
        rt.set_executor(Arc::new(AttemptRecordingExecutor {
            seen: seen.clone(),
            succeed_on: 3,
        }))
        .await;

        let proposal: ActionProposal = serde_json::from_value(serde_json::json!({
            "source": "test",
            "actions": [{
                "id": "a0",
                "type": "tool_call",
                "tool": "flaky",
                "parameters": {},
                "dependencies": [],
                "failure_behavior": "retry",
                "max_retries": 3,
            }],
        }))
        .expect("proposal deserializes");

        let result = rt.execute(&proposal).await;
        assert_eq!(
            result.results[0].status,
            ActionStatus::Succeeded,
            "third attempt succeeds: {:?}",
            result.results[0].error
        );
        assert_eq!(
            *seen.lock().await,
            vec![1, 2, 3],
            "1-based and advancing — a constant here is the #928 bug"
        );
    }

    #[tokio::test(start_paused = true)]
    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). Tokio's
        // paused clock advances the 300ms dispatch without a wall-clock wait.
        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);
    }
}