car_server_core/assistant/

mod.rs

//! Parslee Core — the flagship, general-purpose agent that ships in the `car`
//! binary and works out of the box (`car do`).
//!
//! Unlike the coder (coding-specific) or the create-car-agent skill (build your
//! own), this is a batteries-included assistant: files + a real shell + web +
//! durable memory, sandbox-first for safety, driven by CAR inference through a
//! full [`Runtime`] (validator, policy, permission tiers, event log). One core
//! backs three entry modes — one-shot, REPL, and the conversational
//! `agent.chat` surface.
//!
//! ## Module map
//! - [`executor`] — [`GeneralExecutor`], the substrate-bound tool executor
//!   (agent_basics + `calculate` + `shell` + network delegate).
//! - [`net_tools`] — host-side `http_request` / `web_search` (bypass the
//!   sandbox's `--network none`).
//! - [`substrate`] — sandbox-first environment selection ([`bind_default_substrate`]).
//! - [`identity_tools`] — the gated `set_assistant_name` tool.
//! - [`calendar_tools`] — local EventKit reads plus approval-gated mutations.
//! - [`mail_tools`] — local Mail.app reads/drafts plus approval-gated send.
//! - [`policy`] — the assistant inspector chain (reuses the coder's footgun set).
//! - [`prompt`] — batch vs. conversational system prompts.
//! - [`agent_loop`] — the propose→validate→execute→observe loop.
//! - [`do_json`] — the `car.do/1` envelope: progress events plus the terminal
//!   document, shared by `car do --json` and the MCP run registry.
//!
//! [`Runtime`]: car_engine::Runtime
//! [`GeneralExecutor`]: executor::GeneralExecutor
//! [`bind_default_substrate`]: substrate::bind_default_substrate

pub mod agent_loop;
pub mod automation_tools;
pub mod browser_control;
pub mod browser_producer;
pub mod browser_stream;
pub mod browser_tools;
pub mod calendar_tools;
pub mod chat;
pub mod device_tools;
pub mod do_json;
pub(crate) mod durability;
pub mod executor;
pub mod governance;
pub mod identity_tools;
pub mod m365_tools;
pub mod mail_tools;
pub mod media_tools;
pub mod memory;
pub mod net_tools;
pub mod policy;
pub mod production_gates;
pub mod prompt;
pub mod register;
pub mod studio_tools;
pub mod substrate;
pub mod todo;
pub mod tool_memory;
pub mod value_store;
pub mod vision_tools;

use std::path::PathBuf;
use std::sync::Arc;

use async_trait::async_trait;
use car_engine::{Runtime, ToolEntry, ToolExecutor, ToolSchema};
use car_eventlog::EventLog;
use car_inference::InferenceEngine;
use car_policy::permission::PermissionTier;
use serde_json::Value;

use memory::MemoryTools;
pub use memory::{MemorySync, NoteKind, SyncedFact};

pub use agent_loop::{
    run_assistant_goal_loop, run_assistant_loop, run_assistant_loop_cancellable,
    ungrounded_summary_claims, ApprovalDecision, ApprovalGate, AssistantConfig, AssistantEvent,
    AssistantFailureCause, AssistantModelAttribution, AssistantOutcome, AssistantToolReceipt,
    AuthRequiredReason, GoalLoopResult, AUTH_REQUIRED_EXPIRED_MESSAGE,
    AUTH_REQUIRED_NO_WORKSPACE_MESSAGE, AUTH_REQUIRED_SIGNED_OUT_MESSAGE,
};
pub use chat::{AssistantService, ChatGoal};
pub use device_tools::DeviceProvider;
pub use executor::GeneralExecutor;
pub use net_tools::NetTools;
pub use substrate::{bind_default_substrate, BoundEnvironment, DEFAULT_ASSISTANT_IMAGE};

/// An assembled assistant runtime: the [`Runtime`] to drive, the model-visible
/// tool list, and the environment metadata for the system prompt.
/// One line naming the host OS and the shell the `shell` tool actually uses
/// there, for the local-substrate environment description.
///
/// Windows gets the concrete negative list rather than just "cmd.exe". Saying
/// "this is Windows" is not enough on its own: a model that has seen a million
/// POSIX transcripts will still reach for `grep`, and under `cmd /C` that is not
/// a slightly-wrong command, it is `'grep' is not recognized` — which the coder
/// then reads as its own broken code rather than as a shell mismatch.
fn host_shell_note() -> String {
    if cfg!(windows) {
        "Host platform: Windows. The `shell` tool runs each command through \
         `cmd /C` — this is cmd.exe, NOT a POSIX shell. `ls`, `grep`, `cat`, \
         `head`, `tail`, `rm`, `cp`, `mv`, `which`, `touch` and `export` do not \
         exist, and neither does `$(...)` command substitution or single-quote \
         quoting. Use `dir`, `findstr`, `type`, `del`, `copy`, `move`, `where`, \
         `set` and `%VAR%`. Paths use backslashes and drive letters. Prefer the \
         file tools over shell text-munging wherever they cover the job."
            .to_string()
    } else {
        format!(
            "Host platform: {}. The `shell` tool runs each command through `sh -c`.",
            std::env::consts::OS
        )
    }
}

pub struct AssistantRuntime {
    /// The configured runtime (validator + policy + tiers + event log), whose
    /// tool executor is the [`GeneralExecutor`].
    pub runtime: Runtime,
    /// The model-visible tools (from `GeneralExecutor::all_tool_defs()`).
    pub tools: Vec<Value>,
    /// Environment description for the system prompt: the one-line
    /// substrate sentence, plus — on a local (non-sandboxed) session — an
    /// appended names-only, depth-bounded workspace snapshot (F7/L1).
    pub description: String,
    /// The name this user chose for the assistant, plus the spoken aliases it
    /// answers to. Loaded once here rather than at each prompt-building call
    /// site, so every entry mode — one-shot, REPL, MCP, the coder's discussion
    /// — agrees on who the agent is. Falls back to the shipped default when
    /// `identity.json` is missing or unreadable; `car identity` is the surface
    /// that reports a broken record.
    pub identity: car_identity::AssistantIdentity,
    /// Whether execution is isolated in a container.
    pub sandboxed: bool,
    /// Tools that require human approval before running under the standing tier
    /// (writes/shell on the local host without `--full-access`); empty when the
    /// tier auto-allows everything. Feed into `AssistantConfig::gated_tools`.
    pub gated_tools: Vec<String>,
    /// Shared assistant memory bank used by the loop's proactive memory pass and
    /// by the model-visible `remember` / `recall` tools.
    pub proactive_memory: Arc<MemoryTools>,
    /// The run's learned tool repairs — which call recovered which kind of tool
    /// failure, durable across sessions. Separate from `proactive_memory` on
    /// purpose: that bank holds facts about the USER, this one holds procedural
    /// trivia about the TOOLS, and mixing them would put `shell::exit_1` in
    /// front of a question about their dog. See [`tool_memory`].
    pub tool_memory: Arc<tool_memory::ToolMemory>,
    /// If the sandbox was requested but unavailable, why we fell back to local.
    pub fallback_notice: Option<String>,
    /// The run's browser (Chromium still un-launched until the first browse
    /// call). Exposed so the daemon can publish it as a `browser.view.*`
    /// view for the drawer to watch and drive — the drawer has to reach the
    /// browser an agent ACTUALLY uses, and this is the only handle on it.
    pub browser: Arc<browser_tools::BrowserTools>,
    /// The run's task list (Parslee-ai/car#814), shared with the executor that
    /// `todo_write` mutates. Exposed because rendering it is the loop's job:
    /// #814 item 2 (a per-turn state block) is the consumer, and without a
    /// handle here that change could not reach the state it needs to render.
    pub todos: Arc<tokio::sync::Mutex<todo::TodoList>>,
}

/// A delegate that tries each inner executor in turn, using the `unknown tool`
/// convention to fall through — so several tool families (network, memory) share
/// one `GeneralExecutor` delegate slot.
struct ChainedDelegate(Vec<Arc<dyn ToolExecutor>>);

#[async_trait]
impl ToolExecutor for ChainedDelegate {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        for ex in &self.0 {
            match ex.execute(tool, params).await {
                Err(e) if e.starts_with("unknown tool") => continue,
                other => return other,
            }
        }
        Err(format!("unknown tool: '{tool}'"))
    }

    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
    ) -> Result<Value, String> {
        for ex in &self.0 {
            match ex
                .execute_with_action_in_session(
                    tool, params, action_id, timeout_ms, session_id, attempt,
                )
                .await
            {
                Err(e) if e.starts_with("unknown tool") => continue,
                other => return other,
            }
        }
        Err(format!("unknown tool: '{tool}'"))
    }
}

/// Build a registry [`ToolSchema`] from a model-facing `{name, description,
/// parameters}` def, so any advertised tool can be registered for validation.
fn schema_from_def(def: &Value) -> ToolSchema {
    ToolSchema {
        name: def["name"].as_str().unwrap_or_default().to_string(),
        source: car_ir::ToolSourceKind::UserDefined,
        description: def["description"].as_str().unwrap_or_default().to_string(),
        parameters: def["parameters"].clone(),
        returns: None,
        idempotent: false,
        cache_ttl_secs: None,
        rate_limit: None,
    }
}

/// The names of advertised tools whose self-declared `"tier"` exceeds the
/// standing `tier` — these must be approval-gated (neo leak #3). A tool without
/// a `"tier"` field, or one at/below the standing tier, is not gated here.
fn tier_gated_tool_names(tools: &[Value], standing: PermissionTier) -> Vec<String> {
    tools
        .iter()
        .filter_map(|def| {
            let name = def.get("name").and_then(|v| v.as_str())?;
            let tier = PermissionTier::from_str_opt(def.get("tier").and_then(|v| v.as_str())?)?;
            (tier > standing).then(|| name.to_string())
        })
        .collect()
}

/// Every tool schema that the flagship assistant can advertise on any
/// supported host/configuration. This is the deterministic discoverability
/// catalog; [`build_assistant_runtime`] still filters availability at runtime
/// (models, credentials, platform, host connection, and requested delegation).
pub fn model_tool_catalog() -> Vec<Value> {
    let mut tools = GeneralExecutor::tool_defs();
    tools.extend(net_tools::net_tool_defs());
    tools.extend(MemoryTools::tool_defs());
    tools.extend(media_tools::catalog_tool_defs());
    tools.extend(studio_tools::studio_tool_defs());
    tools.extend(m365_tools::m365_tool_defs());
    tools.extend(vision_tools::catalog_tool_defs());
    tools.extend(automation_tools::catalog_tool_defs());
    tools.extend(browser_tools::browser_tool_defs());
    tools.extend(calendar_tools::calendar_tool_defs());
    tools.extend(mail_tools::mail_tool_defs());
    tools.extend(device_tools::DeviceTools::tool_defs());
    tools.extend(identity_tools::IdentityTools::tool_defs());
    tools.push(GeneralExecutor::events_query_def());
    tools.push(todo::tool_def());
    tools.push(agent_loop::delegate_tool_def(&tools));
    tools.sort_by(|left, right| left["name"].as_str().cmp(&right["name"].as_str()));
    tools.dedup_by(|left, right| left["name"] == right["name"]);
    tools
}

/// Default durable-memory path: `memory/assistant.json` under the CAR state
/// root — `CAR_HOME` when set, otherwise `~/.car` (HOME, or USERPROFILE on
/// Windows), and a relative `.car` when neither resolves.
///
/// Delegates to [`car_memgine::note_store::default_path`] rather than
/// recomputing the join, so the assistant and the MCP server point at the same
/// file **by construction**. They previously agreed only because two copies of
/// the same expression happened to match, and the MCP server did not use its
/// copy at all (car#972 §1). "The editor and `car do` share one memory" is the
/// whole point; it should not depend on nobody editing one of two literals.
pub(crate) fn default_memory_path() -> PathBuf {
    car_memgine::note_store::default_path()
}

/// Assemble an [`AssistantRuntime`] from an engine and a bound environment.
///
/// Registers the model-visible tools so the validator allows them (agent_basics
/// builtins + `shell` + `http_request` + `web_search`), binds the
/// [`GeneralExecutor`] as the executor and the environment's substrate, and
/// attaches an optional event-log journal.
///
/// `trajectories` is the directory for the execution-trace store. The assistant
/// is where most real tool execution happens, so without it the per-tool
/// success rates that `verify.monte_carlo` derives would be built almost
/// entirely from the daemon's `proposal.submit` path and miss the agent that
/// actually runs. It is an explicit `Option<PathBuf>` — mirroring `eventlog`
/// above — rather than defaulting to `~/.car/trajectories/`, because a test
/// that runs a deliberately-broken tool fifty times would otherwise write that
/// into the user's real history and permanently skew the rates the feature
/// reads back.
///
/// Fails only when `<root>/.car/policies/` holds a malformed rule file — a
/// security control that would silently not exist is worse than a startup that
/// refuses; see [`crate::session::apply_project_policies`].
/// `allow_delegate` advertises the loop-intercepted `delegate` sub-agent tool
/// (see `agent_loop::delegate_tool_def`). Pass `true` only for a surface whose
/// operator asked for a delegating run — `car do` one-shot / goal / `--json`.
pub async fn build_assistant_runtime(
    engine: Arc<InferenceEngine>,
    env: BoundEnvironment,
    eventlog: Option<PathBuf>,
    device_provider: Option<Arc<dyn DeviceProvider>>,
    memory_sync: Option<Arc<dyn MemorySync>>,
    trajectories: Option<PathBuf>,
    allow_delegate: bool,
) -> Result<AssistantRuntime, String> {
    build_assistant_runtime_with_tools(
        engine,
        env,
        eventlog,
        device_provider,
        memory_sync,
        trajectories,
        allow_delegate,
        Vec::new(),
        Vec::new(),
    )
    .await
}

/// Surface-scoped tools share the assistant's validator, policies and executor.
/// Callers supply only trusted schemas and executors, never model-provided ones.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn build_assistant_runtime_with_tools(
    engine: Arc<InferenceEngine>,
    env: BoundEnvironment,
    eventlog: Option<PathBuf>,
    device_provider: Option<Arc<dyn DeviceProvider>>,
    memory_sync: Option<Arc<dyn MemorySync>>,
    trajectories: Option<PathBuf>,
    allow_delegate: bool,
    extra_defs: Vec<Value>,
    extra_executors: Vec<Arc<dyn ToolExecutor>>,
) -> Result<AssistantRuntime, String> {
    // Mutations auto-allow unless the standing tier is ReadOnly (local host
    // without --full-access), in which case writes/shell need approval; clamp
    // file paths to root only off-sandbox.
    let mut gated_tools: Vec<String> = if matches!(env.tier, PermissionTier::ReadOnly) {
        ["write_file", "edit_file", "shell"]
            .iter()
            .map(|s| s.to_string())
            .collect()
    } else {
        Vec::new()
    };
    // Renaming the assistant is gated on EVERY session, including
    // `--full-access`. Unlike the writes above, the risk here is not what the
    // session may do — it is where the instruction came from. A rename can
    // arrive inside a fetched page, a file, or a recalled memory, and an
    // assistant that quietly starts answering to a name someone else picked is
    // an identity-spoof surface. One approval tap is the cheaper mistake.
    gated_tools.push("set_assistant_name".to_string());
    let clamp = !env.sandboxed;
    // Names-only, depth-bounded workspace snapshot (F7/L1): orient the model with
    // the repo layout up front instead of only a one-line environment sentence.
    // LOCAL substrate only — for a sandboxed or remote session we keep today's
    // one-liner and never touch the container/VM at prompt-build time (no docker
    // spin-up here). Best-effort: an unreadable/empty root yields nothing.
    let mut description = env.description.clone();
    if !env.sandboxed && env.substrate.is_local() {
        // Name the host platform and the shell it actually gets.
        //
        // Nothing else in the prompt path ever told the model which OS it was on
        // — `std::env::consts::OS` appeared in no prompt builder — while the
        // `shell` tool def said "executed via sh -c" on every platform. On Windows
        // `run_shell_on` dispatches `cmd /C`, so the model was briefed that it held
        // a POSIX shell it does not hold, and `ls`/`grep`/`cat` come back
        // "'grep' is not recognized" (car#1260 audit). The repo already guards this
        // for scripted fixtures — see `coder::test_cmds` — and this is the same
        // hazard one layer up.
        //
        // Local substrate only, deliberately: when a substrate is bound,
        // `run_shell_on` routes to `substrate.run_command`, and inside a Linux
        // container `sh -c` is true even on a Windows host. The sandboxed branch
        // keeps the substrate's own one-line description, which already says so.
        description.push_str("\n\n");
        description.push_str(&host_shell_note());
        let snapshot = substrate::workspace_snapshot(&env.root, 2, 2000);
        if !snapshot.is_empty() {
            description.push_str("\n\n");
            description.push_str(&snapshot);
        }
    }
    let sandboxed = env.sandboxed;
    let fallback_notice = env.fallback_notice.clone();

    // Delegate tools (host-side): network + durable memory. Both bypass the
    // substrate/path-clamp — network needs host egress, memory is CAR's graph.
    let net: Arc<dyn ToolExecutor> = Arc::new(NetTools::new());
    let mem = Arc::new(MemoryTools::open(default_memory_path()).with_sync(memory_sync));
    // Learned tool repairs (see `tool_memory`). Opening only READS the store —
    // every write is driven by the agent loop, and only when a surface opted in
    // by setting `AssistantConfig::tool_memory`. That split is what lets this be
    // unconditional here without a test run teaching the user's real assistant
    // that the way to fix a tool is whatever the fixture did fifty times.
    let tool_memory = Arc::new(tool_memory::ToolMemory::open(tool_memory::default_path()));
    // Media generation (image today) — host-side, backed by the inference
    // engine's local models. Advertises nothing when no image model is
    // available, so it never offers a tool it can't run. The capability a
    // text-only agent structurally cannot have.
    let media = Arc::new(media_tools::MediaTools::new(
        engine.clone(),
        env.root.clone(),
    ));
    // Parslee Studio media (music today) — host-side, via the Studio service on
    // CAR's existing Parslee bearer. Advertises nothing without a Parslee
    // session. Another capability a text-only agent structurally lacks.
    let studio = Arc::new(studio_tools::StudioMediaTools::new(env.root.clone()));
    // Local macOS Calendar — host-side, through EventKit. Unlike `m365_task`,
    // this needs no Parslee bearer or connected Microsoft account; it advertises
    // only after the non-prompting TCC readiness probe reports full access.
    let calendar = Arc::new(calendar_tools::CalendarTools::new());
    // Local macOS Mail — host-side, through Mail.app Automation. Like the local
    // calendar path, this needs no Parslee bearer or connected Microsoft account;
    // it advertises only after the non-prompting target-specific TCC probe grants
    // this process control of Mail.app.
    let mail = Arc::new(mail_tools::MailTools::new());
    // Parslee M365 — host-side, via the Parslee platform on CAR's existing
    // Parslee bearer. Delegates email/calendar/CRM/meeting work to the org's
    // already-agentic M365 employee. Advertises nothing without a Parslee
    // session. Another capability a text-only agent structurally lacks.
    let m365 = Arc::new(m365_tools::M365Tools::new());
    // Vision (image understanding) — host-side, via Apple Vision / Tesseract.
    // The CONSUMER counterpart to the generators: read text from an image (OCR)
    // and classify what it depicts. Advertises nothing when no vision backend is
    // present. A text-only agent can neither make nor read an image.
    let vision = Arc::new(vision_tools::VisionTools::new(env.root.clone()));
    // macOS automation ("control the Mac", AppleScript/JXA) — host-side, CANNOT
    // be sandboxed. Self-declares tier:full_access, so the tier-based gating
    // below routes it through approval unless the session is --full-access. A
    // capability no sandboxed or text-only agent has.
    let automation = Arc::new(automation_tools::AutomationTools::new());
    // Browser driving + session RECORDING — host-side. Chromium launches
    // lazily on first use, so a session that never browses pays nothing. The
    // recorder is what makes this more than automation: it captures the app
    // BEING USED (an answer streaming in, a table filling) rather than a still
    // of its final state. Another capability a text-only agent structurally
    // lacks.
    let browser = Arc::new(browser_tools::BrowserTools::new(env.root.clone()));
    let device_tools = device_provider
        .map(device_tools::DeviceTools::new)
        .map(Arc::new);
    // The assistant's own name. Host-side (it writes the state root), and the
    // one tool gated on every session regardless of tier — see the module docs
    // for why an identity change is not a tier decision.
    let identity_tools = Arc::new(identity_tools::IdentityTools::new());
    let mut delegate_defs = net_tools::net_tool_defs();
    delegate_defs.extend(MemoryTools::tool_defs());
    delegate_defs.extend(media.tool_defs());
    delegate_defs.extend(studio.tool_defs());
    delegate_defs.extend(calendar.tool_defs());
    delegate_defs.extend(mail.tool_defs());
    delegate_defs.extend(m365.tool_defs());
    delegate_defs.extend(vision.tool_defs());
    delegate_defs.extend(automation.tool_defs());
    delegate_defs.extend(browser.tool_defs());
    if device_tools.is_some() {
        delegate_defs.extend(device_tools::DeviceTools::tool_defs());
    }
    delegate_defs.extend(identity_tools::IdentityTools::tool_defs());
    delegate_defs.extend(extra_defs);
    let mem_exec: Arc<dyn ToolExecutor> = mem.clone();
    let media: Arc<dyn ToolExecutor> = media;
    let studio: Arc<dyn ToolExecutor> = studio;
    let calendar: Arc<dyn ToolExecutor> = calendar;
    let mail: Arc<dyn ToolExecutor> = mail;
    let m365: Arc<dyn ToolExecutor> = m365;
    let vision: Arc<dyn ToolExecutor> = vision;
    let automation: Arc<dyn ToolExecutor> = automation;
    let browser_tools = Arc::clone(&browser);
    let browser: Arc<dyn ToolExecutor> = browser;
    let identity_exec: Arc<dyn ToolExecutor> = identity_tools;
    let mut delegates: Vec<Arc<dyn ToolExecutor>> = vec![
        net,
        mem_exec,
        media,
        studio,
        calendar,
        mail,
        m365,
        vision,
        automation,
        browser,
        identity_exec,
    ];
    if let Some(device_tools) = device_tools {
        delegates.push(device_tools);
    }
    delegates.extend(extra_executors);
    let delegate: Arc<dyn ToolExecutor> = Arc::new(ChainedDelegate(delegates));

    // The event log is created HERE, before the executor, so both it and the
    // runtime can hold the same handle. Building it inside `with_event_log`
    // below would leave the executor unable to read what the runtime records,
    // and `events_query` would have nothing to answer from (#815).
    let event_log = if let Some(path) = eventlog.as_ref() {
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let log = if path.exists() {
            let mut loaded = EventLog::load(path).map_err(|e| {
                format!(
                    "cannot resume assistant receipt journal {}: {e}",
                    path.display()
                )
            })?;
            if let Err(index) = loaded.verify_chain() {
                return Err(format!(
                    "assistant receipt journal {} failed its hash chain at event {index}",
                    path.display()
                ));
            }
            loaded.enable_hash_chaining();
            loaded
        } else {
            EventLog::with_journal(path.clone()).with_hash_chaining()
        };
        Some(Arc::new(tokio::sync::Mutex::new(log)))
    } else {
        None
    };

    // The task list is shared, not owned: the executor mutates it via
    // `todo_write` and the loop renders it, so both need the same handle
    // (Parslee-ai/car#814).
    let todos = Arc::new(tokio::sync::Mutex::new(todo::TodoList::new()));

    let mut executor = GeneralExecutor::new(env.substrate.clone(), env.root.clone(), clamp)
        // Scoped opt-in: only the discussion surface sets `clamp_reads`, so the
        // general assistant's read reach is unchanged.
        .with_read_clamp(env.clamp_reads)
        .with_delegate(delegate, delegate_defs)
        .with_todos(Arc::clone(&todos));
    if let Some(log) = &event_log {
        executor = executor.with_event_log(Arc::clone(log));
    }
    // Everything the executor can actually dispatch. The model-visible subset
    // is derived from this AFTER project policy loads below, because a tool the
    // project denies outright should never be advertised in the first place.
    let all_defs = executor.all_tool_defs();
    let executor: Arc<dyn ToolExecutor> = Arc::new(executor);

    // Outbound human messaging, so "text me when the build finishes" is a
    // governed runtime tool here too (validator → policy → rate limit →
    // eventlog) rather than something the model improvises through `shell`.
    //
    // iMessage only, and deliberately NO host fallback. The daemon sets one
    // because it genuinely has a host on the other end of the tool-callback
    // channel; `car do` does not — its executor is the in-process
    // `GeneralExecutor`, which would answer `messaging.channel_send` with
    // `unknown tool` and make `HostChannelAdapter` report "this host does not
    // implement the callback" for every unknown channel. That is a misleading
    // error: the truth is that this process has no host to implement it. The
    // registry's own "unknown messaging channel 'x': registered channels are
    // imessage" is the accurate answer, so we leave the fallback empty.
    //
    // Reuses `RealMessageSender` for the same reason the daemon does, and with
    // the same non-loop argument: it calls the un-gated `messages_send`
    // directly, so nothing here can re-enter the approval transport.
    let outbound = Arc::new(car_messaging::outbound::OutboundRegistry::new());
    outbound.register(Arc::new(
        car_messaging::outbound::ImessageOutboundAdapter::new(
            Arc::new(crate::messaging_orchestrator::RealMessageSender),
            crate::messaging_config::MessagingConfigStore::from_home(),
        ),
    ));

    let mut runtime = Runtime::new()
        .with_inference(engine)
        .with_executor(executor)
        .with_substrate(env.substrate.clone())
        .with_message_sink(outbound);
    if let Some(log) = event_log {
        runtime = runtime.with_shared_event_log(log);
    }
    if let Some(dir) = trajectories {
        runtime = runtime.with_trajectory_store(Arc::new(car_memgine::TrajectoryStore::new(&dir)));
    }

    // OpenClaw-style personal assistants fail dangerously when persistent,
    // high-privilege context can flow straight into outbound tools. Install
    // CAR's verified information-flow gate by default: built-in labels mark
    // network tools as exfiltration sinks, and projects can refine source
    // confidentiality in `.car/tool-labels.json`.
    if let Err(e) = runtime
        .install_information_flow_gate(
            env.project_car_dir
                .clone()
                .unwrap_or_else(|| env.root.join(".car")),
        )
        .await
    {
        tracing::warn!(
            error = %e,
            "assistant could not load project tool labels; falling back to built-in information-flow labels"
        );
        runtime
            .register_admission_gate(Arc::new(
                car_engine::InformationFlowGate::with_builtin_labels(),
            ))
            .await;
    }

    // The declarative half of the same `.car` directory: `policies/*.toml`.
    // Project-scoped, matching its information-flow sibling above — the rules
    // that govern an agent working in this repo are the ones checked into this
    // repo.
    //
    // Note the deliberate asymmetry with that sibling: missing tool labels fall
    // back to a safe built-in default, so warning and continuing is honest
    // there. A malformed policy file has no safe default — the rule it was
    // meant to enforce simply would not exist — so it is fatal.
    // `apply_project_policies` carries the full reasoning; do NOT downgrade it
    // to a warning to match the block above.
    // The DISCOVERED `.car`, not `root.join(".car")`. A `.car/` checked in at a
    // repository root governs the repository, so a run started in a
    // subdirectory is governed by it too — which is what CLAUDE.md has always
    // said and what nothing implemented (car#1288). Falls back to the old form
    // when there is nothing to discover, so a run outside a repository behaves
    // exactly as before.
    let project_car = env
        .project_car_dir
        .clone()
        .unwrap_or_else(|| env.root.join(".car"));
    crate::session::apply_project_policies(&runtime, &project_car).await?;

    // The model-visible tool list: everything the executor offers, minus what
    // project policy denies outright.
    //
    // Enforcement alone was already correct — a denied call is refused at
    // dispatch and the refusal is fed back to the model, which then tries
    // something else. What it was not is *cheap*. Advertising a tool no call
    // can satisfy spends a schema's worth of context on every request and, when
    // the model takes the bait, a whole turn on a refusal. Removing it from the
    // list makes the project's `deny_tool = [...]` mean "this agent does not
    // have that tool" rather than "this agent will be told off for using it".
    //
    // ONLY the `deny_tool` kind, and `PolicyEngine::blanket_denied_tools`
    // carries the reason: it is the one kind whose totality is decidable from
    // the kind alone. Others can forbid a tool outright too (an empty
    // `allow_tool_param`, `max_calls = 0`), but only by inspection, so this
    // deliberately under-reports. Under-reporting is the safe direction — the
    // tool is advertised and then refused, which is the old behavior.
    //
    // Read from the engine rather than re-reading `.car/policies/` so this
    // cannot disagree with what actually enforces, and so any rule that reached
    // this engine by another route is honoured here too. Note what that does
    // NOT include: the daemon's `~/.car/policies` is loaded into the `Runtime`
    // that `session::create_session` builds, not this one, so nothing from
    // there is in scope here.
    //
    // `tools` below is a SNAPSHOT. `blanket_denied_tools` is derived rather
    // than cached so it cannot drift from what `check` enforces, but that is a
    // property of the method, not of this list — nothing recomputes `tools`
    // after build. It is safe here only because no caller mutates this engine's
    // policies afterwards: `build_assistant_runtime` is reached from `car do`,
    // the MCP assistant, and the coder's discussion surface, none of which
    // re-register. A surface that hot-reloads policy must rebuild, not patch.
    // (`car_policy::tool_gate` holds the opposite posture, freshness over
    // caching, for a path where the rules genuinely do change under it.)
    let denied = runtime.policies.read().await.blanket_denied_tools();
    if !denied.is_empty() {
        // The operator needs "never offered" to be distinguishable from "never
        // attempted". Before this filter, every blocked attempt wrote a
        // `PolicyViolation` to the event log, which was incidental proof the
        // rule had loaded. A well-behaved model now never attempts it, so that
        // proof disappears and a policy that silently failed to load would look
        // identical to one working perfectly. Say it once at build time.
        //
        // A LOG LINE, not an event-log record — deliberately. No `EventKind`
        // means "tools withheld at assembly", `PolicyViolation` would be a lie
        // (nothing was violated), and adding a variant changes a serialized
        // event shape that crosses all four binding surfaces. So this reaches
        // an operator watching stderr and does NOT reach `events.query`. If a
        // supervised agent's operator needs it there, that is the change to
        // make, and it is a bigger one than this.
        tracing::info!(
            withdrawn = ?denied,
            "project policy denies these tools outright; withdrawn from the model's advertised list \
             (still registered, so a call naming one is refused by the policy)"
        );
    }
    let mut tools: Vec<Value> = all_defs
        .iter()
        .filter(|def| !denied.contains(def.get("name").and_then(Value::as_str).unwrap_or_default()))
        .cloned()
        .collect();
    // The loop-intercepted sub-agent tool, built over the model-visible set so
    // its `tools` enum names exactly what the parent has — which now excludes
    // the denied ones, so a child is never granted what the project denies the
    // parent. Advertised only where the caller opts in (`car do` foreground
    // runs); the read-only discussion surface, the MCP `run`, and the
    // supervised `--serve` agent leave it off. When advertised it is registered
    // with the validator below like any other def but never dispatched to the
    // executor — `agent_loop` recognizes the name.
    let delegate_def = allow_delegate.then(|| agent_loop::delegate_tool_def(&tools));
    if let Some(def) = &delegate_def {
        // `delegate` is appended rather than drawn from `all_defs`, so the
        // filter above cannot reach it — gate the push explicitly.
        //
        // This is not cosmetic. `delegate` is loop-intercepted: `agent_loop`
        // dispatches it itself when `delegate_advertised`, so it never reaches
        // `runtime.execute` and the policy engine never sees it. Advertising it
        // under a `deny_tool = ["delegate"]` rule therefore gave a project a
        // deny that was neither hidden NOR enforced — the agent kept spawning
        // children, silently. Withholding it flips `delegate_advertised` to
        // false, the call falls through to normal dispatch, and the rule fires
        // like any other. It stays registered with the validator below, so the
        // refusal still names the policy.
        if !denied.contains(agent_loop::DELEGATE_TOOL) {
            tools.push(def.clone());
        }
    }
    // Tier-based approval gating (neo leak #3): any advertised tool that
    // self-declares a `"tier"` ABOVE the standing tier (e.g. a full_access
    // automation tool in a non-full-access session) must route through the
    // approval gate. Derived from the defs, so a new capability gates itself
    // without editing this function.
    gated_tools.extend(tier_gated_tool_names(&tools, env.tier));

    // Register the dispatchable tools so the validator admits them. Execution
    // is owned by the GeneralExecutor above; these registrations are for
    // validation + schema listing. agent_basics covers the file/calculate
    // builtins; everything else advertised (shell, http_request, web_search,
    // remember, recall) is registered from its advertised def.
    runtime.register_agent_basics().await;
    let builtin_names: std::collections::HashSet<String> = car_engine::agent_basic_entries()
        .into_iter()
        .map(|e| e.schema.name)
        .collect();
    // Deliberately the UNFILTERED set plus the delegate meta-tool, not the
    // model-visible list. A tool this project denies is hidden from the model
    // above but stays registered here, so a model that names it anyway — from a
    // stale transcript, a recalled memory, or a plain guess — is refused by the
    // policy with "denied by project policy", the true reason, instead of by
    // the validator with "unregistered tool", which would send it hunting for a
    // spelling mistake that does not exist.
    for def in all_defs.iter().chain(delegate_def.iter()) {
        let name = def["name"].as_str().unwrap_or_default();
        if name.is_empty() || builtin_names.contains(name) {
            continue;
        }
        runtime
            .register_tool_entry(ToolEntry::new(schema_from_def(def)).with_side_effects(true))
            .await;
    }

    // Statically verify proposals before any action dispatches.
    //
    // Be honest about what this buys *here*. The assistant loop submits one
    // action per proposal (`agent_loop::build_proposal`), and `validate_action`
    // already checks tool existence and parameters before that action runs —
    // with a stronger schema validator than car-verify's. So on this runtime the
    // gate is close to inert: rejecting "the proposal" and rejecting "the one
    // action" are the same thing.
    //
    // It is registered anyway for two reasons: the loop may batch actions in
    // future, and a gate that is present everywhere proposals execute is easier
    // to reason about than one that is conditionally absent. The surface where
    // it actually earns its place is the daemon's `proposal.submit` runtime
    // (`session::create_session`), which accepts caller-authored multi-action
    // proposals — there, refusing up front prevents partial execution.
    //
    // Registered after tool registration only for readability; the gate holds
    // the live registry, so tools added later are still checked.
    runtime
        .register_admission_gate(Arc::new(car_engine::StaticVerificationGate::new(
            runtime.tools.clone(),
        )))
        .await;

    Ok(AssistantRuntime {
        runtime,
        tools,
        description,
        identity: car_identity::IdentityStore::from_home().load_or_default(),
        sandboxed,
        gated_tools,
        proactive_memory: mem,
        tool_memory,
        fallback_notice,
        todos,
        browser: browser_tools,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_eventlog::EventKind;
    use car_ir::{Action, ActionProposal, ActionStatus, ActionType};
    use std::collections::HashMap;

    use serde_json::json;

    struct StaticDeviceProvider(Value);

    #[async_trait]
    impl DeviceProvider for StaticDeviceProvider {
        async fn devices(&self) -> Result<Value, String> {
            Ok(self.0.clone())
        }

        async fn notify_device(
            &self,
            device_id: Option<String>,
            title: String,
            body: String,
        ) -> Result<Value, String> {
            Ok(json!({
                "device_id": device_id,
                "title": title,
                "body": body
            }))
        }
    }

    fn test_engine(root: &std::path::Path) -> Arc<InferenceEngine> {
        let mut cfg = car_inference::InferenceConfig::default();
        cfg.models_dir = root.join("models");
        Arc::new(InferenceEngine::new(cfg))
    }

    fn test_env(root: &std::path::Path) -> BoundEnvironment {
        BoundEnvironment {
            substrate: Arc::new(car_engine::LocalSubstrate::new()),
            root: root.to_path_buf(),
            tier: PermissionTier::ReadOnly,
            description: "test local host".to_string(),
            sandboxed: false,
            project_car_dir: None,
            mount: None,
            fallback_notice: None,
            clamp_reads: false,
        }
    }

    fn test_action(tool: &str) -> 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
        }
    }

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

    #[test]
    fn tier_gating_derives_from_self_declared_tier() {
        let tools = vec![
            json!({"name": "read_file"}), // no tier → never gated
            json!({"name": "generate_image", "tier": "sandbox_edit"}),
            json!({"name": "web_search", "tier": "full_access"}),
            json!({"name": "run_applescript", "tier": "full_access"}),
        ];

        // A ReadOnly session gates BOTH the sandbox_edit and full_access tools.
        let g = tier_gated_tool_names(&tools, PermissionTier::ReadOnly);
        assert!(g.contains(&"generate_image".to_string()));
        assert!(g.contains(&"web_search".to_string()));
        assert!(g.contains(&"run_applescript".to_string()));
        assert!(!g.contains(&"read_file".to_string()));

        // A SandboxEdit session gates only the full_access tool.
        let g = tier_gated_tool_names(&tools, PermissionTier::SandboxEdit);
        assert_eq!(
            g,
            vec!["web_search".to_string(), "run_applescript".to_string()]
        );

        // A FullAccess session gates nothing by tier.
        assert!(tier_gated_tool_names(&tools, PermissionTier::FullAccess).is_empty());
    }

    /// `delegate` is advertised by the assembled runtime, built over the
    /// executor's own tool set, and never tier-gated (the CHILD's calls are
    /// what the gates see).
    #[tokio::test]
    async fn assistant_runtime_advertises_delegate_over_its_own_tools() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            true,
        )
        .await
        .unwrap();
        let names: Vec<&str> = rt
            .tools
            .iter()
            .filter_map(|d| d.get("name").and_then(Value::as_str))
            .collect();
        assert!(names.contains(&agent_loop::DELEGATE_TOOL), "{names:?}");
        assert_eq!(
            names
                .iter()
                .filter(|n| **n == agent_loop::DELEGATE_TOOL)
                .count(),
            1,
            "advertised once"
        );
        let def = rt
            .tools
            .iter()
            .find(|d| d["name"] == agent_loop::DELEGATE_TOOL)
            .unwrap();
        let granted: Vec<&str> = def["parameters"]["properties"]["tools"]["items"]["enum"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(Value::as_str)
            .collect();
        let others: Vec<&str> = names
            .iter()
            .copied()
            .filter(|n| *n != agent_loop::DELEGATE_TOOL)
            .collect();
        assert_eq!(granted, others, "the enum names exactly the parent's tools");
        assert!(
            !rt.gated_tools
                .iter()
                .any(|g| g == agent_loop::DELEGATE_TOOL),
            "read_only tier: never gated by tier; got {:?}",
            rt.gated_tools
        );
    }

    /// Surfaces that do not opt in never see the def — not in `tools`, so
    /// not in the prompt and not in the validator either.
    #[tokio::test]
    async fn assistant_runtime_hides_delegate_unless_allowed() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();
        assert!(
            !rt.tools
                .iter()
                .any(|d| d["name"] == agent_loop::DELEGATE_TOOL),
            "delegate must not be advertised without opt-in"
        );
        assert!(!rt
            .gated_tools
            .iter()
            .any(|g| g == agent_loop::DELEGATE_TOOL));
    }

    #[tokio::test]
    async fn assistant_runtime_installs_information_flow_gate_by_default() {
        let dir = tempfile::tempdir().unwrap();
        let engine = test_engine(dir.path());
        let env = test_env(dir.path());

        let rt = build_assistant_runtime(engine, env, None, None, None, None, false)
            .await
            .unwrap();
        let gates = rt.runtime.admission_gate_names().await;
        assert!(
            gates.contains(&"information_flow".to_string()),
            "the information-flow gate must be installed by default, got {gates:?}"
        );
        assert!(
            gates.contains(&"static_verification".to_string()),
            "the static-verification gate must be installed by default, got {gates:?}"
        );
    }

    #[tokio::test]
    async fn assistant_runtime_appends_local_workspace_snapshot() {
        // F7/L1: a local (non-sandboxed) session gets a names-only workspace
        // snapshot appended after the one-line environment sentence.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), "[package]\nname = \"x\"").unwrap();
        std::fs::create_dir_all(dir.path().join("src")).unwrap();
        std::fs::write(dir.path().join("src/main.rs"), "fn main() {}").unwrap();

        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        assert!(
            rt.description.contains("test local host"),
            "keeps the one-line environment sentence"
        );
        assert!(
            rt.description.contains("Workspace contents"),
            "appends the names-only snapshot on a local session: {}",
            rt.description
        );
        assert!(rt.description.contains("Cargo.toml"));
        assert!(rt.description.contains("main.rs"));
    }

    /// The shell fact has to reach the REAL prompt, not just exist as a helper.
    /// Everything about the fix depends on one condition firing
    /// (`!sandboxed && substrate.is_local()`), so assert it end-to-end through
    /// `build_assistant_runtime` rather than unit-testing `host_shell_note` in
    /// isolation, which would pass even if the note were never appended.
    #[tokio::test]
    async fn assistant_runtime_names_the_host_shell_on_a_local_session() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        assert!(
            rt.description.contains("Host platform:"),
            "a local session must be told which host it is on: {}",
            rt.description
        );
        if cfg!(windows) {
            assert!(
                rt.description.contains("cmd /C") && rt.description.contains("findstr"),
                "Windows must get cmd.exe and its substitutions, not `sh -c`: {}",
                rt.description
            );
            assert!(
                !rt.description.contains("through `sh -c`"),
                "Windows must not be told it has a POSIX shell: {}",
                rt.description
            );
        } else {
            assert!(
                rt.description.contains("`sh -c`"),
                "unix keeps its existing wording: {}",
                rt.description
            );
        }
    }

    #[tokio::test]
    async fn assistant_runtime_omits_snapshot_when_sandboxed() {
        // F7/L1: a sandboxed (or remote) session must NOT get the local-fs
        // snapshot — computing it would touch the container/VM at prompt-build
        // time. Only the one-line environment sentence remains.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), "[package]").unwrap();
        let env = BoundEnvironment {
            substrate: Arc::new(car_engine::LocalSubstrate::new()),
            root: dir.path().to_path_buf(),
            tier: PermissionTier::SandboxEdit,
            description: "an isolated Docker sandbox".to_string(),
            sandboxed: true,
            project_car_dir: None,
            mount: None,
            fallback_notice: None,
            clamp_reads: false,
        };
        let rt =
            build_assistant_runtime(test_engine(dir.path()), env, None, None, None, None, false)
                .await
                .unwrap();
        assert!(rt.description.contains("isolated Docker sandbox"));
        assert!(
            !rt.description.contains("Workspace contents"),
            "no snapshot for a sandboxed session: {}",
            rt.description
        );
        assert!(!rt.description.contains("Cargo.toml"));
    }

    #[tokio::test]
    async fn assistant_runtime_gates_external_and_persistent_sinks_by_default() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        assert!(rt.gated_tools.contains(&"http_request".to_string()));
        assert!(rt.gated_tools.contains(&"web_search".to_string()));
        assert!(rt.gated_tools.contains(&"remember".to_string()));
    }

    #[tokio::test]
    async fn assistant_runtime_can_see_linked_devices_when_provider_supplied() {
        let dir = tempfile::tempdir().unwrap();
        let provider: Arc<dyn DeviceProvider> = Arc::new(StaticDeviceProvider(json!([
            {
                "name": "Mia's iPhone",
                "platform": "ios",
                "status": "online",
                "capabilities": ["assistant.chat", "assistant.approvals"]
            }
        ])));
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            Some(provider),
            None,
            None,
            false,
        )
        .await
        .unwrap();

        assert!(rt
            .tools
            .iter()
            .any(|def| def["name"].as_str() == Some("linked_devices")));
        assert!(rt
            .tools
            .iter()
            .any(|def| def["name"].as_str() == Some("notify_linked_device")));
        let result = rt
            .runtime
            .execute(&test_proposal(vec![test_action("linked_devices")]))
            .await;
        assert_eq!(result.results[0].status, ActionStatus::Succeeded);
        assert_eq!(
            result.results[0].output.as_ref().unwrap()[0]["platform"],
            "ios"
        );
    }

    #[tokio::test]
    async fn malformed_assistant_tool_labels_still_install_builtin_flow_gate() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join(".car")).unwrap();
        std::fs::write(dir.path().join(".car/tool-labels.json"), "{not json").unwrap();
        let engine = test_engine(dir.path());
        let env = test_env(dir.path());

        let rt = build_assistant_runtime(engine, env, None, None, None, None, false)
            .await
            .unwrap();
        let gates = rt.runtime.admission_gate_names().await;
        assert!(
            gates.contains(&"information_flow".to_string()),
            "the information-flow gate must be installed by default, got {gates:?}"
        );
        assert!(
            gates.contains(&"static_verification".to_string()),
            "the static-verification gate must be installed by default, got {gates:?}"
        );
    }

    /// The sibling of the test above, and deliberately the OPPOSITE verdict:
    /// tool labels degrade to a safe default, policy rules have none, so a
    /// malformed `policies/*.toml` refuses to start.
    #[tokio::test]
    async fn a_malformed_project_policy_file_fails_the_assistant_startup() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join(".car/policies")).unwrap();
        std::fs::write(
            dir.path().join(".car/policies/broken.toml"),
            "[[deny_tool]\ntool = \"shell\"",
        )
        .unwrap();

        let result = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await;
        let err = match result {
            Ok(_) => panic!("a malformed policy rule must not be silently dropped"),
            Err(e) => e,
        };
        assert!(err.contains("broken.toml"), "{err}");
    }

    /// A project's blanket `deny_tool` removes the tool from the model's view.
    ///
    /// Asserted against the executor's static built-ins so the test cannot pass
    /// vacuously: exactly `shell` must disappear from that stable set while it
    /// remains registered for a policy refusal. This avoids comparing two
    /// independently discovered catalogs, whose host-dependent tools can change
    /// between builds.
    #[tokio::test]
    async fn project_deny_tool_hides_the_tool_from_the_model() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join(".car/policies")).unwrap();
        std::fs::write(
            dir.path().join(".car/policies/deny.toml"),
            "deny_tool = [\"shell\"]\n",
        )
        .unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            true,
        )
        .await
        .unwrap();
        let advertised: std::collections::BTreeSet<String> = rt
            .tools
            .iter()
            .filter_map(|def| def.get("name").and_then(Value::as_str))
            .map(str::to_string)
            .collect();
        let static_names: std::collections::BTreeSet<String> = GeneralExecutor::tool_defs()
            .iter()
            .filter_map(|def| def.get("name").and_then(Value::as_str))
            .map(str::to_string)
            .collect();
        let mut expected = static_names.clone();
        assert!(
            expected.remove("shell"),
            "control catalog must contain the tool denied by the fixture"
        );
        let advertised_static: std::collections::BTreeSet<String> =
            advertised.intersection(&static_names).cloned().collect();
        assert_eq!(
            advertised_static, expected,
            "exactly the denied static tool must leave the model's view"
        );

        // The child cannot be granted what the project denies the parent — the
        // delegate enum is built over the filtered list, not the raw catalog.
        let delegate = rt
            .tools
            .iter()
            .find(|d| d["name"] == agent_loop::DELEGATE_TOOL)
            .expect("delegate advertised");
        let granted: Vec<&str> = delegate["parameters"]["properties"]["tools"]["items"]["enum"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(Value::as_str)
            .collect();
        assert!(
            !granted.contains(&"shell"),
            "delegate must not grant a denied tool: {granted:?}"
        );

        // Hidden from the model, still REGISTERED with the validator. A model
        // that names it anyway (stale transcript, recalled memory, a guess) is
        // then refused by the policy with the true reason rather than by the
        // validator with "unregistered tool".
        assert!(
            rt.runtime.tools.read().await.contains_key("shell"),
            "denied tool stays registered so the refusal names the policy"
        );
    }

    /// `deny_tool = ["delegate"]` must work, and it is the case the filter
    /// cannot reach on its own: `delegate` is appended after the filter, not
    /// drawn from the executor's defs.
    ///
    /// It is also the case where hiding is the ONLY enforcement. `agent_loop`
    /// intercepts the delegate call and dispatches it itself, so it never
    /// reaches the policy engine — advertise it and the deny is inert in both
    /// halves at once: not hidden, and not enforced either.
    #[tokio::test]
    async fn project_deny_tool_withholds_the_delegate_meta_tool() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join(".car/policies")).unwrap();
        std::fs::write(
            dir.path().join(".car/policies/deny.toml"),
            "deny_tool = [\"delegate\"]\n",
        )
        .unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            // The caller DID opt into delegation; policy overrides the opt-in.
            true,
        )
        .await
        .unwrap();

        assert!(
            !rt.tools
                .iter()
                .any(|d| d["name"] == agent_loop::DELEGATE_TOOL),
            "a denied delegate must not be advertised even when the surface opts in"
        );
        // Still registered, so the model naming it anyway is refused by the
        // policy rather than by the validator.
        assert!(
            rt.runtime
                .tools
                .read()
                .await
                .contains_key(agent_loop::DELEGATE_TOOL),
            "the denied delegate stays registered so the refusal names the policy"
        );
    }

    /// A project with no `.car/policies` at all starts normally — the common
    /// case must not pay for the strictness above.
    #[tokio::test]
    async fn a_project_without_policies_starts_normally() {
        let dir = tempfile::tempdir().unwrap();
        build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .expect("no policies directory must be a no-op");
    }

    /// `messaging.send` is executable on the assistant runtime: the sink is
    /// attached, so the schema is registered (`with_message_sink` registers
    /// both together or neither).
    #[tokio::test]
    async fn assistant_runtime_has_the_messaging_send_tool() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        assert!(
            rt.runtime.tools.read().await.contains_key("messaging.send"),
            "the outbound sink must make messaging.send a real tool"
        );
    }

    #[tokio::test]
    async fn assistant_runtime_rejects_confidential_data_to_web_search() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(dir.path().join(".car")).unwrap();
        std::fs::write(
            dir.path().join(".car/tool-labels.json"),
            r#"{"labels":{"read_file":{"capability":"fs_read","confidentiality":"secret"}}}"#,
        )
        .unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        let mut read = test_action("read_file");
        read.expected_effects = [("file_data".to_string(), json!(true))].into();
        let mut search = test_action("web_search");
        search.state_dependencies = vec!["file_data".to_string()];

        let result = rt.runtime.execute(&test_proposal(vec![read, search])).await;

        assert!(result
            .results
            .iter()
            .all(|r| r.status == ActionStatus::Rejected));
        let log = rt.runtime.log.lock().await;
        assert!(log.events().iter().any(|e| {
            e.kind == 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 assistant_runtime_rejects_recalled_memory_to_web_search_by_default() {
        let dir = tempfile::tempdir().unwrap();
        let rt = build_assistant_runtime(
            test_engine(dir.path()),
            test_env(dir.path()),
            None,
            None,
            None,
            None,
            false,
        )
        .await
        .unwrap();

        let mut recall = test_action("recall");
        recall.expected_effects = [("memory_context".to_string(), json!(true))].into();
        let mut search = test_action("web_search");
        search.state_dependencies = vec!["memory_context".to_string()];

        let result = rt
            .runtime
            .execute(&test_proposal(vec![recall, search]))
            .await;

        assert!(result
            .results
            .iter()
            .all(|r| r.status == ActionStatus::Rejected));
        let log = rt.runtime.log.lock().await;
        assert!(log.events().iter().any(|e| {
            e.kind == 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")
        }));
    }
}