car-engine 0.49.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 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)),
        }
    }
}

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.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;
    assert_eq!(result.results[0].status, ActionStatus::Failed);
    assert_eq!(result.results[1].status, ActionStatus::Skipped);
}

#[tokio::test]
async fn test_skip_continues() {
    let rt = setup_runtime().await;
    let mut fail_action = tool_call("fail", HashMap::new());
    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);
}

// --- 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.failure_behavior = FailureBehavior::Abort;

    let p = proposal(vec![state_write("x", Value::from(1)), fail_action]);
    rt.execute(&p).await;

    assert_eq!(rt.state.get("x"), None);
}

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

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

// --- 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 test_expected_effects() {
    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();

    rt.execute(&proposal(vec![action])).await;
    assert_eq!(rt.state.get("result"), Some(Value::from(3)));
}

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

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

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

#[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());
}

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

#[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")
    );
}

#[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;

#[async_trait::async_trait]
impl ToolExecutor for AnyToolExecutor {
    async fn execute(&self, tool: &str, _params: &Value) -> Result<Value, String> {
        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().with_executor(Arc::new(AnyToolExecutor));
    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"));
    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"));
    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"));
    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"));
    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"
    );
}