alkhttp/src/adapters/to_mcp.rs

//! `to_mcp`: 4-tool gateway projection over the local operation registry,
//! exposed to external MCP clients (editors, AI tools) via rmcp's
//! `StreamableHttpService` nested into the axum `Router` at `/mcp`.
//!
//! This is the tool-gateway pattern (ADR-041): the LLM gets a fixed set of
//! meta-tools (`search`, `schema`, `call`, `batch`) and discovers operations
//! on demand — not one MCP tool per registry operation. `Sub` and `Pub`
//! ops are excluded from `search` and cannot be invoked via `call` (MCP
//! tool calls are request/response — ADR-041 §2, ADR-068).
//!
//! `structuredContent` is always a JSON object (strict MCP clients reject
//! non-object structured content): successful `call` results pass object
//! outputs through verbatim and wrap non-object outputs as
//! `{"result": <output>}`; `batch` returns `{"results": [...]}`.
//!
//! `to_mcp` is a pure projection (ADR-017 §5): it consumes the registry and
//! does not produce entries for it. It is not an `OperationAdapter`. The
//! shared dispatch spine (`GatewayDispatch`) is used for the `call` tool; the
//! `ResponseEnvelope` → `CallToolResult` mapping is `to_mcp`-specific.
//!
//! Bearer auth is the shared `bearer_auth_middleware`, applied as an axum
//! layer *around* the nested `StreamableHttpService`. The resolved
//! `Identity` is stashed by the middleware into `http::request::Parts`'s
//! extensions; rmcp injects `Parts` into the `RequestContext<RoleServer>`
//! extensions, so `call_tool` reads it back via
//! `context.extensions.get::<http::request::Parts>()`.
//!
//! Streamable HTTP only (ADR-037 — stdio is not built). Feature-gated behind
//! `mcp`.

use std::borrow::Cow;
use std::sync::Arc;

use alkcall::core::auth::Identity;
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::registration::OperationRegistry;
use alkcall::registry::spec::{AccessResult, Visibility};
use rmcp::model::{
    CallToolRequestParams, CallToolResult, Implementation, JsonObject, ListToolsResult,
    PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool,
};
use rmcp::service::{RequestContext, RoleServer};
use rmcp::transport::{
    streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService},
    StreamableHttpServerConfig,
};
use serde_json::{Map, Value};

use crate::gateway::GatewayDispatch;

const TOOL_SEARCH: &str = "search";
const TOOL_SCHEMA: &str = "schema";
const TOOL_CALL: &str = "call";
const TOOL_BATCH: &str = "batch";

const OP_SERVICES_LIST: &str = "services/list";
const OP_SERVICES_SCHEMA: &str = "services/schema";

fn search_input_schema() -> Value {
    serde_json::json!({
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Optional substring filter on operation name."
            }
        }
    })
}

fn schema_input_schema() -> Value {
    serde_json::json!({
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "description": "The fully-qualified operation name (e.g. `fs/readFile`)."
            }
        },
        "required": ["name"]
    })
}

fn call_input_schema() -> Value {
    serde_json::json!({
        "type": "object",
        "properties": {
            "operation": {
                "type": "string",
                "description": "The fully-qualified operation name to invoke."
            },
            "input": {
                "type": "object",
                "description": "The JSON input object to pass to the operation."
            }
        },
        "required": ["operation"]
    })
}

fn batch_input_schema() -> Value {
    serde_json::json!({
        "type": "object",
        "properties": {
            "calls": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "operation": { "type": "string" },
                        "input": { "type": "object" }
                    },
                    "required": ["operation"]
                },
                "description": "The operations to invoke in this batch."
            }
        },
        "required": ["calls"]
    })
}

/// The `to_mcp` projection (ADR-041): the local call-protocol
/// operations exposed as 4 fixed MCP gateway tools (`search`, `schema`,
/// `call`, `batch`) over the streamable-HTTP transport.
pub struct ToMcpGateway {
    dispatch: Arc<GatewayDispatch>,
}

impl ToMcpGateway {
    /// Assemble the gateway over a dispatch spine.
    pub fn new(dispatch: Arc<GatewayDispatch>) -> Self {
        Self { dispatch }
    }

    /// The dispatch spine the tools invoke through.
    pub fn dispatch(&self) -> &Arc<GatewayDispatch> {
        &self.dispatch
    }

    fn extract_identity(context: &RequestContext<RoleServer>) -> Option<Identity> {
        Self::extract_identity_from_extensions(&context.extensions)
    }

    fn extract_identity_from_extensions(extensions: &rmcp::model::Extensions) -> Option<Identity> {
        let parts = extensions.get::<http::request::Parts>()?;
        parts
            .extensions
            .get::<Option<Identity>>()
            .and_then(Option::clone)
    }

    async fn handle_search(
        &self,
        query: Option<String>,
        identity: Option<Identity>,
    ) -> CallToolResult {
        let response = self
            .dispatch
            .invoke(identity.clone(), OP_SERVICES_LIST, Value::Null)
            .await;
        map_search_response(response, query.as_deref())
    }

    async fn handle_schema(
        &self,
        arguments: Option<JsonObject>,
        identity: Option<Identity>,
    ) -> CallToolResult {
        let name = match arguments
            .and_then(|mut a| a.remove("name"))
            .and_then(|v| v.as_str().map(str::to_string))
        {
            Some(n) => n,
            None => {
                return call_error_to_structured_error(CallError::invalid_input(
                    "invalid arguments: `name` is required and must be a string",
                ));
            }
        };
        if let Some(error) =
            schema_visibility_and_access_denial(self.dispatch.registry(), &name, identity.as_ref())
        {
            return call_error_to_structured_error(error);
        }
        let response = self
            .dispatch
            .invoke(
                identity,
                OP_SERVICES_SCHEMA,
                serde_json::json!({ "name": name }),
            )
            .await;
        envelope_to_call_tool_result(response)
    }

    async fn handle_call(
        &self,
        arguments: Option<JsonObject>,
        identity: Option<Identity>,
    ) -> CallToolResult {
        let (operation, input) = match parse_call_arguments(arguments) {
            Ok(pair) => pair,
            Err(err) => return call_error_to_structured_error(err),
        };
        let response = self.dispatch.invoke(identity, &operation, input).await;
        envelope_to_call_tool_result(response)
    }

    async fn handle_batch(
        &self,
        search_filter: Option<String>,
        arguments: Option<JsonObject>,
        identity: Option<Identity>,
    ) -> CallToolResult {
        let calls = match arguments
            .and_then(|mut a| a.remove("calls"))
            .and_then(|v| v.as_array().cloned())
        {
            Some(arr) => arr,
            None => {
                return call_error_to_structured_error(CallError::invalid_input(
                    "invalid arguments: `calls` is required and must be an array",
                ));
            }
        };

        let mut results: Vec<Value> = Vec::with_capacity(calls.len());
        for call in calls {
            let (operation, input) = match parse_call_arguments(call.as_object().cloned()) {
                Ok(pair) => pair,
                Err(err) => {
                    results.push(serde_json::json!({
                        "isError": true,
                        "error": serde_json::to_value(&err).unwrap_or(Value::Null),
                    }));
                    continue;
                }
            };
            let response = self
                .dispatch
                .invoke(identity.clone(), &operation, input)
                .await;
            results.push(envelope_to_value(response));
        }
        let _ = search_filter;
        CallToolResult::structured(serde_json::json!({ "results": results }))
    }
}

fn parse_call_arguments(arguments: Option<JsonObject>) -> Result<(String, Value), CallError> {
    let mut map = match arguments {
        Some(m) => m,
        None => {
            return Err(CallError::invalid_input(
                "invalid arguments: `operation` is required and must be a string",
            ));
        }
    };
    let operation = match map.remove("operation") {
        Some(Value::String(s)) => s,
        Some(other) => {
            return Err(CallError::invalid_input(format!(
                "invalid arguments: `operation` must be a string, got {}",
                json_type_name(&other)
            )));
        }
        None => {
            return Err(CallError::invalid_input(
                "invalid arguments: missing required field `operation`",
            ));
        }
    };
    let input = map.remove("input").unwrap_or(Value::Object(Map::new()));
    Ok((operation, input))
}

fn json_type_name(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "a boolean",
        Value::Number(_) => "a number",
        Value::String(_) => "a string",
        Value::Array(_) => "an array",
        Value::Object(_) => "an object",
    }
}

/// Symmetric with HTTP `GET /schema` (PRJ-06): internal ops are
/// invisible (NOT_FOUND regardless of caller), ACL-forbidden ops
/// return FORBIDDEN. The MCP `schema` tool must never return the full
/// spec of an op the caller could not call.
fn schema_visibility_and_access_denial(
    registry: &OperationRegistry,
    operation: &str,
    identity: Option<&Identity>,
) -> Option<CallError> {
    let name = operation.strip_prefix('/').unwrap_or(operation);
    let registration = registry.registration(name)?;
    if registration.spec.visibility == Visibility::Internal {
        return Some(CallError::not_found(operation));
    }
    if let AccessResult::Forbidden(message) =
        registration.spec.access_control.check(identity, None, None)
    {
        return Some(CallError::forbidden(message));
    }
    None
}

fn map_search_response(response: ResponseEnvelope, query: Option<&str>) -> CallToolResult {
    match response.result {
        Ok(value) => {
            let operations = value
                .get("operations")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            let filtered: Vec<Value> = operations
                .into_iter()
                .filter(|op| {
                    let op_type = op.get("op_type").and_then(Value::as_str).unwrap_or("");
                    !matches!(op_type, "sub" | "pub")
                })
                .filter(|op| match query {
                    Some(q) => op
                        .get("name")
                        .and_then(Value::as_str)
                        .is_some_and(|name| name.contains(q)),
                    None => true,
                })
                .map(|op| op_to_search_listing(&op))
                .collect();
            CallToolResult::structured(serde_json::json!({ "operations": filtered }))
        }
        Err(err) => call_error_to_structured_error(err),
    }
}

fn op_to_search_listing(op: &Value) -> Value {
    let name = op.get("name").and_then(Value::as_str).unwrap_or("");
    let op_type = op.get("op_type").and_then(Value::as_str).unwrap_or("query");
    let namespace = op.get("namespace").and_then(Value::as_str).unwrap_or("");
    let description = format!("{op_type} operation `{name}` in namespace `{namespace}`");
    serde_json::json!({
        "name": name,
        "description": description,
    })
}

fn envelope_to_call_tool_result(response: ResponseEnvelope) -> CallToolResult {
    match response.result {
        Ok(output) => CallToolResult::structured(object_result(output)),
        Err(err) => call_error_to_structured_error(err),
    }
}

fn call_error_to_structured_error(err: CallError) -> CallToolResult {
    let details = serde_json::to_value(&err).unwrap_or(Value::Null);
    CallToolResult::structured_error(details)
}

fn envelope_to_value(response: ResponseEnvelope) -> Value {
    match response.result {
        Ok(output) => serde_json::json!({
            "isError": false,
            "output": object_result(output),
        }),
        Err(err) => {
            let details = serde_json::to_value(&err).unwrap_or(Value::Null);
            serde_json::json!({
                "isError": true,
                "error": details,
            })
        }
    }
}

/// `structuredContent` must be a JSON object (strict MCP clients reject
/// non-object values). Object outputs pass through verbatim; any other
/// output is wrapped as `{"result": <output>}`.
fn object_result(output: Value) -> Value {
    if output.is_object() {
        output
    } else {
        serde_json::json!({ "result": output })
    }
}

pub(crate) fn gateway_tools() -> Vec<Tool> {
    vec![
        Tool::new(
            Cow::Borrowed(TOOL_SEARCH),
            Cow::Borrowed(
                "List available operations (filtered by the caller's AccessControl). Returns names + descriptions, not full schemas. Subscription and publish operations are excluded.",
            ),
            value_to_object(search_input_schema()),
        ),
        Tool::new(
            Cow::Borrowed(TOOL_SCHEMA),
            Cow::Borrowed(
                "Get the full OperationSpec for an operation (input/output JSON Schemas, error schemas).",
            ),
            value_to_object(schema_input_schema()),
        ),
        Tool::new(
            Cow::Borrowed(TOOL_CALL),
            Cow::Borrowed(
                "Invoke an operation by name with a JSON input. Returns the output as structuredContent (object outputs verbatim; non-object outputs wrapped as {\"result\": <output>}), or isError with typed error details for a CallError.",
            ),
            value_to_object(call_input_schema()),
        ),
        Tool::new(
            Cow::Borrowed(TOOL_BATCH),
            Cow::Borrowed(
                "Invoke multiple operations in one tool call. Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure.",
            ),
            value_to_object(batch_input_schema()),
        ),
    ]
}

fn parse_search_query(arguments: Option<&JsonObject>) -> Option<String> {
    arguments
        .and_then(|a| a.get("query"))
        .and_then(Value::as_str)
        .map(str::to_string)
}

fn value_to_object(value: Value) -> Arc<JsonObject> {
    match value {
        Value::Object(map) => Arc::new(map),
        _ => Arc::new(Map::new()),
    }
}

impl rmcp::handler::server::ServerHandler for ToMcpGateway {
    fn list_tools(
        &self,
        _request: Option<PaginatedRequestParams>,
        _context: RequestContext<RoleServer>,
    ) -> impl futures::Future<Output = Result<ListToolsResult, rmcp::ErrorData>> + Send + '_ {
        let tools = gateway_tools();
        std::future::ready(Ok(ListToolsResult::with_all_items(tools)))
    }

    fn call_tool(
        &self,
        request: CallToolRequestParams,
        context: RequestContext<RoleServer>,
    ) -> impl futures::Future<Output = Result<CallToolResult, rmcp::ErrorData>> + Send + '_ {
        let identity = Self::extract_identity(&context);
        let name = request.name.to_string();
        let arguments = request.arguments;
        let search_filter = parse_search_query(arguments.as_ref());
        let this = self;
        async move {
            let result = match name.as_str() {
                TOOL_SEARCH => this.handle_search(search_filter, identity).await,
                TOOL_SCHEMA => this.handle_schema(arguments, identity).await,
                TOOL_CALL => this.handle_call(arguments, identity).await,
                TOOL_BATCH => this.handle_batch(search_filter, arguments, identity).await,
                unknown => {
                    let err = CallError::new(
                        "NOT_FOUND",
                        format!("unknown gateway tool: {unknown}"),
                        false,
                    );
                    call_error_to_structured_error(err)
                }
            };
            Ok(result)
        }
    }

    fn get_info(&self) -> ServerInfo {
        let capabilities = ServerCapabilities::builder().enable_tools().build();
        ServerInfo::new(capabilities)
            .with_server_info(Implementation::new(
                "alkhttp-to-mcp",
                env!("CARGO_PKG_VERSION"),
            ))
            .with_instructions(
                "alk MCP gateway. Call `search` to discover operations, `schema` for an operation's full spec, `call` to invoke, `batch` to invoke many.",
            )
    }
}

/// The ready-to-mount rmcp service wrapping [`ToMcpGateway`] over the
/// streamable-HTTP transport.
pub type ToMcpService = StreamableHttpService<ToMcpGateway, LocalSessionManager>;

/// Build the mountable MCP service for a dispatch spine.
pub fn to_mcp_service(dispatch: Arc<GatewayDispatch>) -> ToMcpService {
    let gateway = ToMcpGateway::new(dispatch);
    StreamableHttpService::new(
        move || Ok(ToMcpGateway::new(Arc::clone(gateway.dispatch()))),
        LocalSessionManager::default().into(),
        StreamableHttpServerConfig::default(),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::core::types::Capabilities;
    use alkcall::registry::context::{OperationContext, ScopedPeerEnv};
    use alkcall::registry::discovery::{
        services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
    };
    use alkcall::registry::registration::{
        make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
        OperationProvenance, OperationRegistry,
    };
    use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
    use futures::StreamExt;
    use rmcp::model::Extensions;
    use std::collections::HashMap;

    fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
        Identity {
            id: id.to_string(),
            scopes: scopes.iter().map(|s| s.to_string()).collect(),
            resources: HashMap::new(),
        }
    }

    fn external_spec(name: &str, op_type: OperationType, acl: AccessControl) -> OperationSpec {
        OperationSpec::new(
            name,
            op_type,
            Visibility::External,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            acl,
            None,
        )
    }

    fn internal_spec(name: &str) -> OperationSpec {
        OperationSpec::new(
            name,
            OperationType::Query,
            Visibility::Internal,
            serde_json::json!({}),
            serde_json::json!({}),
            vec![],
            AccessControl::default(),
            None,
        )
    }

    fn make_echo_handler() -> alkcall::registry::registration::Handler {
        make_handler(
            |input, context| async move { ResponseEnvelope::ok(context.request_id, input) },
        )
    }

    fn make_echo_streaming_handler() -> alkcall::registry::registration::StreamingHandler {
        make_streaming_handler(|input, context| {
            futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
        })
    }

    fn handler_kind_for(op_type: OperationType) -> HandlerKind {
        match op_type {
            OperationType::Sub => HandlerKind::Stream(make_echo_streaming_handler()),
            OperationType::Query | OperationType::Mutation => {
                HandlerKind::Once(make_echo_handler())
            }
            OperationType::Pub => HandlerKind::Sink(make_echo_sink_handler()),
        }
    }

    fn make_echo_sink_handler() -> alkcall::registry::registration::SinkHandler {
        make_sink_handler(|_input, context: OperationContext, mut stream| async move {
            let mut last = Value::Null;
            while let Some(item) = stream.next().await {
                match item {
                    Ok(chunk) => last = chunk,
                    Err(err) => return ResponseEnvelope::error(context.request_id, err),
                }
            }
            ResponseEnvelope::ok(context.request_id, last)
        })
    }

    fn full_registry_with_ops(
        specs: Vec<(String, OperationType, AccessControl)>,
    ) -> Arc<OperationRegistry> {
        let mut inner = OperationRegistry::new();
        for (name, op_type, acl) in specs {
            inner
                .register(HandlerRegistration::new(
                    external_spec(&name, op_type, acl),
                    handler_kind_for(op_type),
                    OperationProvenance::Local,
                    None,
                    None,
                    Capabilities::new(),
                ))
                .unwrap();
        }
        let inner = Arc::new(inner);

        let mut dispatch_registry = OperationRegistry::new();
        for op in inner.list_operations() {
            dispatch_registry
                .register(HandlerRegistration::new(
                    external_spec(&op.name, op.op_type, op.access_control.clone()),
                    handler_kind_for(op.op_type),
                    OperationProvenance::Local,
                    None,
                    None,
                    Capabilities::new(),
                ))
                .unwrap();
        }
        dispatch_registry
            .register(HandlerRegistration::new(
                services_list_spec(),
                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                Some(ScopedPeerEnv::empty()),
                Capabilities::new(),
            ))
            .unwrap();
        dispatch_registry
            .register(HandlerRegistration::new(
                services_schema_spec(),
                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                Some(ScopedPeerEnv::empty()),
                Capabilities::new(),
            ))
            .unwrap();
        Arc::new(dispatch_registry)
    }

    fn dispatch(registry: Arc<OperationRegistry>) -> Arc<GatewayDispatch> {
        Arc::new(GatewayDispatch::new(registry))
    }

    fn extensions_with_identity(identity: Option<Identity>) -> Extensions {
        let request = http::Request::builder()
            .method(http::Method::POST)
            .uri("/mcp")
            .body(())
            .expect("valid request");
        let (mut parts, _) = request.into_parts();
        parts.extensions.insert(identity);
        let mut extensions = Extensions::new();
        extensions.insert(parts);
        extensions
    }

    async fn invoke_tool(
        gateway: &ToMcpGateway,
        name: &str,
        arguments: Option<JsonObject>,
        identity: Option<Identity>,
    ) -> CallToolResult {
        let search_filter = parse_search_query(arguments.as_ref());
        match name {
            TOOL_SEARCH => gateway.handle_search(search_filter, identity).await,
            TOOL_SCHEMA => gateway.handle_schema(arguments, identity).await,
            TOOL_CALL => gateway.handle_call(arguments, identity).await,
            TOOL_BATCH => {
                gateway
                    .handle_batch(search_filter, arguments, identity)
                    .await
            }
            unknown => {
                let err = CallError::new(
                    "NOT_FOUND",
                    format!("unknown gateway tool: {unknown}"),
                    false,
                );
                call_error_to_structured_error(err)
            }
        }
    }

    #[tokio::test]
    async fn list_tools_returns_exactly_four_gateway_tools() {
        let _gateway = ToMcpGateway::new(dispatch(full_registry_with_ops(vec![])));
        let tools = gateway_tools();
        let names: Vec<String> = tools.iter().map(|t| t.name.to_string()).collect();
        assert_eq!(names.len(), 4);
        assert!(names.contains(&"search".to_string()));
        assert!(names.contains(&"schema".to_string()));
        assert!(names.contains(&"call".to_string()));
        assert!(names.contains(&"batch".to_string()));
    }

    #[tokio::test]
    async fn list_tools_does_not_leak_registry_operations() {
        let registry = full_registry_with_ops(vec![(
            "fs/readFile".to_string(),
            OperationType::Query,
            AccessControl::default(),
        )]);
        let _gateway = ToMcpGateway::new(dispatch(registry));
        let tools = gateway_tools();
        for tool in &tools {
            assert_ne!(tool.name, "fs/readFile");
            assert_ne!(tool.name, "services/list");
            assert_ne!(tool.name, "services/schema");
        }
        assert_eq!(tools.len(), 4);
    }

    #[tokio::test]
    async fn search_returns_access_control_filtered_ops_excluding_subscriptions() {
        let registry = full_registry_with_ops(vec![
            (
                "public/echo".to_string(),
                OperationType::Query,
                AccessControl::default(),
            ),
            (
                "admin/secret".to_string(),
                OperationType::Query,
                AccessControl {
                    required_scopes: vec!["admin".to_string()],
                    ..Default::default()
                },
            ),
            (
                "events/stream".to_string(),
                OperationType::Sub,
                AccessControl::default(),
            ),
        ]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let result = invoke_tool(
            &gateway,
            "search",
            None,
            Some(identity_with_scopes("user", &["user"])),
        )
        .await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        let ops = structured
            .get("operations")
            .and_then(Value::as_array)
            .expect("operations array");
        let names: Vec<&str> = ops
            .iter()
            .filter_map(|o| o.get("name").and_then(Value::as_str))
            .collect();
        assert!(names.contains(&"public/echo"));
        assert!(
            !names.contains(&"admin/secret"),
            "ACL-filtered op must not appear"
        );
        assert!(
            !names.contains(&"events/stream"),
            "Subscription op must be excluded"
        );
        for op in ops {
            assert!(
                op.get("description").is_some(),
                "each entry has a description"
            );
            assert!(
                op.get("input_schema").is_none(),
                "search must not return full schemas"
            );
        }
    }

    #[tokio::test]
    async fn search_excludes_pub_ops() {
        let registry = full_registry_with_ops(vec![
            (
                "public/echo".to_string(),
                OperationType::Query,
                AccessControl::default(),
            ),
            (
                "metrics/ingest".to_string(),
                OperationType::Pub,
                AccessControl::default(),
            ),
        ]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let result = invoke_tool(&gateway, "search", None, None).await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        let names: Vec<String> = structured
            .get("operations")
            .and_then(Value::as_array)
            .expect("operations array")
            .iter()
            .filter_map(|o| o.get("name").and_then(Value::as_str).map(str::to_string))
            .collect();
        assert!(
            !names.contains(&"metrics/ingest".to_string()),
            "publish op must be excluded from search: {names:?}"
        );
        assert_eq!(names, vec!["public/echo".to_string()]);
    }

    #[tokio::test]
    async fn search_honors_query_substring_filter() {
        let registry = full_registry_with_ops(vec![
            (
                "fs/readFile".to_string(),
                OperationType::Query,
                AccessControl::default(),
            ),
            (
                "fs/writeFile".to_string(),
                OperationType::Mutation,
                AccessControl::default(),
            ),
            (
                "mail/send".to_string(),
                OperationType::Query,
                AccessControl::default(),
            ),
        ]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert("query".to_string(), Value::String("fs".to_string()));
        let result = invoke_tool(&gateway, "search", Some(args), None).await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        let mut names: Vec<String> = structured
            .get("operations")
            .and_then(Value::as_array)
            .expect("operations array")
            .iter()
            .filter_map(|o| o.get("name").and_then(Value::as_str).map(str::to_string))
            .collect();
        names.sort();
        assert_eq!(
            names,
            vec!["fs/readFile".to_string(), "fs/writeFile".to_string()],
            "query filter must keep only matching operations"
        );
    }

    #[tokio::test]
    async fn schema_returns_full_operation_spec() {
        let registry = full_registry_with_ops(vec![(
            "fs/readFile".to_string(),
            OperationType::Query,
            AccessControl::default(),
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert("name".to_string(), Value::String("fs/readFile".to_string()));
        let result = invoke_tool(&gateway, "schema", Some(args), None).await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        assert_eq!(
            structured.get("name"),
            Some(&Value::String("fs/readFile".to_string()))
        );
        assert!(structured.get("input_schema").is_some());
        assert!(structured.get("output_schema").is_some());
        assert!(structured.get("error_schemas").is_some());
    }

    #[tokio::test]
    async fn schema_returns_full_spec_for_authorized_identity() {
        let registry = full_registry_with_ops(vec![(
            "fs/readFile".to_string(),
            OperationType::Query,
            AccessControl {
                required_scopes: vec!["fs:read".to_string()],
                ..Default::default()
            },
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert("name".to_string(), Value::String("fs/readFile".to_string()));
        let result = invoke_tool(
            &gateway,
            "schema",
            Some(args),
            Some(identity_with_scopes("reader", &["fs:read"])),
        )
        .await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        assert_eq!(
            structured.get("name"),
            Some(&Value::String("fs/readFile".to_string()))
        );
        assert!(structured.get("access_control").is_some());
    }

    #[tokio::test]
    async fn schema_denies_acl_forbidden_op_symmetrically_with_http() {
        let registry = full_registry_with_ops(vec![(
            "admin/secret".to_string(),
            OperationType::Query,
            AccessControl {
                required_scopes: vec!["admin".to_string()],
                ..Default::default()
            },
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert(
            "name".to_string(),
            Value::String("admin/secret".to_string()),
        );
        let result = invoke_tool(
            &gateway,
            "schema",
            Some(args),
            Some(identity_with_scopes("user", &["user"])),
        )
        .await;
        assert_eq!(result.is_error, Some(true));
        let structured = result.structured_content.expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("FORBIDDEN".to_string()))
        );
        assert!(
            structured.get("input_schema").is_none(),
            "a denied schema lookup must not leak the op's schemas"
        );
    }

    #[tokio::test]
    async fn schema_denies_internal_op_with_not_found_symmetrically_with_http() {
        let mut inner = OperationRegistry::new();
        inner
            .register(HandlerRegistration::new(
                internal_spec("secret/op"),
                HandlerKind::Once(make_echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        let inner = Arc::new(inner);
        let mut dispatch_registry = OperationRegistry::new();
        dispatch_registry
            .register(HandlerRegistration::new(
                internal_spec("secret/op"),
                HandlerKind::Once(make_echo_handler()),
                OperationProvenance::Local,
                None,
                None,
                Capabilities::new(),
            ))
            .unwrap();
        dispatch_registry
            .register(HandlerRegistration::new(
                services_list_spec(),
                HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                Some(ScopedPeerEnv::empty()),
                Capabilities::new(),
            ))
            .unwrap();
        dispatch_registry
            .register(HandlerRegistration::new(
                services_schema_spec(),
                HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
                OperationProvenance::Local,
                None,
                Some(ScopedPeerEnv::empty()),
                Capabilities::new(),
            ))
            .unwrap();
        let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry)));

        let mut args = Map::new();
        args.insert("name".to_string(), Value::String("secret/op".to_string()));
        let unauthenticated = invoke_tool(&gateway, "schema", Some(args.clone()), None).await;
        assert_eq!(unauthenticated.is_error, Some(true));
        let structured = unauthenticated
            .structured_content
            .expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("NOT_FOUND".to_string()))
        );
        assert!(structured.get("input_schema").is_none());

        let unauthorized = invoke_tool(
            &gateway,
            "schema",
            Some(args),
            Some(identity_with_scopes("user", &["user"])),
        )
        .await;
        assert_eq!(unauthorized.is_error, Some(true));
        let structured = unauthorized
            .structured_content
            .expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("NOT_FOUND".to_string()))
        );
    }

    #[tokio::test]
    async fn call_returns_structured_for_success() {
        let registry = full_registry_with_ops(vec![(
            "echo/run".to_string(),
            OperationType::Query,
            AccessControl::default(),
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert(
            "operation".to_string(),
            Value::String("echo/run".to_string()),
        );
        args.insert("input".to_string(), serde_json::json!({ "msg": "hi" }));
        let result = invoke_tool(&gateway, "call", Some(args), None).await;
        assert_eq!(result.is_error, Some(false));
        assert_eq!(
            result.structured_content,
            Some(serde_json::json!({ "msg": "hi" }))
        );
    }

    #[tokio::test]
    async fn call_wraps_non_object_output_into_object_structured_content() {
        for output in [
            Value::String("just a string".to_string()),
            serde_json::json!([1, 2, 3]),
            Value::Null,
            Value::Bool(true),
        ] {
            let registry = full_registry_with_ops(vec![(
                "echo/run".to_string(),
                OperationType::Query,
                AccessControl::default(),
            )]);
            let gateway = ToMcpGateway::new(dispatch(registry));

            let mut args = Map::new();
            args.insert(
                "operation".to_string(),
                Value::String("echo/run".to_string()),
            );
            args.insert("input".to_string(), output.clone());
            let result = invoke_tool(&gateway, "call", Some(args), None).await;
            assert_eq!(result.is_error, Some(false));
            assert_eq!(
                result.structured_content,
                Some(serde_json::json!({ "result": output })),
                "non-object output must be wrapped into an object"
            );
        }
    }

    #[tokio::test]
    async fn call_argument_errors_carry_retryable_and_truthy_messages() {
        let registry = full_registry_with_ops(vec![]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let missing = invoke_tool(&gateway, "call", None, None).await;
        assert_eq!(missing.is_error, Some(true));
        let structured = missing
            .structured_content
            .expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("INVALID_INPUT".to_string()))
        );
        assert_eq!(structured.get("retryable"), Some(&Value::Bool(false)));

        let mut args = Map::new();
        args.insert("operation".to_string(), Value::Number(42.into()));
        let non_string = invoke_tool(&gateway, "call", Some(args), None).await;
        assert_eq!(non_string.is_error, Some(true));
        let structured = non_string
            .structured_content
            .expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("INVALID_INPUT".to_string()))
        );
        assert_eq!(structured.get("retryable"), Some(&Value::Bool(false)));
        let message = structured
            .get("message")
            .and_then(Value::as_str)
            .expect("message");
        assert!(
            message.contains("must be a string"),
            "non-string operation must not report a missing field: {message}"
        );
    }

    #[tokio::test]
    async fn call_returns_structured_error_for_call_error() {
        let registry = full_registry_with_ops(vec![]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert(
            "operation".to_string(),
            Value::String("no/such".to_string()),
        );
        args.insert("input".to_string(), Value::Object(Map::new()));
        let result = invoke_tool(&gateway, "call", Some(args), None).await;
        assert_eq!(result.is_error, Some(true));
        let structured = result.structured_content.expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("NOT_FOUND".to_string()))
        );
    }

    #[tokio::test]
    async fn batch_returns_object_with_result_entries() {
        let registry = full_registry_with_ops(vec![(
            "echo/run".to_string(),
            OperationType::Query,
            AccessControl::default(),
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert(
            "calls".to_string(),
            serde_json::json!([
                { "operation": "echo/run", "input": { "n": 1 } },
                { "operation": "no/such", "input": {} },
                { "operation": 7, "input": {} },
            ]),
        );
        let result = invoke_tool(&gateway, "batch", Some(args), None).await;
        assert_eq!(result.is_error, Some(false));
        let structured = result.structured_content.expect("structured present");
        assert!(
            structured.is_object(),
            "batch structuredContent must be an object, not an array"
        );
        let results = structured
            .get("results")
            .and_then(Value::as_array)
            .expect("results array");
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].get("isError"), Some(&Value::Bool(false)));
        assert_eq!(
            results[0].get("output"),
            Some(&serde_json::json!({ "n": 1 }))
        );
        assert!(results[0].get("error").is_none());
        assert_eq!(results[1].get("isError"), Some(&Value::Bool(true)));
        let not_found = results[1].get("error").expect("error object");
        assert_eq!(
            not_found.get("code"),
            Some(&Value::String("NOT_FOUND".to_string()))
        );
        assert_eq!(
            not_found.get("retryable"),
            Some(&Value::Bool(false)),
            "batch error items carry the full CallError shape"
        );
        assert_eq!(results[2].get("isError"), Some(&Value::Bool(true)));
        assert_eq!(
            results[2].get("error").and_then(|e| e.get("code")),
            Some(&Value::String("INVALID_INPUT".to_string()))
        );
    }

    #[tokio::test]
    async fn call_with_restricted_op_and_unauthorized_identity_returns_forbidden_error() {
        let registry = full_registry_with_ops(vec![(
            "admin/run".to_string(),
            OperationType::Query,
            AccessControl {
                required_scopes: vec!["admin".to_string()],
                ..Default::default()
            },
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let mut args = Map::new();
        args.insert(
            "operation".to_string(),
            Value::String("admin/run".to_string()),
        );
        args.insert("input".to_string(), Value::Object(Map::new()));
        let result = invoke_tool(&gateway, "call", Some(args), None).await;
        assert_eq!(result.is_error, Some(true));
        let structured = result.structured_content.expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("FORBIDDEN".to_string()))
        );
    }

    #[tokio::test]
    async fn unknown_tool_name_returns_not_found_structured_error() {
        let gateway = ToMcpGateway::new(dispatch(Arc::new(OperationRegistry::new())));
        let result = invoke_tool(&gateway, "bogus", None, None).await;
        assert_eq!(result.is_error, Some(true));
        let structured = result.structured_content.expect("structured error present");
        assert_eq!(
            structured.get("code"),
            Some(&Value::String("NOT_FOUND".to_string()))
        );
    }

    #[tokio::test]
    async fn identity_survives_rmcp_framing_into_call_tool() {
        let registry = full_registry_with_ops(vec![(
            "admin/run".to_string(),
            OperationType::Query,
            AccessControl {
                required_scopes: vec!["admin".to_string()],
                ..Default::default()
            },
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let admin_identity = identity_with_scopes("admin-peer", &["admin"]);
        let extensions = extensions_with_identity(Some(admin_identity.clone()));
        let extracted = ToMcpGateway::extract_identity_from_extensions(&extensions);
        assert_eq!(
            extracted.as_ref().map(|i| &i.id),
            Some(&"admin-peer".to_string())
        );

        let mut args = Map::new();
        args.insert(
            "operation".to_string(),
            Value::String("admin/run".to_string()),
        );
        args.insert("input".to_string(), serde_json::json!({ "ok": 1 }));
        let result = gateway.handle_call(Some(args), extracted).await;
        assert_eq!(result.is_error, Some(false));
        assert_eq!(
            result.structured_content,
            Some(serde_json::json!({ "ok": 1 }))
        );
    }

    #[test]
    fn extract_identity_returns_none_when_no_parts_in_extensions() {
        let extensions = Extensions::new();
        assert!(ToMcpGateway::extract_identity_from_extensions(&extensions).is_none());
    }

    #[test]
    fn extract_identity_returns_none_when_parts_have_no_identity() {
        let extensions = extensions_with_identity(None);
        assert!(ToMcpGateway::extract_identity_from_extensions(&extensions).is_none());
    }

    #[test]
    fn extract_identity_reads_stashed_option_identity_from_parts() {
        let id = identity_with_scopes("caller", &["read"]);
        let extensions = extensions_with_identity(Some(id.clone()));
        let extracted = ToMcpGateway::extract_identity_from_extensions(&extensions);
        assert_eq!(
            extracted.as_ref().map(|i| i.id.clone()),
            Some("caller".to_string())
        );
        assert_eq!(
            extracted.as_ref().map(|i| i.scopes.clone()),
            Some(vec!["read".to_string()])
        );
    }

    #[test]
    fn to_mcp_is_not_an_operation_adapter() {
        fn assert_not_adapter<T>() {}
        assert_not_adapter::<ToMcpGateway>();
    }

    #[test]
    fn gateway_tools_definition_is_stable() {
        let tools = gateway_tools();
        assert_eq!(tools.len(), 4);
        assert_eq!(tools[0].name, "search");
        assert_eq!(tools[1].name, "schema");
        assert_eq!(tools[2].name, "call");
        assert_eq!(tools[3].name, "batch");
    }

    #[tokio::test]
    async fn search_schema_call_round_trip() {
        let registry = full_registry_with_ops(vec![(
            "fs/readFile".to_string(),
            OperationType::Query,
            AccessControl::default(),
        )]);
        let gateway = ToMcpGateway::new(dispatch(registry));

        let search_result = invoke_tool(&gateway, "search", None, None).await;
        let ops = search_result
            .structured_content
            .as_ref()
            .and_then(|v| v.get("operations"))
            .and_then(Value::as_array)
            .expect("search ops");
        let first_name = ops[0].get("name").and_then(Value::as_str).expect("name");
        assert_eq!(first_name, "fs/readFile");

        let mut schema_args = Map::new();
        schema_args.insert("name".to_string(), Value::String(first_name.to_string()));
        let schema_result = invoke_tool(&gateway, "schema", Some(schema_args), None).await;
        assert_eq!(
            schema_result
                .structured_content
                .as_ref()
                .and_then(|v| v.get("name"))
                .and_then(Value::as_str),
            Some("fs/readFile")
        );

        let mut call_args = Map::new();
        call_args.insert(
            "operation".to_string(),
            Value::String(first_name.to_string()),
        );
        call_args.insert(
            "input".to_string(),
            serde_json::json!({ "path": "/etc/hosts" }),
        );
        let call_result = invoke_tool(&gateway, "call", Some(call_args), None).await;
        assert_eq!(
            call_result.structured_content,
            Some(serde_json::json!({ "path": "/etc/hosts" }))
        );
    }
}