polyc-tools 2026.8.3

//! Built-in tool executors for polychrome agents.
//!
//! [`ToolRegistry`] is the in-process tool surface advertised every turn: the
//! built-in core the model can call directly. Every tool — built-in or
//! connector (dialed over MCP via [`McpToolSource`]) — is described by the same
//! [`ToolSpec`] shape (name / description / schema / title + `read_only` /
//! `destructive` annotations), so the model and the HITL approval gate treat
//! them uniformly.
//!
//! # Layout
//!
//! - [`ToolRegistry`] — dispatches by tool name, implements
//!   [`polyc_agent::ToolExecutor`]. Pass it to `run_turn`.
//! - [`paid_fetch`] — the advertised spec for the 402-gated payment fetch
//!   (offered only when a wallet is configured; always approval-gated). Like
//!   `web_fetch`, it is advertise-only here — the settlement executor lives in
//!   `polyc-connectors`, and the harness payment proxy owns it in production.
//! - [`mcp_client`] — dial external MCP connectors and compose them with the
//!   built-ins ([`CompositeRegistry`]).
//! - [`peer`] — the advertised spec for `peer_call`, delegating to another
//!   agent over A2A (offered only for the deployment's configured peers;
//!   always approval-gated). Advertise-only, like `paid_fetch` — the harness's
//!   peer-call proxy owns dialing the peer.
//! - [`conversation`] — the advertised specs for reading this conversation's
//!   own committed record: find a moment, read a turn, read a recorded tool
//!   result, list recent turns, list tool calls. Advertise-only; the control
//!   plane runs each call trusted-side, over two backends the model never has
//!   to know about.

use async_trait::async_trait;
use polyc_agent::ToolExecutor;
use polyc_llm::ToolSpec;

pub mod approval;
pub mod ask_question;
pub mod capability;
pub mod coding;
pub mod connection_pool;
pub mod connector_error;
pub mod conversation;
pub mod demote;
pub mod email_link;
pub mod invite;
pub mod list_admins;
pub mod mcp_client;
pub mod mcp_server;
pub mod memory;
pub mod paid_fetch;
pub mod peer;
pub mod provisional_persona;
pub mod revoke;
pub mod routine;
pub mod unlink_identity;
pub mod unlink_self;
pub mod wallet;
pub mod web;

pub use approval::{ApprovalMode, RiskTier, auto_review_eligible, classify_tier, classify_tool};
pub use capability::{
    assert_builtin_requirements_resolved, assert_builtins_classified, builtin_requirements,
    management, unclassified_builtins,
};
pub use coding::{SandboxMode, current_sandbox_mode, sandbox_would_deny};
pub use connection_pool::ConnectionPool;
pub use connector_error::{
    CallRetryPolicy, ConnectorErrorKind, dial_failure_message, failure_json,
    transport_failure_message,
};
pub use mcp_client::{
    ApprovalPolicy, AudienceBoundToken, CALLER_HEADER, CONNECTOR_TOOL_SEPARATOR, CompositeRegistry,
    ConnectOptions, ConnectorProvenance, DEFAULT_CONNECT_TIMEOUT, McpClientError, McpToolSource,
    SpecSource, builtin_admits, core_admits_builtin, is_valid_connector_label,
};
pub use mcp_server::serve;

/// Routes tool calls by name to one of the pure-tool implementations. Pass
/// to [`polyc_agent::run_turn`] in place of `StubTools`.
///
/// `allowed` scopes the built-in surface per agent: `None` advertises and
/// executes the full set (the default / standalone behavior), while `Some(set)`
/// restricts both [`ToolExecutor::specs`] and [`ToolExecutor::execute`] to the
/// named tools — an empty set means none. The control plane derives the set
/// from the resolved Agent and sends it on `TurnInput`; this is the lever that
/// stops a visible built-in (e.g. `grep`) from shadowing a connector tool the
/// model needs.
///
/// `workspace_root` is the coding-tool re-root seam (`#2286`): `None` (the
/// default) means "use [`coding::workspace::root`]", byte-identical to this
/// registry's behavior before the field existed; `Some(root)` runs every
/// coding tool against `root` instead, which is how [`Self::for_worker`]
/// hands a delegated worker its own workspace subtree without any of them
/// touching process env.
#[derive(Clone, Default, Debug)]
pub struct ToolRegistry {
    allowed: Option<std::sync::Arc<std::collections::BTreeSet<String>>>,
    workspace_root: Option<std::path::PathBuf>,
}

impl ToolRegistry {
    /// A registry scoped to exactly `names` — the per-agent built-in allowlist.
    /// Names that aren't real built-ins are simply never matched.
    #[must_use]
    pub fn scoped(names: impl IntoIterator<Item = String>) -> Self {
        Self {
            allowed: Some(std::sync::Arc::new(names.into_iter().collect())),
            workspace_root: None,
        }
    }

    /// A registry rooted at an EXPLICIT workspace directory rather than
    /// [`coding::workspace::root`]'s process-env lookup (`#2286`).
    ///
    /// This is the constructor a test uses to exercise a specific root — the
    /// workspace forbids `std::env::set_var` in tests, so there is no other
    /// way to point a registry at a temp directory without mutating global
    /// process state. `allowed` mirrors [`Self::scoped`]'s semantics (`None`
    /// ⇒ every built-in permitted).
    #[must_use]
    pub const fn rooted_at(
        root: std::path::PathBuf,
        allowed: Option<std::sync::Arc<std::collections::BTreeSet<String>>>,
    ) -> Self {
        Self {
            allowed,
            workspace_root: Some(root),
        }
    }

    /// The effective coding-tool root: [`Self::workspace_root`] when set,
    /// else [`coding::workspace::root`] (the process-env default).
    fn effective_root(&self) -> std::path::PathBuf {
        self.workspace_root
            .clone()
            .unwrap_or_else(coding::workspace::root)
    }

    /// Whether `name` is advertised/executable under the current scope. `true`
    /// for every name when unscoped; membership-gated when scoped.
    fn permits(&self, name: &str) -> bool {
        self.allowed.as_ref().is_none_or(|a| a.contains(name))
    }

    /// Tool specs to advertise to the provider so the model knows what's
    /// callable. Kept in a stable order so prompt hashes stay reproducible.
    ///
    /// `paid_fetch` is advertised **only when machine payments are configured**
    /// (see `payments_configured` on the registry): a deployment with no wallet — e.g. a
    /// read-only chat bot — never offers the payment tool, so the model can't
    /// reach for it on an ordinary "GET this URL" request. The pure tools are
    /// always present and keep their byte-stable order.
    #[must_use]
    pub fn all_specs() -> Vec<ToolSpec> {
        Self::specs_with_payments(payments_configured())
    }

    /// [`Self::all_specs`] with the payments decision injected — the pure inner
    /// function, so the wallet gate is testable without touching process env.
    fn specs_with_payments(payments: bool) -> Vec<ToolSpec> {
        // The always-on coding core (shell + files + search) — the default
        // coding-agent tool surface, advertised every turn.
        let mut specs = coding::specs();
        // `web_fetch` is advertised in the always-on set; execution routes to
        // the harness web proxy (composed first), since the sandbox itself has
        // no egress. In a standalone process with no control plane it has no
        // executor and errors clearly — there is no outbound path to fall back
        // to anyway.
        specs.push(web::fetch_spec());
        // `ask_question` (#1660) is advertise-only, like `web_fetch` — it has no
        // wallet gate and no in-process/proxy executor at all. Calling it
        // always short-circuits the turn loop's own question-pause phase
        // before any `ToolExecutor::execute` is reached, the same way a
        // gated call never reaches `execute` until the HITL approval gate
        // clears it. It is scoped per-agent via `builtin_allow`, not a
        // deployment-wide wallet check, so it is pushed unconditionally here.
        specs.push(ask_question::spec());
        if payments {
            // `paid_fetch` performs IO and may settle an onchain payment — hence
            // the wallet gate (and its intrinsic `destructive` annotation routes
            // it through the HITL approval gate).
            specs.push(paid_fetch::spec());
        }
        specs
    }
}

/// Whether outbound machine payments are configured for this process — i.e. a
/// non-empty `TEMPO_SIGNER_KEY` is set (the same signal `paid_fetch`'s backend
/// reads via `mpp`'s `PaymentsConfig::from_env` at call time). Gates whether
/// `paid_fetch` is advertised at all: with no wallet the payment tool is never
/// offered. Checks the signer's PRESENCE, not validity — a malformed key still
/// surfaces (sanitized) at call time. Read once per process, since payment
/// config is start-time (matching the backend's own cached bring-up).
fn payments_configured() -> bool {
    static CONFIGURED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *CONFIGURED
        .get_or_init(|| std::env::var("TEMPO_SIGNER_KEY").is_ok_and(|v| !v.trim().is_empty()))
}

/// Env var: comma-separated tool names that need HITL approval.
///
/// Set on the harness pod so the operator can flip any tool into the gated
/// path without code changes. Empty / unset keeps the default behaviour (no
/// tool needs approval).
pub const TOOL_NEEDS_APPROVAL_ENV: &str = "POLYCHROME_TOOLS_NEEDS_APPROVAL";

/// Parses a comma-separated approval list (the value of
/// [`TOOL_NEEDS_APPROVAL_ENV`]) into trimmed, non-empty tool names.
///
/// Factored out of [`needs_approval_set`] so the approval seam can be tested
/// without mutating the process-global environment (the workspace forbids the
/// `unsafe` `std::env::set_var` that would otherwise be needed).
fn parse_needs_approval_list(raw: &str) -> Vec<String> {
    raw.split(',')
        .map(str::trim)
        .filter(|t| !t.is_empty())
        .map(str::to_owned)
        .collect()
}

/// Pure approval decision over an explicit spec set and env list, so the seam
/// is testable without mutating process-global environment (the workspace
/// forbids the `unsafe` `std::env::set_var`).
///
/// A tool needs approval when EITHER:
/// - the matching [`ToolSpec`] advertises [`ToolSpec::needs_approval`] (the
///   intrinsic, per-tool side-effect flag — this is where `paid_fetch` is
///   gated, and where an MCP `destructiveHint` lands), OR
/// - the operator named it in `env_list` (the env list only ever ADDS tools;
///   it can never clear a spec's intrinsic flag).
///
/// A `name` with no matching spec is gated only when the env list names it.
fn needs_approval_with(specs: &[ToolSpec], env_list: &[String], name: &str) -> bool {
    let intrinsic = specs
        .iter()
        .find(|s| s.name == name)
        .is_some_and(|s| s.needs_approval);
    intrinsic || env_list.iter().any(|n| n == name)
}

/// Sandbox-mode gate: a `destructive` tool (MCP `destructiveHint`) is routed
/// through HITL only in [`SandboxMode::ReadOnly`](coding::SandboxMode::ReadOnly)
/// — in `workspace-write` / `danger-full-access` its workspace-scoped effects
/// run unattended. Read-only and unknown tools are never gated here; the
/// operator allow-list and intrinsic `needs_approval` are OR-ed in by the caller.
fn sandbox_gated(specs: &[ToolSpec], name: &str, mode: coding::SandboxMode) -> bool {
    mode == coding::SandboxMode::ReadOnly
        && specs
            .iter()
            .find(|s| s.name == name)
            .is_some_and(|s| s.destructive)
}

fn needs_approval_set() -> Vec<String> {
    std::env::var(TOOL_NEEDS_APPROVAL_ENV)
        .ok()
        .map(|s| parse_needs_approval_list(&s))
        .unwrap_or_default()
}

#[async_trait]
impl ToolExecutor for ToolRegistry {
    fn specs(&self) -> Vec<ToolSpec> {
        let mut specs = Self::all_specs();
        // Per-agent scope: advertise only the allowed built-ins (order preserved
        // so prompt hashes stay reproducible). Unscoped → the full set.
        specs.retain(|s| self.permits(&s.name));
        specs
    }

    /// Routes a tool name into the HITL gate.
    ///
    /// A tool is gated when its own [`ToolSpec::needs_approval`] property is set
    /// (the per-tool, intrinsic side-effect flag — `paid_fetch` carries it, so
    /// it is ALWAYS gated even when the operator's [`TOOL_NEEDS_APPROVAL_ENV`]
    /// list is unset). The env list only ADDS further tools. Per-process env
    /// read (cheap; the list is tiny) keeps the operator surface a single env
    /// var.
    fn needs_approval(&self, name: &str) -> bool {
        // Three sources, OR-ed: the per-spec intrinsic flag (e.g. paid_fetch),
        // the operator allow-list, and the sandbox-mode gate — a `destructive`
        // tool gates only in read-only mode. All three are spec/annotation
        // driven (the `destructive` flag is the MCP `destructiveHint`).
        let specs = Self::all_specs();
        needs_approval_with(&specs, &needs_approval_set(), name)
            || sandbox_gated(&specs, name, coding::SandboxMode::from_env())
    }

    // `cacheable_approval` is intentionally NOT overridden: the trait default
    // derives it from `specs()` (== `all_specs()` here), which already carries
    // the per-tool `ToolSpec::cacheable_approval` annotation.

    /// Graduated-approval escalation (`#301`): a path-bearing destructive coding
    /// tool whose target escapes the workspace is a sandbox denial — the gate
    /// escalates it to a human (an unsandboxed retry) rather than running it and
    /// returning the flat "path escapes the workspace" error. Delegates to the
    /// pure [`coding::sandbox_would_deny`] predicate (no filesystem touch).
    fn sandbox_would_deny(&self, name: &str, args_json: &str) -> bool {
        self.permits(name) && coding::sandbox_would_deny(name, args_json)
    }

    /// The one gate-facing classification surface (`#592`): the capabilities a
    /// built-in call requires, derived from its spec annotations plus the
    /// static origin of its name (see [`capability::builtin_origin`]). Scoped
    /// by `permits` — a built-in this agent was not granted fails closed to
    /// the privileged set, exactly like an unknown name.
    fn required_capabilities(&self, name: &str) -> polyc_capability::CapabilitySet {
        if !self.permits(name) {
            return polyc_capability::CapabilitySet::all();
        }
        capability::required_for_builtin(name, &Self::all_specs())
    }

    // `ingests_untrusted_content` is intentionally NOT overridden: the trait
    // default derives it from `specs()` (== `all_specs()` here, scoped by
    // `permits`) via the per-tool `ToolSpec::open_world` annotation — the same
    // spec-as-source-of-truth pattern as `cacheable_approval`. The web fetchers
    // carry `open_world = true`; the sandbox coding tools carry `false`.

    /// Re-root this registry's coding tools for a delegated worker (`#2286`):
    /// hand back a clone of this registry whose effective coding-tool root is
    /// the worker's own subtree under the CURRENT effective root, keyed by
    /// `worker_id` (the delegate call id).
    ///
    /// The subtree is created eagerly — `shell_exec` sets it as the child
    /// process's `current_dir`, which fails to spawn against a directory that
    /// does not exist yet, and `glob`/`grep`/`file_read` would otherwise
    /// error with a confusing "no such file or directory" instead of running
    /// against an empty scratch space.
    ///
    /// A `create_dir_all` failure (a full `sizeLimit` on the workspace
    /// volume, a read-only mount, a file already sitting at the worker path)
    /// still re-roots. Returning `None` there would fail OPEN: the caller
    /// reads `None` as "this executor owns no workspace" and runs the worker
    /// through the SHARED conversation root, which is exactly the clobbering
    /// this method exists to prevent — and the worker is write-capable, so
    /// the fallback is worse than a failed turn. Rooted at a directory that
    /// does not exist, `file_write` recreates it (or fails inside the
    /// worker's own subtree) and the read tools return empty; either way
    /// nothing reaches a sibling's files.
    ///
    /// Once the subtree exists, `scope`'s share-in request is seeded into it
    /// (`#2295`): the parent files this delegation named, copied to the same
    /// relative paths, bounded by the target agent's ceiling. A refused
    /// request returns `Some(Err(_))` — the delegation fails with the reason
    /// rather than running a worker against a view its task never asked for.
    fn for_worker(
        &self,
        scope: &polyc_agent::delegate::WorkerScope<'_>,
    ) -> Option<Result<polyc_agent::delegate::WorkerHandoff, polyc_agent::delegate::ShareInError>>
    {
        let parent_root = self.effective_root();
        let worker_root = coding::workspace::worker_root(&parent_root, scope.worker_id);
        if let Err(err) = std::fs::create_dir_all(&worker_root) {
            tracing::warn!(
                worker_id = scope.worker_id,
                root = %worker_root.display(),
                error = %err,
                "could not create delegated worker's workspace subtree; re-rooting anyway so the worker cannot reach the shared root"
            );
        }
        // `#2295`: seed BEFORE handing the executor back, so the worker's
        // first tool call already sees the files its task is about. A refusal
        // fails the delegation rather than degrading to an unseeded worker —
        // see this module's `share_in` docs for why a partial view is worse
        // than no run at all.
        let seeded =
            match coding::share_in::seed(&parent_root, &worker_root, scope.share_in, scope.ceiling)
            {
                Ok(seeded) => seeded,
                Err(err) => return Some(Err(err)),
            };
        Some(Ok(polyc_agent::delegate::WorkerHandoff {
            tools: std::sync::Arc::new(Self {
                allowed: self.allowed.clone(),
                workspace_root: Some(worker_root),
            }),
            seeded,
        }))
    }

    async fn execute(&self, name: &str, args_json: &str) -> String {
        // Defense in depth: a scoped-out built-in is never advertised, so the
        // model can't call it — but refuse here too rather than execute a tool
        // this agent was not granted.
        if !self.permits(name) {
            return serde_json::json!({
                "error": format!("tool not available to this agent: {name}"),
            })
            .to_string();
        }
        // `web_fetch` and `paid_fetch` are advertise-only here: the registry
        // offers their specs but owns no executor (the harness proxy intercepts
        // them, composed first; `polyc-connectors` carries the standalone
        // settlement). Standalone, they fall through to the clear "unknown tool"
        // error below — there is no local egress path to run them on anyway.
        // Coding tools (shell/file/search) dispatch through their module,
        // rooted at this registry's effective root (`#2286`) rather than
        // always re-reading process env.
        if let Some(out) = coding::execute_rooted(&self.effective_root(), name, args_json).await {
            return out;
        }
        connector_error::failure_json(
            connector_error::ConnectorErrorKind::Application,
            &format!("unknown tool: {name}"),
        )
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]
    use super::*;

    use serde_json::Value;

    #[test]
    fn specs_always_include_ask_question_regardless_of_payments() {
        // ask_question has no wallet gate and is always advertised (the
        // per-agent grant, not a global wallet check, scopes it) — assert
        // this for both payment settings so a future wallet-gate copy/paste
        // mistake can't silently fold it under `if payments`.
        for payments in [true, false] {
            let specs = ToolRegistry::specs_with_payments(payments);
            assert!(
                specs.iter().any(|s| s.name == ask_question::TOOL_NAME),
                "ask_question must be advertised regardless of payments={payments}"
            );
        }
    }

    #[test]
    fn specs_include_paid_fetch_last_when_payments_configured() {
        let specs = ToolRegistry::specs_with_payments(true);
        assert!(
            specs.iter().any(|s| s.name == "paid_fetch"),
            "paid_fetch is advertised when a wallet is configured"
        );
        // Appended last to keep existing prompt hashes stable.
        assert_eq!(
            specs.last().map(|s| s.name.as_str()),
            Some("paid_fetch"),
            "paid_fetch must be the last spec"
        );
    }

    /// Pin: every tool spec whose execution can itself move money or
    /// change spend authority carries `.destructive()` — a `blanket_below_high`
    /// routine grant covers everything below the High risk tier
    /// (`polyc_tools::approval::classify_tier`), so a payment-capable tool
    /// that forgot this annotation would slip under it silently, letting an
    /// unattended routine fire spend money nobody rehearsed.
    ///
    /// `paid_fetch` (settles an onchain payment) and `unlink_self` (removes
    /// the caller's own linked wallet or email) are asserted directly. Every
    /// OTHER wallet-family spec is asserted non-destructive against an
    /// explicit allowlist — the family's module doc states none of them
    /// moves money — so a NEW wallet tool that isn't added to the allowlist
    /// fails this test instead of silently inheriting "safe".
    #[test]
    fn every_payment_capable_tool_spec_is_destructive() {
        assert!(
            paid_fetch::spec().destructive,
            "paid_fetch settles a real payment and must be destructive"
        );
        assert!(
            unlink_self::unlink_self_spec().destructive,
            "unlink_self can remove the caller's own linked wallet and must be destructive"
        );

        // wallet_set_policy changes spend authority (limit/expiry/host
        // allowlist) — the one wallet-family tool that mutates anything.
        const MUTATES_SPEND_AUTHORITY: &[&str] = &[wallet::WALLET_SET_POLICY];
        for spec in wallet::all_specs() {
            let should_be_destructive = MUTATES_SPEND_AUTHORITY.contains(&spec.name.as_str());
            assert_eq!(
                spec.destructive, should_be_destructive,
                "{}: destructive={}, expected {should_be_destructive} — a wallet tool that \
                 moves money or spend authority must be destructive, and a new wallet tool \
                 must be added to MUTATES_SPEND_AUTHORITY here if it does",
                spec.name, spec.destructive
            );
        }
    }

    #[test]
    fn specs_omit_paid_fetch_without_a_wallet() {
        // The default for a read-only deployment: no TEMPO_SIGNER_KEY => the
        // payment tool is never advertised, so the model can't pick it for an
        // ordinary URL GET.
        let specs = ToolRegistry::specs_with_payments(false);
        assert!(
            !specs.iter().any(|s| s.name == "paid_fetch"),
            "paid_fetch must NOT be advertised without a wallet"
        );
        // The always-on coding core is still advertised regardless of wallet.
        assert!(specs.iter().any(|s| s.name == "shell_exec"));
        assert!(specs.iter().any(|s| s.name == "file_read"));
    }

    #[test]
    fn cacheable_approval_marks_idempotent_reads_not_paid_fetch() {
        let registry = ToolRegistry::default();
        // Idempotent read-only tools may have their approval remembered.
        assert!(registry.cacheable_approval("file_read"));
        assert!(registry.cacheable_approval("grep"));
        // Non-idempotent / destructive tools must NOT be cacheable.
        assert!(!registry.cacheable_approval("paid_fetch"));
        assert!(!registry.cacheable_approval("shell_exec"));
        // Unknown tool → not cacheable.
        assert!(!registry.cacheable_approval("does_not_exist"));
    }

    #[test]
    fn only_ask_question_is_interactive() {
        // Pin today's interactive membership so a future second member is
        // caught here rather than silently widening the fire-conversation
        // exclusion — walks every built-in spec, never a name list.
        let interactive: Vec<String> = ToolRegistry::all_specs()
            .into_iter()
            .filter(|s| s.interactive)
            .map(|s| s.name)
            .collect();
        assert_eq!(interactive, vec![ask_question::TOOL_NAME.to_owned()]);
    }

    #[test]
    fn unscoped_registry_advertises_the_full_builtin_set() {
        // The default (no allowlist) path is unchanged: every built-in is
        // advertised, so a standalone process / an agent that grants all
        // built-ins behaves exactly as before.
        let names: Vec<String> = ToolRegistry::default()
            .specs()
            .into_iter()
            .map(|s| s.name)
            .collect();
        assert_eq!(
            names,
            ToolRegistry::all_specs()
                .into_iter()
                .map(|s| s.name)
                .collect::<Vec<_>>()
        );
        assert!(names.iter().any(|n| n == "grep"));
        assert!(names.iter().any(|n| n == "shell_exec"));
    }

    #[test]
    fn scoped_registry_advertises_only_allowed_builtins() {
        // The keystone: an agent scoped to {grep} sees grep and nothing else —
        // so a visible built-in can't shadow a connector tool the model needs.
        let names: Vec<String> = ToolRegistry::scoped(["grep".to_owned()])
            .specs()
            .into_iter()
            .map(|s| s.name)
            .collect();
        assert_eq!(names, vec!["grep".to_owned()]);
    }

    #[test]
    fn empty_scope_advertises_no_builtins() {
        // An orchestrator with an empty allowlist exposes zero built-ins.
        assert!(ToolRegistry::scoped(std::iter::empty()).specs().is_empty());
    }

    #[tokio::test]
    async fn scoped_registry_refuses_a_disallowed_builtin() {
        // Defense in depth: even if a scoped-out tool were somehow invoked, the
        // registry refuses it rather than executing an ungranted built-in.
        let registry = ToolRegistry::scoped(["grep".to_owned()]);
        let out = registry
            .execute("shell_exec", r#"{"command":"echo hi"}"#)
            .await;
        let v: Value = serde_json::from_str(&out).expect("output must be JSON");
        assert!(
            v["error"]
                .as_str()
                .unwrap_or_default()
                .contains("not available to this agent"),
            "a disallowed built-in must be refused, got: {out}"
        );
    }

    #[test]
    fn composite_registry_delegates_cacheable_approval() {
        use crate::mcp_client::CompositeRegistry;
        use std::sync::Arc;
        // Regression: the harness runs tools through a CompositeRegistry. If it
        // doesn't delegate `cacheable_approval`, it falls back to the trait
        // default `false` and a session ("don't ask again") approval is never
        // honored for a local cacheable tool — exactly the bug an e2e Slack test
        // surfaced.
        let composite = CompositeRegistry::new().with(Arc::new(ToolRegistry::default()));
        assert!(
            composite.cacheable_approval("file_read"),
            "composite must delegate cacheable_approval to the owning source"
        );
        assert!(composite.cacheable_approval("grep"));
        // A tool no source owns is not cacheable.
        assert!(!composite.cacheable_approval("does_not_exist"));
    }

    #[test]
    fn composite_registry_classifies_through_the_owner() {
        use crate::mcp_client::CompositeRegistry;
        use polyc_agent::ToolExecutor as _;
        use polyc_capability::{Capability, CapabilitySet};
        use std::sync::Arc;
        // The harness always runs tools through a CompositeRegistry. The gate
        // reads one derived required-capability set per tool, delegated to the
        // owning source: the local coding tools are sandbox-confined, the
        // fetchers reach model-controlled destinations.
        let composite = CompositeRegistry::new().with(Arc::new(ToolRegistry::default()));
        assert_eq!(
            composite.required_capabilities("web_fetch"),
            CapabilitySet::of(Capability::ArbitraryEgress)
        );
        assert_eq!(
            composite.required_capabilities("file_read"),
            CapabilitySet::of(Capability::LocalRead)
        );
        assert_eq!(
            composite.required_capabilities("shell_exec"),
            CapabilitySet::of(Capability::LocalRead)
                .with(Capability::LocalWrite)
                .with(Capability::ArbitraryEgress)
        );
        // An unowned name (the model invented it, or a wallet-gated paid_fetch
        // on a no-wallet deployment) fails CLOSED to the privileged set: under
        // taint it escalates rather than sliding through to the execute-time
        // error un-gated.
        assert_eq!(
            composite.required_capabilities("does_not_exist"),
            CapabilitySet::all()
        );
    }

    #[test]
    fn own_conversation_reads_classify_taint_immune_through_their_proxy() {
        use crate::mcp_client::CompositeRegistry;
        use polyc_agent::ToolExecutor as _;
        use polyc_capability::{Capability, CapabilitySet};
        use std::sync::Arc;

        /// The shape of the harness's control-plane tool proxy: advertises the
        /// own-conversation read specs and classifies them via the shared
        /// built-in derivation, exactly like the production proxy.
        struct ReadProxyStub;
        #[async_trait::async_trait]
        impl polyc_agent::ToolExecutor for ReadProxyStub {
            fn specs(&self) -> Vec<polyc_llm::ToolSpec> {
                crate::conversation::all_specs()
            }
            fn required_capabilities(&self, name: &str) -> CapabilitySet {
                self.specs()
                    .iter()
                    .find(|s| s.name == name)
                    .map_or_else(CapabilitySet::all, crate::capability::required_for_spec)
            }
            async fn execute(&self, _name: &str, _args: &str) -> String {
                "{}".to_owned()
            }
        }

        // The successor of a live-hit regression: a read-only recall of the
        // caller's OWN history must never gate under taint. Under the
        // capability model that is structural — the history reads require only
        // the fixed-connector read, which taint never revokes — instead of a
        // hand-written carve-out list in an egress classifier. The composite's
        // fail-safe floor adds nothing for these names.
        let composite = CompositeRegistry::new().with(Arc::new(ReadProxyStub));
        for tool in crate::conversation::ALL {
            assert_eq!(
                composite.required_capabilities(tool),
                CapabilitySet::of(Capability::FixedConnectorRead),
                "{tool} is an own-history read: taint must not revoke it"
            );
        }
    }

    #[test]
    fn registry_classification_matches_the_builtin_table() {
        use polyc_agent::ToolExecutor as _;
        use polyc_capability::{Capability, CapabilitySet};
        let registry = ToolRegistry::default();
        assert_eq!(
            registry.required_capabilities("web_fetch"),
            CapabilitySet::of(Capability::ArbitraryEgress)
        );
        assert_eq!(
            registry.required_capabilities("file_read"),
            CapabilitySet::of(Capability::LocalRead)
        );
        assert_eq!(
            registry.required_capabilities("shell_exec"),
            CapabilitySet::of(Capability::LocalRead)
                .with(Capability::LocalWrite)
                .with(Capability::ArbitraryEgress)
        );
        // A scoped-out built-in fails CLOSED for this agent — it is never
        // advertised, and even a forged call classifies to the privileged set.
        let scoped = ToolRegistry::scoped(["grep".to_owned()]);
        assert_eq!(
            scoped.required_capabilities("web_fetch"),
            CapabilitySet::all()
        );
        assert_eq!(
            scoped.required_capabilities("grep"),
            CapabilitySet::of(Capability::LocalRead)
        );
    }

    #[test]
    fn registry_flags_open_world_builtins_as_untrusted_ingress() {
        use polyc_agent::ToolExecutor as _;
        // The untrusted-content (trifecta) leg is spec-derived from the MCP
        // `openWorldHint`: the web fetchers reach an open world of external
        // entities, the sandbox-confined coding reads do not. This is the dual of
        // the egress leg and NOT the same set — `file_read` egresses nothing AND
        // ingests nothing untrusted, but a connector read (egress) may be either.
        let registry = ToolRegistry::default();
        assert!(registry.ingests_untrusted_content("web_fetch"));
        // (paid_fetch is payments-gated out of the default spec set, so it is
        // simply not advertised here; when advertised it carries open_world too.)
        assert!(!registry.ingests_untrusted_content("file_read"));
        assert!(!registry.ingests_untrusted_content("grep"));
        assert!(!registry.ingests_untrusted_content("shell_exec"));
        // A spec that declares `openWorldHint` follows the annotation, whatever
        // the tool's name — the classification is the declared property, never a
        // hardcoded list.
        let closed = ToolSpec::new("remote_read", "d", serde_json::json!({}));
        assert!(!closed.open_world);
        let open = ToolSpec::new("remote_read", "d", serde_json::json!({})).open_world();
        assert!(open.open_world);
    }

    #[test]
    fn composite_registry_delegates_sandbox_would_deny() {
        use crate::mcp_client::CompositeRegistry;
        use std::sync::Arc;
        // Regression (#301): the harness ALWAYS runs tools through a
        // CompositeRegistry (`build_tool_executor` returns one). If it does not
        // delegate `sandbox_would_deny` it inherits the trait default `false`,
        // and the per-caller escalation is a silent no-op on the only production
        // path — a sandbox-denied destructive write would run-then-flat-deny
        // instead of pausing for an unsandboxed retry.
        let composite = CompositeRegistry::new().with(Arc::new(ToolRegistry::default()));
        // A destructive write whose path escapes the workspace must escalate.
        assert!(
            composite.sandbox_would_deny("file_write", r#"{"path":"../etc/passwd","content":"x"}"#),
            "composite must delegate sandbox_would_deny to the owning source"
        );
        // A contained destructive write is NOT a denial — it runs in the box.
        assert!(
            !composite.sandbox_would_deny("file_write", r#"{"path":"src/main.rs","content":"x"}"#)
        );
        // A tool no source owns never escalates.
        assert!(!composite.sandbox_would_deny("does_not_exist", r#"{"path":"../x"}"#));
    }

    #[tokio::test]
    async fn composite_registry_delegates_for_worker() {
        use crate::mcp_client::CompositeRegistry;
        use std::sync::Arc;
        // Same regression class as `composite_registry_delegates_sandbox_would_deny`
        // (#301, and `cacheable_approval` before it): the harness ALWAYS runs
        // tools through a CompositeRegistry, so a composite that fails to
        // delegate `for_worker` inherits the trait default `None` — and #2286's
        // per-worker fencing becomes a silent no-op on the only production
        // path, with every worker back on the shared root and every test here
        // still green.
        let root = coding::tmp_dir("composite-for-worker");
        let composite =
            CompositeRegistry::new().with(Arc::new(ToolRegistry::rooted_at(root.clone(), None)));
        let rerooted = composite
            .for_worker(&polyc_agent::delegate::WorkerScope::bare("call-a"))
            .expect("composite must delegate for_worker to the owning source")
            .expect("a bare scope requests no seeding, so nothing can be refused")
            .tools;

        // Delegation must be REAL, not merely non-None: a write through the
        // re-rooted composite has to land in a per-worker subdirectory, not at
        // the conversation root. Asserting only that the Arc differs would pass
        // even if the source handed back an unchanged clone.
        let out = rerooted
            .execute("file_write", r#"{"path":"out.txt","content":"x"}"#)
            .await;
        assert!(
            !out.contains("\"error\""),
            "write through re-rooted composite failed: {out}"
        );
        assert!(
            !root.join("out.txt").exists(),
            "the write landed at the shared conversation root — for_worker was not honored"
        );
        let nested: Vec<_> = std::fs::read_dir(&root)
            .expect("conversation root is readable")
            .flatten()
            .filter(|e| e.path().is_dir())
            .collect();
        assert_eq!(
            nested.len(),
            1,
            "expected exactly one per-worker subdirectory"
        );
        assert!(nested[0].path().join("out.txt").exists());

        // A composite whose sources own no workspace has nothing to re-root and
        // must report that honestly, so the caller falls back cleanly.
        let empty = CompositeRegistry::new();
        assert!(
            empty
                .for_worker(&polyc_agent::delegate::WorkerScope::bare("call-a"))
                .is_none(),
            "a composite with no re-rootable source must return None"
        );

        std::fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn needs_approval_via_env_paid_fetch() {
        // The approval seam: when the operator lists "paid_fetch" in
        // POLYCHROME_TOOLS_NEEDS_APPROVAL, the registry routes it through HITL.
        // Tested via the pure parser so we never mutate process-global env
        // (the workspace forbids the unsafe std::env::set_var).
        let list = parse_needs_approval_list("hash, paid_fetch ,calculator");
        assert!(list.iter().any(|n| n == "paid_fetch"));
        assert!(parse_needs_approval_list("").is_empty());
        assert!(
            !parse_needs_approval_list("calculator")
                .iter()
                .any(|n| n == "paid_fetch")
        );
    }

    /// Finding #2/#7: `paid_fetch` settles real onchain payments, so it must
    /// be approval-required INTRINSICALLY — even when the operator's env list is
    /// empty/unset. The gate now reads the per-tool [`ToolSpec::needs_approval`]
    /// property rather than a hard-coded name list. Pure tools stay unguarded,
    /// and the env list still ADDS extra tools on top of the intrinsic set.
    #[test]
    fn paid_fetch_is_intrinsically_approval_required() {
        // Use the payments-configured spec set so paid_fetch is present to gate.
        let specs = ToolRegistry::specs_with_payments(true);
        // Empty env list (the unset default) must still gate paid_fetch via its
        // own spec property.
        assert!(
            needs_approval_with(&specs, &[], "paid_fetch"),
            "paid_fetch must need approval even with no env list"
        );
        // A pure tool stays unguarded under the same empty list.
        assert!(
            !needs_approval_with(&specs, &[], "calculator"),
            "pure tools must not be approval-gated by default"
        );
        // The env list ADDS extra tools without dropping the intrinsic set.
        let extra = parse_needs_approval_list("calculator");
        assert!(
            needs_approval_with(&specs, &extra, "calculator"),
            "env list must still add extra tools"
        );
        assert!(
            needs_approval_with(&specs, &extra, "paid_fetch"),
            "intrinsic paid_fetch gate must survive a non-empty env list"
        );
        // An unknown tool the operator did not list stays unguarded.
        assert!(!needs_approval_with(&specs, &extra, "hash"));
    }

    /// Finding #7: the gate is driven by each spec's own `needs_approval`
    /// property — a tool that advertises it is gated regardless of the env
    /// list, and one that does not stays ungated unless the env list adds it.
    #[test]
    fn gate_reads_per_spec_property() {
        let gated = ToolSpec::new("delete_file", "d", serde_json::json!({})).approval_required();
        let pure = ToolSpec::new("calculator", "d", serde_json::json!({}));
        let specs = [gated, pure];
        // Spec-advertised destructive tool is gated with an empty env list.
        assert!(needs_approval_with(&specs, &[], "delete_file"));
        // Pure spec stays ungated until the env list names it.
        assert!(!needs_approval_with(&specs, &[], "calculator"));
        let env = parse_needs_approval_list("calculator");
        assert!(needs_approval_with(&specs, &env, "calculator"));
        // A name with no matching spec is only gated when the env list adds it.
        assert!(!needs_approval_with(&specs, &[], "mystery"));
        assert!(needs_approval_with(
            &specs,
            &parse_needs_approval_list("mystery"),
            "mystery"
        ));
    }

    /// Sandbox-mode gate: a `destructive` tool gates only in read-only mode;
    /// read-only tools never gate; the decision is driven by the spec's
    /// `destructive` annotation, not a hardcoded name list.
    #[test]
    fn sandbox_gate_reads_destructive_annotation_per_mode() {
        use coding::SandboxMode::{DangerFullAccess, ReadOnly, WorkspaceWrite};
        let writer = ToolSpec::new("file_write", "d", serde_json::json!({})).destructive();
        let reader = ToolSpec::new("file_read", "d", serde_json::json!({})).read_only();
        let specs = [writer, reader];
        assert!(sandbox_gated(&specs, "file_write", ReadOnly));
        assert!(!sandbox_gated(&specs, "file_write", WorkspaceWrite));
        assert!(!sandbox_gated(&specs, "file_write", DangerFullAccess));
        assert!(!sandbox_gated(&specs, "file_read", ReadOnly));
        assert!(!sandbox_gated(&specs, "unknown", ReadOnly));
    }
}