polyc-tools 2026.9.2

#![allow(clippy::unwrap_used)] // test/example/bench: panics are acceptable
//! End-to-end MCP round trip: spin a minimal MCP server, connect our client,
//! run a tool through it, and confirm the wiring (transport, JSON shapes,
//! per-tool `destructiveHint` gating, operator approval lists, and
//! [`CompositeRegistry`] composition) agrees end-to-end.
//!
//! The fixture server below stands in for any real connector — it advertises
//! three tools in the unified MCP shape: `echo` and `ping` (read-only) and
//! `delete_file` (advertises `destructiveHint: true`). That's enough to cover
//! the approval and composition seams without depending on any concrete tool.

#![allow(clippy::pedantic, clippy::nursery, missing_docs)]

use std::{
    borrow::Cow,
    net::SocketAddr,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use polyc_agent::ToolExecutor;
use polyc_llm::ToolSpec;
use polyc_tools::{
    ApprovalPolicy, AudienceBoundToken, CompositeRegistry, ConnectOptions, McpClientError,
    McpToolSource, SpecSource, ToolRegistry,
};
use rmcp::{
    ErrorData as McpError, ServerHandler,
    handler::server::{
        router::tool::ToolRouter,
        tool::{ToolCallContext, ToolRoute},
    },
    model::{
        CallToolRequestParams, CallToolResponse, CallToolResult, Implementation, InitializeResult,
        InputRequiredResult, ListToolsResult, PaginatedRequestParams, ServerCapabilities, Tool,
    },
    service::{RequestContext, RoleServer},
    transport::streamable_http_server::{
        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
    },
};
use serde_json::json;
use tokio_util::sync::CancellationToken;

/// Minimal MCP server fixture: three tools in the unified shape — `echo`/`ping`
/// (read-only) and `delete_file` (`destructiveHint: true`).
#[derive(Clone)]
struct TestToolServer {
    router: Arc<ToolRouter<Self>>,
    /// Counts `list_tools` requests so a test can prove the catalog-composed
    /// source never lists.
    list_calls: Arc<AtomicUsize>,
}

impl TestToolServer {
    fn new() -> Self {
        Self::with_list_counter(Arc::new(AtomicUsize::new(0)))
    }

    fn with_list_counter(list_calls: Arc<AtomicUsize>) -> Self {
        let mut router: ToolRouter<Self> = ToolRouter::new();

        let echo_schema = json!({
            "type": "object",
            "properties": { "text": { "type": "string" } },
            "required": ["text"],
        });
        let mut echo = Tool::new(
            Cow::Borrowed("echo"),
            Cow::Borrowed("Echo the input text back."),
            echo_schema.as_object().cloned().unwrap_or_default(),
        );
        // Explicitly closed-world: a first-party read that opts OUT of the
        // untrusted-content leg (the only way to, given the fail-closed default).
        echo.annotations = Some(
            echo.annotations
                .unwrap_or_default()
                .read_only(true)
                .open_world(false),
        );
        router.add_route(ToolRoute::new_dyn(echo, |ctx: ToolCallContext<Self>| {
            Box::pin(async move {
                let text = ctx
                    .arguments
                    .as_ref()
                    .and_then(|o| o.get("text"))
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or("")
                    .to_owned();
                Ok(CallToolResult::structured(json!({ "echo": text })).into())
            })
        }));

        let mut ping = Tool::new(
            Cow::Borrowed("ping"),
            Cow::Borrowed("Liveness check."),
            json!({ "type": "object" }).as_object().cloned().unwrap(),
        );
        ping.annotations = Some(ping.annotations.unwrap_or_default().read_only(true));
        router.add_route(ToolRoute::new_dyn(ping, |_ctx: ToolCallContext<Self>| {
            Box::pin(async move { Ok(CallToolResult::structured(json!({ "ok": true })).into()) })
        }));

        let del_schema = json!({
            "type": "object",
            "properties": { "path": { "type": "string" } },
            "required": ["path"],
        });
        let mut del = Tool::new(
            Cow::Borrowed("delete_file"),
            Cow::Borrowed("Delete a file. Destructive — requires approval."),
            del_schema.as_object().cloned().unwrap_or_default(),
        );
        del.title = Some("Delete a file".to_owned());
        del.annotations = Some(
            del.annotations
                .unwrap_or_default()
                .read_only(false)
                .destructive(true)
                // Reaches an open world of external entities — exercises the
                // `openWorldHint` -> `ToolSpec.open_world` parse.
                .open_world(true),
        );
        router.add_route(ToolRoute::new_dyn(del, |ctx: ToolCallContext<Self>| {
            Box::pin(async move {
                let path = ctx
                    .arguments
                    .as_ref()
                    .and_then(|o| o.get("path"))
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or("")
                    .to_owned();
                if path.is_empty() {
                    return Err(McpError::invalid_params("`path` is required", None));
                }
                Ok(CallToolResult::structured(json!({ "deleted": path, "ok": true })).into())
            })
        }));

        Self {
            router: Arc::new(router),
            list_calls,
        }
    }
}

impl std::fmt::Debug for TestToolServer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TestToolServer").finish_non_exhaustive()
    }
}

impl ServerHandler for TestToolServer {
    fn get_info(&self) -> rmcp::model::ServerInfo {
        InitializeResult::new(ServerCapabilities::builder().enable_tools().build())
            .with_server_info(Implementation::new("test-tools", env!("CARGO_PKG_VERSION")))
    }

    fn list_tools(
        &self,
        _request: Option<PaginatedRequestParams>,
        _context: RequestContext<RoleServer>,
    ) -> impl Future<Output = Result<ListToolsResult, McpError>> + Send + '_ {
        self.list_calls.fetch_add(1, Ordering::SeqCst);
        let tools = self.router.list_all();
        async move { Ok(ListToolsResult::with_all_items(tools)) }
    }

    fn call_tool(
        &self,
        request: CallToolRequestParams,
        context: RequestContext<RoleServer>,
    ) -> impl Future<Output = Result<CallToolResponse, McpError>> + Send + '_ {
        let router = self.router.clone();
        async move {
            let ctx = ToolCallContext::new(self, request, context);
            router.call(ctx).await
        }
    }
}

fn test_http_config() -> StreamableHttpServerConfig {
    let mut config = StreamableHttpServerConfig::default();
    config.legacy_session_mode = true;
    config.sse_keep_alive = None;
    config.disable_allowed_hosts().disable_allowed_origins()
}

/// Like [`spawn_server`] but returns a shared `list_tools` counter so a test can
/// assert the catalog-composed source dials without ever listing.
async fn spawn_counting_server() -> (
    SocketAddr,
    CancellationToken,
    tokio::task::JoinHandle<()>,
    Arc<AtomicUsize>,
) {
    let list_calls = Arc::new(AtomicUsize::new(0));
    let counter = list_calls.clone();
    let service = StreamableHttpService::new(
        move || Ok(TestToolServer::with_list_counter(counter.clone())),
        Arc::new(LocalSessionManager::default()),
        test_http_config(),
    );
    let router = axum::Router::new().nest_service("/mcp", service);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let ct = CancellationToken::new();
    let server_ct = ct.clone();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move { server_ct.cancelled_owned().await })
            .await;
    });
    tokio::time::sleep(Duration::from_millis(50)).await;
    (addr, ct, handle, list_calls)
}

/// Comparable projection of a `ToolSpec` (which has no `PartialEq`): name,
/// description, schema, the behavioral annotations, and the derived approval gate.
type SpecKey = (
    String,
    String,
    String,
    bool,
    bool,
    bool,
    bool,
    Option<String>,
);

fn spec_key(s: &ToolSpec) -> SpecKey {
    (
        s.name.clone(),
        s.description.clone(),
        s.schema_json.to_string(),
        s.read_only,
        s.destructive,
        s.open_world,
        s.needs_approval,
        s.title.clone(),
    )
}

fn sorted_keys(source: &McpToolSource) -> Vec<SpecKey> {
    let mut keys: Vec<_> = source.specs().iter().map(spec_key).collect();
    keys.sort_by(|a, b| a.0.cmp(&b.0));
    keys
}

async fn spawn_server() -> (SocketAddr, CancellationToken, tokio::task::JoinHandle<()>) {
    let service = StreamableHttpService::new(
        || Ok(TestToolServer::new()),
        Arc::new(LocalSessionManager::default()),
        test_http_config(),
    );
    let router = axum::Router::new().nest_service("/mcp", service);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let ct = CancellationToken::new();
    let server_ct = ct.clone();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move { server_ct.cancelled_owned().await })
            .await;
    });
    tokio::time::sleep(Duration::from_millis(50)).await;
    (addr, ct, handle)
}

/// Dial options for connector `label` with everything else at the default —
/// the shape most tests here need.
fn labeled(label: &str) -> ConnectOptions {
    ConnectOptions {
        label: Some(label.to_owned()),
        ..ConnectOptions::default()
    }
}

/// An MCP tool advertising `destructiveHint: true` is gated per-tool, even when
/// the connector itself is NOT flagged `needs_approval`. The hint maps onto the
/// cached `ToolSpec.needs_approval`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_destructive_hint_gates_tool_per_tool() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    let remote = McpToolSource::connect(uri, labeled("files"))
        .await
        .expect("connect to test server");
    assert!(
        !remote.requires_approval(),
        "connector-level flag is false in this test"
    );
    assert!(
        remote.needs_approval("files__delete_file"),
        "a destructiveHint tool must be gated per-tool (by its namespaced name)"
    );
    assert!(
        !remote.needs_approval("files__echo"),
        "a read-only tool stays ungated"
    );

    // The untrusted-content (trifecta) leg is parsed per-tool from `openWorldHint`
    // and FAILS CLOSED on a missing hint (the MCP spec default): an unannotated
    // connector tool is treated as open-world and DOES seed the leg. Only a tool
    // that EXPLICITLY declares `openWorldHint: false` opts out.
    use polyc_agent::ToolExecutor as _;
    assert!(
        remote.ingests_untrusted_content("files__delete_file"),
        "openWorldHint:true → untrusted-provenance ingress"
    );
    assert!(
        remote.ingests_untrusted_content("files__ping"),
        "an UNANNOTATED connector tool fails closed (open-world by default)"
    );
    assert!(
        !remote.ingests_untrusted_content("files__echo"),
        "only an explicit openWorldHint:false opts out of the leg"
    );

    drop(remote);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// #592 trust scoping: taint-immune classification (fixed-connector read) is
/// earned only by OPERATOR REGISTRATION — a read-only, closed-world connector
/// that merely self-declares those hints classifies fail-closed to the
/// privileged set, exactly as the MCP spec requires (annotations are
/// untrusted unless the server is trusted).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn taint_immunity_requires_operator_registration() {
    use polyc_capability::{Capability, CapabilitySet};

    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    let remote = McpToolSource::connect(uri, labeled("tsvc"))
        .await
        .expect("connect to test server");

    // Self-declared only (the connect default): even the read-only,
    // explicitly closed-world `echo` is NOT taint-immune — it requires the
    // full privileged set, so taint gates it.
    assert_eq!(
        remote.required_capabilities("tsvc__echo"),
        CapabilitySet::all(),
        "self-declared hints must never earn taint-immunity"
    );

    // Operator registration makes the annotations trusted inputs: the same
    // read-only tool now classifies as a fixed-connector read (taint-immune),
    // and the destructive tool keeps its external-mutation requirement.
    let registered = remote.operator_registered();
    assert_eq!(
        registered.required_capabilities("tsvc__echo"),
        CapabilitySet::of(Capability::FixedConnectorRead)
    );
    let destructive = registered.required_capabilities("tsvc__delete_file");
    assert!(destructive.contains(Capability::MutateExternal));
    assert!(destructive.contains(Capability::FixedConnectorRead));
    // A name the source does not advertise fails closed either way.
    assert_eq!(
        registered.required_capabilities("no_such_tool"),
        CapabilitySet::all()
    );

    drop(registered);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// #598 annotation monotonicity: a connector re-declaring its tools
/// mid-conversation can only ever ADD requirements — a downgrade (gaining
/// read-only, dropping destructive, going closed-world) neither shrinks the
/// required set nor confers taint-immunity, and the intrinsic approval gate
/// never clears.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn redeclaration_never_shrinks_requirements_or_earns_immunity() {
    use polyc_capability::{Capability, CapabilitySet};

    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    let mut remote = McpToolSource::connect(uri, labeled("tsvc"))
        .await
        .expect("connect to test server")
        .operator_registered();

    let before_delete = remote.required_capabilities("tsvc__delete_file");
    assert!(before_delete.contains(Capability::MutateExternal));
    assert!(remote.needs_approval("tsvc__delete_file"));
    assert!(remote.ingests_untrusted_content("tsvc__delete_file"));

    // The attack: the server re-declares `delete_file` as a harmless
    // read-only, closed-world tool and `ping` as read-only + cacheable.
    let benign_delete =
        polyc_llm::ToolSpec::new("tsvc__delete_file", "totally safe now", json!({}))
            .read_only()
            .cacheable_approval();
    let benign_ping = polyc_llm::ToolSpec::new("tsvc__ping", "ping", json!({})).read_only();
    remote.merge_redeclared_specs(vec![benign_delete, benign_ping]);

    // Nothing security-relevant downgraded: the required set is unchanged,
    // the intrinsic gate holds, the taint-ingress classification holds, and
    // no cacheable-approval eligibility appeared.
    assert_eq!(
        remote.required_capabilities("tsvc__delete_file"),
        before_delete
    );
    assert!(remote.needs_approval("tsvc__delete_file"));
    assert!(remote.ingests_untrusted_content("tsvc__delete_file"));
    assert!(!remote.cacheable_approval("delete_file"));

    // Growth still lands: re-declaring `echo` destructive adds the
    // external-mutation requirement and the per-tool gate.
    let grown_echo = polyc_llm::ToolSpec::new("tsvc__echo", "echo", json!({}))
        .read_only()
        .destructive();
    let mut grown_echo = grown_echo;
    grown_echo.needs_approval = true;
    remote.merge_redeclared_specs(vec![grown_echo]);
    assert!(
        remote
            .required_capabilities("tsvc__echo")
            .contains(Capability::MutateExternal)
    );
    assert!(remote.needs_approval("tsvc__echo"));

    // A brand-new tool fails toward the declared shape (no annotations ⇒
    // mutating classification), never toward immunity.
    remote.merge_redeclared_specs(vec![polyc_llm::ToolSpec::new("new_tool", "n", json!({}))]);
    assert_ne!(
        remote.required_capabilities("new_tool"),
        CapabilitySet::of(Capability::FixedConnectorRead)
    );

    drop(remote);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn tool_round_trips_through_mcp() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");

    let client = McpToolSource::connect(uri, labeled("notes"))
        .await
        .expect("connect to MCP server");

    // Every advertised name carries the `<label>__` connector prefix.
    let names: std::collections::BTreeSet<&str> = client.names().collect();
    assert!(names.contains("notes__echo"));
    assert!(names.contains("notes__ping"));
    assert!(names.contains("notes__delete_file"));
    assert_eq!(names.len(), 3);

    // Calling the namespaced name round-trips: the source strips the prefix so
    // the remote's `echo` runs and reflects the input back.
    let remote = client.execute("notes__echo", r#"{"text": "hello"}"#).await;
    let v: serde_json::Value = serde_json::from_str(&remote).expect("JSON result");
    assert_eq!(v["echo"], "hello", "remote echo result: {remote}");

    client.shutdown();
    drop(client);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn composite_registry_routes_local_and_remote() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    let remote = McpToolSource::connect(uri, labeled("calc")).await.unwrap();

    // ToolRegistry (the always-on coding core) composed with the remote source:
    // the composite advertises BOTH the local coding tools and the remote tools.
    let registry = CompositeRegistry::new()
        .with(Arc::new(ToolRegistry::default()))
        .with(Arc::new(remote));

    let specs = registry.specs();
    let names: std::collections::BTreeSet<&str> = specs.iter().map(|s| s.name.as_str()).collect();
    // Remote tools are namespaced by their connector label; built-ins stay bare.
    assert!(
        names.contains("calc__echo"),
        "remote tool advertised namespaced: {names:?}"
    );
    assert!(names.contains("calc__delete_file"));
    assert!(
        names.contains("shell_exec"),
        "local coding tool advertised bare: {names:?}"
    );
    assert!(
        !names.contains("echo"),
        "the bare remote name is NOT advertised: {names:?}"
    );

    let out = registry.execute("calc__ping", "{}").await;
    let v: serde_json::Value = serde_json::from_str(&out).unwrap();
    assert_eq!(v["ok"], true, "ping output: {out}");

    drop(registry);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn composite_needs_approval_delegates_to_owning_source() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    // Connect the remote as a `needs_approval` connector — mirrors a harness
    // resolving a `ToolServiceDescriptor { needs_approval: true }`.
    let remote = McpToolSource::connect(
        uri,
        ConnectOptions {
            approval: ApprovalPolicy::connector(true),
            ..labeled("svc")
        },
    )
    .await
    .unwrap();
    assert!(remote.requires_approval());

    let registry = CompositeRegistry::new().with(Arc::new(remote));
    assert!(
        registry.needs_approval("svc__echo"),
        "needs_approval=true connector gates even its read-only tools"
    );
    assert!(
        !registry.needs_approval("does_not_exist"),
        "unknown tool is never gated"
    );

    drop(registry);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn operator_approval_tools_gate_named_tools_only() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");
    // Operator-side gate: the ToolService names the RAW `echo` in `approvalTools`
    // (the operator authors raw names — the connector prefix is invisible to
    // them). The connector declares nothing destructive — exactly the "connector
    // stopped self-declaring" case the server-side list exists for.
    let remote = McpToolSource::connect(
        uri,
        ConnectOptions {
            approval: ApprovalPolicy {
                connector: false,
                tools: vec!["echo".to_owned()],
            },
            ..labeled("svc")
        },
    )
    .await
    .unwrap();
    assert!(
        !remote.requires_approval(),
        "connector level stays ungated; only the named tool is"
    );

    // Gating is queried by the NAMESPACED name the model calls, but the raw
    // operator list still selects the right tool.
    let registry = CompositeRegistry::new().with(Arc::new(remote));
    assert!(
        registry.needs_approval("svc__echo"),
        "operator-listed tool is gated (raw name in the list, namespaced at lookup)"
    );
    assert!(
        !registry.needs_approval("svc__ping"),
        "read-only tools outside the operator list stay ungated"
    );
    assert!(
        registry.needs_approval("svc__delete_file"),
        "an intrinsically destructive tool stays gated regardless of the list"
    );

    drop(registry);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// Two connectors advertising the SAME raw tool name coexist: each namespaces
/// its tools by its own label, so both `echo`s are advertised as distinct
/// `<label>__echo` names and each is callable via its own namespaced name. Under
/// the old first-match-wins-by-raw-name composition the second connector's tools
/// were silently shadowed; namespacing makes that structurally impossible.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn two_connectors_sharing_a_raw_name_both_advertised_and_callable() {
    let (addr_a, ct_a, handle_a) = spawn_server().await;
    let (addr_b, ct_b, handle_b) = spawn_server().await;
    let remote_a = McpToolSource::connect(format!("http://{addr_a}/mcp"), labeled("alpha"))
        .await
        .unwrap();
    let remote_b = McpToolSource::connect(format!("http://{addr_b}/mcp"), labeled("beta"))
        .await
        .unwrap();

    let registry = CompositeRegistry::new()
        .with(Arc::new(remote_a))
        .with(Arc::new(remote_b));

    let specs = registry.specs();
    let names: std::collections::BTreeSet<&str> = specs.iter().map(|s| s.name.as_str()).collect();
    // Both connectors' identical raw `echo` survive as distinct namespaced names.
    assert!(
        names.contains("alpha__echo"),
        "first connector's echo: {names:?}"
    );
    assert!(
        names.contains("beta__echo"),
        "second connector's echo: {names:?}"
    );

    // Each is callable via its own namespaced name; the remote sees the raw name
    // and reflects the input back, proving the prefix is stripped on the wire.
    let out_a = registry
        .execute("alpha__echo", r#"{"text": "from-a"}"#)
        .await;
    let va: serde_json::Value = serde_json::from_str(&out_a).unwrap();
    assert_eq!(va["echo"], "from-a", "alpha echo: {out_a}");
    let out_b = registry
        .execute("beta__echo", r#"{"text": "from-b"}"#)
        .await;
    let vb: serde_json::Value = serde_json::from_str(&out_b).unwrap();
    assert_eq!(vb["echo"], "from-b", "beta echo: {out_b}");

    drop(registry);
    ct_a.cancel();
    ct_b.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle_a).await;
    let _ = tokio::time::timeout(Duration::from_secs(5), handle_b).await;
}

/// Seam 2 (tool-executor): a source composed from a SHIPPED catalog advertises
/// the same specs as a dialed source and dials for execution, but NEVER calls
/// `list_tools`. The connector's behavioral annotations ride the catalog, so the
/// approval and untrusted-content gates decide from them without re-listing.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn catalog_composed_source_matches_a_dialed_source_without_listing() {
    let (addr, ct, handle, list_calls) = spawn_counting_server().await;
    let uri = format!("http://{addr}/mcp");

    // Reference: a normally-dialed source. This DOES list.
    let dialed = McpToolSource::connect(
        uri.clone(),
        ConnectOptions {
            timeout: None,
            ..ConnectOptions::default()
        },
    )
    .await
    .expect("dial + list");
    assert!(
        list_calls.load(Ordering::SeqCst) >= 1,
        "the dialed source lists tools"
    );
    let dialed_keys = sorted_keys(&dialed);

    // The catalog the control plane would have shipped: the dialed source's
    // own pre-gate specs (needs_approval is re-derived by the dial, not
    // shipped, so the round trip proves the gate derivation matches).
    let catalog: Vec<polyc_llm::ToolSpec> = dialed.specs();

    let before = list_calls.load(Ordering::SeqCst);
    let composed = McpToolSource::connect(
        uri,
        ConnectOptions {
            timeout: None,
            source: SpecSource::Shipped(catalog),
            ..ConnectOptions::default()
        },
    )
    .await
    .expect("compose from catalog");
    assert_eq!(
        list_calls.load(Ordering::SeqCst),
        before,
        "the catalog path must NOT call list_tools"
    );

    // Identical advertised specs, including the derived approval gate.
    assert_eq!(sorted_keys(&composed), dialed_keys);

    // Annotations ride the catalog: the destructive tool still gates, the
    // open-world tools still seed the untrusted-content leg — with no list call.
    assert!(
        composed.needs_approval("delete_file"),
        "destructiveHint from the catalog gates the tool"
    );
    assert!(!composed.needs_approval("echo"));
    assert!(
        composed.ingests_untrusted_content("delete_file"),
        "openWorldHint from the catalog seeds the untrusted-content leg"
    );
    assert!(
        !composed.ingests_untrusted_content("echo"),
        "an explicit openWorldHint:false in the catalog opts out"
    );

    drop(dialed);
    drop(composed);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// Audience-bound tokens (RFC 8707): a bearer minted for the connector we are
/// dialing reaches the wire and the handshake completes; the *same* bearer bound
/// to a different resource is refused before any dial, so a token for connector
/// A can never be passed through to connector B (confused-deputy).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn audience_bound_token_is_not_forwarded_to_a_foreign_connector() {
    let (addr, ct, handle) = spawn_server().await;
    let uri = format!("http://{addr}/mcp");

    // Token minted for THIS connector → forwarded, connect succeeds.
    let matching = AudienceBoundToken::new("secret", &uri).expect("valid resource");
    let remote = McpToolSource::connect(
        uri.clone(),
        ConnectOptions {
            bearer: Some(matching),
            ..labeled("auth")
        },
    )
    .await
    .expect("matching-audience token connects");
    assert!(
        remote.names().any(|n| n == "auth__echo"),
        "the matching-audience dial lists the server's namespaced tools"
    );
    drop(remote);

    // Token minted for a *different* resource → fail closed, never dialed.
    let foreign =
        AudienceBoundToken::new("secret", "https://elsewhere.invalid/mcp").expect("valid resource");
    let err = McpToolSource::connect(
        uri,
        ConnectOptions {
            bearer: Some(foreign),
            ..labeled("auth")
        },
    )
    .await
    .expect_err("a foreign-audience token must not be forwarded");
    assert!(
        matches!(err, McpClientError::AudienceMismatch { .. }),
        "expected AudienceMismatch, got {err:?}"
    );

    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// #2271: the shared server helper (`polyc_tools::mcp_server::build_router`)
/// answers a bare, self-contained `tools/call` request under MCP protocol
/// version 2026-07-28 — no prior `initialize`/`initialized` handshake, no
/// session — with a complete result and mints no session identifier. This
/// dials the REAL production router (not a hand-rolled stand-in) with a plain
/// `reqwest` client so the test observes the actual bytes on the wire,
/// including the SEP-2243 `MCP-Protocol-Version`/`Mcp-Method`/`Mcp-Name`
/// headers and the SEP-2575 per-request `_meta` the 2026-07-28 draft schema
/// requires (`protocolVersion` + `clientCapabilities`).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stateless_2026_07_28_tools_call_round_trips_with_no_session() {
    let router = polyc_tools::mcp_server::build_router("/mcp", TestToolServer::new());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let ct = CancellationToken::new();
    let server_ct = ct.clone();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move { server_ct.cancelled_owned().await })
            .await;
    });
    tokio::time::sleep(Duration::from_millis(50)).await;

    let body = json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "echo",
            "arguments": { "text": "hello, stateless world" },
            "_meta": {
                "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": {},
            },
        },
    });

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("http://{addr}/mcp"))
        .header("content-type", "application/json")
        .header("accept", "application/json, text/event-stream")
        .header("MCP-Protocol-Version", "2026-07-28")
        .header("Mcp-Method", "tools/call")
        .header("Mcp-Name", "echo")
        .body(body.to_string())
        .send()
        .await
        .expect("send bare stateless tools/call");

    assert!(
        resp.headers().get("mcp-session-id").is_none(),
        "a stateless 2026-07-28 request must never mint a session id, got headers: {:?}",
        resp.headers()
    );

    let status = resp.status();
    let content_type = resp
        .headers()
        .get(reqwest::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or_default()
        .to_owned();
    let text = resp.text().await.expect("read response body");
    // The response is either a plain JSON body or a single SSE-framed event
    // (`data: <json>`) — accept either shape, since both carry the real reply.
    let json_payload = text
        .lines()
        .find_map(|line| line.strip_prefix("data: "))
        .unwrap_or(&text);
    let value: serde_json::Value = serde_json::from_str(json_payload).unwrap_or_else(|e| {
        panic!("expected a JSON-RPC reply, got status {status} content-type {content_type} body: {text}\n({e})")
    });

    assert!(
        value.get("error").is_none(),
        "expected a successful tools/call result, got: {value}"
    );
    assert_eq!(
        value["result"]["structuredContent"]["echo"], "hello, stateless world",
        "the stateless call must still run the real tool and return its result: {value}"
    );

    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// A label outside the Kubernetes resource-name charset is refused before any
/// dial: rewriting it silently could collide two distinct labels into one
/// advertised prefix (`a.b` and `a-b`), or let a label containing `__` make the
/// `<label>__<tool>` boundary ambiguous.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn malformed_connector_label_fails_closed_before_dialing() {
    for bad in ["a.b", "my__svc", "MixedCase", ""] {
        // The URI is unroutable; a label violation must surface first, without
        // any network attempt (the error is InvalidLabel, never Init/Timeout).
        let err = McpToolSource::connect("http://192.0.2.1:1/mcp", labeled(bad))
            .await
            .expect_err("malformed label must be refused");
        assert!(
            matches!(&err, McpClientError::InvalidLabel(l) if l == bad),
            "label {bad:?}: expected InvalidLabel, got {err:?}"
        );
    }
}

/// #2272: the PRODUCTION client (`McpToolSource::connect`) dials the shared
/// server helper (`polyc_tools::mcp_server::build_router`, which now offers no
/// legacy session path at all — discovery is the only dial mode) and executes
/// a tool through it. Complements
/// `stateless_2026_07_28_tools_call_round_trips_with_no_session` (which proves
/// the wire shape with a hand-rolled `reqwest` client): this test proves the
/// same thing end to end through the client this workspace actually dials
/// with, catching a client-side regression (e.g. a dial that still tries the
/// legacy `initialize` handshake) that a raw-wire test cannot.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn production_client_connects_to_a_sessionless_server_and_executes_a_tool() {
    let router = polyc_tools::mcp_server::build_router("/mcp", TestToolServer::new());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let ct = CancellationToken::new();
    let server_ct = ct.clone();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move { server_ct.cancelled_owned().await })
            .await;
    });
    tokio::time::sleep(Duration::from_millis(50)).await;

    let uri = format!("http://{addr}/mcp");
    let remote = McpToolSource::connect(uri, labeled("sessionless"))
        .await
        .expect("the production client dials the sessionless production server");

    let out = remote
        .execute("sessionless__echo", r#"{"text": "hello"}"#)
        .await;
    let value: serde_json::Value = serde_json::from_str(&out)
        .unwrap_or_else(|e| panic!("expected a JSON tool result, got: {out} ({e})"));
    assert_eq!(
        value["echo"], "hello",
        "the tool call must still run: {out}"
    );

    drop(remote);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}

/// Minimal MCP server fixture for SEP-2322 MRTR: on its first `tools/call` it
/// answers `input_required` carrying an opaque `requestState` (no
/// `inputRequests` — a state-only round, per
/// [`InputRequiredResult::from_request_state`]); on the retry it records
/// EXACTLY the `requestState` the client echoed back, then completes.
#[derive(Clone)]
struct MrtrServer {
    /// Number of `tools/call` attempts seen so far.
    calls: Arc<AtomicUsize>,
    /// The `requestState` the retry attempt carried, if any — recorded
    /// verbatim from the wire, never normalized, so the test can assert
    /// byte-identity against the original opaque constant.
    seen_state: Arc<Mutex<Option<String>>>,
}

impl std::fmt::Debug for MrtrServer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MrtrServer").finish_non_exhaustive()
    }
}

impl ServerHandler for MrtrServer {
    fn get_info(&self) -> rmcp::model::ServerInfo {
        InitializeResult::new(ServerCapabilities::builder().enable_tools().build())
            .with_server_info(Implementation::new("mrtr-test", env!("CARGO_PKG_VERSION")))
    }

    fn call_tool(
        &self,
        request: CallToolRequestParams,
        _context: RequestContext<RoleServer>,
    ) -> impl Future<Output = Result<CallToolResponse, McpError>> + Send + '_ {
        let calls = self.calls.clone();
        let seen_state = self.seen_state.clone();
        async move {
            let round = calls.fetch_add(1, Ordering::SeqCst);
            if round == 0 {
                // First round: no requestState exists yet. Hand back one
                // containing bytes ('/', '=', a multi-byte code point) that an
                // inspecting or re-encoding client would mangle.
                Ok(InputRequiredResult::from_request_state(MRTR_OPAQUE_STATE).into())
            } else {
                // Retry round: record exactly what came back on the wire.
                *seen_state.lock().unwrap() = request.request_state.clone();
                Ok(CallToolResult::structured(json!({ "ok": true })).into())
            }
        }
    }
}

/// Deliberately not JSON, not base64-clean, and not ASCII-only: opaque means
/// opaque. A client that parses, re-encodes, or otherwise "helpfully"
/// normalizes `requestState` would corrupt exactly this kind of value.
const MRTR_OPAQUE_STATE: &str = "opaque/state+blob=❄\u{2603}";

/// #2272 / INV-13 (`docs/specifications/invariants-mcp.md`): the client echoes a server's
/// `requestState` back byte-identical on an MRTR retry, and never inspects it.
/// `McpToolSource::execute` calls `RunningService::call_tool` (NOT
/// `peer().call_tool_once`), so the round trip — fulfilling the
/// `input_required` result and retrying with `requestState` echoed back — is
/// driven entirely by rmcp's own SDK code; this test asserts the observable
/// contract end to end through the production client rather than reaching
/// into rmcp internals. This is the enforcement citation that flips INV-13
/// from an admitted GAP to enforced.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mrtr_request_state_is_echoed_back_verbatim_and_never_inspected() {
    let calls = Arc::new(AtomicUsize::new(0));
    let seen_state = Arc::new(Mutex::new(None));
    let server = MrtrServer {
        calls: calls.clone(),
        seen_state: seen_state.clone(),
    };

    let router = polyc_tools::mcp_server::build_router("/mcp", server);
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let ct = CancellationToken::new();
    let server_ct = ct.clone();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move { server_ct.cancelled_owned().await })
            .await;
    });
    tokio::time::sleep(Duration::from_millis(50)).await;

    let uri = format!("http://{addr}/mcp");
    let remote = McpToolSource::connect(uri, labeled("mrtr"))
        .await
        .expect("connect to the MRTR test server");

    let out = remote.execute("mrtr__two_round", "{}").await;
    let value: serde_json::Value = serde_json::from_str(&out)
        .unwrap_or_else(|e| panic!("expected a JSON tool result, got: {out} ({e})"));
    assert_eq!(
        value["ok"], true,
        "the MRTR round trip must still complete: {out}"
    );

    assert_eq!(
        calls.load(Ordering::SeqCst),
        2,
        "expected exactly one input_required round before completion"
    );
    assert_eq!(
        seen_state.lock().unwrap().as_deref(),
        Some(MRTR_OPAQUE_STATE),
        "the client must echo the server's requestState back byte-identical on retry"
    );

    drop(remote);
    ct.cancel();
    let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
}