car_engine/

executor.rs

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

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

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

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

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

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

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

/// 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. cost_budget (RwLock, read-only during execution)
/// 5. log (TokioMutex, acquired/released per event)
/// 6. tool_executor (TokioMutex, clone Arc and drop before await)
/// 7. 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>>,
    pub log: Arc<TokioMutex<EventLog>>,
    pub rate_limiter: Arc<RateLimiter>,
    pub result_cache: Arc<ResultCache>,
    tool_executor: TokioMutex<Option<Arc<dyn ToolExecutor>>>,
    idempotency_cache: TokioMutex<HashMap<String, ActionResult>>,
    cost_budget: TokioRwLock<Option<CostBudget>>,
    capabilities: TokioRwLock<Option<CapabilitySet>>,
    inference_engine: Option<Arc<car_inference::InferenceEngine>>,
    /// Optional memgine for skill learning (auto-distillation after execution).
    memgine: Option<Arc<TokioMutex<car_memgine::MemgineEngine>>>,
    /// Whether to auto-distill skills after each proposal execution.
    auto_distill: bool,
}

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())),
            log: Arc::new(TokioMutex::new(EventLog::new())),
            rate_limiter: Arc::new(RateLimiter::new()),
            result_cache: Arc::new(ResultCache::new()),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            memgine: None,
            auto_distill: false,
        }
    }

    /// 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,
            log,
            rate_limiter: Arc::new(RateLimiter::new()),
            result_cache: Arc::new(ResultCache::new()),
            tool_executor: TokioMutex::new(None),
            idempotency_cache: TokioMutex::new(HashMap::new()),
            cost_budget: TokioRwLock::new(None),
            capabilities: TokioRwLock::new(None),
            inference_engine: None,
            memgine: None,
            auto_distill: false,
        }
    }

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

    /// Attach a memgine for automatic skill learning after execution.
    /// Enables auto-distillation of execution traces into skills.
    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
    }

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

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

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

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

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

    /// 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 full proposal through the runtime loop.
    pub async fn execute(&self, proposal: &ActionProposal) -> ProposalResult {
        // Generate trace_id for this proposal execution
        let trace_id = Uuid::new_v4().to_string();

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

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

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

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

        let mut results: Vec<ActionResult> = Vec::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);

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

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

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

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

                    let (ar, action_retries) = self.process_action(&proposal.actions[idx], &proposal.id, &trace_id, &root_span_id).await;
                    total_retries += action_retries;

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

                    if ar.status == ActionStatus::Failed
                        && proposal.actions[idx].failure_behavior == FailureBehavior::Abort
                    {
                        aborted = true;
                    }
                    results.push(ar);
                }
            } else {
                // Concurrent execution via futures::join_all
                let futs: Vec<_> = level.iter()
                    .map(|&idx| self.process_action(&proposal.actions[idx], &proposal.id, &trace_id, &root_span_id))
                    .collect();
                let level_results = futures::future::join_all(futs).await;

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

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

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

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

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

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

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

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

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

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

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

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

                    // Check if any domains need evolution
                    let threshold = 0.6; // TODO: make configurable
                    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
    }

    /// Process a single action: idempotency → validate → policy → execute.
    /// Returns (ActionResult, retries_count).
    async fn process_action(
        &self,
        action: &Action,
        proposal_id: &str,
        trace_id: &str,
        parent_span_id: &str,
    ) -> (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).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).
    async fn process_action_inner(&self, action: &Action, proposal_id: &str) -> (ActionResult, u32) {
        // Idempotency check
        if action.idempotent {
            let key = idempotency_key(action);
            let cache = self.idempotency_cache.lock().await;
            if let Some(cached) = cache.get(&key) {
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::ActionDeduplicated,
                    Some(&action.id),
                    Some(proposal_id),
                    [("cached_action_id".to_string(), Value::from(cached.action_id.as_str()))].into(),
                );
                return (ActionResult {
                    action_id: action.id.clone(),
                    status: cached.status.clone(),
                    output: cached.output.clone(),
                    error: cached.error.clone(),
                    state_changes: cached.state_changes.clone(),
                    duration_ms: Some(0.0),
                    timestamp: chrono::Utc::now(),
                }, 0);
            }
        }

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

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

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

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

        // Policy check
        {
            let policies = self.policies.read().await;
            let violations = policies.check(action, &self.state);
            if !violations.is_empty() {
                let error = violations
                    .iter()
                    .map(|v| format!("policy '{}': {}", v.policy_name, v.reason))
                    .collect::<Vec<_>>()
                    .join("; ");
                let mut log = self.log.lock().await;
                log.append(
                    EventKind::PolicyViolation,
                    Some(&action.id),
                    Some(proposal_id),
                    HashMap::new(),
                );
                return (rejected_result(&action.id, error), 0);
            }
        }

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

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

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

        (result, retries)
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

                // Try built-in inference tools first when the inference engine is available.
                if matches!(tool_name, "infer" | "infer.grounded" | "embed" | "classify") {
                    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;
                    }
                }

                let executor = {
                    let guard = self.tool_executor.lock().await;
                    guard.as_ref().ok_or("no tool executor configured")?.clone()
                };
                let result = executor.execute(tool_name, &params).await;

                // Store successful results in the cross-proposal cache.
                if let Ok(ref value) = result {
                    self.result_cache
                        .put(tool_name, &params, value.clone())
                        .await;
                }

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

    // --- Checkpoint and resume ---

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

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

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

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

    /// Restore runtime state from a checkpoint.
    pub async fn restore_checkpoint(&self, checkpoint: &Checkpoint) {
        // Restore state
        for (key, value) in &checkpoint.state {
            self.state.set(key, value.clone(), "checkpoint_restore");
        }
        // Restore tools (as name-only schemas; full schemas are not persisted in checkpoint)
        let mut tools = self.tools.write().await;
        tools.clear();
        for tool_name in &checkpoint.tools {
            let schema = ToolSchema {
                name: tool_name.clone(),
                description: String::new(),
                parameters: serde_json::Value::Object(Default::default()),
                returns: None,
                idempotent: false,
                cache_ttl_secs: None,
                rate_limit: None,
            };
            tools.insert(tool_name.clone(), schema);
        }
    }
}

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