car-engine 0.52.0

//! Tests for the core runtime engine.

use crate::{
    capabilities::CapabilitySet, checkpoint::Checkpoint, CostBudget, Runtime, ToolExecutor,
    TransactionCheckMode,
};
use car_ir::*;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;

/// A simple tool executor for tests.
struct TestExecutor;

#[async_trait::async_trait]
impl ToolExecutor for TestExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "add" => {
                let a = params.get("a").and_then(|v| v.as_i64()).unwrap_or(0);
                let b = params.get("b").and_then(|v| v.as_i64()).unwrap_or(0);
                Ok(Value::from(a + b))
            }
            "echo" => {
                let msg = params.get("message").and_then(|v| v.as_str()).unwrap_or("");
                Ok(Value::from(msg))
            }
            "fail" => Err("boom".to_string()),
            "slow" => {
                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
                Ok(Value::from("done"))
            }
            _ => Err(format!("unknown tool: {}", tool)),
        }
    }
}

/// Forces two same-level tool actions to overlap after each action has taken
/// its transition-log watermark, then records one state mutation per action.
/// This makes sibling-transition contamination deterministic.
struct ConcurrentStateExecutor {
    state: Arc<car_state::StateStore>,
    barrier: tokio::sync::Barrier,
}

/// Coordinates two proposal executions without assuming they are allowed to
/// enter dispatch concurrently. The first call waits; an unserialized second
/// call writes before the first is released, while a serialized runtime keeps
/// the second outside dispatch until the first completes.
struct CrossProposalBarrierExecutor {
    state: Arc<car_state::StateStore>,
    calls: AtomicU32,
    first_entered: tokio::sync::Notify,
    second_entered: tokio::sync::Notify,
    release_first: tokio::sync::Notify,
}

/// Makes the first planning candidate wait while a separate proposal queues
/// for the shared StateStore transaction guard. Candidate identity controls
/// whether dispatch fails, succeeds, or commits the concurrent sentinel.
struct PlanTransactionExecutor {
    state: Arc<car_state::StateStore>,
    first_candidate_entered: tokio::sync::Notify,
    release_first_candidate: tokio::sync::Notify,
}

#[async_trait::async_trait]
impl ToolExecutor for PlanTransactionExecutor {
    async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
        Err("plan transaction executor requires action identity".to_string())
    }

    async fn execute_with_action(
        &self,
        _tool: &str,
        _params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        match action_id {
            "plan-action-a" => {
                self.first_candidate_entered.notify_one();
                self.release_first_candidate.notified().await;
                Err("first candidate failed".to_string())
            }
            "plan-action-b" => Err("second candidate failed".to_string()),
            "plan-action-c" => Ok(Value::from("fallback succeeded")),
            "outside-write" => {
                self.state
                    .set("concurrent_sentinel", Value::from("committed"), action_id);
                Ok(Value::from("committed"))
            }
            other => Err(format!("unexpected controlled action: {other}")),
        }
    }
}

/// Produces a partial-success `Skip` candidate: one idempotent action mutates
/// state while its sibling fails without triggering execute's Abort rollback.
struct FallbackRollbackExecutor {
    state: Arc<car_state::StateStore>,
    effect_dispatches: AtomicU32,
}

#[async_trait::async_trait]
impl ToolExecutor for FallbackRollbackExecutor {
    async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
        Err("fallback rollback executor requires action identity".to_string())
    }

    async fn execute_with_action(
        &self,
        _tool: &str,
        _params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        match action_id {
            "partial-effect" => {
                self.effect_dispatches.fetch_add(1, Ordering::SeqCst);
                self.state
                    .set("partial_dirty", Value::from(true), action_id);
                Ok(Value::from("effect applied"))
            }
            "partial-skip" => Err("skip candidate failure".to_string()),
            "fallback-good" => Ok(Value::from("fallback succeeded")),
            other => Err(format!("unexpected fallback action: {other}")),
        }
    }
}

#[async_trait::async_trait]
impl ToolExecutor for CrossProposalBarrierExecutor {
    async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
        Err("cross-proposal executor requires action identity".to_string())
    }

    async fn execute_with_action(
        &self,
        _tool: &str,
        params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        let call = self.calls.fetch_add(1, Ordering::SeqCst);
        if call == 0 {
            self.first_entered.notify_one();
            self.release_first.notified().await;
        }
        let key = params
            .get("key")
            .and_then(Value::as_str)
            .ok_or("missing key")?;
        let value = params.get("value").cloned().ok_or("missing value")?;
        self.state.set(key, value.clone(), action_id);
        if call == 1 {
            self.second_entered.notify_one();
        }
        Ok(value)
    }
}

struct RuntimeMutationExecutor {
    state: Arc<car_state::StateStore>,
}

#[async_trait::async_trait]
impl ToolExecutor for RuntimeMutationExecutor {
    async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
        Err("mutation executor requires action identity".to_string())
    }

    async fn execute_with_action(
        &self,
        _tool: &str,
        params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        let key = params
            .get("key")
            .and_then(Value::as_str)
            .ok_or("missing key")?;
        match params.get("op").and_then(Value::as_str) {
            Some("delete") => {
                self.state.delete(key, action_id);
            }
            Some("set_null") => {
                self.state.set(key, Value::Null, action_id);
            }
            other => return Err(format!("unsupported mutation op: {other:?}")),
        }
        Ok(Value::from("mutated"))
    }
}

#[async_trait::async_trait]
impl ToolExecutor for ConcurrentStateExecutor {
    async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
        Err("concurrent state executor requires action identity".to_string())
    }

    async fn execute_with_action(
        &self,
        _tool: &str,
        params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        self.barrier.wait().await;
        let key = params
            .get("key")
            .and_then(Value::as_str)
            .ok_or("missing key")?;
        let value = params.get("value").cloned().ok_or("missing value")?;
        self.state.set(key, value.clone(), action_id);
        Ok(value)
    }
}

fn make_runtime() -> Runtime {
    Runtime::new().with_executor(Arc::new(TestExecutor))
}

async fn setup_runtime() -> Runtime {
    let rt = make_runtime();
    rt.register_tool("add").await;
    rt.register_tool("echo").await;
    rt.register_tool("fail").await;
    rt.register_tool("slow").await;
    rt
}

fn tool_call(tool: &str, params: HashMap<String, Value>) -> Action {
    {
        let mut a = Action::new(ActionType::ToolCall);
        a.id = uuid::Uuid::new_v4().simple().to_string()[..12].to_string();
        a.tool = Some(tool.to_string());
        a.parameters = params;
        a
    }
}

fn state_write(key: &str, value: Value) -> Action {
    {
        let mut a = Action::new(ActionType::StateWrite);
        a.id = uuid::Uuid::new_v4().simple().to_string()[..12].to_string();
        a.parameters = [
            ("key".to_string(), Value::from(key)),
            ("value".to_string(), value),
        ]
        .into();
        a
    }
}

fn proposal(actions: Vec<Action>) -> ActionProposal {
    ActionProposal {
        id: "test-proposal".to_string(),
        source: "test".to_string(),
        actions,
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    }
}

// --- Basic execution ---

#[tokio::test]
async fn test_tool_call_succeeds() {
    let rt = setup_runtime().await;
    let p = proposal(vec![tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    )]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
    assert_eq!(result.results[0].output, Some(Value::from(3)));
}

#[tokio::test]
async fn test_agent_basics_read_file() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("note.txt");
    fs::write(&path, "alpha\nbeta\ngamma\n").unwrap();

    let rt = Runtime::new();
    rt.register_agent_basics().await;

    let result = rt
        .execute(&proposal(vec![tool_call(
            "read_file",
            [("path".to_string(), Value::from(path.display().to_string()))].into(),
        )]))
        .await;

    assert!(result.all_succeeded());
    let output = result.results[0].output.as_ref().unwrap();
    // Content is line-numbered `cat -n` style (H1/F4-remainder, audit 2026-07-06);
    // `total_lines` still reflects the full file.
    assert_eq!(
        output.get("content"),
        Some(&Value::from("     1\talpha\n     2\tbeta\n     3\tgamma"))
    );
    assert_eq!(output.get("total_lines"), Some(&Value::from(3)));
}

#[tokio::test]
async fn test_agent_basics_write_and_edit_file() {
    let dir = tempfile::tempdir().unwrap();
    let file_path = dir.path().join("story.txt");
    let rt = Runtime::new();
    rt.register_agent_basics().await;

    let write = tool_call(
        "write_file",
        [
            (
                "path".to_string(),
                Value::from(file_path.display().to_string()),
            ),
            ("content".to_string(), Value::from("hello world")),
        ]
        .into(),
    );
    let edit = tool_call(
        "edit_file",
        [
            (
                "path".to_string(),
                Value::from(file_path.display().to_string()),
            ),
            ("old_text".to_string(), Value::from("world")),
            ("new_text".to_string(), Value::from("car")),
        ]
        .into(),
    );

    let result = rt.execute(&proposal(vec![write, edit])).await;

    assert!(result.all_succeeded());
    assert_eq!(fs::read_to_string(file_path).unwrap(), "hello car");
}

/// (#1d) On the bare-Runtime path (no configured executor → the agent_basics
/// fallback with the Runtime's own read ledger), an `edit_file` on an existing
/// file the session never read is gated: the action fails, its result carries
/// the read-first error, and the file is untouched.
#[tokio::test]
async fn test_agent_basics_edit_requires_prior_read() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("f.txt");
    fs::write(&path, "hello world").unwrap();

    let rt = Runtime::new();
    rt.register_agent_basics().await;

    let result = rt
        .execute(&proposal(vec![tool_call(
            "edit_file",
            [
                ("path".to_string(), Value::from(path.display().to_string())),
                ("old_text".to_string(), Value::from("hello")),
                ("new_text".to_string(), Value::from("hi")),
            ]
            .into(),
        )]))
        .await;

    assert!(!result.all_succeeded());
    let err = result.results[0]
        .error
        .as_ref()
        .expect("a gated edit must carry an error");
    assert!(err.contains("before editing it"), "{err}");
    assert_eq!(fs::read_to_string(&path).unwrap(), "hello world");
}

#[tokio::test]
async fn test_agent_basics_read_ledger_isolated_by_runtime_session() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("f.txt");
    fs::write(&path, "hello world").unwrap();

    let rt = Runtime::new();
    rt.register_agent_basics().await;
    let first = rt.open_session().await;
    let second = rt.open_session().await;
    let path_s = path.display().to_string();

    let read = rt
        .execute_with_session(
            &proposal(vec![tool_call(
                "read_file",
                [("path".to_string(), Value::from(path_s.clone()))].into(),
            )]),
            &first,
        )
        .await;
    assert!(read.all_succeeded(), "{:?}", read.results[0].error);

    let edit = rt
        .execute_with_session(
            &proposal(vec![tool_call(
                "edit_file",
                [
                    ("path".to_string(), Value::from(path_s)),
                    ("old_text".to_string(), Value::from("hello")),
                    ("new_text".to_string(), Value::from("hi")),
                ]
                .into(),
            )]),
            &second,
        )
        .await;
    assert!(!edit.all_succeeded());
    assert!(edit.results[0]
        .error
        .as_deref()
        .unwrap_or_default()
        .contains("before editing it"));
    assert_eq!(fs::read_to_string(path).unwrap(), "hello world");
}

#[tokio::test]
async fn test_agent_basics_substrate_swap_clears_read_ledger() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("f.txt");
    fs::write(&path, "hello world").unwrap();

    let rt = Runtime::new();
    rt.register_agent_basics().await;
    let path_s = path.display().to_string();
    let read = rt
        .execute(&proposal(vec![tool_call(
            "read_file",
            [("path".to_string(), Value::from(path_s.clone()))].into(),
        )]))
        .await;
    assert!(read.all_succeeded(), "{:?}", read.results[0].error);

    rt.set_substrate(Arc::new(crate::substrate::LocalSubstrate::new()))
        .await;
    let edit = rt
        .execute(&proposal(vec![tool_call(
            "edit_file",
            [
                ("path".to_string(), Value::from(path_s)),
                ("old_text".to_string(), Value::from("hello")),
                ("new_text".to_string(), Value::from("hi")),
            ]
            .into(),
        )]))
        .await;
    assert!(!edit.all_succeeded());
    assert!(edit.results[0]
        .error
        .as_deref()
        .unwrap_or_default()
        .contains("before editing it"));
    assert_eq!(fs::read_to_string(path).unwrap(), "hello world");
}

/// (#6b) A `ToolExecutor` that routes the six builtins through
/// `execute_with_ledger` with its OWN ledger (mirroring coder/assistant/bench).
/// Under `runtime.execute`, a read then edit through this executor SUCCEEDS —
/// proving one ledger serves both, i.e. the Runtime never re-dispatches a
/// builtin to its second ledger (the "unknown tool" fall-through never fires) —
/// while an unread edit surfaces the executor's own gate error.
#[tokio::test]
async fn runtime_routes_builtins_through_single_executor_ledger() {
    struct LedgerExecutor {
        substrate: Arc<dyn crate::substrate::Substrate>,
        ledger: crate::agent_basics::ReadLedger,
    }
    #[async_trait::async_trait]
    impl ToolExecutor for LedgerExecutor {
        async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
            match crate::agent_basics::execute_with_ledger(
                &self.substrate,
                &self.ledger,
                tool,
                params,
            )
            .await
            {
                Some(r) => r,
                None => Err(format!("unknown tool: {tool}")),
            }
        }
    }

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("f.txt");
    fs::write(&path, "alpha beta").unwrap();
    let path_s = path.display().to_string();

    let substrate: Arc<dyn crate::substrate::Substrate> =
        Arc::new(crate::substrate::LocalSubstrate::new());
    let rt = Runtime::new().with_executor(Arc::new(LedgerExecutor {
        substrate: substrate.clone(),
        ledger: crate::agent_basics::ReadLedger::new(),
    }));
    rt.register_agent_basics().await;

    // Read records in the executor's ledger; the subsequent edit is licensed.
    let read = rt
        .execute(&proposal(vec![tool_call(
            "read_file",
            [("path".to_string(), Value::from(path_s.clone()))].into(),
        )]))
        .await;
    assert!(
        read.all_succeeded(),
        "read failed: {:?}",
        read.results[0].error
    );

    let edit = rt
        .execute(&proposal(vec![tool_call(
            "edit_file",
            [
                ("path".to_string(), Value::from(path_s.clone())),
                ("old_text".to_string(), Value::from("alpha")),
                ("new_text".to_string(), Value::from("ALPHA")),
            ]
            .into(),
        )]))
        .await;
    assert!(
        edit.all_succeeded(),
        "read then edit must succeed through one executor ledger: {:?}",
        edit.results[0].error
    );

    // A DIFFERENT, unread file surfaces the executor's gate error verbatim.
    let other = dir.path().join("g.txt");
    fs::write(&other, "gamma").unwrap();
    let unread = rt
        .execute(&proposal(vec![tool_call(
            "edit_file",
            [
                ("path".to_string(), Value::from(other.display().to_string())),
                ("old_text".to_string(), Value::from("gamma")),
                ("new_text".to_string(), Value::from("GAMMA")),
            ]
            .into(),
        )]))
        .await;
    assert!(!unread.all_succeeded());
    let err = unread.results[0]
        .error
        .as_ref()
        .expect("the unread edit must carry the gate error");
    assert!(err.contains("before editing it"), "{err}");
}

#[tokio::test]
async fn test_agent_basics_find_and_grep_files() {
    let dir = tempfile::tempdir().unwrap();
    let src_dir = dir.path().join("src");
    fs::create_dir_all(&src_dir).unwrap();
    fs::write(src_dir.join("lib.rs"), "fn main() {}\n").unwrap();
    fs::write(src_dir.join("data.txt"), "hello car runtime\n").unwrap();

    let rt = Runtime::new();
    rt.register_agent_basics().await;

    let find = tool_call(
        "find_files",
        [
            ("pattern".to_string(), Value::from("*.rs")),
            (
                "path".to_string(),
                Value::from(dir.path().display().to_string()),
            ),
        ]
        .into(),
    );
    let grep = tool_call(
        "grep_files",
        [
            ("pattern".to_string(), Value::from("car")),
            (
                "path".to_string(),
                Value::from(dir.path().display().to_string()),
            ),
        ]
        .into(),
    );

    let result = rt.execute(&proposal(vec![find, grep])).await;

    assert!(result.all_succeeded());
    assert_eq!(
        result.results[0].output.as_ref().unwrap()["count"],
        Value::from(1)
    );
    assert_eq!(
        result.results[1].output.as_ref().unwrap()["count"],
        Value::from(1)
    );
}

#[tokio::test]
async fn test_state_write_and_read() {
    let rt = setup_runtime().await;
    let p = proposal(vec![state_write("x", Value::from(42)), {
        let mut a = state_write("unused", Value::Null);
        a.action_type = ActionType::StateRead;
        a.parameters = [("key".to_string(), Value::from("x"))].into();
        a.state_dependencies = vec!["x".to_string()];
        a
    }]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
    assert_eq!(result.results[1].output, Some(Value::from(42)));
}

#[tokio::test]
async fn test_unknown_tool_rejected() {
    let rt = Runtime::new().with_executor(Arc::new(TestExecutor));
    // Don't register any tools
    let p = proposal(vec![tool_call("nonexistent", HashMap::new())]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
}

#[tokio::test]
async fn test_precondition_blocks() {
    let rt = setup_runtime().await;
    let mut action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    action.preconditions = vec![Precondition {
        key: "auth".to_string(),
        operator: "eq".to_string(),
        value: Value::Bool(true),
        description: String::new(),
    }];
    let result = rt.execute(&proposal(vec![action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
}

#[tokio::test]
async fn test_precondition_passes() {
    let rt = setup_runtime().await;
    rt.state.set("auth", Value::Bool(true), "setup");
    let mut action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    action.preconditions = vec![Precondition {
        key: "auth".to_string(),
        operator: "eq".to_string(),
        value: Value::Bool(true),
        description: String::new(),
    }];
    let result = rt.execute(&proposal(vec![action])).await;
    assert!(result.all_succeeded());
}

// --- Failure behavior ---

#[tokio::test]
async fn test_abort_stops_remaining() {
    let rt = setup_runtime().await;
    let mut fail_action = tool_call("fail", HashMap::new());
    fail_action.id = "abort-first".to_string();
    fail_action.failure_behavior = FailureBehavior::Abort;
    let mut echo_action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    echo_action.id = "abort-skipped".to_string();

    let result = rt.execute(&proposal(vec![fail_action, echo_action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Failed);
    assert_eq!(result.results[1].status, ActionStatus::Skipped);
    let log = rt.log.lock().await;
    let skipped = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::ActionSkipped
                && event.action_id.as_deref() == Some("abort-skipped")
        })
        .expect("downstream abort skip is journaled");
    assert_eq!(skipped.data["attempted"], false);
    assert_eq!(skipped.data["stage"], "dependency_abort");
    assert!(!skipped.data.contains_key("attempt"));
}

#[tokio::test]
async fn test_skip_continues() {
    let rt = setup_runtime().await;
    let mut fail_action = tool_call("fail", HashMap::new());
    fail_action.id = "skip-after-failure".to_string();
    fail_action.failure_behavior = FailureBehavior::Skip;
    let echo_action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());

    let result = rt.execute(&proposal(vec![fail_action, echo_action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Skipped);
    assert_eq!(result.results[1].status, ActionStatus::Succeeded);
    let log = rt.log.lock().await;
    let skipped = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::ActionSkipped
                && event.action_id.as_deref() == Some("skip-after-failure")
        })
        .expect("attempted skip terminal is journaled");
    assert_eq!(skipped.data["attempted"], true);
    assert_eq!(skipped.data["attempt"], 1);
    assert_eq!(skipped.data["stage"], "failure_behavior");
}

// --- Rollback ---

#[tokio::test]
async fn test_abort_rolls_back_state() {
    let rt = setup_runtime().await;
    let mut fail_action = tool_call("fail", HashMap::new());
    fail_action.id = "rollback-failure".to_string();
    fail_action.failure_behavior = FailureBehavior::Abort;

    let mut write = state_write("x", Value::from(1));
    write.id = "rolled-back-write".to_string();

    let p = proposal(vec![write, fail_action]);
    let result = rt.execute(&p).await;

    assert_eq!(rt.state.get("x"), None);
    let rolled_back = result
        .results
        .iter()
        .find(|action| action.action_id == "rolled-back-write")
        .unwrap();
    assert_eq!(rolled_back.status, ActionStatus::Failed);
    assert!(rolled_back.state_changes.is_empty());
    assert!(rolled_back
        .error
        .as_deref()
        .unwrap()
        .contains("external effects may remain"));

    let log = rt.log.lock().await;
    let provisional = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::StateChanged
                && event.action_id.as_deref() == Some("rolled-back-write")
        })
        .expect("provisional state mutation");
    assert_eq!(
        provisional.data["changes_semantics"],
        "provisional_state_mutations"
    );
    let rollback = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::StateRollback)
        .expect("terminal rollback evidence");
    assert_eq!(rollback.data["stage"], "proposal_rollback");
    assert_eq!(rollback.data["attempted"], true);
    assert_eq!(
        rollback.data["affected_actions"],
        serde_json::json!(["rolled-back-write"])
    );
    assert_eq!(
        rollback.data["rolled_back_changes"]["rolled-back-write"]["x"],
        serde_json::json!({"op":"set","value":1})
    );
    assert_eq!(
        rollback.data["changes_semantics"],
        "rolled_back_provisional_state_mutations"
    );
}

#[tokio::test]
async fn scoped_abort_rollback_survives_state_store_reopen() {
    let dir = tempfile::tempdir().unwrap();
    let state_path = dir.path().join("state.jsonl");
    {
        let state = Arc::new(car_state::StateStore::durable(&state_path).unwrap());
        state
            .scoped(Some("acme"))
            .set("existing", Value::from("old"), "setup");
        state
            .scoped(Some("globex"))
            .set("survivor", Value::from("kept"), "setup");
        state.sync().unwrap();

        let rt = Runtime::with_shared(
            state.clone(),
            Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
            Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
        )
        .with_executor(Arc::new(TestExecutor));
        rt.register_tool("fail").await;

        let mut overwrite = state_write("existing", Value::from("new"));
        overwrite.id = "durable-overwrite".to_string();
        let mut create = state_write("created", Value::from(true));
        create.id = "durable-create".to_string();
        let mut fail = tool_call("fail", HashMap::new());
        fail.id = "durable-abort".to_string();
        fail.failure_behavior = FailureBehavior::Abort;
        let scope = crate::scope::RuntimeScope {
            caller_id: Some("alice".to_string()),
            tenant_id: Some("acme".to_string()),
            claims: Default::default(),
        };

        let result = rt
            .execute_scoped(&proposal(vec![overwrite, create, fail]), &scope)
            .await;
        assert!(!result.all_succeeded());
        assert_eq!(
            state.scoped(Some("acme")).get("existing"),
            Some(Value::from("old"))
        );
        assert!(!state.scoped(Some("acme")).exists("created"));
        state.sync().unwrap();
    }

    let reopened = car_state::StateStore::durable(&state_path).unwrap();
    assert_eq!(
        reopened.scoped(Some("acme")).get("existing"),
        Some(Value::from("old"))
    );
    assert!(!reopened.scoped(Some("acme")).exists("created"));
    assert_eq!(
        reopened.scoped(Some("globex")).get("survivor"),
        Some(Value::from("kept")),
        "tenant-scoped abort rollback must preserve unrelated durable state"
    );
}

#[tokio::test]
async fn durable_abort_rollback_failure_preserves_committed_state_and_idempotency() {
    let dir = tempfile::tempdir().unwrap();
    let state_path = dir.path().join("state.jsonl");
    let idempotency_path = dir.path().join("idempotency.jsonl");
    let failures = car_state::RestoreFailureInjector::default();
    let state = Arc::new(
        car_state::StateStore::durable_with_restore_failure_injector(
            &state_path,
            Some(failures.clone()),
        )
        .unwrap(),
    );
    let executor = Arc::new(FallbackRollbackExecutor {
        state: state.clone(),
        effect_dispatches: AtomicU32::new(0),
    });
    let rt = Runtime::with_shared(
        state.clone(),
        Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
        Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
    )
    .with_executor(executor);
    rt.register_tool("partial-effect").await;
    rt.register_tool("partial-skip").await;
    rt.set_idempotency_cache_path(&idempotency_path)
        .await
        .unwrap();

    let mut effect = tool_call("partial-effect", HashMap::new());
    effect.id = "partial-effect".to_string();
    effect.idempotent = true;
    let mut fail = tool_call("partial-skip", HashMap::new());
    fail.id = "partial-skip".to_string();
    fail.failure_behavior = FailureBehavior::Abort;

    failures.fail_next(car_state::RestoreFailurePoint::BeforePublication);
    let result = rt.execute(&proposal(vec![effect, fail])).await;

    let effect_result = result
        .results
        .iter()
        .find(|result| result.action_id == "partial-effect")
        .unwrap();
    assert_eq!(effect_result.status, ActionStatus::Succeeded);
    assert_eq!(state.get("partial_dirty"), Some(Value::from(true)));
    assert!(result.results.iter().any(|result| {
        result.error.as_deref().is_some_and(|error| {
            error.contains("durable state rollback failed before publication")
                && error.contains("idempotency entries were preserved")
        })
    }));

    let log = rt.log.lock().await;
    assert!(!log.events().iter().any(|event| {
        event.kind == car_eventlog::EventKind::StateRollback
            && event.data.get("stage") == Some(&Value::from("proposal_rollback"))
    }));
    assert!(!log.events().iter().any(|event| {
        matches!(
            event.kind,
            car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
        )
    }));
    let failure = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::ActionFailed
                && event.data.get("stage") == Some(&Value::from("proposal_rollback"))
        })
        .expect("failed rollback must record pre-publication failure evidence");
    assert_eq!(failure.data["publication_succeeded"], false);
    assert_eq!(failure.data["rollback_succeeded"], false);
    assert_eq!(failure.data["idempotency_entries_preserved"], true);
    drop(log);

    let journal_lines: Vec<Value> = fs::read_to_string(&idempotency_path)
        .unwrap()
        .lines()
        .map(|line| serde_json::from_str(line).unwrap())
        .collect();
    assert_eq!(
        journal_lines.len(),
        1,
        "rollback failure must not tombstone"
    );
    assert!(journal_lines[0]["result"].is_object());
    state.sync().unwrap();
    drop(rt);
    drop(state);

    let reopened = car_state::StateStore::durable(&state_path).unwrap();
    assert_eq!(reopened.get("partial_dirty"), Some(Value::from(true)));
}

#[tokio::test]
async fn parent_sync_failure_adopts_rollback_state_and_invalidates_idempotency() {
    let dir = tempfile::tempdir().unwrap();
    let state_path = dir.path().join("state.jsonl");
    let idempotency_path = dir.path().join("idempotency.jsonl");
    let failures = car_state::RestoreFailureInjector::default();
    let state = Arc::new(
        car_state::StateStore::durable_with_restore_failure_injector(
            &state_path,
            Some(failures.clone()),
        )
        .unwrap(),
    );
    let rt = Runtime::with_shared(
        state.clone(),
        Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
        Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
    )
    .with_executor(Arc::new(FallbackRollbackExecutor {
        state: state.clone(),
        effect_dispatches: AtomicU32::new(0),
    }));
    rt.register_tool("partial-effect").await;
    rt.register_tool("partial-skip").await;
    rt.set_idempotency_cache_path(&idempotency_path)
        .await
        .unwrap();

    let mut effect = tool_call("partial-effect", HashMap::new());
    effect.id = "partial-effect".to_string();
    effect.idempotent = true;
    let mut fail = tool_call("partial-skip", HashMap::new());
    fail.id = "partial-skip".to_string();
    fail.failure_behavior = FailureBehavior::Abort;
    failures.fail_next(car_state::RestoreFailurePoint::ParentDirectorySync);

    let result = rt.execute(&proposal(vec![effect.clone(), fail])).await;
    assert_eq!(state.get("partial_dirty"), None);
    assert_eq!(
        result
            .results
            .iter()
            .find(|result| result.action_id == "partial-effect")
            .unwrap()
            .status,
        ActionStatus::Failed
    );
    assert!(
        result.results.iter().any(|result| {
            result.error.as_deref().is_some_and(|error| {
                error.contains("parent-directory durability is unknown")
                    && error.contains("idempotency entries were invalidated")
            })
        }),
        "durability-unknown result evidence missing: {result:?}"
    );

    let log = rt.log.lock().await;
    let rollback = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::StateRollback
                && event.data.get("stage") == Some(&Value::from("proposal_rollback"))
        })
        .expect("published rollback must have explicit durability-unknown evidence");
    assert_eq!(rollback.data["publication_succeeded"], true);
    assert_eq!(rollback.data["rollback_succeeded"], true);
    assert_eq!(rollback.data["durability_unknown"], true);
    drop(log);

    let journal_lines: Vec<Value> = fs::read_to_string(&idempotency_path)
        .unwrap()
        .lines()
        .map(|line| serde_json::from_str(line).unwrap())
        .collect();
    assert_eq!(journal_lines.len(), 2, "insert plus rollback tombstone");
    assert!(journal_lines[0]["result"].is_object());
    assert!(journal_lines[1]["result"].is_null());

    state.set("after_unknown", Value::from("safe"), "later");
    state.sync().unwrap();
    drop(rt);
    drop(state);

    let reopened = Arc::new(car_state::StateStore::durable(&state_path).unwrap());
    assert_eq!(reopened.get("partial_dirty"), None);
    assert_eq!(reopened.get("after_unknown"), Some(Value::from("safe")));
    let retry_executor = Arc::new(FallbackRollbackExecutor {
        state: reopened.clone(),
        effect_dispatches: AtomicU32::new(0),
    });
    let retry_rt = Runtime::with_shared(
        reopened,
        Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
        Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
    )
    .with_executor(retry_executor.clone());
    retry_rt.register_tool("partial-effect").await;
    retry_rt
        .set_idempotency_cache_path(&idempotency_path)
        .await
        .unwrap();
    let retry = ActionProposal {
        id: "parent-sync-retry".to_string(),
        source: "test".to_string(),
        actions: vec![effect],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };
    assert!(retry_rt.execute(&retry).await.all_succeeded());
    assert_eq!(retry_executor.effect_dispatches.load(Ordering::SeqCst), 1);
}

// --- Timeout ---

#[tokio::test]
async fn test_timeout() {
    let rt = setup_runtime().await;
    let mut action = tool_call("slow", HashMap::new());
    action.timeout_ms = Some(50);
    let result = rt.execute(&proposal(vec![action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Failed);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("timed out"));
}

// --- Idempotency ---

#[tokio::test]
async fn test_idempotent_cached() {
    let rt = setup_runtime().await;
    let mut action1 = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    );
    action1.idempotent = true;

    rt.execute(&proposal(vec![action1.clone()])).await;

    let mut action2 = action1.clone();
    action2.id = "second".to_string();
    let result = rt.execute(&proposal(vec![action2])).await;

    assert_eq!(result.results[0].output, Some(Value::from(3)));
    assert_eq!(result.results[0].duration_ms, Some(0.0));
}

#[tokio::test]
async fn deduplicated_result_does_not_reclaim_historical_state_mutations() {
    let rt = setup_runtime().await;
    let mut first_action = state_write("cached-key", Value::from(1));
    first_action.id = "cached-original".to_string();
    first_action.idempotent = true;
    let first = rt
        .execute(&ActionProposal {
            id: "cached-first-proposal".to_string(),
            ..proposal(vec![first_action.clone()])
        })
        .await;
    assert_eq!(
        first.results[0].state_changes["cached-key"],
        StateMutation::Set {
            value: Value::from(1),
        }
        .encode()
    );

    first_action.id = "cached-replay".to_string();
    let replay = rt
        .execute(&ActionProposal {
            id: "cached-second-proposal".to_string(),
            ..proposal(vec![first_action])
        })
        .await;
    assert!(replay.results[0].state_changes.is_empty());
    let log = rt.log.lock().await;
    let committed = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::StateCommitted
                && event.proposal_id.as_deref() == Some("cached-second-proposal")
        })
        .expect("deduplicated proposal commit boundary");
    assert_eq!(committed.data["affected_actions"], serde_json::json!([]));
    assert_eq!(committed.data["committed_changes"], serde_json::json!({}));
}

// --- Retry ---

#[tokio::test]
async fn test_retry_succeeds() {
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct FlakyExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for FlakyExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            let n = self.0.fetch_add(1, Ordering::SeqCst) + 1;
            if n < 3 {
                Err("not yet".to_string())
            } else {
                Ok(Value::from("ok"))
            }
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(FlakyExecutor(cc)));
    rt.register_tool("flaky").await;

    let mut action = tool_call("flaky", HashMap::new());
    action.id = "retry-action".to_string();
    action.failure_behavior = FailureBehavior::Retry;
    action.max_retries = 3;

    let result = rt.execute(&proposal(vec![action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Succeeded);
    assert_eq!(call_count.load(Ordering::SeqCst), 3);

    let log = rt.log.lock().await;
    let attempts = |kind| {
        log.events()
            .iter()
            .filter(|event| {
                event.kind == kind && event.action_id.as_deref() == Some("retry-action")
            })
            .map(|event| event.data["attempt"].as_u64())
            .collect::<Vec<_>>()
    };
    assert_eq!(
        attempts(car_eventlog::EventKind::ActionExecuting),
        vec![Some(1), Some(2), Some(3)]
    );
    assert_eq!(
        attempts(car_eventlog::EventKind::ActionFailed),
        vec![Some(1), Some(2)]
    );
    assert_eq!(
        attempts(car_eventlog::EventKind::ActionRetrying),
        vec![Some(2), Some(3)]
    );
    assert_eq!(
        attempts(car_eventlog::EventKind::ActionSucceeded),
        vec![Some(3)]
    );
}

#[tokio::test]
async fn retry_overflow_is_rejected_before_proposal_receipt_or_dispatch() {
    struct CountingExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(Value::from("must not run"))
        }
    }

    let calls = Arc::new(AtomicU32::new(0));
    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor(calls.clone())));
    rt.register_tool("count").await;
    let mut action = tool_call("count", HashMap::new());
    action.id = "overflow-retry".to_string();
    action.failure_behavior = FailureBehavior::Retry;
    action.max_retries = u32::MAX;
    let proposal = ActionProposal {
        id: "overflow-proposal".to_string(),
        ..proposal(vec![action])
    };

    let result = rt.execute(&proposal).await;

    assert_eq!(calls.load(Ordering::SeqCst), 0);
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_deref()
        .unwrap()
        .contains("retry attempt count overflows u32"));
    assert_eq!(result.final_proposal.as_ref(), Some(&proposal));
    let log = rt.log.lock().await;
    assert!(log
        .events()
        .iter()
        .all(|event| event.kind != car_eventlog::EventKind::ProposalReceived));
    let boundaries: Vec<_> = log
        .events()
        .iter()
        .filter(|event| {
            matches!(
                event.kind,
                car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
            ) && event.proposal_id.as_deref() == Some("overflow-proposal")
        })
        .collect();
    assert_eq!(boundaries.len(), 1);
    assert_eq!(boundaries[0].kind, car_eventlog::EventKind::StateRollback);
    assert_eq!(boundaries[0].data["attempted"], false);
    assert_eq!(boundaries[0].data["stage"], "proposal_rejection");
}

#[tokio::test]
async fn non_jcs_tool_output_fails_before_success_or_state_commit() {
    struct ExternalEffectExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for ExternalEffectExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(Value::from(9_007_199_254_740_992_u64))
        }
    }

    let effects = Arc::new(AtomicU32::new(0));
    let rt = Runtime::new().with_executor(Arc::new(ExternalEffectExecutor(effects.clone())));
    rt.register_tool("unsafe_output").await;
    let mut action = tool_call("unsafe_output", HashMap::new());
    action.id = "unsafe-output-action".to_string();
    action.max_retries = 0;
    action
        .expected_effects
        .insert("must_not_commit".to_string(), Value::from(true));

    let result = rt.execute(&proposal(vec![action])).await;

    assert_eq!(effects.load(Ordering::SeqCst), 1);
    assert_eq!(result.results[0].status, ActionStatus::Failed);
    assert!(result.results[0].state_changes.is_empty());
    let error = result.results[0].error.as_deref().unwrap();
    assert!(error.contains("tool output failed JCS/I-JSON validation"));
    assert!(error.contains("external effect may have occurred and was not undone"));
    assert_eq!(rt.state.get("must_not_commit"), None);

    let log = rt.log.lock().await;
    assert!(log.events().iter().all(|event| {
        !(event.kind == car_eventlog::EventKind::ActionSucceeded
            && event.action_id.as_deref() == Some("unsafe-output-action"))
    }));
    assert!(log
        .events()
        .iter()
        .all(|event| event.kind != car_eventlog::EventKind::StateCommitted));
    let failed = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::ActionFailed
                && event.action_id.as_deref() == Some("unsafe-output-action")
        })
        .expect("deterministic terminal failure evidence");
    assert_eq!(failed.data["attempt"], 1);
    assert_eq!(failed.data["attempted"], true);
    assert_eq!(failed.data["stage"], "output_validation");
    assert_eq!(
        failed.data["external_effect_status"],
        "may_have_occurred_not_undone"
    );
}

#[tokio::test]
async fn proposal_journal_authenticates_plan_and_separates_runtime_state() {
    let rt = setup_runtime().await;

    let mut prepare = state_write("runtime_input", Value::from(7));
    prepare.id = "prepare".to_string();
    prepare.expected_effects = [
        ("runtime_input".to_string(), Value::from(7)),
        ("declared_input".to_string(), Value::from(7)),
    ]
    .into();

    let mut report = tool_call(
        "echo",
        [("message".to_string(), Value::from("ready"))].into(),
    );
    report.id = "report".to_string();
    report.state_dependencies = vec!["runtime_input".to_string()];
    report.preconditions = vec![Precondition {
        key: "runtime_input".to_string(),
        operator: "eq".to_string(),
        value: Value::from(7),
        description: "prepare supplied the input".to_string(),
    }];

    let proposal = ActionProposal {
        id: "journal-dag".to_string(),
        source: "journal-contract-test".to_string(),
        actions: vec![prepare, report],
        timestamp: chrono::DateTime::parse_from_rfc3339("2026-08-28T10:00:00Z")
            .unwrap()
            .with_timezone(&chrono::Utc),
        context: [("edition".to_string(), Value::from("daily"))].into(),
    };

    let result = rt.execute(&proposal).await;
    assert!(result.all_succeeded(), "DAG should execute: {result:?}");

    let expected_preimage = serde_json::to_value(&proposal).unwrap();
    let canonical = car_inference::catalog_identity::canonical_json(&expected_preimage).unwrap();
    let expected_digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
    let log = rt.log.lock().await;
    let received = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::ProposalReceived)
        .expect("proposal_received");
    assert_eq!(received.proposal_id.as_deref(), Some("journal-dag"));
    assert_eq!(received.data["proposal"], expected_preimage);
    assert_eq!(received.data["proposal_digest"], expected_digest);
    assert_eq!(result.original_proposal_id, "journal-dag");
    assert_eq!(result.proposal_id, "journal-dag");
    assert_eq!(result.final_proposal.as_ref(), Some(&proposal));
    assert_eq!(result.replan_lineage.len(), 1);
    assert_eq!(result.replan_lineage[0].generation, 0);
    assert_eq!(
        result.replan_lineage[0].status,
        car_ir::ProposalLineageStatus::Accepted
    );
    assert_eq!(
        result.replan_lineage[0].proposal_digest.as_deref(),
        Some(expected_digest.as_str())
    );

    let state_event = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::StateChanged
                && event.action_id.as_deref() == Some("prepare")
        })
        .expect("state_changed for prepare");
    assert_eq!(
        state_event.data["declared_expected_effects"],
        serde_json::json!({"declared_input": 7, "runtime_input": 7})
    );
    assert_eq!(
        state_event.data["runtime_state_mutations"],
        serde_json::json!({"runtime_input": {"op": "set", "value": 7}})
    );
    assert_eq!(
        state_event.data["changes"],
        serde_json::json!({"runtime_input": {"op": "set", "value": 7}})
    );
    assert_eq!(
        state_event.data["changes_semantics"], "provisional_state_mutations",
        "the compatibility field must explicitly name its meaning"
    );
    assert_eq!(rt.state.get("declared_input"), None);
    let committed = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::StateCommitted)
        .expect("proposal-level commit evidence");
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| {
                matches!(
                    event.kind,
                    car_eventlog::EventKind::StateCommitted
                        | car_eventlog::EventKind::StateRollback
                ) && event.proposal_id.as_deref() == Some("journal-dag")
            })
            .count(),
        1,
        "one proposal generation must have exactly one aggregate state boundary"
    );
    assert_eq!(committed.data["stage"], "proposal_commit");
    assert_eq!(committed.data["attempted"], true);
    assert_eq!(committed.data["proposal_digest"], expected_digest);
    assert_eq!(
        committed.data["affected_actions"],
        serde_json::json!(["prepare"])
    );
    assert_eq!(
        committed.data["committed_changes"]["prepare"]["runtime_input"],
        serde_json::json!({"op": "set", "value": 7})
    );

    let prepare_result = result
        .results
        .iter()
        .find(|action| action.action_id == "prepare")
        .unwrap();
    assert_eq!(
        prepare_result.state_changes,
        [(
            "runtime_input".to_string(),
            StateMutation::Set {
                value: Value::from(7),
            }
            .encode(),
        )]
        .into()
    );
}

#[tokio::test]
async fn concurrent_actions_attribute_only_their_own_runtime_state_changes() {
    let rt = Runtime::new();
    let executor = Arc::new(ConcurrentStateExecutor {
        state: rt.state.clone(),
        barrier: tokio::sync::Barrier::new(2),
    });
    let rt = rt.with_executor(executor);
    rt.register_tool("mutate").await;

    let mut first = tool_call(
        "mutate",
        [
            ("key".to_string(), Value::from("first_key")),
            ("value".to_string(), Value::from(1)),
        ]
        .into(),
    );
    first.id = "first_action".to_string();
    first.failure_behavior = FailureBehavior::Skip;
    let mut second = tool_call(
        "mutate",
        [
            ("key".to_string(), Value::from("second_key")),
            ("value".to_string(), Value::from(2)),
        ]
        .into(),
    );
    second.id = "second_action".to_string();
    second.failure_behavior = FailureBehavior::Skip;

    let result = rt.execute(&proposal(vec![first, second])).await;
    assert!(
        result.all_succeeded(),
        "parallel actions should run: {result:?}"
    );

    let log = rt.log.lock().await;
    for (action_id, expected) in [
        (
            "first_action",
            serde_json::json!({"first_key": {"op": "set", "value": 1}}),
        ),
        (
            "second_action",
            serde_json::json!({"second_key": {"op": "set", "value": 2}}),
        ),
    ] {
        let state_event = log
            .events()
            .iter()
            .find(|event| {
                event.kind == car_eventlog::EventKind::StateChanged
                    && event.action_id.as_deref() == Some(action_id)
            })
            .unwrap_or_else(|| panic!("state_changed for {action_id}"));
        assert_eq!(
            state_event.data["runtime_state_mutations"], expected,
            "a same-level sibling's transition must not be attributed to {action_id}"
        );
        assert_eq!(
            state_event.data["changes"], expected,
            "the compatibility projection must keep the same action boundary"
        );
    }
}

#[tokio::test]
async fn concurrent_proposals_reusing_ids_cannot_cross_attribute_transitions() {
    let state = Arc::new(car_state::StateStore::new());
    let executor = Arc::new(CrossProposalBarrierExecutor {
        state: state.clone(),
        calls: AtomicU32::new(0),
        first_entered: tokio::sync::Notify::new(),
        second_entered: tokio::sync::Notify::new(),
        release_first: tokio::sync::Notify::new(),
    });
    let runtime = || {
        Runtime::with_shared(
            state.clone(),
            Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
            Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
        )
        .with_executor(executor.clone())
    };
    let first_runtime = Arc::new(runtime());
    let second_runtime = Arc::new(runtime());
    first_runtime.register_tool("mutate").await;
    second_runtime.register_tool("mutate").await;

    let action = |key: &str, value: i64| {
        let mut action = tool_call(
            "mutate",
            [
                ("key".to_string(), Value::from(key)),
                ("value".to_string(), Value::from(value)),
            ]
            .into(),
        );
        action.id = "reused-action-id".to_string();
        action.failure_behavior = FailureBehavior::Skip;
        action
    };
    let first = ActionProposal {
        id: "concurrent-proposal-one".to_string(),
        ..proposal(vec![action("first_key", 1)])
    };
    let second = ActionProposal {
        id: "concurrent-proposal-two".to_string(),
        ..proposal(vec![action("second_key", 2)])
    };

    let first_task = tokio::spawn(async move { first_runtime.execute(&first).await });
    executor.first_entered.notified().await;
    let second_task = tokio::spawn(async move { second_runtime.execute(&second).await });

    // Under the unsafe implementation proposal two reaches dispatch while
    // proposal one is paused. A proposal-execution serialization boundary
    // deliberately keeps it outside; release proposal one after the short
    // observation window so either implementation completes deterministically.
    let second_entered_early = tokio::time::timeout(
        std::time::Duration::from_millis(100),
        executor.second_entered.notified(),
    )
    .await
    .is_ok();
    executor.release_first.notify_one();

    let first_result = first_task.await.unwrap();
    let second_result = second_task.await.unwrap();
    assert!(first_result.all_succeeded(), "{first_result:?}");
    assert!(second_result.all_succeeded(), "{second_result:?}");
    assert_eq!(
        first_result.results[0].state_changes,
        [(
            "first_key".to_string(),
            StateMutation::Set {
                value: Value::from(1),
            }
            .encode(),
        )]
        .into(),
        "proposal one must not claim proposal two's same-ID transition; second_entered_early={second_entered_early}"
    );
    assert_eq!(
        second_result.results[0].state_changes,
        [(
            "second_key".to_string(),
            StateMutation::Set {
                value: Value::from(2),
            }
            .encode()
        )]
        .into(),
        "proposal two must retain its own exact transition"
    );
}

#[tokio::test]
async fn deletion_and_set_null_have_distinct_result_and_event_mutations() {
    let rt = Runtime::new();
    rt.state.set("delete_me", Value::from("present"), "setup");
    let state = rt.state.clone();
    let rt = rt.with_executor(Arc::new(RuntimeMutationExecutor { state }));
    rt.register_tool("mutate").await;

    let mutation = |id: &str, key: &str, op: &str| {
        let mut action = tool_call(
            "mutate",
            [
                ("key".to_string(), Value::from(key)),
                ("op".to_string(), Value::from(op)),
            ]
            .into(),
        );
        action.id = id.to_string();
        action
    };
    let delete_result = rt
        .execute(&ActionProposal {
            id: "delete-proposal".to_string(),
            ..proposal(vec![mutation("delete-action", "delete_me", "delete")])
        })
        .await;
    let null_result = rt
        .execute(&ActionProposal {
            id: "null-proposal".to_string(),
            ..proposal(vec![mutation("null-action", "null_key", "set_null")])
        })
        .await;

    assert_eq!(
        delete_result.results[0].state_changes["delete_me"],
        StateMutation::Delete.encode()
    );
    assert_eq!(
        null_result.results[0].state_changes["null_key"],
        StateMutation::Set { value: Value::Null }.encode()
    );
    assert_eq!(rt.state.get("delete_me"), None);
    assert_eq!(rt.state.get("null_key"), Some(Value::Null));

    let log = rt.log.lock().await;
    let mutation_for = |action_id: &str, key: &str| {
        log.events()
            .iter()
            .find(|event| {
                event.kind == car_eventlog::EventKind::StateChanged
                    && event.action_id.as_deref() == Some(action_id)
            })
            .unwrap_or_else(|| panic!("missing state event for {action_id}"))
            .data["runtime_state_mutations"][key]
            .clone()
    };
    assert_eq!(
        mutation_for("delete-action", "delete_me"),
        serde_json::json!({"op":"delete"})
    );
    assert_eq!(
        mutation_for("null-action", "null_key"),
        serde_json::json!({"op":"set","value":null})
    );
}

#[tokio::test]
async fn duplicate_same_level_action_ids_fail_closed_before_any_side_effect() {
    let rt = setup_runtime().await;
    let mut first = state_write("first_key", Value::from(1));
    first.id = "duplicate-action".to_string();
    first.failure_behavior = FailureBehavior::Skip;
    let mut second = state_write("second_key", Value::from(2));
    second.id = "duplicate-action".to_string();
    second.failure_behavior = FailureBehavior::Skip;

    let result = rt.execute(&proposal(vec![first, second])).await;
    assert!(
        result
            .results
            .iter()
            .all(|action| action.status == ActionStatus::Rejected),
        "duplicate action identity must reject the whole proposal: {result:?}"
    );
    assert!(result.results.iter().all(|action| action
        .error
        .as_deref()
        .unwrap()
        .contains("duplicate action id 'duplicate-action'")));
    assert!(
        rt.state.snapshot().is_empty(),
        "no state mutation is admitted"
    );
    let log = rt.log.lock().await;
    assert!(log.events().iter().all(|event| {
        event.kind != car_eventlog::EventKind::ProposalReceived
            && event.kind != car_eventlog::EventKind::ActionExecuting
    }));
    let boundaries: Vec<_> = log
        .events()
        .iter()
        .filter(|event| {
            matches!(
                event.kind,
                car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
            )
        })
        .collect();
    assert_eq!(boundaries.len(), 1);
    assert_eq!(boundaries[0].kind, car_eventlog::EventKind::StateRollback);
    assert_eq!(boundaries[0].data["attempted"], false);
}

#[tokio::test]
async fn duplicate_writer_ids_fail_closed_before_dependent_dag_execution() {
    let rt = setup_runtime().await;
    let mut first_writer = state_write("first_dependency", Value::from(1));
    first_writer.id = "duplicate-writer".to_string();
    first_writer.failure_behavior = FailureBehavior::Skip;
    let mut second_writer = state_write("second_dependency", Value::from(2));
    second_writer.id = "duplicate-writer".to_string();
    second_writer.failure_behavior = FailureBehavior::Skip;
    let mut dependent = Action::new(ActionType::StateRead);
    dependent.id = "dependent-reader".to_string();
    dependent.parameters = [("key".to_string(), Value::from("first_dependency"))].into();
    dependent.state_dependencies = vec![
        "first_dependency".to_string(),
        "second_dependency".to_string(),
    ];

    let result = rt
        .execute(&proposal(vec![first_writer, second_writer, dependent]))
        .await;
    assert!(
        result
            .results
            .iter()
            .all(|action| action.status == ActionStatus::Rejected),
        "duplicate writer identity must reject before dependency scheduling: {result:?}"
    );
    assert!(rt.state.snapshot().is_empty(), "neither writer may execute");
    let log = rt.log.lock().await;
    assert!(log.events().iter().all(|event| {
        event.kind != car_eventlog::EventKind::ProposalReceived
            && event.kind != car_eventlog::EventKind::ActionExecuting
    }));
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| event.kind == car_eventlog::EventKind::StateRollback)
            .count(),
        1,
        "invalid proposal has one explicit no-commit boundary"
    );
}

#[tokio::test]
async fn proposal_identity_rejects_non_interoperable_numbers_before_execution() {
    let rt = setup_runtime().await;
    let mut invalid = proposal(vec![state_write("must_not_change", Value::from(1))]);
    invalid.context.insert(
        "unsafe_integer".to_string(),
        Value::from(9_007_199_254_740_992_u64),
    );

    let result = rt.execute(&invalid).await;
    assert_eq!(result.results.len(), 1);
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_deref()
        .unwrap()
        .contains("RFC 8785 canonicalization failed"));
    assert_eq!(result.replan_lineage.len(), 1);
    assert_eq!(
        result.replan_lineage[0].status,
        car_ir::ProposalLineageStatus::Rejected
    );
    assert!(result.replan_lineage[0].proposal_digest.is_none());
    assert_eq!(
        result.replan_lineage[0].rejection_reason,
        result.results[0].error
    );
    assert_eq!(rt.state.get("must_not_change"), None);
    let log = rt.log.lock().await;
    assert_eq!(result.final_proposal.as_ref(), Some(&invalid));
    assert!(log
        .events()
        .iter()
        .all(|event| event.kind != car_eventlog::EventKind::ProposalReceived));
    let boundary = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::StateRollback)
        .expect("JCS rejection no-commit boundary");
    assert_eq!(boundary.data["attempted"], false);
    assert_eq!(
        boundary.data["proposal"],
        serde_json::to_value(&invalid).unwrap()
    );
    assert!(!boundary.data.contains_key("proposal_digest"));
}

#[tokio::test]
async fn jcs_impossible_duplicate_proposal_has_truthful_undigested_lineage() {
    let rt = setup_runtime().await;
    let mut action = state_write("must_not_change", Value::from(1));
    action.id = "duplicate".to_string();
    let mut invalid = proposal(vec![action.clone(), action]);
    invalid.context.insert(
        "unsafe_integer".to_string(),
        Value::from(9_007_199_254_740_992_u64),
    );

    let result = rt.execute(&invalid).await;
    assert_eq!(result.replan_lineage.len(), 1);
    let lineage = &result.replan_lineage[0];
    assert_eq!(lineage.status, car_ir::ProposalLineageStatus::Rejected);
    assert!(lineage.proposal_digest.is_none());
    assert!(lineage
        .rejection_reason
        .as_deref()
        .unwrap()
        .starts_with("RFC 8785 canonicalization failed"));
    assert!(serde_json::to_value(&result).is_ok());
    assert_eq!(result.final_proposal.as_ref(), Some(&invalid));
    let log = rt.log.lock().await;
    assert!(log
        .events()
        .iter()
        .all(|event| event.kind != car_eventlog::EventKind::ProposalReceived));
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| event.kind == car_eventlog::EventKind::StateRollback)
            .count(),
        1
    );
}

// --- Policy ---

#[tokio::test]
async fn test_policy_blocks() {
    let rt = setup_runtime().await;
    {
        let mut policies = rt.policies.write().await;
        policies.register(
            "no_echo",
            Box::new(|action, _state| {
                if action.tool.as_deref() == Some("echo") {
                    Some("echo forbidden".to_string())
                } else {
                    None
                }
            }),
            "",
        );
    }

    let action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    let result = rt.execute(&proposal(vec![action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("forbidden"));
}

// --- Session-scoped policies ---
//
// Hosts that multiplex multiple agent sessions (IDE per-project rules,
// multi-tenant servers) over one Runtime register policies under a
// session id; validation walks both global and session layers and
// both must pass. Closing the session drops its registry.
// docs/proposals/per-session-policy-scoping.md

/// Parslee-ai/car#623 — a *global* policy had no removal path. Session-scoped
/// ones could always be dropped wholesale by `close_session`, but a global rule
/// lived until the process exited, so a mistyped or over-broad one could only be
/// cleared by restarting the daemon and losing all other in-memory state.
#[tokio::test]
async fn test_unregister_policy_stops_enforcement() {
    let rt = setup_runtime().await;
    let echo = || {
        proposal(vec![tool_call(
            "echo",
            [("message".to_string(), Value::from("hi"))].into(),
        )])
    };

    {
        let mut policies = rt.policies.write().await;
        policies.register(
            "no_echo",
            Box::new(|action, _state| {
                if action.tool.as_deref() == Some("echo") {
                    Some("echo forbidden".to_string())
                } else {
                    None
                }
            }),
            "global echo deny",
        );
    }
    let denied = rt.execute(&echo()).await;
    assert_eq!(denied.results[0].status, ActionStatus::Rejected);

    assert_eq!(
        rt.list_policies(None).await.unwrap(),
        vec![("no_echo".to_string(), "global echo deny".to_string())]
    );

    assert_eq!(rt.unregister_policy("no_echo", None).await.unwrap(), 1);
    assert!(rt.list_policies(None).await.unwrap().is_empty());

    let allowed = rt.execute(&echo()).await;
    assert_eq!(
        allowed.results[0].status,
        ActionStatus::Succeeded,
        "unregistering the policy must actually stop enforcement"
    );

    // Removing something that isn't there reports 0 rather than erroring, so
    // cleanup can run unconditionally.
    assert_eq!(rt.unregister_policy("no_echo", None).await.unwrap(), 0);
}

/// Unregistering inside a session must not touch the global set, and an unknown
/// session id is an error rather than a silent no-op.
#[tokio::test]
async fn test_unregister_policy_is_scope_aware() {
    let rt = setup_runtime().await;
    let session_id = rt.open_session().await;
    {
        let mut policies = rt.policies.write().await;
        policies.register("shared_name", Box::new(|_, _| None), "global one");
    }
    rt.register_policy_in_session(
        &session_id,
        "shared_name",
        Box::new(|_, _| None),
        "session one",
    )
    .await
    .unwrap();

    assert_eq!(
        rt.unregister_policy("shared_name", Some(&session_id))
            .await
            .unwrap(),
        1
    );
    assert!(rt
        .list_policies(Some(&session_id))
        .await
        .unwrap()
        .is_empty());
    assert_eq!(
        rt.list_policies(None).await.unwrap().len(),
        1,
        "removing a session policy must leave the global one alone"
    );

    assert!(rt.unregister_policy("x", Some("ghost")).await.is_err());
    assert!(rt.list_policies(Some("ghost")).await.is_err());
}

#[tokio::test]
async fn test_session_policy_denies_what_global_allows() {
    let rt = setup_runtime().await;
    let session_id = rt.open_session().await;
    rt.register_policy_in_session(
        &session_id,
        "no_echo_in_session",
        Box::new(|action, _state| {
            if action.tool.as_deref() == Some("echo") {
                Some("echo forbidden in this session".to_string())
            } else {
                None
            }
        }),
        "Session-scoped echo deny",
    )
    .await
    .unwrap();

    // Outside the session, echo runs (no global policy).
    let global = rt
        .execute(&proposal(vec![tool_call(
            "echo",
            [("message".to_string(), Value::from("hi"))].into(),
        )]))
        .await;
    assert_eq!(global.results[0].status, ActionStatus::Succeeded);

    // Inside the session, the same action is rejected.
    let scoped = rt
        .execute_with_session(
            &proposal(vec![tool_call(
                "echo",
                [("message".to_string(), Value::from("hi"))].into(),
            )]),
            &session_id,
        )
        .await;
    assert_eq!(scoped.results[0].status, ActionStatus::Rejected);
    assert!(scoped.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("forbidden in this session"));
}

#[tokio::test]
async fn test_global_policy_still_enforced_under_session() {
    let rt = setup_runtime().await;
    {
        let mut policies = rt.policies.write().await;
        policies.register(
            "global_no_fail",
            Box::new(|action, _state| {
                if action.tool.as_deref() == Some("fail") {
                    Some("fail tool blocked globally".to_string())
                } else {
                    None
                }
            }),
            "Global block on fail",
        );
    }
    let session_id = rt.open_session().await;
    // Even though no session policy mentions fail, the global one
    // still applies under the session — sessions add to the deny
    // list, they don't bypass it.
    let result = rt
        .execute_with_session(
            &proposal(vec![tool_call("fail", HashMap::new())]),
            &session_id,
        )
        .await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("blocked globally"));
}

#[tokio::test]
async fn test_close_session_drops_its_policies() {
    let rt = setup_runtime().await;
    let session_id = rt.open_session().await;
    rt.register_policy_in_session(
        &session_id,
        "no_add",
        Box::new(|action, _state| {
            if action.tool.as_deref() == Some("add") {
                Some("add denied in session".to_string())
            } else {
                None
            }
        }),
        "",
    )
    .await
    .unwrap();

    // Confirm the policy bites while the session is live.
    let denied = rt
        .execute_with_session(
            &proposal(vec![tool_call(
                "add",
                [
                    ("a".to_string(), Value::from(1)),
                    ("b".to_string(), Value::from(2)),
                ]
                .into(),
            )]),
            &session_id,
        )
        .await;
    assert_eq!(denied.results[0].status, ActionStatus::Rejected);

    // Close — the session and its registry vanish.
    let removed = rt.close_session(&session_id).await;
    assert!(removed);
    assert!(!rt.session_exists(&session_id).await);

    // Idempotent close: second call returns false but doesn't panic.
    assert!(!rt.close_session(&session_id).await);

    // Without the session, the same proposal runs against global only —
    // and there are no global policies, so it succeeds.
    let allowed = rt
        .execute(&proposal(vec![tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        )]))
        .await;
    assert_eq!(allowed.results[0].status, ActionStatus::Succeeded);
}

#[tokio::test]
async fn test_unknown_session_id_rejects_actions() {
    // Hosts that pass a stale or never-opened id should NOT silently
    // get the global-only path — that would let a typo escape the
    // session boundary they thought they were inside. The action
    // rejects with a clear message.
    let rt = setup_runtime().await;
    let result = rt
        .execute_with_session(
            &proposal(vec![tool_call(
                "echo",
                [("message".to_string(), Value::from("x"))].into(),
            )]),
            "no-such-session",
        )
        .await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("unknown session id"));
}

#[tokio::test]
async fn test_register_in_unknown_session_errors() {
    let rt = setup_runtime().await;
    let err = rt
        .register_policy_in_session("ghost", "p", Box::new(|_, _| None), "")
        .await
        .expect_err("should refuse to register against unknown session");
    assert!(err.contains("unknown session id"));
}

#[tokio::test]
async fn test_global_only_path_unchanged_when_no_session_used() {
    // Backwards-compat: code paths that never call open_session must
    // observe identical behaviour to before this change.
    let rt = setup_runtime().await;
    let action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    let result = rt.execute(&proposal(vec![action])).await;
    assert_eq!(result.results[0].status, ActionStatus::Succeeded);
}

// --- Expected effects ---

#[tokio::test]
async fn declared_expected_effects_are_assertions_not_runtime_writes() {
    let rt = setup_runtime().await;
    let mut action = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    );
    action.expected_effects = [("result".to_string(), Value::from(3))].into();

    let result = rt.execute(&proposal(vec![action])).await;
    assert!(result.all_succeeded());
    assert_eq!(rt.state.get("result"), None);
    assert!(result.results[0].state_changes.is_empty());
    let log = rt.log.lock().await;
    let declaration = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::StateChanged)
        .expect("declared assertion remains auditable");
    assert_eq!(
        declaration.data["declared_expected_effects"],
        serde_json::json!({"result": 3})
    );
    assert_eq!(
        declaration.data["runtime_state_mutations"],
        serde_json::json!({})
    );
    assert_eq!(declaration.data["changes"], serde_json::json!({}));
}

// --- Cost tracking ---

#[tokio::test]
async fn test_cost_summary_computed() {
    let rt = setup_runtime().await;
    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        state_write("x", Value::from(42)),
        tool_call("echo", [("message".to_string(), Value::from("hi"))].into()),
    ]);
    let result = rt.execute(&p).await;

    assert!(result.all_succeeded());
    assert_eq!(result.cost.tool_calls, 2);
    assert_eq!(result.cost.actions_executed, 3);
    assert_eq!(result.cost.actions_skipped, 0);
    assert!(result.cost.total_duration_ms > 0.0);
    assert_eq!(result.cost.retries, 0);
}

#[tokio::test]
async fn test_cost_summary_with_skipped() {
    let rt = setup_runtime().await;
    let mut fail_action = tool_call("fail", HashMap::new());
    fail_action.failure_behavior = FailureBehavior::Abort;
    let echo_action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());

    let result = rt.execute(&proposal(vec![fail_action, echo_action])).await;

    // First action failed (executed), second was skipped
    assert_eq!(result.cost.actions_executed, 1);
    assert_eq!(result.cost.actions_skipped, 1);
    assert_eq!(result.cost.tool_calls, 0); // none succeeded
}

/// Parslee-ai/car#624 — a rejected action never ran, so it must not inflate
/// `actions_executed`. The reported case: a two-action proposal naming a tool
/// that isn't registered came back `{actions_executed: 2, tool_calls: 0}` while
/// `metrics.summary` correctly said `actions_succeeded: 0, actions_rejected: 2`
/// — the two surfaces disagreed, and `cost` had no field that could express
/// "rejected" at all.
#[tokio::test]
async fn test_cost_summary_counts_rejected_separately() {
    let rt = setup_runtime().await;
    // `nope` is deliberately not registered — the validator rejects both.
    let p = proposal(vec![
        tool_call("nope", HashMap::new()),
        tool_call("nope", HashMap::new()),
    ]);
    let result = rt.execute(&p).await;

    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    assert_eq!(
        result.cost.actions_executed, 0,
        "nothing ran, so nothing was executed"
    );
    assert_eq!(result.cost.actions_rejected, 2);
    assert_eq!(result.cost.tool_calls, 0);
}

#[tokio::test]
async fn test_cost_budget_max_actions() {
    let rt = setup_runtime().await;
    rt.set_cost_budget(CostBudget {
        max_tool_calls: None,
        max_duration_ms: None,
        max_actions: Some(1),
    })
    .await;

    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        tool_call("echo", [("message".to_string(), Value::from("hi"))].into()),
    ]);
    let result = rt.execute(&p).await;

    assert_eq!(result.results[0].status, ActionStatus::Succeeded);
    assert_eq!(result.results[1].status, ActionStatus::Skipped);
    assert!(result.results[1]
        .error
        .as_ref()
        .unwrap()
        .contains("cost budget exceeded"));
    assert_eq!(result.cost.actions_executed, 1);
    assert_eq!(result.cost.actions_skipped, 1);
}

#[tokio::test]
async fn test_cost_budget_max_tool_calls() {
    let rt = setup_runtime().await;
    rt.set_cost_budget(CostBudget {
        max_tool_calls: Some(1),
        max_duration_ms: None,
        max_actions: None,
    })
    .await;

    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        tool_call("echo", [("message".to_string(), Value::from("hi"))].into()),
    ]);
    let result = rt.execute(&p).await;

    assert_eq!(result.results[0].status, ActionStatus::Succeeded);
    assert_eq!(result.results[1].status, ActionStatus::Skipped);
    assert_eq!(result.cost.tool_calls, 1);
}

// --- Cross-proposal result cache ---

#[tokio::test]
async fn test_result_cache_hit_returns_stored_result() {
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct CountingExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor {
        async fn execute(&self, _tool: &str, params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            let a = params.get("a").and_then(|v| v.as_i64()).unwrap_or(0);
            let b = params.get("b").and_then(|v| v.as_i64()).unwrap_or(0);
            Ok(Value::from(a + b))
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor(cc)));
    rt.register_tool("add").await;
    rt.enable_tool_cache("add", 60).await;

    let action = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    );

    // First call executes the tool.
    let r1 = rt.execute(&proposal(vec![action.clone()])).await;
    assert!(r1.all_succeeded());
    assert_eq!(r1.results[0].output, Some(Value::from(3)));
    assert_eq!(call_count.load(Ordering::SeqCst), 1);

    // Second call (new proposal) should hit cache, not execute again.
    let r2 = rt.execute(&proposal(vec![action])).await;
    assert!(r2.all_succeeded());
    assert_eq!(r2.results[0].output, Some(Value::from(3)));
    assert_eq!(call_count.load(Ordering::SeqCst), 1); // Still 1 — cache hit
}

#[tokio::test]
async fn test_result_cache_expired_entries_return_none() {
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct CountingExecutor2(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor2 {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(Value::from("result"))
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor2(cc)));
    rt.register_tool("echo").await;
    // TTL of 0 means entries expire immediately.
    rt.enable_tool_cache("echo", 0).await;

    let action = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());

    rt.execute(&proposal(vec![action.clone()])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 1);

    // Cache entry should be expired, so this executes again.
    rt.execute(&proposal(vec![action])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 2);
}

#[tokio::test]
async fn test_result_cache_different_params_different_keys() {
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct CountingExecutor3(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor3 {
        async fn execute(&self, _tool: &str, params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            let a = params.get("a").and_then(|v| v.as_i64()).unwrap_or(0);
            let b = params.get("b").and_then(|v| v.as_i64()).unwrap_or(0);
            Ok(Value::from(a + b))
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor3(cc)));
    rt.register_tool("add").await;
    rt.enable_tool_cache("add", 60).await;

    let action_a = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    );
    let action_b = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(3)),
            ("b".to_string(), Value::from(4)),
        ]
        .into(),
    );

    let r1 = rt.execute(&proposal(vec![action_a.clone()])).await;
    assert_eq!(r1.results[0].output, Some(Value::from(3)));

    let r2 = rt.execute(&proposal(vec![action_b.clone()])).await;
    assert_eq!(r2.results[0].output, Some(Value::from(7)));

    // Both calls should have executed (different params = different cache keys).
    assert_eq!(call_count.load(Ordering::SeqCst), 2);

    // Repeating action_a should hit cache.
    rt.execute(&proposal(vec![action_a])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 2);
}

// --- Capability permissions ---

#[tokio::test]
async fn test_capability_default_allows_everything() {
    let rt = setup_runtime().await;
    rt.set_capabilities(CapabilitySet::new()).await;

    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        state_write("x", Value::from(42)),
    ]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_capability_allowed_tool_passes() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().allow_tool("add");
    rt.set_capabilities(caps).await;

    let p = proposal(vec![tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    )]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_capability_denied_tool_rejected() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().deny_tool("echo");
    rt.set_capabilities(caps).await;

    let mut denied = tool_call("echo", [("message".to_string(), Value::from("hi"))].into());
    denied.id = "denied-tool".to_string();
    let p = proposal(vec![denied]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("capability denied: tool 'echo' not allowed"));
    let log = rt.log.lock().await;
    let rejected = log
        .events()
        .iter()
        .find(|event| {
            event.kind == car_eventlog::EventKind::ActionRejected
                && event.action_id.as_deref() == Some("denied-tool")
        })
        .expect("rejection is journaled");
    assert_eq!(rejected.data["attempted"], false);
    assert_eq!(rejected.data["stage"], "capability");
    assert!(rejected.data["reason"]
        .as_str()
        .unwrap()
        .contains("capability denied"));
    assert!(!rejected.data.contains_key("attempt"));
}

#[tokio::test]
async fn test_capability_unlisted_tool_rejected() {
    let rt = setup_runtime().await;
    // Only allow "add", so "echo" should be rejected
    let caps = CapabilitySet::new().allow_tool("add");
    rt.set_capabilities(caps).await;

    let p = proposal(vec![tool_call(
        "echo",
        [("message".to_string(), Value::from("hi"))].into(),
    )]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("capability denied: tool 'echo' not allowed"));
}

#[tokio::test]
async fn test_capability_deny_overrides_allow() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().allow_tool("echo").deny_tool("echo");
    rt.set_capabilities(caps).await;

    let p = proposal(vec![tool_call(
        "echo",
        [("message".to_string(), Value::from("hi"))].into(),
    )]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
}

#[tokio::test]
async fn test_capability_state_key_allowed() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().allow_state_key("x");
    rt.set_capabilities(caps).await;

    let p = proposal(vec![state_write("x", Value::from(42))]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_capability_state_key_denied() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().allow_state_key("x");
    rt.set_capabilities(caps).await;

    let p = proposal(vec![state_write("secret", Value::from("nope"))]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("capability denied: state key 'secret' not allowed"));
}

#[tokio::test]
async fn test_capability_state_read_key_denied() {
    let rt = setup_runtime().await;
    rt.state.set("secret", Value::from(42), "setup");
    let caps = CapabilitySet::new().allow_state_key("x");
    rt.set_capabilities(caps).await;

    let mut read_action = state_write("unused", Value::Null);
    read_action.action_type = ActionType::StateRead;
    read_action.parameters = [("key".to_string(), Value::from("secret"))].into();

    let p = proposal(vec![read_action]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("capability denied: state key 'secret' not allowed"));
}

#[tokio::test]
async fn test_capability_max_actions_enforced() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().with_max_actions(1);
    rt.set_capabilities(caps).await;

    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        tool_call("echo", [("message".to_string(), Value::from("hi"))].into()),
    ]);
    let result = rt.execute(&p).await;

    // Entire proposal rejected because 2 actions > max_actions of 1
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert_eq!(result.results[1].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("capability denied"));
    assert_eq!(result.final_proposal.as_ref(), Some(&p));
    let expected_preimage = serde_json::to_value(&p).unwrap();
    let expected_digest = format!(
        "{:x}",
        Sha256::digest(
            car_inference::catalog_identity::canonical_json(&expected_preimage)
                .unwrap()
                .as_bytes()
        )
    );
    let log = rt.log.lock().await;
    let boundaries: Vec<_> = log
        .events()
        .iter()
        .filter(|event| {
            matches!(
                event.kind,
                car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
            ) && event.proposal_id.as_deref() == Some(p.id.as_str())
        })
        .collect();
    assert_eq!(boundaries.len(), 1);
    assert_eq!(boundaries[0].kind, car_eventlog::EventKind::StateRollback);
    assert_eq!(boundaries[0].data["proposal"], expected_preimage);
    assert_eq!(boundaries[0].data["proposal_digest"], expected_digest);
    assert_eq!(
        boundaries[0].data["affected_actions"],
        serde_json::json!([])
    );
    assert_eq!(
        boundaries[0].data["rolled_back_changes"],
        serde_json::json!({})
    );
    assert_eq!(
        boundaries[0].data["changes_semantics"],
        "proposal_rejected_no_state_commit"
    );
    assert_eq!(boundaries[0].data["attempted"], false);
    assert_eq!(boundaries[0].data["stage"], "proposal_rejection");
    assert!(boundaries[0].data["rejection_reason"]
        .as_str()
        .is_some_and(|reason| reason.contains("capability denied")));
}

#[tokio::test]
async fn test_capability_max_actions_within_budget() {
    let rt = setup_runtime().await;
    let caps = CapabilitySet::new().with_max_actions(2);
    rt.set_capabilities(caps).await;

    let p = proposal(vec![tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    )]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_no_capabilities_allows_everything() {
    // No capabilities set at all (None) should allow everything
    let rt = setup_runtime().await;

    let p = proposal(vec![
        tool_call(
            "add",
            [
                ("a".to_string(), Value::from(1)),
                ("b".to_string(), Value::from(2)),
            ]
            .into(),
        ),
        state_write("anything", Value::from(99)),
    ]);
    let result = rt.execute(&p).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_result_cache_invalidate_clears_correctly() {
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct CountingExecutor4(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor4 {
        async fn execute(&self, _tool: &str, params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            let a = params.get("a").and_then(|v| v.as_i64()).unwrap_or(0);
            let b = params.get("b").and_then(|v| v.as_i64()).unwrap_or(0);
            Ok(Value::from(a + b))
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor4(cc)));
    rt.register_tool("add").await;
    rt.enable_tool_cache("add", 60).await;

    let action = tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    );

    rt.execute(&proposal(vec![action.clone()])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 1);

    // Invalidate cache for "add".
    rt.result_cache.invalidate("add").await;

    // Should re-execute after invalidation.
    rt.execute(&proposal(vec![action])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 2);
}

// --- Checkpoint and resume ---

#[tokio::test]
async fn test_save_checkpoint_captures_state_and_tools() {
    let rt = setup_runtime().await;
    rt.state.set("key1", Value::from("value1"), "test");
    rt.state.set("key2", Value::from(42), "test");

    let cp = rt.save_checkpoint().await;

    assert!(!cp.checkpoint_id.is_empty());
    assert_eq!(cp.state.get("key1"), Some(&Value::from("value1")));
    assert_eq!(cp.state.get("key2"), Some(&Value::from(42)));
    assert!(cp.tools.contains(&"add".to_string()));
    assert!(cp.tools.contains(&"echo".to_string()));
    assert!(cp.tools.contains(&"fail".to_string()));
    assert!(cp.tools.contains(&"slow".to_string()));
    assert_eq!(cp.tools.len(), 4);
}

#[tokio::test]
async fn test_restore_checkpoint_restores_state_and_tools() {
    let rt = Runtime::new();

    let cp = Checkpoint {
        checkpoint_id: "test-cp".to_string(),
        created_at: chrono::Utc::now(),
        state: [("restored_key".to_string(), Value::from("restored_value"))].into(),
        events: vec![],
        tools: vec!["tool_a".to_string(), "tool_b".to_string()],
        metadata: HashMap::new(),
    };

    rt.restore_checkpoint(&cp).await;

    assert_eq!(
        rt.state.get("restored_key"),
        Some(Value::from("restored_value"))
    );
    let tools = rt.tools.read().await;
    assert!(tools.contains_key("tool_a"));
    assert!(tools.contains_key("tool_b"));
    assert_eq!(tools.len(), 2);
}

#[tokio::test]
async fn test_restore_checkpoint_clears_previous_tools() {
    let rt = Runtime::new();
    rt.register_tool("old_tool").await;

    let cp = Checkpoint {
        checkpoint_id: "test-cp".to_string(),
        created_at: chrono::Utc::now(),
        state: HashMap::new(),
        events: vec![],
        tools: vec!["new_tool".to_string()],
        metadata: HashMap::new(),
    };

    rt.restore_checkpoint(&cp).await;

    let tools = rt.tools.read().await;
    assert!(!tools.contains_key("old_tool"));
    assert!(tools.contains_key("new_tool"));
    assert_eq!(tools.len(), 1);
}

#[tokio::test]
async fn test_checkpoint_file_roundtrip() {
    let rt = setup_runtime().await;
    rt.state
        .set("persist_key", Value::from("persist_value"), "test");

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("checkpoint.json");
    let path_str = path.to_str().unwrap();

    rt.save_checkpoint_to_file(path_str).await.unwrap();
    assert!(path.exists());

    // Create a fresh runtime and load the checkpoint
    let rt2 = Runtime::new();
    let loaded = rt2.load_checkpoint_from_file(path_str).await.unwrap();

    assert_eq!(
        rt2.state.get("persist_key"),
        Some(Value::from("persist_value"))
    );
    let tools = rt2.tools.read().await;
    assert!(tools.contains_key("add"));
    assert!(tools.contains_key("echo"));

    // Verify loaded checkpoint data
    assert!(!loaded.checkpoint_id.is_empty());
    assert_eq!(
        loaded.state.get("persist_key"),
        Some(&Value::from("persist_value"))
    );
}

#[tokio::test]
async fn test_checkpoint_serializes_as_json() {
    let cp = Checkpoint {
        checkpoint_id: "json-test".to_string(),
        created_at: chrono::Utc::now(),
        state: [("k".to_string(), Value::from(1))].into(),
        events: vec![serde_json::json!({"kind": "test"})],
        tools: vec!["mytool".to_string()],
        metadata: [("meta_key".to_string(), Value::from("meta_val"))].into(),
    };

    let json = serde_json::to_string(&cp).unwrap();
    let deserialized: Checkpoint = serde_json::from_str(&json).unwrap();

    assert_eq!(deserialized.checkpoint_id, "json-test");
    assert_eq!(deserialized.state.get("k"), Some(&Value::from(1)));
    assert_eq!(deserialized.tools, vec!["mytool".to_string()]);
    assert_eq!(
        deserialized.metadata.get("meta_key"),
        Some(&Value::from("meta_val"))
    );
    assert_eq!(deserialized.events.len(), 1);
}

#[tokio::test]
async fn test_checkpoint_captures_events() {
    let rt = setup_runtime().await;

    // Execute a proposal to generate some events
    let p = proposal(vec![tool_call(
        "add",
        [
            ("a".to_string(), Value::from(1)),
            ("b".to_string(), Value::from(2)),
        ]
        .into(),
    )]);
    rt.execute(&p).await;

    let cp = rt.save_checkpoint().await;
    assert!(!cp.events.is_empty());
}

#[tokio::test]
async fn test_load_checkpoint_from_nonexistent_file() {
    let rt = Runtime::new();
    let result = rt
        .load_checkpoint_from_file("/tmp/nonexistent_checkpoint_file.json")
        .await;
    assert!(result.is_err());
    assert!(result.unwrap_err().contains("read error"));
}

#[tokio::test]
async fn restore_checkpoint_replaces_state_completely() {
    let rt = Runtime::new();
    rt.state
        .set("pre_existing", serde_json::json!("should_be_gone"), "setup");
    rt.state
        .set("shared_key", serde_json::json!("old_value"), "setup");

    let checkpoint = Checkpoint {
        checkpoint_id: "cp-replace".to_string(),
        created_at: chrono::Utc::now(),
        state: {
            let mut m = std::collections::HashMap::new();
            m.insert(
                "shared_key".to_string(),
                serde_json::json!("checkpoint_value"),
            );
            m.insert("checkpoint_only".to_string(), serde_json::json!(true));
            m
        },
        events: vec![],
        tools: vec![],
        metadata: HashMap::new(),
    };

    rt.restore_checkpoint(&checkpoint).await;

    assert_eq!(
        rt.state.get("shared_key"),
        Some(serde_json::json!("checkpoint_value"))
    );
    assert_eq!(
        rt.state.get("checkpoint_only"),
        Some(serde_json::json!(true))
    );
    // pre_existing should be GONE (not merged)
    assert_eq!(rt.state.get("pre_existing"), None);
}

#[tokio::test]
async fn restore_checkpoint_does_not_emit_synthetic_state_transitions() {
    let rt = Runtime::new();
    rt.state
        .set("pre_existing", serde_json::json!("old"), "setup");
    let _before = rt.state.transition_count();

    let checkpoint = Checkpoint {
        checkpoint_id: "cp-transitions".to_string(),
        created_at: chrono::Utc::now(),
        state: [("replacement".to_string(), serde_json::json!(1))].into(),
        events: vec![],
        tools: vec![],
        metadata: HashMap::new(),
    };

    rt.restore_checkpoint(&checkpoint).await;
    // Checkpoint restore clears transition history (clean slate)
    assert_eq!(rt.state.transition_count(), 0);
}

#[tokio::test]
async fn restore_checkpoint_clears_idempotency_cache() {
    // Use a counting executor so we can tell whether the tool actually ran
    // vs. returning a cached result (duration_ms == 0).
    let call_count = Arc::new(AtomicU32::new(0));
    let cc = call_count.clone();

    struct CountingExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(Value::from(42))
        }
    }

    let rt = Runtime::new().with_executor(Arc::new(CountingExecutor(cc)));
    rt.register_tool("counter").await;

    // Execute an idempotent action — populates the cache
    let mut action = tool_call("counter", HashMap::new());
    action.idempotent = true;
    rt.execute(&proposal(vec![action.clone()])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 1);

    // Re-execute the same action — should hit cache (no new executor call)
    let mut action2 = action.clone();
    action2.id = "second".to_string();
    let cached_result = rt.execute(&proposal(vec![action2])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 1); // still 1 — cached
    assert_eq!(cached_result.results[0].duration_ms, Some(0.0));

    // Restore a checkpoint — must clear the idempotency cache
    let checkpoint = Checkpoint {
        checkpoint_id: "cp-idem".to_string(),
        created_at: chrono::Utc::now(),
        state: HashMap::new(),
        events: vec![],
        tools: vec!["counter".to_string()],
        metadata: HashMap::new(),
    };
    rt.restore_checkpoint(&checkpoint).await;

    // Re-execute the same idempotent action — cache was cleared, so the
    // executor must be called again (not a stale cache hit)
    let mut action3 = action.clone();
    action3.id = "third".to_string();
    let fresh_result = rt.execute(&proposal(vec![action3])).await;
    assert_eq!(call_count.load(Ordering::SeqCst), 2); // actually called
    assert_eq!(fresh_result.results[0].status, ActionStatus::Succeeded);
}

// --- ToolSchema registry ---

#[tokio::test]
async fn test_register_tool_backward_compat() {
    let rt = Runtime::new();
    rt.register_tool("echo").await;
    let tools = rt.tools.read().await;
    assert!(tools.contains_key("echo"));
    let schema = tools.get("echo").unwrap();
    assert_eq!(schema.name, "echo");
    assert_eq!(schema.description, "");
    assert!(!schema.idempotent);
    assert!(schema.cache_ttl_secs.is_none());
    assert!(schema.rate_limit.is_none());
}

#[tokio::test]
async fn test_register_tool_schema_full() {
    let rt = make_runtime();
    let schema = ToolSchema {
        name: "add".to_string(),
        description: "Add two numbers".to_string(),
        parameters: serde_json::json!({
            "type": "object",
            "properties": {
                "a": {"type": "number"},
                "b": {"type": "number"}
            },
            "required": ["a", "b"]
        }),
        returns: Some(serde_json::json!({"type": "number"})),
        idempotent: true,
        cache_ttl_secs: None,
        rate_limit: None,
    };
    rt.register_tool_schema(schema).await;

    let tools = rt.tools.read().await;
    let s = tools.get("add").unwrap();
    assert_eq!(s.description, "Add two numbers");
    assert!(s.idempotent);
    assert_eq!(
        s.parameters
            .get("required")
            .unwrap()
            .as_array()
            .unwrap()
            .len(),
        2
    );
}

#[tokio::test]
async fn test_tool_schemas_returns_all() {
    let rt = Runtime::new();
    rt.register_tool("a").await;
    rt.register_tool("b").await;
    rt.register_tool("c").await;
    let schemas = rt.tool_schemas().await;
    assert_eq!(schemas.len(), 3);
    let names: Vec<String> = schemas.iter().map(|s| s.name.clone()).collect();
    assert!(names.contains(&"a".to_string()));
    assert!(names.contains(&"b".to_string()));
    assert!(names.contains(&"c".to_string()));
}

#[tokio::test]
async fn test_schema_auto_configures_cache() {
    let rt = make_runtime();
    let schema = ToolSchema {
        name: "cached_tool".to_string(),
        description: String::new(),
        parameters: Value::Object(Default::default()),
        returns: None,
        idempotent: true,
        cache_ttl_secs: Some(120),
        rate_limit: None,
    };
    rt.register_tool_schema(schema).await;

    // The cache should now be configured for "cached_tool".
    // We can verify by putting/getting a value through the cache.
    let params = serde_json::json!({"x": 1});
    rt.result_cache
        .put("cached_tool", &params, Value::from(42))
        .await;
    let cached = rt.result_cache.get("cached_tool", &params).await;
    assert_eq!(cached, Some(Value::from(42)));
}

#[tokio::test]
async fn test_schema_auto_configures_rate_limit() {
    let rt = make_runtime();
    let schema = ToolSchema {
        name: "limited_tool".to_string(),
        description: String::new(),
        parameters: Value::Object(Default::default()),
        returns: None,
        idempotent: false,
        cache_ttl_secs: None,
        rate_limit: Some(ToolRateLimit {
            max_calls: 5,
            interval_secs: 1.0,
        }),
    };
    rt.register_tool_schema(schema).await;

    // Verify rate limiter was configured by acquiring a token (should not block).
    rt.rate_limiter.acquire("limited_tool").await;
}

#[tokio::test]
async fn test_schema_parameter_validation_rejects_missing_required() {
    let rt = make_runtime();
    let schema = ToolSchema {
        name: "strict_tool".to_string(),
        description: String::new(),
        parameters: serde_json::json!({
            "type": "object",
            "properties": {
                "x": {"type": "number"}
            },
            "required": ["x"]
        }),
        returns: None,
        idempotent: false,
        cache_ttl_secs: None,
        rate_limit: None,
    };
    rt.register_tool_schema(schema).await;

    // Call without required param "x"
    let action = tool_call("strict_tool", HashMap::new());
    let p = proposal(vec![action]);
    let result = rt.execute(&p).await;
    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    assert!(result.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("missing required parameter 'x'"));
}

// --- Replan tests ---

use crate::executor::{ReplanCallback, ReplanConfig, ReplanContext};

/// A mock replan callback that returns a fixed alternative proposal.
struct MockReplanner {
    /// How many times replan was called.
    call_count: AtomicU32,
}

#[async_trait::async_trait]
impl ReplanCallback for MockReplanner {
    async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
        self.call_count.fetch_add(1, Ordering::SeqCst);
        // Return a proposal that succeeds (calls "echo" instead of "fail")
        Ok(ActionProposal {
            id: "replan-proposal".to_string(),
            source: "replanner".to_string(),
            actions: vec![tool_call(
                "echo",
                [("message".to_string(), Value::from("recovered"))].into(),
            )],
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        })
    }
}

/// Replanner that always fails.
struct FailingReplanner;

#[async_trait::async_trait]
impl ReplanCallback for FailingReplanner {
    async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
        Err("replanner failed".to_string())
    }
}

/// Replanner that returns another failing proposal.
struct PersistentlyFailingReplanner {
    call_count: AtomicU32,
}

#[async_trait::async_trait]
impl ReplanCallback for PersistentlyFailingReplanner {
    async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
        self.call_count.fetch_add(1, Ordering::SeqCst);
        // Return a proposal that also fails
        Ok(ActionProposal {
            id: format!("replan-{}", self.call_count.load(Ordering::SeqCst)),
            source: "replanner".to_string(),
            actions: vec![tool_call("fail", HashMap::new())],
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        })
    }
}

/// Replanner that returns a proposal using an unregistered tool.
struct BadToolReplanner;

#[async_trait::async_trait]
impl ReplanCallback for BadToolReplanner {
    async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
        Ok(ActionProposal {
            id: "bad-replan".to_string(),
            source: "test".to_string(),
            actions: vec![tool_call("nonexistent_tool", HashMap::new())],
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        })
    }
}

/// Every candidate is invalid because two actions reuse one id. The original
/// action has an externally visible attempt counter so re-dispatching it while
/// rejecting candidates is observable.
struct DuplicateIdReplanner {
    call_count: AtomicU32,
}

#[async_trait::async_trait]
impl ReplanCallback for DuplicateIdReplanner {
    async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
        let generation = self.call_count.fetch_add(1, Ordering::SeqCst) + 1;
        let mut first = tool_call("effect_then_fail", HashMap::new());
        first.id = "duplicate-candidate-action".to_string();
        let mut second = tool_call("effect_then_fail", HashMap::new());
        second.id = "duplicate-candidate-action".to_string();
        Ok(ActionProposal {
            id: format!("duplicate-candidate-{generation}"),
            source: "duplicate-replanner".to_string(),
            actions: vec![first, second],
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        })
    }
}

#[tokio::test]
async fn test_replan_recovers_from_failure() {
    let replanner = Arc::new(MockReplanner {
        call_count: AtomicU32::new(0),
    });

    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 3,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("fail").await;

    // First proposal fails (calls "fail" tool)
    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let original_preimage = p.clone();
    let result = rt.execute(&p).await;

    // Should have recovered via replan
    assert!(
        result.all_succeeded(),
        "expected success after replan, got: {:?}",
        result.results
    );
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 1);
    let final_proposal = result
        .final_proposal
        .as_ref()
        .expect("active execution returns the exact final proposal");
    assert_eq!(final_proposal.id, "replan-proposal");
    assert_eq!(result.proposal_id, final_proposal.id);
    let expected_preimage = serde_json::to_value(final_proposal).unwrap();
    let expected_digest = format!(
        "{:x}",
        Sha256::digest(
            car_inference::catalog_identity::canonical_json(&expected_preimage)
                .unwrap()
                .as_bytes()
        )
    );
    assert_eq!(
        result
            .replan_lineage
            .last()
            .unwrap()
            .proposal_digest
            .as_deref(),
        Some(expected_digest.as_str())
    );
    assert_eq!(result.accepted_proposal_preimages.len(), 2);
    assert_eq!(result.accepted_proposal_preimages[0].generation, 0);
    assert_eq!(
        result.accepted_proposal_preimages[0].proposal,
        original_preimage
    );
    assert_eq!(result.accepted_proposal_preimages[1].generation, 1);
    assert_eq!(
        result.accepted_proposal_preimages[1].proposal,
        *final_proposal
    );
    assert_eq!(
        result.accepted_proposal_preimages[1].proposal_digest,
        expected_digest
    );
    let log = rt.log.lock().await;
    let received = log
        .events()
        .iter()
        .find(|event| event.kind == car_eventlog::EventKind::ReplanProposalReceived)
        .expect("accepted replan preimage event");
    assert_eq!(received.data["proposal"], expected_preimage);
    assert_eq!(received.data["proposal_digest"], expected_digest);
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| {
                matches!(
                    event.kind,
                    car_eventlog::EventKind::StateCommitted
                        | car_eventlog::EventKind::StateRollback
                )
            })
            .count(),
        2,
        "failed original rolls back once and accepted replan commits once"
    );
}

#[tokio::test]
async fn accepted_proposal_preimages_survive_preconfigured_event_retention() {
    let replanner = Arc::new(MockReplanner {
        call_count: AtomicU32::new(0),
    });
    let rt = make_runtime().with_replan(
        replanner,
        ReplanConfig {
            max_replans: 1,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("fail").await;
    rt.log
        .lock()
        .await
        .set_retention(Some(car_eventlog::RetentionPolicy {
            max_events: Some(1),
            max_age_secs: None,
        }));

    let original = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&original).await;
    assert!(result.all_succeeded());
    assert_eq!(
        result
            .accepted_proposal_preimages
            .iter()
            .map(|entry| entry.generation)
            .collect::<Vec<_>>(),
        vec![0, 1]
    );
    assert_eq!(result.accepted_proposal_preimages[0].proposal, original);
    assert_eq!(
        result.accepted_proposal_preimages[1].proposal,
        result.final_proposal.clone().unwrap()
    );
    for accepted in &result.accepted_proposal_preimages {
        let lineage = result
            .replan_lineage
            .iter()
            .find(|entry| {
                entry.generation == accepted.generation
                    && entry.status == car_ir::ProposalLineageStatus::Accepted
            })
            .expect("every retained preimage has an accepted lineage entry");
        assert_eq!(
            lineage.proposal_digest.as_deref(),
            Some(accepted.proposal_digest.as_str())
        );
    }

    let log = rt.log.lock().await;
    assert_eq!(log.events().len(), 1, "retention must be active during run");
    assert!(
        log.events()
            .iter()
            .all(|event| event.kind != car_eventlog::EventKind::ReplanProposalReceived),
        "accepted preimages must not depend on a retained proposal event"
    );
}

#[tokio::test]
async fn invalid_replan_candidates_never_redispatch_the_failed_original() {
    struct EffectThenFailExecutor(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for EffectThenFailExecutor {
        async fn execute(&self, _tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Err("external effect happened before failure".to_string())
        }
    }

    let dispatches = Arc::new(AtomicU32::new(0));
    let replanner = Arc::new(DuplicateIdReplanner {
        call_count: AtomicU32::new(0),
    });
    let rt = Runtime::new()
        .with_executor(Arc::new(EffectThenFailExecutor(dispatches.clone())))
        .with_replan(
            replanner.clone(),
            ReplanConfig {
                max_replans: 2,
                delay_ms: 0,
                verify_before_execute: false,
                replan_on_rejected: false,
            },
        );
    rt.register_tool("effect_then_fail").await;
    let mut original = tool_call("effect_then_fail", HashMap::new());
    original.id = "original-effect".to_string();
    original.max_retries = 0;
    original.failure_behavior = FailureBehavior::Abort;

    let submitted = proposal(vec![original]);
    let submitted_preimage = serde_json::to_value(&submitted).unwrap();
    let submitted_digest = format!(
        "{:x}",
        Sha256::digest(
            car_inference::catalog_identity::canonical_json(&submitted_preimage)
                .unwrap()
                .as_bytes()
        )
    );
    let result = rt.execute(&submitted).await;

    assert_eq!(result.results[0].status, ActionStatus::Failed);
    assert_eq!(
        dispatches.load(Ordering::SeqCst),
        1,
        "candidate rejection must not re-enter execution of the prior failed proposal"
    );
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 2);
    assert_eq!(result.original_proposal_id, result.proposal_id);
    assert_eq!(result.final_proposal.as_ref(), Some(&submitted));
    assert_eq!(
        result.replan_lineage[0].proposal_digest.as_deref(),
        Some(submitted_digest.as_str())
    );
    assert_eq!(result.replan_lineage.len(), 3);
    assert_eq!(
        result
            .replan_lineage
            .iter()
            .map(|entry| entry.generation)
            .collect::<Vec<_>>(),
        vec![0, 1, 2]
    );
    assert_eq!(
        result
            .replan_lineage
            .iter()
            .map(|entry| entry.status)
            .collect::<Vec<_>>(),
        vec![
            car_ir::ProposalLineageStatus::Accepted,
            car_ir::ProposalLineageStatus::Rejected,
            car_ir::ProposalLineageStatus::Rejected,
        ]
    );
    assert!(result.replan_lineage.iter().all(|entry| entry
        .proposal_digest
        .as_deref()
        .is_some_and(|digest| {
            digest.len() == 64
                && digest
                    .bytes()
                    .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        })));
    let log = rt.log.lock().await;
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| {
                event.kind == car_eventlog::EventKind::ActionExecuting
                    && event.action_id.as_deref() == Some("original-effect")
            })
            .count(),
        1
    );
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| event.kind == car_eventlog::EventKind::ReplanRejected)
            .count(),
        2
    );
    assert_eq!(
        log.events()
            .iter()
            .filter(|event| {
                matches!(
                    event.kind,
                    car_eventlog::EventKind::StateCommitted
                        | car_eventlog::EventKind::StateRollback
                )
            })
            .count(),
        1,
        "rejected candidates must not create execution boundaries"
    );
}

#[tokio::test]
async fn active_run_rejects_replan_that_changes_claimed_proposal_id_before_dispatch() {
    struct CountingFailThenEffect(Arc<AtomicU32>);

    #[async_trait::async_trait]
    impl ToolExecutor for CountingFailThenEffect {
        async fn execute(&self, tool: &str, _params: &Value) -> Result<Value, String> {
            self.0.fetch_add(1, Ordering::SeqCst);
            if tool == "fail" {
                Err("original failed".to_string())
            } else {
                Ok(serde_json::json!({"effect": true}))
            }
        }
    }

    struct CollidingIdReplanner;

    #[async_trait::async_trait]
    impl ReplanCallback for CollidingIdReplanner {
        async fn replan(&self, _ctx: &ReplanContext) -> Result<ActionProposal, String> {
            Ok(ActionProposal {
                id: "some-other-durable-claim".to_string(),
                source: "replanner".to_string(),
                actions: vec![tool_call("effect", HashMap::new())],
                timestamp: chrono::Utc::now(),
                context: HashMap::new(),
            })
        }
    }

    let dispatches = Arc::new(AtomicU32::new(0));
    let rt = Runtime::new()
        .with_executor(Arc::new(CountingFailThenEffect(dispatches.clone())))
        .with_replan(
            Arc::new(CollidingIdReplanner),
            ReplanConfig {
                max_replans: 1,
                delay_ms: 0,
                verify_before_execute: false,
                replan_on_rejected: false,
            },
        );
    rt.register_tool("fail").await;
    rt.register_tool("effect").await;
    let submitted = proposal(vec![tool_call("fail", HashMap::new())]);

    let result = rt.execute_with_stable_replan_id(&submitted).await;

    assert!(!result.all_succeeded());
    assert_eq!(result.proposal_id, submitted.id);
    assert_eq!(dispatches.load(Ordering::SeqCst), 1);
    assert_eq!(result.replan_lineage.len(), 2);
    assert_eq!(
        result.replan_lineage[1].status,
        car_ir::ProposalLineageStatus::Rejected
    );
    assert!(result.replan_lineage[1]
        .rejection_reason
        .as_deref()
        .is_some_and(|reason| reason.contains("does not retain authenticated proposal id")));
}

#[tokio::test]
async fn test_replan_disabled_by_default() {
    let rt = setup_runtime().await;

    // No replan callback, default config (max_replans=0)
    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // Should fail normally — no replan
    assert!(!result.all_succeeded());
}

#[tokio::test]
async fn test_replan_exhausted_after_max_attempts() {
    let replanner = Arc::new(PersistentlyFailingReplanner {
        call_count: AtomicU32::new(0),
    });

    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 2,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // Should have tried 2 replans then given up
    assert!(!result.all_succeeded());
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 2);
}

#[tokio::test]
async fn test_replan_callback_failure_returns_original() {
    let rt = make_runtime().with_replan(
        Arc::new(FailingReplanner),
        ReplanConfig {
            max_replans: 3,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // Replan callback itself failed — should return original failure
    assert!(!result.all_succeeded());
}

#[tokio::test]
async fn test_rejected_does_not_replan_by_default() {
    // A capability/policy rejection (ActionStatus::Rejected) must NOT trigger
    // replan when replan_on_rejected is left at its conservative default.
    let replanner = Arc::new(MockReplanner {
        call_count: AtomicU32::new(0),
    });
    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 3,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("blocked").await;
    // Deny the tool so the action is Rejected (not Failed).
    rt.set_capabilities(CapabilitySet::new().deny_tool("blocked"))
        .await;

    let p = proposal(vec![tool_call("blocked", HashMap::new())]);
    let result = rt.execute(&p).await;

    assert_eq!(result.results[0].status, ActionStatus::Rejected);
    // Default behavior: rejection does not enter the replan path.
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 0);
}

#[tokio::test]
async fn test_rejected_replans_when_opted_in() {
    // With replan_on_rejected=true, a rejection triggers rollback + replan;
    // the replanner's substitute (echo, allowed) then succeeds.
    let replanner = Arc::new(MockReplanner {
        call_count: AtomicU32::new(0),
    });
    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 3,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: true,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("blocked").await;
    rt.set_capabilities(CapabilitySet::new().deny_tool("blocked"))
        .await;

    let p = proposal(vec![tool_call("blocked", HashMap::new())]);
    let result = rt.execute(&p).await;

    // The rejection was treated as recoverable; the replanned echo succeeded.
    assert!(
        result.all_succeeded(),
        "expected success after replan-on-rejected, got: {:?}",
        result.results
    );
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn test_replan_context_contains_failure_info() {
    use std::sync::Mutex;

    struct CapturingReplanner {
        captured: Mutex<Option<ReplanContext>>,
    }

    #[async_trait::async_trait]
    impl ReplanCallback for CapturingReplanner {
        async fn replan(&self, ctx: &ReplanContext) -> Result<ActionProposal, String> {
            *self.captured.lock().unwrap() = Some(ctx.clone());
            // Return success to stop the loop
            Ok(ActionProposal {
                id: "recovery".to_string(),
                source: "test".to_string(),
                actions: vec![],
                timestamp: chrono::Utc::now(),
                context: HashMap::new(),
            })
        }
    }

    let replanner = Arc::new(CapturingReplanner {
        captured: Mutex::new(None),
    });

    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 1,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let _result = rt.execute(&p).await;

    let ctx = replanner.captured.lock().unwrap().take().unwrap();
    assert_eq!(ctx.attempt, 1);
    assert_eq!(ctx.proposal_id, "test-proposal");
    assert_eq!(ctx.original_action_count, 1);
    assert!(!ctx.failed_actions.is_empty());
    assert_eq!(ctx.failed_actions[0].error, "boom");
    assert_eq!(ctx.replans_remaining, 0);
}

#[tokio::test]
async fn test_replan_state_is_clean_between_attempts() {
    // Verify that state written by attempt 1 is rolled back before attempt 2.
    use std::sync::Mutex;

    struct StateCheckingReplanner {
        captured_state: Mutex<Option<HashMap<String, Value>>>,
    }

    #[async_trait::async_trait]
    impl ReplanCallback for StateCheckingReplanner {
        async fn replan(&self, ctx: &ReplanContext) -> Result<ActionProposal, String> {
            *self.captured_state.lock().unwrap() = Some(ctx.state_snapshot.clone());
            // Return a succeeding proposal
            Ok(ActionProposal {
                id: "replan-clean".to_string(),
                source: "test".to_string(),
                actions: vec![tool_call(
                    "echo",
                    [("message".to_string(), Value::from("clean"))].into(),
                )],
                timestamp: chrono::Utc::now(),
                context: HashMap::new(),
            })
        }
    }

    let replanner = Arc::new(StateCheckingReplanner {
        captured_state: Mutex::new(None),
    });

    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 1,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("fail").await;

    // First proposal writes state then fails
    let mut write_then_fail = vec![
        state_write("dirty_key", Value::from("dirty_value")),
        tool_call("fail", HashMap::new()),
    ];
    // Make the fail action depend on the write so they run in order
    write_then_fail[1]
        .state_dependencies
        .push("dirty_key".to_string());

    let p = proposal(write_then_fail);
    let result = rt.execute(&p).await;

    // Should have recovered
    assert!(result.all_succeeded());

    // The state snapshot seen by the replanner should NOT contain "dirty_key"
    // because rollback should have cleaned it
    let snapshot = replanner.captured_state.lock().unwrap().take().unwrap();
    assert!(
        !snapshot.contains_key("dirty_key"),
        "state should be clean after rollback, but found dirty_key: {:?}",
        snapshot
    );
}

#[tokio::test]
async fn test_replan_with_unregistered_tool_is_rejected() {
    // Replan returns a proposal using a tool that was never registered.
    // The validation pipeline should reject it, not crash.
    let rt = make_runtime().with_replan(
        Arc::new(BadToolReplanner),
        ReplanConfig {
            max_replans: 2,
            delay_ms: 0,
            verify_before_execute: false,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // Should fail gracefully — the replan proposal gets rejected, not crash
    assert!(!result.all_succeeded());
}

#[tokio::test]
async fn test_replan_quality_gate_rejects_bad_proposal() {
    // With verify_before_execute=true, a replan that uses an unregistered tool
    // should be rejected by the quality gate without executing.
    let replanner = Arc::new(MockReplanner {
        call_count: AtomicU32::new(0),
    });

    // Note: verify_before_execute=true is the default, but "echo" IS registered
    // so MockReplanner's proposal (which uses "echo") should pass verification.
    let rt = make_runtime().with_replan(
        replanner.clone(),
        ReplanConfig {
            max_replans: 1,
            delay_ms: 0,
            verify_before_execute: true,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("echo").await;
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // MockReplanner returns echo tool which IS registered → passes gate → succeeds
    assert!(result.all_succeeded());
    assert_eq!(replanner.call_count.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn test_replan_quality_gate_blocks_invalid_replan() {
    // BadToolReplanner returns a proposal with "nonexistent_tool".
    // With verify_before_execute=true, this should be caught by the quality gate
    // and rejected without executing, consuming the replan attempt.
    let rt = make_runtime().with_replan(
        Arc::new(BadToolReplanner),
        ReplanConfig {
            max_replans: 2,
            delay_ms: 0,
            verify_before_execute: true,
            replan_on_rejected: false,
        },
    );
    rt.register_tool("fail").await;

    let p = proposal(vec![tool_call("fail", HashMap::new())]);
    let result = rt.execute(&p).await;

    // Both replan attempts should be consumed (both fail verification)
    // and the original failure should be returned
    assert!(!result.all_succeeded());
}

// --- plan_and_execute tests ---

#[tokio::test]
async fn test_plan_and_execute_picks_best() {
    let rt = setup_runtime().await;

    // Two candidates: one uses "echo" (succeeds), one uses "fail" (fails)
    let good = proposal(vec![tool_call(
        "echo",
        [("message".to_string(), Value::from("hello"))].into(),
    )]);
    let bad = proposal(vec![tool_call("fail", HashMap::new())]);

    // Good should be picked (valid, higher score)
    let result = rt.plan_and_execute(&[bad, good], None, None).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_plan_and_execute_falls_back() {
    let rt = setup_runtime().await;

    // First candidate fails at execution, second succeeds
    let fails = proposal(vec![tool_call("fail", HashMap::new())]);
    let succeeds = proposal(vec![tool_call(
        "echo",
        [("message".to_string(), Value::from("fallback"))].into(),
    )]);

    // Both are valid (tools registered), but "fail" will abort at runtime.
    // plan_and_execute should fall back to the echo proposal.
    let result = rt.plan_and_execute(&[fails, succeeds], None, None).await;
    assert!(result.all_succeeded());
}

#[tokio::test]
async fn test_plan_and_execute_empty_candidates() {
    let rt = setup_runtime().await;
    let result = rt.plan_and_execute(&[], None, None).await;
    assert_eq!(result.proposal_id, "empty");
    assert!(result.results.is_empty());
}

#[tokio::test]
async fn plan_and_execute_does_not_erase_a_concurrent_proposal_commit() {
    let state = Arc::new(car_state::StateStore::new());
    let executor = Arc::new(PlanTransactionExecutor {
        state: state.clone(),
        first_candidate_entered: tokio::sync::Notify::new(),
        release_first_candidate: tokio::sync::Notify::new(),
    });
    let rt = Arc::new(
        Runtime::with_shared(
            state.clone(),
            Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
            Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
        )
        .with_executor(executor.clone()),
    );
    rt.register_tool("controlled").await;

    let timestamp = chrono::Utc::now();
    let candidate = |proposal_id: &str, action_id: &str| {
        let mut action = tool_call("controlled", HashMap::new());
        action.id = action_id.to_string();
        ActionProposal {
            id: proposal_id.to_string(),
            source: "planner-source".to_string(),
            actions: vec![action],
            timestamp,
            context: HashMap::new(),
        }
    };
    let candidates = vec![
        candidate("plan-choice-a", "plan-action-a"),
        candidate("plan-choice-b", "plan-action-b"),
        candidate("plan-choice-c", "plan-action-c"),
    ];

    let plan_rt = rt.clone();
    let plan = tokio::spawn(async move { plan_rt.plan_and_execute(&candidates, None, None).await });
    tokio::time::timeout(
        std::time::Duration::from_secs(1),
        executor.first_candidate_entered.notified(),
    )
    .await
    .expect("first planning candidate must reach dispatch");

    let mut outside_action = tool_call("controlled", HashMap::new());
    outside_action.id = "outside-write".to_string();
    let outside_proposal = ActionProposal {
        id: "outside-plan".to_string(),
        source: "outside-source".to_string(),
        actions: vec![outside_action],
        timestamp,
        context: HashMap::new(),
    };
    let outside_rt = rt.clone();
    let outside = tokio::spawn(async move { outside_rt.execute(&outside_proposal).await });

    // Let the outside proposal reach the transaction mutex while candidate A
    // still owns it. Tokio's FIFO mutex then makes the old unlock/relock gap
    // deterministically admit the outside proposal before candidate B.
    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
    executor.release_first_candidate.notify_one();

    let plan_result = tokio::time::timeout(std::time::Duration::from_secs(2), plan)
        .await
        .expect("planning must complete")
        .expect("planning task must not panic");
    let outside_result = tokio::time::timeout(std::time::Duration::from_secs(2), outside)
        .await
        .expect("outside proposal must complete")
        .expect("outside proposal task must not panic");
    assert!(plan_result.all_succeeded());
    assert!(outside_result.all_succeeded());
    assert_eq!(
        state.get("concurrent_sentinel"),
        Some(Value::from("committed")),
        "a committed concurrent proposal must survive planning fallback restores"
    );
}

#[tokio::test]
async fn plan_fallback_rolls_back_partial_skip_state_journal_and_idempotency() {
    let dir = tempfile::tempdir().unwrap();
    let journal = dir.path().join("idempotency.jsonl");
    let state_path = dir.path().join("state.jsonl");
    let state = Arc::new(car_state::StateStore::durable(&state_path).unwrap());
    state.set("baseline", Value::from("preserved"), "test-setup");
    let baseline_transitions = state.transition_count();
    let executor = Arc::new(FallbackRollbackExecutor {
        state: state.clone(),
        effect_dispatches: AtomicU32::new(0),
    });
    let rt = Runtime::with_shared(
        state.clone(),
        Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
        Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
    )
    .with_executor(executor.clone());
    for tool in ["partial-effect", "partial-skip", "fallback-good"] {
        rt.register_tool(tool).await;
    }
    rt.set_idempotency_cache_path(&journal).await.unwrap();

    let mut effect = tool_call("partial-effect", HashMap::new());
    effect.id = "partial-effect".to_string();
    effect.idempotent = true;
    let mut skip = tool_call("partial-skip", HashMap::new());
    skip.id = "partial-skip".to_string();
    skip.failure_behavior = FailureBehavior::Skip;
    let mut fallback = tool_call("fallback-good", HashMap::new());
    fallback.id = "fallback-good".to_string();

    let timestamp = chrono::Utc::now();
    let partial = ActionProposal {
        id: "partial-candidate".to_string(),
        source: "planner".to_string(),
        actions: vec![effect.clone(), skip],
        timestamp,
        context: HashMap::new(),
    };
    let fallback = ActionProposal {
        id: "fallback-candidate".to_string(),
        source: "planner".to_string(),
        actions: vec![fallback],
        timestamp,
        context: HashMap::new(),
    };
    let planner_config = car_planner::PlannerConfig {
        feedback_weight: 1.0,
        ..Default::default()
    };
    let feedback = car_planner::ToolFeedback {
        tool_success_rates: [
            ("partial-effect".to_string(), 1.0),
            ("partial-skip".to_string(), 1.0),
            ("fallback-good".to_string(), 0.0),
        ]
        .into(),
        ..Default::default()
    };
    let ranked = car_planner::Planner::new(planner_config.clone()).rank_with_feedback(
        &[partial.clone(), fallback.clone()],
        Some(&state.snapshot()),
        Some(
            &["partial-effect", "partial-skip", "fallback-good"]
                .into_iter()
                .map(str::to_string)
                .collect(),
        ),
        Some(&feedback),
    );
    assert_eq!(ranked[0].index, 0, "partial candidate must run first");

    let result = rt
        .plan_and_execute(
            &[partial.clone(), fallback],
            Some(planner_config),
            Some(&feedback),
        )
        .await;
    assert!(result.all_succeeded(), "fallback must succeed: {result:?}");
    assert_eq!(state.get("baseline"), Some(Value::from("preserved")));
    assert_eq!(
        state.get("partial_dirty"),
        None,
        "failed candidate state must be restored before fallback"
    );
    assert_eq!(
        state.transition_count(),
        baseline_transitions,
        "failed candidate transitions must be discarded"
    );
    assert_eq!(executor.effect_dispatches.load(Ordering::SeqCst), 1);

    {
        let log = rt.log.lock().await;
        let commit = log
            .events()
            .iter()
            .position(|event| {
                event.proposal_id.as_deref() == Some("partial-candidate")
                    && event.kind == car_eventlog::EventKind::StateCommitted
            })
            .expect("partial execution records its provisional commit boundary");
        let rollback = log
            .events()
            .iter()
            .position(|event| {
                event.proposal_id.as_deref() == Some("partial-candidate")
                    && event.kind == car_eventlog::EventKind::StateRollback
                    && event.data.get("stage") == Some(&Value::from("plan_fallback_rollback"))
            })
            .expect("fallback rejection must have compensating rollback evidence");
        assert!(
            rollback > commit,
            "rollback evidence must follow the commit"
        );
    }

    let journal_lines: Vec<Value> = fs::read_to_string(&journal)
        .unwrap()
        .lines()
        .map(|line| serde_json::from_str(line).unwrap())
        .collect();
    assert_eq!(journal_lines.len(), 2, "cache insert plus tombstone");
    assert!(journal_lines[0]["result"].is_object());
    assert!(journal_lines[1]["result"].is_null());
    assert_eq!(journal_lines[0]["key"], journal_lines[1]["key"]);

    state.sync().unwrap();
    drop(rt);
    drop(executor);
    drop(state);

    let reopened = Arc::new(car_state::StateStore::durable(&state_path).unwrap());
    assert_eq!(
        reopened.get("baseline"),
        Some(Value::from("preserved")),
        "pre-plan state must survive reopening"
    );
    assert_eq!(
        reopened.get("partial_dirty"),
        None,
        "failed planning candidate state must not replay after reopening"
    );
    let retry_executor = Arc::new(FallbackRollbackExecutor {
        state: reopened.clone(),
        effect_dispatches: AtomicU32::new(0),
    });
    let retry_rt = Runtime::with_shared(
        reopened,
        Arc::new(tokio::sync::Mutex::new(car_eventlog::EventLog::new())),
        Arc::new(tokio::sync::RwLock::new(car_policy::PolicyEngine::new())),
    )
    .with_executor(retry_executor.clone());
    retry_rt.register_tool("partial-effect").await;
    retry_rt.set_idempotency_cache_path(&journal).await.unwrap();

    let retry = ActionProposal {
        id: "partial-retry".to_string(),
        source: "test".to_string(),
        actions: vec![effect],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };
    let retry_result = retry_rt.execute(&retry).await;
    assert!(retry_result.all_succeeded());
    assert_eq!(
        retry_executor.effect_dispatches.load(Ordering::SeqCst),
        1,
        "rolled-back idempotency result must not deduplicate an exact retry"
    );
}

// --- Cancellation token tests --------------------------------------

#[tokio::test]
async fn execute_with_cancel_emits_canceled_results_when_token_tripped() {
    use tokio_util::sync::CancellationToken;

    let rt = setup_runtime().await;

    // Three state-writes chained by state_dependencies. The DAG
    // builder puts each in its own level, so execute_inner walks
    // three level boundaries and re-checks the cancellation token
    // between each one.
    let mut a1 = state_write("x", Value::from(1));
    a1.id = "a1".into();
    let mut a2 = state_write("y", Value::from(2));
    a2.id = "a2".into();
    a2.state_dependencies = vec!["x".into()];
    let mut a3 = state_write("z", Value::from(3));
    a3.id = "a3".into();
    a3.state_dependencies = vec!["y".into()];

    let proposal = ActionProposal {
        id: "p-cancel".into(),
        source: "test".into(),
        actions: vec![a1, a2, a3],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };

    // Trip the token BEFORE execution. The first level-boundary
    // check sees `is_cancelled = true` and emits canceled results
    // for every action.
    let token = CancellationToken::new();
    token.cancel();

    let result = rt.execute_with_cancel(&proposal, &token).await;
    assert_eq!(result.results.len(), 3, "all 3 actions accounted for");
    for r in &result.results {
        let err = r.error.as_deref().unwrap_or("");
        assert!(
            err.contains("canceled"),
            "action {} should be canceled, got status={:?} error={:?}",
            r.action_id,
            r.status,
            r.error
        );
    }
}

#[tokio::test]
async fn execute_with_cancel_skips_later_levels_when_tripped_mid_execute() {
    use tokio_util::sync::CancellationToken;

    let rt = setup_runtime().await;

    // a1 (level 0): a slow tool that sleeps 200ms.
    // a2 (level 1): depends on `x` (which a0 writes synchronously).
    // a3 (level 2): depends on `y` (which a2 writes).
    //
    // We trip the token AFTER a1 has started. Since a1 is in the
    // first level, it runs to completion (the engine doesn't
    // interrupt mid-action). Levels 1 and 2 hit the cancel check at
    // their level boundary and emit canceled results.
    let mut a0 = state_write("x", Value::from(1));
    a0.id = "a0".into();
    let mut a1 = tool_call("slow", HashMap::new());
    a1.id = "a1".into();
    a1.state_dependencies = vec!["x".into()];
    let mut a2 = state_write("y", Value::from(2));
    a2.id = "a2".into();
    a2.state_dependencies = vec!["x".into()];
    let mut a3 = state_write("z", Value::from(3));
    a3.id = "a3".into();
    a3.state_dependencies = vec!["y".into()];

    let proposal = ActionProposal {
        id: "p-mid".into(),
        source: "test".into(),
        actions: vec![a0, a1, a2, a3],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };

    let token = CancellationToken::new();
    let token_for_trip = token.clone();
    // Trip the token shortly after execute starts, while a1's
    // 5-second sleep is still pending. The level-boundary check
    // fires when the executor tries to advance to level 2.
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        token_for_trip.cancel();
    });

    let result = rt.execute_with_cancel(&proposal, &token).await;
    // a0 succeeded (level 0 finished before cancel arrived).
    let a0_result = result.results.iter().find(|r| r.action_id == "a0").unwrap();
    assert_eq!(a0_result.status, ActionStatus::Succeeded);

    // At least one of a2 / a3 must be canceled — they're at later
    // levels and the cancel landed before they ran. (a1 is the slow
    // one that may or may not have completed by the time the cancel
    // check fires; we don't assert on it.)
    let later_canceled = result
        .results
        .iter()
        .filter(|r| matches!(r.action_id.as_str(), "a2" | "a3"))
        .filter(|r| r.error.as_deref().is_some_and(|e| e.contains("canceled")))
        .count();
    assert!(
        later_canceled >= 1,
        "expected ≥1 of a2/a3 to be canceled mid-execute, got results: {:?}",
        result
            .results
            .iter()
            .map(|r| (&r.action_id, &r.status, &r.error))
            .collect::<Vec<_>>()
    );
}

#[tokio::test]
async fn execute_with_cancel_no_op_when_token_never_tripped() {
    use tokio_util::sync::CancellationToken;
    let rt = setup_runtime().await;

    let mut params = HashMap::new();
    params.insert("a".into(), Value::from(1));
    params.insert("b".into(), Value::from(2));
    let action = tool_call("add", params);
    let proposal = ActionProposal {
        id: "p-no-cancel".into(),
        source: "test".into(),
        actions: vec![action],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };

    let token = CancellationToken::new();
    let result = rt.execute_with_cancel(&proposal, &token).await;
    // Token never tripped → behaves identically to plain execute().
    assert_eq!(result.results.len(), 1);
    assert_eq!(result.results[0].status, ActionStatus::Succeeded);
    assert_eq!(result.results[0].output, Some(Value::from(3)));
}

#[tokio::test]
async fn execute_falls_through_to_execute_with_cancel_with_inert_token() {
    // Verifies the additive contract: existing execute() callers
    // see no behaviour change after the cancellation refactor.
    let rt = setup_runtime().await;
    let mut params = HashMap::new();
    params.insert("a".into(), Value::from(40));
    params.insert("b".into(), Value::from(2));
    let proposal = ActionProposal {
        id: "p-default".into(),
        source: "test".into(),
        actions: vec![tool_call("add", params)],
        timestamp: chrono::Utc::now(),
        context: HashMap::new(),
    };
    let result = rt.execute(&proposal).await;
    assert_eq!(result.results[0].output, Some(Value::from(42)));
}

// Tenant scoping end-to-end through Runtime::execute_scoped
// (Parslee-ai/car#187 phase 3 enforcement). A StateWrite under
// tenant "acme" must NOT be visible to a StateRead under tenant
// "globex", and vice-versa. The legacy unscoped path also can't
// see scoped writes.

#[tokio::test]
async fn execute_scoped_isolates_state_writes_between_tenants() {
    let rt = setup_runtime().await;

    // Tenant acme writes "config" = "A"
    let acme_scope = crate::scope::RuntimeScope {
        caller_id: Some("alice".into()),
        tenant_id: Some("acme".into()),
        claims: Default::default(),
    };
    let _ = rt
        .execute_scoped(
            &proposal(vec![state_write("config", Value::from("A"))]),
            &acme_scope,
        )
        .await;

    // Tenant globex writes "config" = "G"
    let globex_scope = crate::scope::RuntimeScope {
        caller_id: Some("bob".into()),
        tenant_id: Some("globex".into()),
        claims: Default::default(),
    };
    let _ = rt
        .execute_scoped(
            &proposal(vec![state_write("config", Value::from("G"))]),
            &globex_scope,
        )
        .await;

    // Each tenant reads back their own value.
    assert_eq!(
        rt.state.scoped(Some("acme")).get("config"),
        Some(Value::from("A"))
    );
    assert_eq!(
        rt.state.scoped(Some("globex")).get("config"),
        Some(Value::from("G"))
    );

    // The unscoped namespace doesn't see either.
    assert!(rt.state.get("config").is_none());
}

#[tokio::test]
async fn execute_scoped_state_read_returns_tenant_value() {
    let rt = setup_runtime().await;
    let scope = crate::scope::RuntimeScope {
        caller_id: Some("alice".into()),
        tenant_id: Some("acme".into()),
        claims: Default::default(),
    };

    let mut params = HashMap::new();
    params.insert("key".into(), Value::from("greeting"));

    let read_action = {
        let mut a = Action::new(ActionType::StateRead);
        a.id = "r1".into();
        a.parameters = params;
        a.max_retries = 0;
        a
    };

    // Write under acme, then read under acme.
    let _ = rt
        .execute_scoped(
            &proposal(vec![state_write("greeting", Value::from("hello acme"))]),
            &scope,
        )
        .await;

    let result = rt
        .execute_scoped(&proposal(vec![read_action]), &scope)
        .await;
    assert_eq!(result.results[0].output, Some(Value::from("hello acme")));
}

// --- transactional pre-execution conflict check (survey §4.3/§5.2.4) ---

fn conflicting_proposal() -> ActionProposal {
    // Two unordered tool calls that both write key "k" via expected_effects
    // → a write-write conflict the transaction checker flags.
    let mut a = tool_call("echo", HashMap::new());
    a.expected_effects = [("k".to_string(), Value::from(1))].into();
    let mut b = tool_call("echo", HashMap::new());
    b.expected_effects = [("k".to_string(), Value::from(2))].into();
    proposal(vec![a, b])
}

#[tokio::test]
async fn transaction_strict_rejects_conflicting_proposal() {
    let rt = setup_runtime().await;
    rt.set_transaction_check_mode(TransactionCheckMode::Strict)
        .await;
    let result = rt.execute(&conflicting_proposal()).await;
    // The whole proposal is rejected before any action runs.
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    assert!(result.results.iter().any(|r| r
        .error
        .as_deref()
        .unwrap_or("")
        .contains("transactional conflict")));
    // Conflict telemetry was emitted, with the documented snake_case kind
    // (neo review #4: must be "write_write", not Debug's "writewrite").
    let log = rt.log.lock().await;
    let conflict = log
        .events()
        .iter()
        .find(|e| e.kind == car_eventlog::EventKind::TransactionConflict)
        .expect("a TransactionConflict event");
    assert_eq!(
        conflict.data.get("kind").and_then(|v| v.as_str()),
        Some("write_write")
    );
    let boundaries: Vec<_> = log
        .events()
        .iter()
        .filter(|event| {
            matches!(
                event.kind,
                car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
            )
        })
        .collect();
    assert_eq!(boundaries.len(), 1);
    assert_eq!(boundaries[0].kind, car_eventlog::EventKind::StateRollback);
    assert_eq!(boundaries[0].data["attempted"], false);
    assert_eq!(boundaries[0].data["stage"], "proposal_rejection");
    assert_eq!(
        boundaries[0].data["changes_semantics"],
        "proposal_rejected_no_state_commit"
    );
}

#[tokio::test]
async fn transaction_off_by_default_skips_check() {
    let rt = setup_runtime().await;
    // Default Off: the conflicting proposal executes (echo succeeds) and no
    // TransactionConflict telemetry is emitted.
    let result = rt.execute(&conflicting_proposal()).await;
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Succeeded));
    let log = rt.log.lock().await;
    assert!(!log
        .events()
        .iter()
        .any(|e| e.kind == car_eventlog::EventKind::TransactionConflict));
}

#[tokio::test]
async fn transaction_warn_records_but_executes() {
    let rt = setup_runtime().await;
    rt.set_transaction_check_mode(TransactionCheckMode::Warn)
        .await;
    let result = rt.execute(&conflicting_proposal()).await;
    // Warn executes anyway...
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Succeeded));
    // ...but records the conflict as telemetry.
    let log = rt.log.lock().await;
    assert!(log
        .events()
        .iter()
        .any(|e| e.kind == car_eventlog::EventKind::TransactionConflict));
}

// --- admission gates: the pre-execution safety seam (EPIC A / A1) ---

/// A test gate that blocks any proposal containing an action whose tool
/// name matches `forbidden_tool`, naming that action in the rejection.
struct BlockToolGate {
    forbidden_tool: String,
}

#[async_trait::async_trait]
impl crate::admission::AdmissionGate for BlockToolGate {
    fn name(&self) -> &str {
        "block_tool"
    }
    async fn check(
        &self,
        proposal: &ActionProposal,
        _ctx: &crate::admission::GateContext<'_>,
    ) -> crate::admission::GateOutcome {
        let blocked: Vec<String> = proposal
            .actions
            .iter()
            .filter(|a| a.tool.as_deref() == Some(self.forbidden_tool.as_str()))
            .map(|a| a.id.clone())
            .collect();
        if blocked.is_empty() {
            crate::admission::GateOutcome::Allow
        } else {
            crate::admission::GateOutcome::reject_actions(
                blocked,
                format!("tool '{}' is forbidden", self.forbidden_tool),
            )
        }
    }
}

/// A test gate that escalates the whole proposal to approval.
struct AlwaysApprovalGate;

#[async_trait::async_trait]
impl crate::admission::AdmissionGate for AlwaysApprovalGate {
    fn name(&self) -> &str {
        "always_approval"
    }
    async fn check(
        &self,
        _proposal: &ActionProposal,
        _ctx: &crate::admission::GateContext<'_>,
    ) -> crate::admission::GateOutcome {
        crate::admission::GateOutcome::NeedsApproval {
            actions: std::collections::HashSet::new(),
            fingerprint: "fp-test".to_string(),
            reason: "needs review".to_string(),
        }
    }
}

#[tokio::test]
async fn admission_no_gates_is_zero_overhead() {
    let rt = setup_runtime().await;
    assert_eq!(rt.admission_gate_count().await, 0);
    // With no gates the echo proposal executes exactly as before.
    let result = rt
        .execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Succeeded));
    // No admission events emitted when no gates are registered.
    let log = rt.log.lock().await;
    assert!(!log
        .events()
        .iter()
        .any(|e| e.kind == car_eventlog::EventKind::AdmissionGateDecision));
}

#[tokio::test]
async fn admission_gate_blocks_proposal_before_any_action_runs() {
    let rt = setup_runtime().await;
    rt.register_admission_gate(Arc::new(BlockToolGate {
        forbidden_tool: "fail".to_string(),
    }))
    .await;
    // A proposal that touches the forbidden tool is rejected wholesale —
    // even the otherwise-fine echo action does not run.
    let result = rt
        .execute(&proposal(vec![
            tool_call("echo", HashMap::new()),
            tool_call("fail", HashMap::new()),
        ]))
        .await;
    assert!(
        result
            .results
            .iter()
            .all(|r| r.status == ActionStatus::Rejected),
        "every action should be rejected at admission"
    );
    assert!(result.results.iter().any(|r| r
        .error
        .as_deref()
        .unwrap_or("")
        .contains("tool 'fail' is forbidden")));
    // The decision was audited and attributed to the gate.
    let log = rt.log.lock().await;
    let ev = log
        .events()
        .iter()
        .find(|e| e.kind == car_eventlog::EventKind::AdmissionGateDecision)
        .expect("an AdmissionGateDecision event");
    assert_eq!(
        ev.data.get("gate").and_then(|v| v.as_str()),
        Some("block_tool")
    );
    assert_eq!(
        ev.data.get("decision").and_then(|v| v.as_str()),
        Some("reject")
    );
    let boundaries: Vec<_> = log
        .events()
        .iter()
        .filter(|event| {
            matches!(
                event.kind,
                car_eventlog::EventKind::StateCommitted | car_eventlog::EventKind::StateRollback
            )
        })
        .collect();
    assert_eq!(boundaries.len(), 1);
    assert_eq!(boundaries[0].kind, car_eventlog::EventKind::StateRollback);
    assert_eq!(boundaries[0].data["attempted"], false);
    assert_eq!(boundaries[0].data["stage"], "proposal_rejection");
}

#[tokio::test]
async fn admission_gate_allows_clean_proposal() {
    let rt = setup_runtime().await;
    rt.register_admission_gate(Arc::new(BlockToolGate {
        forbidden_tool: "fail".to_string(),
    }))
    .await;
    // No forbidden tool → gate allows → execution proceeds.
    let result = rt
        .execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Succeeded));
    let log = rt.log.lock().await;
    let ev = log
        .events()
        .iter()
        .find(|e| e.kind == car_eventlog::EventKind::AdmissionGateDecision)
        .expect("an allow decision is still audited");
    assert_eq!(
        ev.data.get("decision").and_then(|v| v.as_str()),
        Some("allow")
    );
}

#[tokio::test]
async fn admission_needs_approval_fails_closed_without_transport() {
    let rt = setup_runtime().await;
    rt.register_admission_gate(Arc::new(AlwaysApprovalGate))
        .await;
    // Until A7 wires the approval transport, an escalation is a
    // fail-closed block with an explanatory reason — never fail-open.
    let result = rt
        .execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    assert!(result.results.iter().any(|r| r
        .error
        .as_deref()
        .unwrap_or("")
        .contains("requires human approval")));
}

#[tokio::test]
async fn admission_clear_gates_restores_default() {
    let rt = setup_runtime().await;
    rt.register_admission_gate(Arc::new(BlockToolGate {
        forbidden_tool: "fail".to_string(),
    }))
    .await;
    assert_eq!(rt.admission_gate_count().await, 1);
    rt.clear_admission_gates().await;
    assert_eq!(rt.admission_gate_count().await, 0);
    // After clearing, the previously-blocked proposal runs.
    let result = rt
        .execute(&proposal(vec![tool_call("fail", HashMap::new())]))
        .await;
    // 'fail' tool errors at execution (not admission) — the point is it
    // was admitted past the gate, i.e. not Rejected-at-admission.
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Failed));
}

#[tokio::test]
async fn load_project_policies_registers_rules() {
    let rt = setup_runtime().await;
    // A .car directory with a policies/ subdir holding one rule file.
    let car_dir = std::env::temp_dir().join(format!("car_a2_{}", uuid::Uuid::new_v4().simple()));
    let pol = car_dir.join("policies");
    fs::create_dir_all(&pol).unwrap();
    fs::write(
        pol.join("security.toml"),
        "deny_tool = [\"deploy\"]\ndeny_keyword = [\"rm -rf /\"]\n",
    )
    .unwrap();

    let count = rt.load_project_policies(&car_dir).await.unwrap();
    assert_eq!(count, 2);

    let names = rt.policies.read().await.policy_names();
    assert!(names.iter().any(|n| n == "deny_tool:deploy"));
    assert!(names.iter().any(|n| n == "deny_keyword:rm -rf /"));

    fs::remove_dir_all(&car_dir).ok();
}

#[tokio::test]
async fn information_flow_gate_blocks_secret_to_sink_in_executor() {
    let rt = setup_runtime().await;
    rt.register_tool("read_secret").await;
    rt.register_tool("http_request").await;
    // Label read_secret as Secret; register the info-flow gate from the
    // resulting config (http_request is a built-in sink).
    let mut labels = crate::flow::builtin_tool_labels();
    labels.insert(
        "read_secret".to_string(),
        car_verify::infoflow::ToolLabels {
            capability: Some("fs_read".to_string()),
            confidentiality: car_verify::infoflow::Confidentiality::Secret,
            ..Default::default()
        },
    );
    rt.register_admission_gate(Arc::new(crate::flow::InformationFlowGate::new(
        crate::flow::ToolLabelConfig {
            labels,
            ..Default::default()
        },
    )))
    .await;

    // a1 reads a secret into "data"; a2 (sink) depends on "data".
    let mut a1 = tool_call("read_secret", HashMap::new());
    a1.expected_effects = [("data".to_string(), Value::from(1))].into();
    let mut a2 = tool_call("http_request", HashMap::new());
    a2.state_dependencies = vec!["data".to_string()];

    let result = rt.execute(&proposal(vec![a1, a2])).await;
    // Blocked at admission — nothing runs.
    assert!(result
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    // Audited as an information_flow gate rejection.
    let log = rt.log.lock().await;
    assert!(log.events().iter().any(|e| {
        e.kind == car_eventlog::EventKind::AdmissionGateDecision
            && e.data.get("gate").and_then(|v| v.as_str()) == Some("information_flow")
            && e.data.get("decision").and_then(|v| v.as_str()) == Some("reject")
    }));
}

#[tokio::test]
async fn install_information_flow_gate_from_missing_dir_uses_builtins() {
    let rt = setup_runtime().await;
    rt.install_information_flow_gate("/nonexistent/.car")
        .await
        .unwrap();
    assert_eq!(rt.admission_gate_count().await, 1);
}

/// A gate that always escalates with a fixed fingerprint, to exercise the
/// durable-approval resolution (A7) deterministically.
struct FixedApprovalGate {
    fp: String,
}

#[async_trait::async_trait]
impl crate::admission::AdmissionGate for FixedApprovalGate {
    fn name(&self) -> &str {
        "fixed_approval"
    }
    async fn check(
        &self,
        _proposal: &ActionProposal,
        _ctx: &crate::admission::GateContext<'_>,
    ) -> crate::admission::GateOutcome {
        crate::admission::GateOutcome::NeedsApproval {
            actions: std::collections::HashSet::new(),
            fingerprint: self.fp.clone(),
            reason: "needs review".to_string(),
        }
    }
}

#[tokio::test]
async fn admission_approval_resolves_via_durable_ledger() {
    let rt = setup_runtime().await;
    rt.set_approval_ledger_in_memory().await;
    rt.register_admission_gate(Arc::new(FixedApprovalGate {
        fp: "fp-xyz".to_string(),
    }))
    .await;
    let p = || proposal(vec![tool_call("echo", HashMap::new())]);

    // 1. Unseen escalation → fail-closed (pending).
    let r1 = rt.execute(&p()).await;
    assert!(r1
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    assert!(r1.results.iter().any(|r| r
        .error
        .as_deref()
        .unwrap_or("")
        .contains("requires human approval")));

    // 2. Operator approves the fingerprint → proposal now runs.
    rt.approve_admission("fp-xyz", "operator", "looks fine")
        .await
        .unwrap();
    let r2 = rt.execute(&p()).await;
    assert!(
        r2.results
            .iter()
            .all(|r| r.status == ActionStatus::Succeeded),
        "approved escalation should execute"
    );

    // 3. A rejected fingerprint blocks (use a second gate/fp).
    rt.clear_admission_gates().await;
    rt.register_admission_gate(Arc::new(FixedApprovalGate {
        fp: "fp-no".to_string(),
    }))
    .await;
    rt.reject_admission("fp-no", "operator", "nope")
        .await
        .unwrap();
    let r3 = rt.execute(&p()).await;
    assert!(r3.results.iter().any(|r| r
        .error
        .as_deref()
        .unwrap_or("")
        .contains("rejected by operator")));
}

#[tokio::test]
async fn approval_ledger_persists_across_reload() {
    let dir = std::env::temp_dir().join(format!("car_a7_{}", uuid::Uuid::new_v4().simple()));
    fs::create_dir_all(&dir).unwrap();
    let path = dir.join("approvals.jsonl");

    // First runtime records an approval to the journal.
    {
        let rt = setup_runtime().await;
        rt.set_approval_ledger_path(&path).await.unwrap();
        rt.approve_admission("fp-persist", "op", "ok")
            .await
            .unwrap();
    }
    // A fresh runtime loading the same journal sees the decision.
    {
        let rt = setup_runtime().await;
        rt.set_approval_ledger_path(&path).await.unwrap();
        assert_eq!(
            rt.admission_decision("fp-persist").await,
            Some(car_policy::ApprovalDecision::Approved)
        );
    }
    fs::remove_dir_all(&dir).ok();
}

#[tokio::test]
async fn tool_receipts_catch_fabricated_claim_and_ground_true_claim() {
    use car_eventlog::tool_receipts::{ClaimKind, ToolClaim};
    let rt = setup_runtime().await;
    // Actually run the echo tool so a real receipt is logged.
    let r = rt
        .execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    assert!(r
        .results
        .iter()
        .all(|x| x.status == ActionStatus::Succeeded));

    // A truthful claim ("I invoked echo") is grounded — scoped to the
    // proposal the claim is about.
    let grounded = rt
        .verify_tool_receipts(
            &[ToolClaim {
                kind: ClaimKind::Invoked,
                tool: "echo".to_string(),
                call_id: None,
                count: None,
                text: None,
            }],
            Some("test-proposal"),
        )
        .await;
    assert!(grounded.grounded, "a real invocation must be grounded");

    // A fabricated claim ("I invoked deploy") is caught.
    let report = rt
        .verify_tool_receipts(
            &[ToolClaim {
                kind: ClaimKind::Invoked,
                tool: "deploy".to_string(),
                call_id: None,
                count: None,
                text: Some("I deployed it".to_string()),
            }],
            Some("test-proposal"),
        )
        .await;
    assert!(!report.grounded);
    assert_eq!(
        report.hallucinations[0].kind,
        car_eventlog::tool_receipts::HallucinationKind::FabricatedToolReference
    );
    // The hallucination was audited.
    let log = rt.log.lock().await;
    assert!(log
        .events()
        .iter()
        .any(|e| e.kind == car_eventlog::EventKind::ToolReceiptHallucination));
}

#[tokio::test]
async fn tool_receipts_after_retention_trim_are_ungroundable_not_fabricated() {
    // Regression (review A6): the receipts projection used to read the
    // trimmed in-memory log, so after a retention trim a *truthful* claim
    // about an evicted tool call was flagged `fabricated_tool_reference`.
    // It must come back as the distinct, non-accusatory "window evicted".
    use car_eventlog::tool_receipts::{ClaimKind, ToolClaim};
    let rt = setup_runtime().await;
    let r = rt
        .execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    assert!(r
        .results
        .iter()
        .all(|x| x.status == ActionStatus::Succeeded));

    // Evict the proposal's events (including its ProposalReceived marker).
    {
        let mut log = rt.log.lock().await;
        let removed = log.enforce_retention(
            &car_eventlog::RetentionPolicy {
                max_events: Some(0),
                max_age_secs: None,
            },
            chrono::Utc::now(),
        );
        assert!(removed > 0, "retention must have trimmed the run's events");
    }

    let truthful = ToolClaim {
        kind: ClaimKind::Invoked,
        tool: "echo".to_string(),
        call_id: None,
        count: None,
        text: None,
    };
    // Scoped to the evicted proposal: ungroundable, never fabricated.
    let report = rt
        .verify_tool_receipts(std::slice::from_ref(&truthful), Some("test-proposal"))
        .await;
    assert!(
        report.hallucinations.is_empty(),
        "an evicted-window claim must not be flagged: {:?}",
        report.hallucinations
    );
    assert!(report.grounded, "non-accusatory outcome");
    assert_eq!(report.ungroundable.len(), 1);
    assert!(report.ungroundable[0]
        .explanation
        .contains("window evicted"));

    // Unscoped over a trimmed log is equally unknowable — same outcome.
    let report = rt
        .verify_tool_receipts(std::slice::from_ref(&truthful), None)
        .await;
    assert!(report.hallucinations.is_empty());
    assert_eq!(report.ungroundable.len(), 1);
}

#[tokio::test]
async fn idempotency_cache_persists_across_restart() {
    let dir = std::env::temp_dir().join(format!("car_c3_{}", uuid::Uuid::new_v4().simple()));
    fs::create_dir_all(&dir).unwrap();
    let path = dir.join("idempotency.jsonl");

    let mut idem = tool_call("echo", HashMap::new());
    idem.idempotent = true;

    // First runtime: durable cache, execute the idempotent action once.
    {
        let rt = setup_runtime().await;
        rt.set_idempotency_cache_path(&path).await.unwrap();
        let r = rt.execute(&proposal(vec![idem.clone()])).await;
        assert!(r
            .results
            .iter()
            .all(|x| x.status == ActionStatus::Succeeded));
    }

    // Second runtime ("restart"): same journal. Re-submitting the same
    // idempotent action returns the cached result (ActionDeduplicated)
    // rather than re-executing it.
    {
        let rt = setup_runtime().await;
        let loaded = rt.set_idempotency_cache_path(&path).await.unwrap();
        assert_eq!(loaded, 1, "one cached entry should load from the journal");
        let r = rt.execute(&proposal(vec![idem.clone()])).await;
        assert!(r
            .results
            .iter()
            .all(|x| x.status == ActionStatus::Succeeded));
        let log = rt.log.lock().await;
        assert!(
            log.events()
                .iter()
                .any(|e| e.kind == car_eventlog::EventKind::ActionDeduplicated),
            "the reloaded entry should dedupe the re-submitted action"
        );
    }
    fs::remove_dir_all(&dir).ok();
}

#[tokio::test]
async fn event_log_hash_chain_verifies_after_execution() {
    let rt = setup_runtime().await;
    rt.enable_event_log_hash_chaining().await;
    // Execute a couple of proposals so real events are appended + chained.
    rt.execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    rt.execute(&proposal(vec![tool_call("echo", HashMap::new())]))
        .await;
    // The chain verifies and covers a non-trivial number of events.
    let verified = rt.verify_event_log_chain().await.expect("chain intact");
    assert!(verified > 0);
}

#[tokio::test]
async fn load_project_policies_missing_dir_is_ok() {
    let rt = setup_runtime().await;
    let count = rt
        .load_project_policies("/nonexistent/project/.car")
        .await
        .unwrap();
    assert_eq!(count, 0);
}

// --- harness config drives live retry behavior (Evolution Agent, §3.5) ---

#[tokio::test]
async fn harness_config_caps_retries() {
    let rt = setup_runtime().await;
    let mut a = tool_call("fail", HashMap::new());
    a.failure_behavior = FailureBehavior::Retry;
    a.max_retries = 3;

    // No config installed → prior behavior: 3 retries (4 attempts).
    rt.execute(&proposal(vec![a.clone()])).await;
    let baseline_retries = {
        let log = rt.log.lock().await;
        log.events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::ActionRetrying)
            .count()
    };
    assert_eq!(baseline_retries, 3);

    // Install a harness config capping retries to 0 (an applied retry-config
    // mutation) → the same action now makes a single attempt.
    rt.set_harness_config(car_memgine::HarnessConfig {
        max_retries: 0,
        ..Default::default()
    })
    .await;
    let before = rt.log.lock().await.events().len();
    rt.execute(&proposal(vec![a])).await;
    let log = rt.log.lock().await;
    let capped_retries = log.events()[before..]
        .iter()
        .filter(|e| e.kind == car_eventlog::EventKind::ActionRetrying)
        .count();
    assert_eq!(capped_retries, 0, "harness config must cap retries live");
}

#[tokio::test]
async fn harness_config_is_none_until_installed() {
    let rt = setup_runtime().await;
    assert!(rt.harness_config().await.is_none());
    rt.set_harness_config(car_memgine::HarnessConfig::default())
        .await;
    assert!(rt.harness_config().await.is_some());
}

// --- C2: detached (streaming / long-running) tool dispatch ---

/// Executor whose `execute_stream` yields controllable chunks: emits one
/// Text chunk, then holds the stream open until the sender is dropped or
/// the runtime cancels (receiver drop). Never terminates on its own —
/// the point is proving the DAG does not wait for it.
struct StreamingExecutor {
    started: Arc<AtomicU32>,
}

#[async_trait::async_trait]
impl ToolExecutor for StreamingExecutor {
    async fn execute(&self, tool: &str, _params: &Value) -> Result<Value, String> {
        match tool {
            "echo" => Ok(Value::from("oneshot")),
            _ => Err(format!("unknown tool: {tool}")),
        }
    }

    async fn execute_stream(
        &self,
        tool: &str,
        _params: &Value,
        _action_id: &str,
    ) -> Result<tokio::sync::mpsc::Receiver<car_ir::ToolStreamChunk>, String> {
        if tool != "tail_forever" {
            return Err(format!("tool '{tool}': streaming unsupported"));
        }
        self.started.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = tokio::sync::mpsc::channel(8);
        tokio::spawn(async move {
            let _ = tx
                .send(car_ir::ToolStreamChunk::Text {
                    text: "first line".into(),
                })
                .await;
            // Hold the stream open "forever" (test-bounded): a cancel
            // drops the receiver, which closes this send loop.
            loop {
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                if tx
                    .send(car_ir::ToolStreamChunk::Progress {
                        fraction: 0.1,
                        message: None,
                    })
                    .await
                    .is_err()
                {
                    break;
                }
            }
        });
        Ok(rx)
    }
}

#[tokio::test]
async fn detached_tool_returns_handle_without_blocking_dag() {
    let started = Arc::new(AtomicU32::new(0));
    let rt = Runtime::new().with_executor(Arc::new(StreamingExecutor {
        started: started.clone(),
    }));
    rt.register_tool("tail_forever").await;
    rt.register_tool("echo").await;

    let mut detached = tool_call("tail_forever", HashMap::new());
    detached.invocation_mode = car_ir::ToolInvocationMode::LongRunning;
    let oneshot = tool_call("echo", HashMap::new());
    let p = proposal(vec![detached.clone(), oneshot]);

    // The whole proposal must complete promptly even though the detached
    // tool never finishes — that's the C2 contract.
    let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), rt.execute(&p))
        .await
        .expect("proposal must not block on the long-running tool");
    let results = outcome.results;
    assert_eq!(results.len(), 2);
    assert_eq!(started.load(Ordering::SeqCst), 1);

    // The detached action's output is a live handle, not a result.
    let detached_result = results.iter().find(|r| r.action_id == detached.id).unwrap();
    assert_eq!(detached_result.status, ActionStatus::Succeeded);
    let handle_id = detached_result
        .output
        .as_ref()
        .and_then(|o| o.get("tool_handle"))
        .and_then(|h| h.as_str())
        .expect("output carries tool_handle")
        .to_string();

    // Poll: first chunk is there, status still running.
    let mut saw_text = false;
    for _ in 0..100 {
        if let Some(poll) = rt.tool_poll(&handle_id).await {
            if poll
                .chunks
                .iter()
                .any(|c| matches!(c, car_ir::ToolStreamChunk::Text { .. }))
            {
                saw_text = true;
                assert_eq!(poll.status, car_ir::ToolStatus::Running);
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    assert!(saw_text, "streamed chunk reaches poll while running");

    // Mid-stream cancel seals the handle and the executor's send loop
    // observes the dropped receiver.
    assert!(rt.tool_cancel(&handle_id).await);
    let poll = rt
        .tool_poll(&handle_id)
        .await
        .expect("handle still queryable");
    assert_eq!(poll.status, car_ir::ToolStatus::Cancelled);
}

#[tokio::test]
async fn detached_tool_against_oneshot_executor_is_rejected() {
    // TestExecutor doesn't implement execute_stream — the default
    // "unsupported" error must surface as a failed action, not a hang.
    let rt = setup_runtime().await;
    let mut a = tool_call("echo", HashMap::new());
    a.invocation_mode = car_ir::ToolInvocationMode::Streaming;
    a.max_retries = 0;
    let p = proposal(vec![a]);
    let outcome = rt.execute(&p).await;
    assert_eq!(outcome.results[0].status, ActionStatus::Failed);
    assert!(outcome.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("does not support streaming"));
}

// --- Outbound human messaging (`messaging.send`) ---
//
// The point of the built-in is that reaching a human travels the SAME governed
// path as any other side effect. These tests pin the three properties that
// make that true: it is only advertised when it can actually run, policy can
// stop it, and an unconfigured runtime refuses rather than quietly handing the
// send to the host executor.

/// Records what the runtime handed the sink, so the tests can assert the tool
/// parameters were parsed into a message rather than passed through raw.
struct FakeSink {
    sent: std::sync::Mutex<Vec<crate::messaging::OutboundMessage>>,
}

impl FakeSink {
    fn new() -> Self {
        Self {
            sent: std::sync::Mutex::new(Vec::new()),
        }
    }

    fn sent(&self) -> Vec<crate::messaging::OutboundMessage> {
        self.sent.lock().unwrap().clone()
    }
}

#[async_trait::async_trait]
impl crate::messaging::MessageSink for FakeSink {
    async fn channels(&self) -> Vec<String> {
        vec!["imessage".to_string()]
    }

    async fn send(
        &self,
        msg: &crate::messaging::OutboundMessage,
    ) -> Result<crate::messaging::MessageReceipt, String> {
        self.sent.lock().unwrap().push(msg.clone());
        Ok(crate::messaging::MessageReceipt::delivered("imessage").with_message_id("m-1"))
    }
}

fn messaging_action(params: HashMap<String, Value>) -> Action {
    let mut a = tool_call("messaging.send", params);
    a.max_retries = 0;
    a
}

fn message_params() -> HashMap<String, Value> {
    [
        ("channel".to_string(), Value::from("imessage")),
        ("to".to_string(), Value::from("+15551112222")),
        ("body".to_string(), Value::from("build is green")),
        ("idempotency_key".to_string(), Value::from("run-42")),
    ]
    .into()
}

#[tokio::test]
async fn test_messaging_send_is_only_advertised_with_a_sink() {
    let bare = Runtime::new();
    assert!(
        !bare
            .tool_schemas()
            .await
            .iter()
            .any(|s| s.name == "messaging.send"),
        "a runtime with no sink must not advertise a tool it cannot execute"
    );

    let wired = Runtime::new().with_message_sink(Arc::new(FakeSink::new()));
    assert!(wired
        .tool_schemas()
        .await
        .iter()
        .any(|s| s.name == "messaging.send"));
    let entry = wired.registry.get("messaging.send").await.unwrap();
    assert_eq!(entry.permission, crate::registry::ToolPermission::AskUser);
    assert!(entry.side_effects);
    assert_eq!(entry.category.as_deref(), Some("messaging"));
}

#[tokio::test]
async fn test_messaging_send_without_sink_is_refused() {
    // The schema is registered by hand so validation passes and dispatch is
    // actually reached — otherwise this would only prove the validator works.
    let rt = make_runtime();
    rt.register_tool_schema(car_ir::builtins::messaging_send())
        .await;

    let outcome = rt
        .execute(&proposal(vec![messaging_action(message_params())]))
        .await;
    assert_eq!(outcome.results[0].status, ActionStatus::Failed);
    let err = outcome.results[0].error.as_ref().unwrap();
    assert!(
        err.contains("messaging is not configured on this runtime"),
        "{err}"
    );
    // No fall-through: TestExecutor's own "unknown tool" error (and the
    // agent_basics fallback behind it) must never be reached, or an ungoverned
    // host transport could answer the call.
    assert!(!err.contains("unknown tool"), "{err}");
    assert!(!err.contains("no handler"), "{err}");
}

#[tokio::test]
async fn test_messaging_send_returns_the_serialized_receipt() {
    let sink = Arc::new(FakeSink::new());
    let rt = Runtime::new().with_message_sink(sink.clone());

    let outcome = rt
        .execute(&proposal(vec![messaging_action(message_params())]))
        .await;
    assert!(outcome.all_succeeded(), "{:?}", outcome.results[0].error);
    assert_eq!(
        outcome.results[0].output,
        Some(serde_json::json!({
            "channel": "imessage",
            "message_id": "m-1",
            "deduplicated": false,
        }))
    );

    let sent = sink.sent();
    assert_eq!(sent.len(), 1);
    assert_eq!(sent[0].channel, "imessage");
    assert_eq!(
        sent[0].to,
        crate::messaging::Recipient::Direct("+15551112222".to_string())
    );
    assert_eq!(sent[0].body, "build is green");
    assert_eq!(sent[0].idempotency_key.as_deref(), Some("run-42"));
}

#[tokio::test]
async fn test_messaging_send_bad_params_never_reach_the_sink() {
    let sink = Arc::new(FakeSink::new());
    let rt = Runtime::new().with_message_sink(sink.clone());

    // A blank body satisfies the JSON schema (it is a string) but is a bug —
    // the parser is the second line of defence, and it runs before the wire.
    let mut params = message_params();
    params.insert("body".to_string(), Value::from("   "));
    let outcome = rt.execute(&proposal(vec![messaging_action(params)])).await;
    assert_eq!(outcome.results[0].status, ActionStatus::Failed);
    assert!(outcome.results[0]
        .error
        .as_ref()
        .unwrap()
        .contains("'body' must not be empty"));
    assert!(sink.sent().is_empty());
}

#[tokio::test]
async fn test_messaging_send_is_blocked_by_policy() {
    let sink = Arc::new(FakeSink::new());
    let rt = Runtime::new().with_message_sink(sink.clone());
    {
        let mut policies = rt.policies.write().await;
        policies.register(
            "no_outbound",
            Box::new(|action, _state| {
                if action.tool.as_deref() == Some("messaging.send") {
                    Some("outbound messaging denied by project policy".to_string())
                } else {
                    None
                }
            }),
            "deny messaging.send",
        );
    }

    let outcome = rt
        .execute(&proposal(vec![messaging_action(message_params())]))
        .await;
    assert_eq!(outcome.results[0].status, ActionStatus::Rejected);
    assert!(
        sink.sent().is_empty(),
        "a policy-denied send must never reach the transport"
    );
}

// --- A policy file an operator writes governs a real send (end to end) ---
//
// The tests above register a policy closure in code. These do not: they write
// a real `.car/policies/*.toml` on disk, load it with
// `Runtime::load_project_policies`, and then execute real `messaging.send`
// actions down the runtime's ordinary action path — the same path a proposal
// takes. That is the whole claim: governance sits ON the execution route, not
// merely beside it. Every refusal is checked twice, once for the verdict and
// once against the fake transport's call count, because "refused but delivered
// anyway" is the failure that matters.

/// The one recipient the fixture policy permits. Anyone else — and a send that
/// names nobody — is denied by the `allow_tool_param` rule.
const POLICY_ALLOWLISTED_RECIPIENT: &str = "+15551112222";

/// A recipient the fixture policy does not list.
const POLICY_UNLISTED_RECIPIENT: &str = "+15559998888";

/// How many `messaging.send` calls the fixture's `rate_limit_tool` rule admits.
/// Deliberately tiny so the cap is reachable in a test, paired with an hour-long
/// window so nothing here depends on wall-clock timing and nothing sleeps.
const POLICY_MAX_CALLS: usize = 2;

/// A `messaging.send` action addressed to `to` carrying `body`. Each send gets a
/// distinct idempotency key so a dedup path can never be mistaken for a policy
/// refusal.
fn governed_send(to: &str, body: &str, key: &str) -> Action {
    messaging_action(
        [
            ("channel".to_string(), Value::from("imessage")),
            ("to".to_string(), Value::from(to)),
            ("body".to_string(), Value::from(body)),
            ("idempotency_key".to_string(), Value::from(key)),
        ]
        .into(),
    )
}

/// A runtime governed by a policy file that exists on disk, plus the fake
/// transport behind it.
struct GovernedMessagingProject {
    /// Held so the tempdir outlives the runtime that loaded from it.
    _dir: tempfile::TempDir,
    rt: Runtime,
    sink: Arc<FakeSink>,
}

/// Write a real `.car/policies/outbound-messaging.toml` into a fresh tempdir,
/// build a runtime whose `messaging.send` is backed by a counting fake
/// transport, and load that file into it.
///
/// The file carries all three of the new declarative rule kinds, scoped to
/// `messaging.send`: an allowlist on the recipient, a regex content rule on the
/// body, and a call cap. It is written out longhand rather than constructed
/// through `PolicyRules`, so what these tests exercise is the same TOML an
/// operator would commit — a rename or a serde change in the file format fails
/// here.
///
/// Each test gets its own tempdir and its own runtime: the rate-limit rule holds
/// its window inside the registered closure, so a shared runtime would let one
/// test's sends spend another's budget.
async fn governed_messaging_project() -> GovernedMessagingProject {
    let dir = tempfile::tempdir().unwrap();
    let policies = dir.path().join(".car").join("policies");
    fs::create_dir_all(&policies).unwrap();
    fs::write(
        policies.join("outbound-messaging.toml"),
        format!(
            r#"
# Only this one person may be messaged. Any other recipient — or a send that
# names none — is denied.
[[allow_tool_param]]
tool  = "messaging.send"
param = "to"
allow = ["{POLICY_ALLOWLISTED_RECIPIENT}"]

# Never let a credential-shaped string leave in a message body. The shape here
# is a synthetic marker invented for this fixture, not any real credential
# format.
[[deny_tool_param_matching]]
tool    = "messaging.send"
param   = "body"
matches = "EXAMPLE-NOT-A-REAL-SECRET-[A-Z0-9]{{6}}"

# Bound how much outbound messaging a run can produce. The window is an hour so
# the cap is reached by call count alone, never by elapsed time.
[[rate_limit_tool]]
tool          = "messaging.send"
max_calls     = {POLICY_MAX_CALLS}
interval_secs = 3600.0
"#
        ),
    )
    .unwrap();

    let sink = Arc::new(FakeSink::new());
    let rt = Runtime::new().with_message_sink(sink.clone());
    let loaded = rt
        .load_project_policies(dir.path().join(".car"))
        .await
        .expect("the fixture policy file must parse");
    assert_eq!(loaded, 3, "all three rules in the file must be loaded");

    GovernedMessagingProject {
        _dir: dir,
        rt,
        sink,
    }
}

#[tokio::test]
async fn project_policy_allowlist_admits_the_allowlisted_recipient() {
    let p = governed_messaging_project().await;

    let outcome =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "build is green",
            "run-1",
        )]))
        .await;

    assert!(outcome.all_succeeded(), "{:?}", outcome.results[0].error);
    let sent = p.sink.sent();
    assert_eq!(
        sent.len(),
        1,
        "the allowlisted send must reach the transport"
    );
    assert_eq!(
        sent[0].to,
        crate::messaging::Recipient::Direct(POLICY_ALLOWLISTED_RECIPIENT.to_string())
    );
    assert_eq!(sent[0].body, "build is green");
}

#[tokio::test]
async fn project_policy_allowlist_refuses_an_unlisted_recipient_without_sending() {
    let p = governed_messaging_project().await;

    let outcome =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_UNLISTED_RECIPIENT,
            "build is green",
            "run-1",
        )]))
        .await;

    assert_eq!(outcome.results[0].status, ActionStatus::Rejected);
    assert!(
        p.sink.sent().is_empty(),
        "a refused send must never be handed to a transport — the verdict alone \
         does not prove the message did not go out"
    );
}

#[tokio::test]
async fn project_policy_content_rule_refuses_a_credential_shaped_body_without_sending() {
    let p = governed_messaging_project().await;

    // Addressed to the allowlisted recipient, so only the content rule can be
    // what refuses this.
    let leak =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "deploy key is EXAMPLE-NOT-A-REAL-SECRET-AB12CD",
            "run-1",
        )]))
        .await;
    assert_eq!(leak.results[0].status, ActionStatus::Rejected);
    assert!(
        p.sink.sent().is_empty(),
        "a body matching the denied pattern must never reach the transport"
    );

    // And the rule is not a blanket ban on the tool: an ordinary body to the
    // same recipient still goes through.
    let ordinary =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "deploy finished",
            "run-2",
        )]))
        .await;
    assert!(ordinary.all_succeeded(), "{:?}", ordinary.results[0].error);
    let sent = p.sink.sent();
    assert_eq!(sent.len(), 1);
    assert_eq!(sent[0].body, "deploy finished");
}

#[tokio::test]
async fn project_policy_rate_cap_refuses_the_send_after_max_calls_without_sending() {
    let p = governed_messaging_project().await;

    for i in 0..POLICY_MAX_CALLS {
        let outcome =
            p.rt.execute(&proposal(vec![governed_send(
                POLICY_ALLOWLISTED_RECIPIENT,
                "status update",
                &format!("run-{i}"),
            )]))
            .await;
        assert!(
            outcome.all_succeeded(),
            "send {i} is within the cap: {:?}",
            outcome.results[0].error
        );
    }

    let over =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "status update",
            "run-over",
        )]))
        .await;
    assert_eq!(over.results[0].status, ActionStatus::Rejected);
    assert_eq!(
        p.sink.sent().len(),
        POLICY_MAX_CALLS,
        "the transport must have seen exactly the calls the cap admits, and not \
         the one over it"
    );
}

#[tokio::test]
async fn project_policy_refusal_names_the_rule_that_fired() {
    let p = governed_messaging_project().await;

    // An operator reading a refusal has to be able to tell WHICH of their rules
    // stopped the send. The engine registers each declarative rule under a
    // stable name and the rejection reason carries it, so all three are checked
    // here against the names the loader produced.
    let names = p.rt.policies.read().await.policy_names();
    for expected in [
        "allow_tool_param:messaging.send.to",
        "deny_tool_param_matching:messaging.send.body",
        "rate_limit_tool:messaging.send",
    ] {
        assert!(
            names.iter().any(|n| n == expected),
            "expected a rule registered as '{expected}', got {names:?}"
        );
    }

    let refusal = |outcome: &ProposalResult| {
        assert_eq!(outcome.results[0].status, ActionStatus::Rejected);
        outcome.results[0].error.clone().unwrap()
    };

    let unlisted =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_UNLISTED_RECIPIENT,
            "build is green",
            "run-1",
        )]))
        .await;
    let err = refusal(&unlisted);
    assert!(
        err.contains("policy 'allow_tool_param:messaging.send.to'"),
        "{err}"
    );

    let leak =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "deploy key is EXAMPLE-NOT-A-REAL-SECRET-AB12CD",
            "run-2",
        )]))
        .await;
    let err = refusal(&leak);
    assert!(
        err.contains("policy 'deny_tool_param_matching:messaging.send.body'"),
        "{err}"
    );
    // The refusal names the rule without echoing the matched text — the matched
    // text is precisely the secret the rule exists to keep out of the log.
    assert!(!err.contains("EXAMPLE-NOT-A-REAL-SECRET-AB12CD"), "{err}");

    // Both refusals above still spent their slot in the window (budget is
    // consumed at admission), so the cap is now full and an otherwise-clean
    // send is refused by the third rule.
    let capped =
        p.rt.execute(&proposal(vec![governed_send(
            POLICY_ALLOWLISTED_RECIPIENT,
            "build is green",
            "run-3",
        )]))
        .await;
    let err = refusal(&capped);
    assert!(
        err.contains("policy 'rate_limit_tool:messaging.send'"),
        "{err}"
    );

    assert!(
        p.sink.sent().is_empty(),
        "not one of these three refusals may have reached the transport"
    );
}

// --- VIGIL runtime taint provenance (crate::taint) ---

/// Succeeds for every tool, so a proposal's admission verdict — not a
/// tool's availability — is what the taint tests observe.
struct AnyToolExecutor {
    state: Arc<car_state::StateStore>,
}

#[async_trait::async_trait]
impl ToolExecutor for AnyToolExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        if let (Some(key), Some(value)) = (
            params.get("runtime_state_key").and_then(Value::as_str),
            params.get("runtime_state_value"),
        ) {
            self.state.set(key, value.clone(), tool);
        }
        Ok(Value::from(format!("{tool} ok")))
    }

    async fn execute_with_action(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        _timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        if let (Some(key), Some(value)) = (
            params.get("runtime_state_key").and_then(Value::as_str),
            params.get("runtime_state_value"),
        ) {
            self.state.set(key, value.clone(), action_id);
        }
        Ok(Value::from(format!("{tool} ok")))
    }
}

fn taint_intent_config() -> crate::intent_gate::IntentGateConfig {
    crate::intent_gate::IntentGateConfig {
        intent: car_verify::intent::IntentSpec {
            allowed_tools: vec!["fetch_web".to_string(), "summarize".to_string()],
            ..Default::default()
        },
        untrusted_tools: vec!["fetch_web".to_string()],
        on_untainted_drift: None,
    }
}

async fn taint_runtime() -> Runtime {
    let rt = Runtime::new();
    let state = rt.state.clone();
    let rt = rt.with_executor(Arc::new(AnyToolExecutor { state }));
    rt.register_tool("fetch_web").await;
    rt.register_tool("summarize").await;
    rt.register_tool("send_payment").await;
    rt
}

/// Proposal 1: the untrusted fetch, writing state key `page`.
fn poisoning_proposal() -> ActionProposal {
    let mut fetch = tool_call("fetch_web", HashMap::new());
    fetch.max_retries = 0;
    fetch
        .expected_effects
        .insert("page".to_string(), Value::from("attacker text"));
    fetch
        .parameters
        .insert("runtime_state_key".to_string(), Value::from("page"));
    fetch.parameters.insert(
        "runtime_state_value".to_string(),
        Value::from("attacker text"),
    );
    ActionProposal {
        id: "p-fetch".to_string(),
        ..proposal(vec![fetch])
    }
}

/// Proposal 2: a SEPARATE proposal — no dependency edge to proposal 1 —
/// whose out-of-intent action reads the key proposal 1 wrote. This is the
/// replan shape.
fn replanned_payment_proposal() -> ActionProposal {
    let mut pay = tool_call("send_payment", HashMap::new());
    pay.max_retries = 0;
    pay.state_dependencies.push("page".to_string());
    ActionProposal {
        id: "p-replan".to_string(),
        ..proposal(vec![pay])
    }
}

/// The intent gate's own verdict for `proposal_id`, off the audit trail.
async fn intent_verdict(rt: &Runtime, proposal_id: &str) -> String {
    let log = rt.log.lock().await;
    log.events()
        .iter()
        .find(|e| {
            e.kind == car_eventlog::EventKind::AdmissionGateDecision
                && e.proposal_id.as_deref() == Some(proposal_id)
                && e.data.get("gate").and_then(|v| v.as_str()) == Some("intent")
        })
        .and_then(|e| e.data.get("decision").and_then(|v| v.as_str()))
        .unwrap_or("<no intent decision>")
        .to_string()
}

#[tokio::test]
async fn replanned_out_of_intent_action_is_hard_rejected_via_runtime_taint() {
    let rt = taint_runtime().await;
    rt.install_intent_gate(taint_intent_config()).await;

    // Proposal 1 is in intent and runs; the executor records that
    // `fetch_web`'s result landed in state key `page`.
    let first = rt.execute(&poisoning_proposal()).await;
    assert!(first.all_succeeded(), "{:?}", first.results);
    assert_eq!(intent_verdict(&rt, "p-fetch").await, "allow");
    let ledger = rt.taint_ledger().await.expect("intent gate installs one");
    assert!(ledger.tainted_keys(None).await.contains("page"));

    // Proposal 2 carries no edge back to proposal 1, but it READS `page`.
    // Runtime provenance makes it tool-stream-influenced, which is the
    // injection signature: a hard reject, not an approvable escalation.
    let second = rt.execute(&replanned_payment_proposal()).await;
    assert!(
        second
            .results
            .iter()
            .all(|r| r.status == ActionStatus::Rejected),
        "the replanned proposal must not execute: {:?}",
        second.results
    );
    assert_eq!(
        intent_verdict(&rt, "p-replan").await,
        "reject",
        "an out-of-intent action reading a tainted key is the injection signature"
    );
}

#[tokio::test]
async fn replanned_out_of_intent_action_only_escalates_without_runtime_taint() {
    // The control that proves what the ledger fixes. Same runtime, same
    // proposals, same intent config — but the gate registered WITHOUT the
    // ledger (the pre-change behavior). Proposal 2's taint lives in the
    // previous proposal, so the dependency DAG cannot see it and the
    // injection reads as the model's own drift: merely approvable.
    let rt = taint_runtime().await;
    rt.register_admission_gate(Arc::new(crate::intent_gate::IntentGate::new(
        taint_intent_config(),
    )))
    .await;
    assert!(
        rt.taint_ledger().await.is_none(),
        "registering the gate directly installs no ledger"
    );

    let first = rt.execute(&poisoning_proposal()).await;
    assert!(first.all_succeeded(), "{:?}", first.results);

    let second = rt.execute(&replanned_payment_proposal()).await;
    assert!(second
        .results
        .iter()
        .all(|r| r.status == ActionStatus::Rejected));
    assert_eq!(
        intent_verdict(&rt, "p-replan").await,
        "needs_approval",
        "without runtime provenance this is only an approval escalation"
    );
}

#[tokio::test]
async fn a_trusted_tool_laundering_a_tainted_key_is_still_caught() {
    // The other hole: `summarize` is in `allowed_tools` and is not an
    // untrusted tool, but it reads `page` and writes `digest`. An
    // out-of-intent action reading only `digest` must still hard-reject.
    let rt = taint_runtime().await;
    rt.install_intent_gate(taint_intent_config()).await;
    assert!(rt.execute(&poisoning_proposal()).await.all_succeeded());

    let mut summarize = tool_call("summarize", HashMap::new());
    summarize.max_retries = 0;
    summarize.state_dependencies.push("page".to_string());
    summarize
        .expected_effects
        .insert("digest".to_string(), Value::from("summary"));
    summarize
        .parameters
        .insert("runtime_state_key".to_string(), Value::from("digest"));
    summarize
        .parameters
        .insert("runtime_state_value".to_string(), Value::from("summary"));
    let laundered = ActionProposal {
        id: "p-launder".to_string(),
        ..proposal(vec![summarize])
    };
    assert!(rt.execute(&laundered).await.all_succeeded());

    let mut pay = tool_call("send_payment", HashMap::new());
    pay.max_retries = 0;
    pay.state_dependencies.push("digest".to_string());
    let p = ActionProposal {
        id: "p-after-launder".to_string(),
        ..proposal(vec![pay])
    };
    rt.execute(&p).await;
    assert_eq!(intent_verdict(&rt, "p-after-launder").await, "reject");
}

#[tokio::test]
async fn no_intent_gate_means_no_taint_ledger() {
    // The ledger is strictly opt-in with VIGIL: a runtime that never
    // installs the intent gate carries no ledger, so the per-action record
    // is one Option read and nothing more.
    let rt = taint_runtime().await;
    assert!(rt.taint_ledger().await.is_none());
    assert!(rt.execute(&poisoning_proposal()).await.all_succeeded());
    assert!(
        rt.taint_ledger().await.is_none(),
        "executing must not conjure a ledger"
    );

    rt.install_intent_gate(taint_intent_config()).await;
    assert!(rt.taint_ledger().await.is_some());
}

#[tokio::test]
async fn reinstalling_the_intent_gate_replaces_it_and_rebinds_the_ledger() {
    // Reinstalling is a real operation — an operator edits
    // `.car/intent.json` and re-runs `install_intent_gate_from_project`.
    // Each install builds a FRESH ledger, so appending a second gate would
    // leave the first one registered holding a ledger the executor no
    // longer writes to. Gate aggregation is fail-closed, so that frozen
    // gate's verdict would win: every key tainted before the reload would
    // stay tainted forever, no trusted overwrite could ever clear it, and
    // legitimate out-of-intent work would hard-refuse with no approval path.
    let rt = taint_runtime().await;
    rt.install_intent_gate(taint_intent_config()).await;
    assert!(rt.execute(&poisoning_proposal()).await.all_succeeded());
    let first_ledger = rt.taint_ledger().await.expect("intent gate installs one");
    assert!(first_ledger.tainted_keys(None).await.contains("page"));

    rt.install_intent_gate(taint_intent_config()).await;
    assert_eq!(
        rt.admission_gate_names()
            .await
            .iter()
            .filter(|n| n.as_str() == "intent")
            .count(),
        1,
        "a reinstall must replace the intent gate, not add a second one: {:?}",
        rt.admission_gate_names().await
    );

    // The live ledger is a NEW one. Re-poison through it so the clear below
    // proves something — with a fresh empty ledger the key is trivially
    // untainted, which is not the property under test.
    let live_ledger = rt.taint_ledger().await.expect("still installed");
    assert!(
        !Arc::ptr_eq(&first_ledger, &live_ledger),
        "each install builds a fresh ledger"
    );
    let repoison = ActionProposal {
        id: "p-fetch-2".to_string(),
        ..poisoning_proposal()
    };
    assert!(rt.execute(&repoison).await.all_succeeded());
    assert!(
        live_ledger.tainted_keys(None).await.contains("page"),
        "the executor must write to the ledger the reinstall bound"
    );

    // A trusted, in-intent overwrite of `page` executed AFTER the reinstall
    // must clear the taint, and the gate the runtime consults must see it.
    let mut overwrite = tool_call("summarize", HashMap::new());
    overwrite.max_retries = 0;
    overwrite
        .expected_effects
        .insert("page".to_string(), Value::from("operator-authored text"));
    overwrite
        .parameters
        .insert("runtime_state_key".to_string(), Value::from("page"));
    overwrite.parameters.insert(
        "runtime_state_value".to_string(),
        Value::from("operator-authored text"),
    );
    let clean = ActionProposal {
        id: "p-overwrite".to_string(),
        ..proposal(vec![overwrite])
    };
    assert!(rt.execute(&clean).await.all_succeeded());
    assert!(
        !live_ledger.tainted_keys(None).await.contains("page"),
        "a trusted overwrite must clear the key in the LIVE ledger"
    );

    // A stale gate bound to the orphaned ledger would still hard-reject
    // here — forever, with no approval path.
    rt.execute(&replanned_payment_proposal()).await;
    assert_eq!(
        intent_verdict(&rt, "p-replan").await,
        "needs_approval",
        "after a trusted overwrite the reader is untainted drift, not an injection"
    );
}

/// The same intent as [`taint_intent_config`], plus a resource scope: the
/// task may write `page` / `digest` and nothing else. `send_payment` is out
/// of intent by TOOL; an `outbox/…` write is out of intent by TARGET even
/// when the tool making it is allowed and trusted.
fn scoped_taint_intent_config() -> crate::intent_gate::IntentGateConfig {
    crate::intent_gate::IntentGateConfig {
        intent: car_verify::intent::IntentSpec {
            allowed_tools: vec!["fetch_web".to_string(), "summarize".to_string()],
            allowed_resources: vec!["page".to_string(), "digest".to_string()],
            ..Default::default()
        },
        untrusted_tools: vec!["fetch_web".to_string()],
        on_untainted_drift: None,
    }
}

/// A single action whose tool IS in `allowed_tools` and is NOT in
/// `untrusted_tools`, which reads the poisoned key and writes a target
/// outside the intent's resource scope.
fn trusted_tool_out_of_scope_proposal() -> ActionProposal {
    let mut summarize = tool_call("summarize", HashMap::new());
    summarize.max_retries = 0;
    summarize.state_dependencies.push("page".to_string());
    summarize
        .expected_effects
        .insert("outbox/exfil".to_string(), Value::from("attacker text"));
    summarize
        .parameters
        .insert("runtime_state_key".to_string(), Value::from("outbox/exfil"));
    summarize.parameters.insert(
        "runtime_state_value".to_string(),
        Value::from("attacker text"),
    );
    ActionProposal {
        id: "p-scoped".to_string(),
        ..proposal(vec![summarize])
    }
}

#[tokio::test]
async fn a_trusted_tool_writing_out_of_scope_is_rejected_when_tainted() {
    // The literal claim: an out-of-intent action influenced by a tainted
    // result is flagged even when its TOOL is otherwise trusted. Every other
    // test here flags an action whose tool is out of intent (`send_payment`).
    // Here `summarize` is in `allowed_tools` and is not untrusted; what is out
    // of intent is its TARGET (`car_verify::intent` raises TargetOutOfIntent
    // when an action's effective write set escapes `allowed_resources`). It
    // hard-rejects only because the ledger marks it as reading a tainted key.
    let rt = taint_runtime().await;
    rt.install_intent_gate(scoped_taint_intent_config()).await;

    // The poisoning fetch is fully in intent — allowed tool, and `page` is
    // inside the resource scope — so it admits and runs.
    assert!(rt.execute(&poisoning_proposal()).await.all_succeeded());
    assert_eq!(intent_verdict(&rt, "p-fetch").await, "allow");

    let p = trusted_tool_out_of_scope_proposal();
    let outcome = rt.execute(&p).await;
    assert!(
        outcome
            .results
            .iter()
            .all(|r| r.status == ActionStatus::Rejected),
        "the out-of-scope write must not execute: {:?}",
        outcome.results
    );
    assert_eq!(
        intent_verdict(&rt, "p-scoped").await,
        "reject",
        "a trusted tool writing out of scope while reading a tainted key is the injection signature"
    );
}

#[tokio::test]
async fn a_trusted_tool_writing_out_of_scope_only_escalates_without_the_ledger() {
    // The paired control that makes the claim above mean something: same
    // runtime, same intent, same proposal — but the gate registered WITHOUT
    // the ledger. Nothing marks `summarize` (its tool is trusted and the
    // poisoned write is in a previous proposal), so the out-of-scope write
    // reads as the model's own drift: approvable, not refused.
    let rt = taint_runtime().await;
    rt.register_admission_gate(Arc::new(crate::intent_gate::IntentGate::new(
        scoped_taint_intent_config(),
    )))
    .await;
    assert!(rt.taint_ledger().await.is_none());

    assert!(rt.execute(&poisoning_proposal()).await.all_succeeded());
    rt.execute(&trusted_tool_out_of_scope_proposal()).await;
    assert_eq!(
        intent_verdict(&rt, "p-scoped").await,
        "needs_approval",
        "without runtime provenance this is only an approval escalation"
    );
}