car-server-core 0.55.0

//! MCP HTTP-streamable transport for the daemon.
//!
//! Wraps `car_mcp::Server` in an axum Router so MCP-aware clients
//! (Claude Desktop, Cursor, Claude Code's `--mcp-config`, Codex when
//! MCP-aware, custom GPTs, third-party agents) can call CAR's tools
//! through one governance pipeline. Same dispatch logic as the
//! `car-mcp-server` stdio binary; what changes is the transport
//! and the shared engine — the daemon binds the same
//! `Arc<Mutex<MemgineEngine>>` it uses for WS traffic, so facts
//! ingested via MCP show up in WS-served queries and vice versa.
//!
//! ## Wire shape (MCP over HTTP; revisions per [`car_mcp::SUPPORTED_VERSIONS`])
//!
//! - **POST `/mcp`** — body is one JSON-RPC 2.0 request, response
//!   is the JSON-RPC reply. A notification (no `id`) has no reply,
//!   so it is answered `202 Accepted` with an empty body. The HTTP
//!   layer never invents protocol semantics; everything routes
//!   through [`car_mcp::Server::handle`].
//! - **`initialize` negotiates.** The requested `protocolVersion`
//!   comes back when it is one this server supports; absent,
//!   malformed, and unsupported all get `car_mcp::PROTOCOL_VERSION`
//!   and the client decides whether to proceed. On the daemon endpoint,
//!   the response also mints `MCP-Session-Id` for peer attribution and a
//!   polling inbox.
//! - **DELETE `/mcp`** — closes the protocol session named by
//!   `MCP-Session-Id` and discards its unread peer inbox.
//! - **GET `/mcp/health`** — liveness probe. Returns
//!   `{"status":"ok","protocol_version":"…"}` interpolating
//!   [`car_mcp::PROTOCOL_VERSION`] (`2025-06-18` today). Useful for
//!   hosts that want to confirm the endpoint is up without sending
//!   a JSON-RPC initialize.
//!
//! There is **no server→client push channel, by decision** (car#972
//! Track B): the SSE registry and its `push_to_session` primitive were
//! removed after weeks with zero production callers, and `GET /mcp`
//! answers `405 Method Not Allowed` — the streamable-HTTP spec's branch
//! for a server that offers no SSE stream. The small protocol-session
//! record used for peer identity and the polled `peer_inbox` has no sink
//! and does not reverse that decision. Push-dependent MCP features
//! (`listChanged`, `resources/subscribe`, `logging/setLevel`) remain
//! declined at the capabilities literal in `car_mcp::Server::handle`.
//!
//! ## Origin validation (DNS-rebinding protection)
//!
//! The MCP HTTP transport requires the server to validate `Origin`
//! (car#972 §3). Without it, any web page the user happens to have
//! open can POST to `http://127.0.0.1:9102/mcp` and drive CAR's
//! tools — the browser attaches the user's loopback reachability,
//! and a rebound DNS name defeats the loopback bind on its own.
//! The rule this module implements is **absent-permissive,
//! present-strict**:
//!
//! - **No `Origin` header → allowed.** A browser ALWAYS sends
//!   `Origin` on a cross-origin request, so its absence cannot be a
//!   browser cross-origin attack. This is what keeps every real
//!   client working: `curl`, the `car-mcp` stdio binary's HTTP
//!   cousin, Claude Desktop, Cursor, Claude Code `--mcp-config`,
//!   Codex, and CAR's own `car-connectors` client (which sends
//!   only `content-type`, `accept`, `mcp-session-id` and
//!   `MCP-Protocol-Version`) send no `Origin` at all. Requiring one
//!   would break all of them and buy nothing.
//! - **`Origin` present and loopback → allowed.** `http`/`https`
//!   with a host of `localhost`, anything in `127.0.0.0/8`, or
//!   `::1` (bracketed or not), on any port. Covers a local dev page
//!   on `http://localhost:3000` and the daemon's own dashboard.
//! - **Anything else → `403`** with `{"error":"origin not
//!   allowed"}`. The literal `null` origin (sandboxed iframe,
//!   `file://`, a redirected cross-origin request) is opaque, not
//!   safe, and is rejected with the rest.
//!
//! The rule does **not** widen for a non-loopback `--mcp-bind`. A
//! wildcard bind's host is `0.0.0.0`, which never appears as an
//! `Origin` value, so matching against the bound address would be
//! dead code in exactly the case it was written for; and a wider
//! exposure makes the guard more important, not less. Deployments
//! that genuinely want browser traffic from another origin should
//! front the daemon with a reverse proxy that owns its own
//! CORS/`Origin` policy.
//!
//! `GET /mcp/health` is deliberately **exempt** — see
//! [`handle_health`].
//!
//! ## `MCP-Protocol-Version` (car#972 §3)
//!
//! Once a client has negotiated a revision on `initialize`, the
//! HTTP transport asks it to name that revision on every later
//! request, and asks the server to reject a value it does not
//! speak. Same shape as the `Origin` rule above — **absent-
//! permissive, present-strict** — and guarded on `POST /mcp` and
//! `DELETE /mcp`. The constant `GET /mcp` response is exempt:
//!
//! - **No header → allowed.** Absent cannot be an error here.
//!   Requiring the header would break `curl`, Claude Desktop,
//!   Cursor, and Claude Code for no gain — and now that
//!   [`car_mcp::SUPPORTED_VERSIONS`] holds two entries, a silent
//!   client is simply served the revision it negotiated on
//!   `initialize`: the 2025-06-18 delta on this surface is additive
//!   result keys (`structuredContent`) a 2024-11-05 client ignores,
//!   so dispatch needs no per-request version to answer correctly.
//! - **A value in [`car_mcp::SUPPORTED_VERSIONS`] → allowed.**
//!   `car-connectors` is the in-repo reference client: it echoes the
//!   version it negotiated on `initialize`,
//!   which is exactly what this guard expects to see.
//! - **Anything else → `400`** with `{"error":"unsupported
//!   MCP-Protocol-Version", "requested":…, "supported":[…]}`. An
//!   empty or non-UTF-8 value is rejected with the rest, the same
//!   way `check_origin` treats a non-UTF-8 `Origin`.
//!
//! A foreign value is a `400` rather than a silent downshift
//! because the client has *told us* which dialect it will read the
//! reply in. Answering it in a revision it never agreed to is the
//! same inversion the `initialize` negotiation exists to avoid,
//! only without a handshake to catch it — the reply would parse as
//! that other revision and mean something else. Better to fail the
//! request and name the versions we do speak, which is why the
//! body carries the supported list: a client that guessed wrong
//! has no other way to learn what to send.
//!
//! `GET /mcp/health` is exempt from this guard too, for the same
//! liveness-probe reason — see [`handle_health`].
//!
//! ## Lifecycle
//!
//! [`start_mcp_with_peer_sessions`] returns a [`JoinHandle`] for the axum task
//! and binds protocol sessions to daemon state. [`start_mcp`] is the sessionless
//! embedder/test form. The daemon's main holds the handle for its process
//! lifetime; on shutdown
//! the handle is dropped and the task winds down. Failure to bind
//! is reported synchronously — the daemon's startup path can decide
//! whether to abort or continue without MCP.

use std::net::SocketAddr;
use std::sync::Arc;

use axum::{
    extract::State,
    http::{HeaderMap, StatusCode, Uri},
    response::{IntoResponse, Json},
    routing::{get, post},
    Router,
};
use serde_json::{json, Value};
use tokio::task::JoinHandle;

use car_mcp::error_codes::PARSE as E_PARSE;
use car_mcp::{Request as McpRequest, Server as McpServer};

/// Header a client uses to name the protocol revision it negotiated
/// on `initialize` — per the 2025-06-18 spec. Lowercase deliberately;
/// `HeaderMap` lookup is case-insensitive, so a client sending
/// `MCP-Protocol-Version` still matches.
const PROTOCOL_HEADER: &str = "mcp-protocol-version";
/// Streamable-HTTP protocol session header.
const SESSION_HEADER: &str = "mcp-session-id";
/// Query marker added only to MCP URLs CAR generates for batch children.
const BATCH_MODE_QUERY: &str = "car-mcp-mode=batch";

tokio::task_local! {
    static MCP_PEER_PRINCIPAL: Option<String>;
}

/// The peer principal bound to the MCP request currently executing a tool.
pub(crate) fn current_mcp_peer_principal() -> Option<String> {
    MCP_PEER_PRINCIPAL.try_with(Clone::clone).ok().flatten()
}

/// Is this `Origin` value one we accept? See the module doc for the
/// rule and why it is shaped this way.
///
/// Hand-parses the `Origin` grammar — `scheme "://" host [":" port]`
/// — rather than pulling in a URL crate for fifteen lines.
/// `car-server-core` has no `url` dependency and this is not a
/// reason to add one; the grammar is fixed and tiny, and every
/// deviation from it is something we want to reject anyway.
///
/// Deliberately strict about the host: an exact match against the
/// loopback set, never a substring test. `http://localhost.evil.com`
/// and `http://sub.localhost.evil.example` both *contain*
/// `localhost` and both must be rejected.
fn origin_allowed(origin: &str) -> bool {
    // The opaque origin. Sent by sandboxed iframes, `file://` pages,
    // and cross-origin redirects — i.e. exactly the contexts we have
    // the least reason to trust. Opaque is not the same as safe.
    if origin.is_empty() || origin == "null" {
        return false;
    }

    // scheme "://" rest — anything without the separator is not an
    // origin at all.
    let Some((scheme, rest)) = origin.split_once("://") else {
        return false;
    };
    if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
        return false;
    }

    // Strip an optional `:port`. The bracketed IPv6 form (`[::1]:3000`)
    // puts colons inside the host, so split after the closing bracket
    // rather than on the last colon.
    let host = match rest.strip_prefix('[') {
        Some(after_bracket) => match after_bracket.split_once(']') {
            // Whatever follows `]` must be empty or a `:port`.
            Some((inner, tail)) if tail.is_empty() || tail.starts_with(':') => inner,
            _ => return false,
        },
        None => match rest.split_once(':') {
            Some((h, _port)) => h,
            None => rest,
        },
    };
    let host = host.to_ascii_lowercase();

    if host == "localhost" || host == "::1" {
        return true;
    }
    // The whole of 127.0.0.0/8 is loopback, not just 127.0.0.1 —
    // glibc resolves `localhost` to 127.0.0.53 on systemd-resolved
    // hosts, for one.
    matches!(host.parse::<std::net::IpAddr>(), Ok(ip) if ip.is_loopback())
}

/// Gate one request on its `Origin` header.
///
/// `Ok(())` when the header is absent or names an allowed origin;
/// `Err(response)` carrying a `403` when it names anything else. A
/// non-UTF-8 header value is rejected with the rest — it cannot be a
/// well-formed origin, and we will not guess at it.
///
/// Rejections are logged at `warn`: a browser page trying to drive
/// the daemon is a real signal, not routine noise.
fn check_origin(headers: &HeaderMap) -> Result<(), axum::response::Response> {
    let Some(raw) = headers.get(axum::http::header::ORIGIN) else {
        return Ok(());
    };
    let allowed = raw.to_str().map(origin_allowed).unwrap_or(false);
    if allowed {
        return Ok(());
    }
    let shown = raw.to_str().unwrap_or("<non-utf8>");
    tracing::warn!(origin = %shown, "MCP request rejected: origin not allowed");
    Err((
        StatusCode::FORBIDDEN,
        Json(json!({ "error": "origin not allowed" })),
    )
        .into_response())
}

/// Gate one request on its `MCP-Protocol-Version` header. See the
/// module doc for the rule and why absent cannot be an error.
///
/// `Ok(())` when the header is absent or names a revision in
/// [`car_mcp::SUPPORTED_VERSIONS`]; `Err(response)` carrying a `400`
/// for anything else. An empty or non-UTF-8 value is rejected with
/// the rest — it names no revision we could honour, and we will not
/// guess at it.
///
/// The `400` body names the supported list: a client that guessed
/// wrong has no other way to learn what to send, and the header is
/// sent *after* the handshake that would otherwise have told it.
///
/// Rejections are logged at `warn`, same as the origin rejection —
/// a client speaking a revision we don't is worth seeing.
fn check_protocol_version(headers: &HeaderMap) -> Result<(), axum::response::Response> {
    let Some(raw) = headers.get(PROTOCOL_HEADER) else {
        return Ok(());
    };
    if raw
        .to_str()
        .map(|v| car_mcp::SUPPORTED_VERSIONS.contains(&v))
        .unwrap_or(false)
    {
        return Ok(());
    }
    let shown = raw.to_str().unwrap_or("<non-utf8>");
    tracing::warn!(version = %shown, "MCP request rejected: unsupported protocol version");
    Err((
        StatusCode::BAD_REQUEST,
        Json(json!({
            "error": "unsupported MCP-Protocol-Version",
            "requested": shown,
            "supported": car_mcp::SUPPORTED_VERSIONS,
        })),
    )
        .into_response())
}

#[derive(Clone)]
struct McpState {
    server: Arc<McpServer>,
    peer_state: Option<Arc<crate::ServerState>>,
}

/// Start the MCP HTTP listener bound to `addr`. Returns `Ok` with
/// `(bound_addr, join_handle)` on success — the bound address may
/// differ from the requested one when port `0` is supplied (the OS
/// picks).
///
/// Returns `Err` synchronously when binding fails so the daemon's
/// startup path can log and decide whether to continue.
///
/// This used to return a third element: an SSE sink registry a handler could
/// `push_to_session` on. It was removed with the car#972 Track B decision (see
/// the module doc). The optional daemon state here holds identities and polled
/// inboxes only; it cannot send a server-initiated MCP message.
pub async fn start_mcp(
    server: Arc<McpServer>,
    addr: SocketAddr,
) -> Result<(SocketAddr, JoinHandle<()>), String> {
    start_mcp_inner(server, addr, None).await
}

/// Start MCP with daemon-backed protocol sessions for peer identity/inboxes.
pub async fn start_mcp_with_peer_sessions(
    server: Arc<McpServer>,
    addr: SocketAddr,
    peer_state: Arc<crate::ServerState>,
) -> Result<(SocketAddr, JoinHandle<()>), String> {
    start_mcp_inner(server, addr, Some(peer_state)).await
}

async fn start_mcp_inner(
    server: Arc<McpServer>,
    addr: SocketAddr,
    peer_state: Option<Arc<crate::ServerState>>,
) -> Result<(SocketAddr, JoinHandle<()>), String> {
    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .map_err(|e| format!("bind {addr}: {e}"))?;
    let bound = listener
        .local_addr()
        .map_err(|e| format!("local_addr: {e}"))?;

    let state = McpState { server, peer_state };
    let app: Router = Router::new()
        .route(
            "/mcp",
            post(handle_mcp_post)
                .get(handle_mcp_get)
                .delete(handle_mcp_delete),
        )
        .route("/mcp/health", get(handle_health))
        .with_state(state);

    let task = tokio::spawn(async move {
        if let Err(e) = axum::serve(listener, app).await {
            tracing::warn!(error = %e, "mcp HTTP server exited");
        }
    });

    Ok((bound, task))
}

/// Liveness probe. Deliberately **not** `Origin`-guarded, and
/// exempt from the `MCP-Protocol-Version` check for the same
/// reason.
///
/// It has no side effect and returns no secret — status, protocol
/// version, server name — and the canonical
/// `curl http://127.0.0.1:9102/mcp/health` in the authoring guide,
/// plus any browser-based uptime check, must keep working. Guarding
/// it would add no security and break documented usage. The
/// protocol guard would be actively counterproductive here: a
/// client whose version we reject is precisely the one that needs
/// to read `protocol_version` off this route to find out what we
/// speak.
async fn handle_health() -> impl IntoResponse {
    Json(json!({
        "status": "ok",
        "protocol_version": car_mcp::PROTOCOL_VERSION,
        "server_name": car_mcp::SERVER_NAME,
    }))
}

/// `GET /mcp` — **405 Method Not Allowed**, deliberately.
///
/// The streamable-HTTP transport lets a client GET the endpoint to
/// open an SSE stream for server-initiated messages, and lets a
/// server that has none answer `405` instead — the spec's own branch
/// for "this server does not offer an SSE stream at this endpoint".
/// CAR has no server-initiated MCP messages by decision (car#972
/// Track B; see the module doc), so this is that branch. The SSE
/// stream that used to live here carried only a session-id
/// announcement and keepalives — no production code ever pushed a
/// payload down it.
///
/// No `Origin` or protocol-version guard: the response is a constant
/// with no side effect and no secrets, so there is nothing to guard.
/// The `Allow` header names POST and DELETE, per HTTP semantics.
async fn handle_mcp_get() -> impl IntoResponse {
    (
        StatusCode::METHOD_NOT_ALLOWED,
        [(axum::http::header::ALLOW, "POST, DELETE")],
        Json(json!({
            "error": "this server offers no SSE stream; POST JSON-RPC requests to /mcp",
        })),
    )
}

// `HeaderMap` must precede `body: String`: `String` is the body
// extractor and axum requires it last.
async fn handle_mcp_post(
    State(state): State<McpState>,
    headers: HeaderMap,
    uri: Uri,
    body: String,
) -> axum::response::Response {
    // First, before the parse — a rejected request must never reach
    // `Server::handle`, or the guard blocks the reply while the side
    // effect still lands.
    if let Err(resp) = check_origin(&headers) {
        return resp;
    }
    if let Err(resp) = check_protocol_version(&headers) {
        return resp;
    }

    // Parse the JSON-RPC envelope. A failed parse returns the
    // standard `-32700` parse error, mirroring what the stdio
    // transport does.
    let req: McpRequest = match serde_json::from_str(&body) {
        Ok(req) => req,
        Err(e) => {
            let resp = json!({
                "jsonrpc": "2.0",
                "id": Value::Null,
                "error": {
                    "code": E_PARSE,
                    "message": format!("parse error: {e}"),
                },
            });
            return (StatusCode::OK, Json(resp)).into_response();
        }
    };

    // `initialize` creates the protocol session. Later requests recover that
    // exact principal from MCP-Session-Id. An absent header remains allowed for
    // non-peer tools so older MCP clients keep working; the peer tools refuse
    // without a bound session rather than falling back to a shared identity.
    let is_initialize = req.method == "initialize";
    let mut minted_session_id = None;
    let principal = if let Some(peer_state) = state.peer_state.as_ref() {
        if is_initialize {
            let receive_capable = uri
                .query()
                .is_none_or(|query| !query.split('&').any(|part| part == BATCH_MODE_QUERY));
            let (session_id, principal) =
                crate::peers::open_mcp_peer_session(peer_state, receive_capable).await;
            minted_session_id = Some(session_id);
            Some(principal)
        } else if let Some(session_id) = headers
            .get(SESSION_HEADER)
            .and_then(|value| value.to_str().ok())
        {
            match crate::peers::touch_mcp_peer_session(peer_state, session_id).await {
                Some(principal) => Some(principal),
                None => {
                    return (
                        StatusCode::NOT_FOUND,
                        Json(json!({ "error": "unknown or expired MCP session" })),
                    )
                        .into_response();
                }
            }
        } else {
            None
        }
    } else {
        None
    };

    // Dispatch through the same handler the stdio transport uses. The task-local
    // principal reaches daemon-registered peer tools without changing car-mcp's
    // transport-neutral ToolHandler trait.
    let handled = MCP_PEER_PRINCIPAL
        .scope(
            principal,
            crate::mcp_daemon::handle_request(&state.server, req, &headers),
        )
        .await;
    let mut response = match handled {
        Some(resp) => match serde_json::to_value(&resp) {
            Ok(v) => (StatusCode::OK, Json(v)).into_response(),
            Err(e) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({
                    "jsonrpc": "2.0",
                    "id": Value::Null,
                    "error": {
                        "code": -32603,
                        "message": format!("response serialization failed: {e}"),
                    },
                })),
            )
                .into_response(),
        },
        None => StatusCode::ACCEPTED.into_response(),
    };
    if let Some(session_id) = minted_session_id {
        if let Ok(value) = session_id.parse() {
            response.headers_mut().insert(SESSION_HEADER, value);
        }
    }
    response
}

/// End one protocol session and discard its unread peer inbox.
async fn handle_mcp_delete(
    State(state): State<McpState>,
    headers: HeaderMap,
) -> axum::response::Response {
    if let Err(resp) = check_origin(&headers) {
        return resp;
    }
    if let Err(resp) = check_protocol_version(&headers) {
        return resp;
    }
    let Some(peer_state) = state.peer_state.as_ref() else {
        return StatusCode::METHOD_NOT_ALLOWED.into_response();
    };
    let Some(session_id) = headers
        .get(SESSION_HEADER)
        .and_then(|value| value.to_str().ok())
    else {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": "missing MCP-Session-Id" })),
        )
            .into_response();
    };
    if crate::peers::close_mcp_peer_session(peer_state, session_id).await {
        StatusCode::NO_CONTENT.into_response()
    } else {
        (
            StatusCode::NOT_FOUND,
            Json(json!({ "error": "unknown or expired MCP session" })),
        )
            .into_response()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    /// Build a server with a fresh memgine, start it on an OS-picked
    /// port, return the bound address and the join handle so the
    /// caller can shut down. Used for in-process integration tests.
    async fn boot_test_server() -> (SocketAddr, JoinHandle<()>) {
        let server = Arc::new(McpServer::new());
        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let (bound, task) = start_mcp(server, addr).await.expect("start_mcp");
        (bound, task)
    }

    async fn http_post(addr: SocketAddr, body: &str) -> (StatusCode, Value) {
        http_post_with_origin(addr, body, None).await
    }

    /// POST with an optional `Origin`. `None` is what every real MCP
    /// client sends (see the module doc), so it is also what
    /// [`http_post`] sends.
    async fn http_post_with_origin(
        addr: SocketAddr,
        body: &str,
        origin: Option<&str>,
    ) -> (StatusCode, Value) {
        let (status, text) = http_post_raw(addr, body, origin, None).await;
        let value: Value = serde_json::from_str(&text).expect("json");
        (status, value)
    }

    /// POST naming a protocol revision, the way a client that has
    /// negotiated one does on every later request. Same absent-is-
    /// normal default as [`http_post_with_origin`].
    async fn http_post_with_protocol(
        addr: SocketAddr,
        body: &str,
        version: Option<&str>,
    ) -> (StatusCode, Value) {
        let (status, text) = http_post_raw(addr, body, None, version).await;
        let value: Value = serde_json::from_str(&text).expect("json");
        (status, value)
    }

    /// The same POST without the JSON parse, for the responses that carry
    /// no body at all — [`http_post_with_origin`] would panic on those.
    async fn http_post_raw(
        addr: SocketAddr,
        body: &str,
        origin: Option<&str>,
        version: Option<&str>,
    ) -> (StatusCode, String) {
        let url = format!("http://{}/mcp", addr);
        let client = reqwest::Client::new();
        let mut req = client
            .post(&url)
            .header("Content-Type", "application/json")
            .body(body.to_string());
        if let Some(origin) = origin {
            req = req.header("Origin", origin);
        }
        if let Some(version) = version {
            // Sent in the spec's canonical casing, not the lowercase
            // constant, so the case-insensitive lookup is exercised.
            req = req.header("MCP-Protocol-Version", version);
        }
        let resp = req.send().await.expect("post");
        let status = resp.status();
        let text = resp.text().await.expect("body");
        (status, text)
    }

    async fn initialize_peer_session(client: &reqwest::Client, url: &str) -> (String, Value) {
        let response = client
            .post(url)
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {}
            }))
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        let session_id = response
            .headers()
            .get(SESSION_HEADER)
            .and_then(|value| value.to_str().ok())
            .expect("initialize minted MCP-Session-Id")
            .to_string();
        let body = response.json().await.unwrap();
        (session_id, body)
    }

    async fn call_peer_tool(
        client: &reqwest::Client,
        url: &str,
        session_id: &str,
        name: &str,
        arguments: Value,
    ) -> Value {
        client
            .post(url)
            .header(SESSION_HEADER, session_id)
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/call",
                "params": { "name": name, "arguments": arguments }
            }))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn concurrent_mcp_sessions_send_as_distinct_principals() {
        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(crate::ServerState::standalone(temp.path().into()));
        state
            .attached_agents
            .lock()
            .await
            .insert("milo".into(), "detached-test-client".into());
        let mut server = McpServer::new();
        crate::peers::register_peer_tools(&mut server, state.clone()).unwrap();
        let (addr, task) = start_mcp_with_peer_sessions(
            Arc::new(server),
            "127.0.0.1:0".parse().unwrap(),
            state.clone(),
        )
        .await
        .unwrap();
        let client = reqwest::Client::new();
        let url = format!("http://{addr}/mcp");

        let (first, second) = tokio::join!(
            initialize_peer_session(&client, &url),
            initialize_peer_session(&client, &url)
        );
        assert_ne!(first.0, second.0, "each initialize mints a session");
        let (sent_first, sent_second) = tokio::join!(
            call_peer_tool(
                &client,
                &url,
                &first.0,
                "peer_message",
                json!({"to":"milo", "body":"same body"})
            ),
            call_peer_tool(
                &client,
                &url,
                &second.0,
                "peer_message",
                json!({"to":"milo", "body":"same body"})
            )
        );
        assert_eq!(sent_first["result"]["isError"], true, "{sent_first}");
        assert_eq!(sent_second["result"]["isError"], true, "{sent_second}");

        let audit = std::fs::read_to_string(&state.peer_audit_journal).unwrap();
        let senders: std::collections::HashSet<String> = audit
            .lines()
            .map(|line| {
                serde_json::from_str::<Value>(line).unwrap()["from"]
                    .as_str()
                    .unwrap()
                    .to_string()
            })
            .collect();
        assert_eq!(senders.len(), 2, "audit rows: {audit}");
        assert!(senders.iter().all(|sender| sender.starts_with("mcp:")));
        assert!(!senders.contains("mcp:external-cli"));
        task.abort();
    }

    #[tokio::test]
    async fn batch_session_is_send_only_and_delete_ends_a_live_session() {
        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(crate::ServerState::standalone(temp.path().into()));
        let mut server = McpServer::new();
        crate::peers::register_peer_tools(&mut server, state.clone()).unwrap();
        let (addr, task) = start_mcp_with_peer_sessions(
            Arc::new(server),
            "127.0.0.1:0".parse().unwrap(),
            state.clone(),
        )
        .await
        .unwrap();
        let client = reqwest::Client::new();
        let base = format!("http://{addr}/mcp");

        let (batch_id, _) =
            initialize_peer_session(&client, &format!("{base}?{BATCH_MODE_QUERY}")).await;
        let inbox = call_peer_tool(&client, &base, &batch_id, "peer_inbox", json!({})).await;
        assert_eq!(inbox["result"]["isError"], true, "{inbox}");
        assert!(inbox["result"]["content"][0]["text"]
            .as_str()
            .is_some_and(|text| text.contains("send-only")));
        assert!(crate::peers::snapshot_mcp_sessions(&state).await.is_empty());

        let (live_id, _) = initialize_peer_session(&client, &base).await;
        assert_eq!(crate::peers::snapshot_mcp_sessions(&state).await.len(), 1);
        let deleted = client
            .delete(&base)
            .header(SESSION_HEADER, &live_id)
            .send()
            .await
            .unwrap();
        assert_eq!(deleted.status(), StatusCode::NO_CONTENT);
        assert!(crate::peers::snapshot_mcp_sessions(&state).await.is_empty());
        task.abort();
    }

    #[tokio::test]
    async fn health_endpoint_returns_ok() {
        let (addr, _task) = boot_test_server().await;
        // Tiny sleep so the listener is ready before the client
        // connects on slow CI runners — axum::serve needs a yield
        // before accepting.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp/health", addr);
        let resp = reqwest::get(&url).await.expect("get");
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Value = resp.json().await.expect("json");
        assert_eq!(body["status"], "ok");
        assert_eq!(body["protocol_version"], car_mcp::PROTOCOL_VERSION);
    }

    #[tokio::test]
    async fn initialize_round_trips_over_http() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#;
        let (status, body) = http_post(addr, req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body["jsonrpc"], "2.0");
        assert_eq!(body["id"], 1);
        assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);
    }

    #[tokio::test]
    async fn initialize_negotiates_over_http() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = format!(
            r#"{{"jsonrpc":"2.0","id":1,"method":"initialize","params":{{"protocolVersion":"{}","capabilities":{{}},"clientInfo":{{"name":"c","version":"0"}}}}}}"#,
            car_mcp::PROTOCOL_VERSION
        );
        let (status, body) = http_post(addr, &req).await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);

        // An unsupported request is still a result, not an error: the
        // server names its own latest and the client decides.
        let req = r#"{"jsonrpc":"2.0","id":2,"method":"initialize","params":{"protocolVersion":"1999-01-01"}}"#;
        let (status, body) = http_post(addr, req).await;
        assert_eq!(status, StatusCode::OK);
        assert!(body.get("error").is_none(), "{body}");
        assert_eq!(body["result"]["protocolVersion"], car_mcp::PROTOCOL_VERSION);
    }

    /// A notification has no JSON-RPC reply, so the Streamable HTTP transport
    /// answers `202 Accepted` with nothing in the body — not a `200` carrying
    /// an invented `{}`.
    #[tokio::test]
    async fn notification_over_http_returns_202_and_no_body() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#;
        let (status, body) = http_post_raw(addr, req, None, None).await;
        assert_eq!(status, StatusCode::ACCEPTED);
        assert!(body.is_empty(), "expected an empty body, got {body:?}");
    }

    #[tokio::test]
    async fn tools_list_round_trips_over_http() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":2,"method":"tools/list"}"#;
        let (status, body) = http_post(addr, req).await;
        assert_eq!(status, StatusCode::OK);
        let tools = body["result"]["tools"].as_array().expect("tools array");
        // Two independent properties, both worth keeping (this test previously
        // pinned a hardcoded `6`, which silently rotted when the proactive-memory
        // tools took the real count to 12):
        //
        // 1. The transport surfaces exactly the tools the server DEFINES. Asserted
        //    against the schema source, not a magic number, so it tracks the tool
        //    set as it grows instead of going stale again.
        assert_eq!(
            tools.len(),
            car_mcp::cached_tool_schemas().len(),
            "HTTP tools/list must surface every defined tool, and no others"
        );
        // 2. Every tool name is DISTINCT — a duplicate would be a registration bug
        //    that a count alone can't see.
        let names: std::collections::HashSet<&str> =
            tools.iter().filter_map(|t| t["name"].as_str()).collect();
        assert_eq!(
            names.len(),
            tools.len(),
            "tool names must be unique: {tools:?}"
        );
    }

    /// A tool the daemon can serve and the stdio binary cannot. Trivial here,
    /// but the dependency direction is the point: `car-server-core` depends on
    /// `car-mcp`, so this is where a tool needing the daemon's live `Runtime`
    /// (the assistant, the scheduler, external agents — car#972 §6) gets
    /// written, without `car-mcp` growing a dependency on the daemon.
    struct ProbeTool;

    #[async_trait::async_trait]
    impl car_mcp::ToolHandler for ProbeTool {
        async fn call(&self, _args: Value) -> Result<String, car_mcp::ToolError> {
            Ok("served by the daemon".to_string())
        }
    }

    /// The per-`Server` tool surface reaches the wire, not just `Server::handle`.
    #[tokio::test]
    async fn a_registered_tool_is_reachable_over_http() {
        let mut server = McpServer::new();
        server
            .register_tool(
                json!({
                    "name": "daemon_probe",
                    "description": "a daemon-only tool",
                    "inputSchema": { "type": "object", "properties": {} },
                    "annotations": {
                        "readOnlyHint": true,
                        "destructiveHint": false,
                        "idempotentHint": true,
                        "openWorldHint": false,
                    },
                }),
                Arc::new(ProbeTool),
            )
            .expect("registers");
        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let (addr, _task) = start_mcp(Arc::new(server), addr).await.expect("start_mcp");
        tokio::time::sleep(Duration::from_millis(50)).await;

        let (status, body) =
            http_post(addr, r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#).await;
        assert_eq!(status, StatusCode::OK);
        let tools = body["result"]["tools"].as_array().expect("tools array");
        assert_eq!(tools.len(), car_mcp::cached_tool_schemas().len() + 1);
        assert!(tools
            .iter()
            .any(|t| t["name"].as_str() == Some("daemon_probe")));

        let (status, body) = http_post(
            addr,
            r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"daemon_probe","arguments":{}}}"#,
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body["result"]["isError"], false);
        assert_eq!(body["result"]["content"][0]["text"], "served by the daemon");
    }

    #[tokio::test]
    async fn malformed_json_returns_parse_error() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let (status, body) = http_post(addr, "{not valid").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body["error"]["code"], -32700);
    }

    #[tokio::test]
    async fn shared_memgine_lets_facts_persist_across_requests() {
        // Build a server with a known memgine, ingest via MCP HTTP,
        // then query via MCP HTTP — both calls hit the same engine.
        let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
            None,
        )));
        let server = Arc::new(McpServer::with_memgine(memgine));
        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let (addr, _task) = start_mcp(server, addr).await.expect("start");
        tokio::time::sleep(Duration::from_millis(50)).await;

        let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"daemon","body":"shared engine works"}}}"#;
        let (_, _) = http_post(addr, add).await;

        let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"daemon","k":5}}}"#;
        let (_, body) = http_post(addr, query).await;
        let text = body["result"]["content"][0]["text"].as_str().expect("text");
        assert!(
            text.contains("daemon"),
            "expected query to find ingested fact: {text}"
        );
    }

    /// car#972 Track B: with the push channel declined, `GET /mcp` is the
    /// spec's "no SSE stream offered" branch — `405`, with `Allow: POST, DELETE`.
    /// Protocol-session identity does not turn GET into a response stream.
    #[tokio::test]
    async fn get_mcp_returns_405_method_not_allowed() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp", addr);
        let resp = reqwest::Client::new()
            .get(&url)
            .header("mcp-session-id", "hopeful-subscriber")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
        assert_eq!(
            resp.headers()
                .get(axum::http::header::ALLOW)
                .and_then(|v| v.to_str().ok()),
            Some("POST, DELETE"),
            "a 405 names what IS allowed"
        );
        let body: Value = resp.json().await.expect("json body");
        assert!(
            body["error"]
                .as_str()
                .is_some_and(|e| e.contains("no SSE stream")),
            "the body should say why: {body}"
        );
    }

    #[test]
    fn origin_allowed_unit() {
        // Loopback, across schemes, with and without a port. 127.0.0.53
        // is in 127.0.0.0/8 — systemd-resolved hands it out for
        // `localhost` — so the whole /8 has to pass, not just .1.
        for ok in [
            "http://localhost",
            "http://localhost:3000",
            "https://localhost:8443",
            "HTTP://LocalHost:3000",
            "http://127.0.0.1",
            "http://127.0.0.1:9102",
            "https://127.0.0.53",
            "http://[::1]",
            "http://[::1]:3000",
        ] {
            assert!(origin_allowed(ok), "should be allowed: {ok}");
        }

        // The two that catch a naive `contains("localhost")`, plus the
        // opaque origin and the malformed cases.
        for bad in [
            "https://evil.example",
            "http://sub.localhost.evil.example",
            "http://localhost.evil.com",
            "https://127.0.0.1.evil.com",
            "null",
            "",
            "file://",
            "file:///etc/passwd",
            "ws://localhost:3000",
            "localhost:3000",
            "http://[::1",
            "http://[::1]x",
            "http://10.0.0.5",
        ] {
            assert!(!origin_allowed(bad), "should be rejected: {bad}");
        }
    }

    #[tokio::test]
    async fn mcp_post_without_origin_is_allowed() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
        let (status, body) = http_post_with_origin(addr, req, None).await;
        assert_eq!(status, StatusCode::OK);
        assert!(
            body["result"]["tools"].as_array().is_some(),
            "absent Origin must be allowed — no real MCP client sends one: {body}"
        );
    }

    #[tokio::test]
    async fn mcp_post_with_loopback_origin_is_allowed() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
        for origin in [
            "http://localhost:3000",
            "http://127.0.0.1:9102",
            "http://[::1]",
        ] {
            let (status, body) = http_post_with_origin(addr, req, Some(origin)).await;
            assert_eq!(status, StatusCode::OK, "origin {origin} must be allowed");
            assert!(body["result"]["tools"].as_array().is_some(), "{body}");
        }
    }

    #[tokio::test]
    async fn mcp_post_with_foreign_origin_is_rejected() {
        // The point of the test is the SIDE EFFECT, not the status
        // code: a guard that 403s the response but still runs the tool
        // has not protected anything.
        let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
            None,
        )));
        let server = Arc::new(McpServer::with_memgine(memgine));
        let (addr, _task) = start_mcp(server, "127.0.0.1:0".parse().unwrap())
            .await
            .expect("start");
        tokio::time::sleep(Duration::from_millis(50)).await;

        let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"rebind","body":"attacker planted this"}}}"#;
        let (status, body) = http_post_with_origin(addr, add, Some("https://evil.example")).await;
        assert_eq!(status, StatusCode::FORBIDDEN);
        assert_eq!(body["error"], "origin not allowed");

        // Query with no Origin — allowed — and confirm the write never
        // landed.
        let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"rebind","k":5}}}"#;
        let (_, body) = http_post(addr, query).await;
        let text = body["result"]["content"][0]["text"]
            .as_str()
            .unwrap_or_default();
        assert!(
            !text.contains("attacker planted this"),
            "rejected origin must not reach the tool: {text}"
        );
    }

    #[tokio::test]
    async fn mcp_post_with_null_origin_is_rejected() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
        let (status, body) = http_post_with_origin(addr, req, Some("null")).await;
        assert_eq!(
            status,
            StatusCode::FORBIDDEN,
            "the opaque origin is not a safe origin"
        );
        assert_eq!(body["error"], "origin not allowed");
    }

    /// A foreign-origin GET is still refused work — there is no work: the
    /// route is a constant 405 whatever the Origin, so a rebinding page
    /// learns nothing and registers nothing.
    #[tokio::test]
    async fn mcp_get_with_foreign_origin_gets_the_same_405() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp", addr);
        let resp = reqwest::Client::new()
            .get(&url)
            .header("Origin", "https://evil.example")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
    }

    #[tokio::test]
    async fn health_stays_reachable_from_any_origin() {
        // Deliberate exemption — the documented uptime curl and any
        // browser probe must keep working. See `handle_health`.
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp/health", addr);
        let resp = reqwest::Client::new()
            .get(&url)
            .header("Origin", "https://evil.example")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::OK);
    }
    /// The header is absent on every request every shipped client
    /// sends today. Fails against a required-by-default guard, which
    /// is the regression this test exists to pin.
    #[tokio::test]
    async fn mcp_post_without_protocol_version_is_allowed() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
        let (status, body) = http_post_with_protocol(addr, req, None).await;
        assert_eq!(status, StatusCode::OK);
        assert!(
            body["result"]["tools"].as_array().is_some(),
            "an absent MCP-Protocol-Version must be allowed: {body}"
        );
    }

    #[tokio::test]
    async fn mcp_post_with_supported_protocol_version_is_allowed() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let req = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
        // Against the constant, never the literal, so this floats with
        // a future bump instead of pinning a stale revision.
        let (status, body) =
            http_post_with_protocol(addr, req, Some(car_mcp::PROTOCOL_VERSION)).await;
        assert_eq!(status, StatusCode::OK);
        assert!(body["result"]["tools"].as_array().is_some(), "{body}");
    }

    #[tokio::test]
    async fn mcp_post_with_unsupported_protocol_version_is_rejected() {
        // As with the foreign-origin test, the SIDE EFFECT is the
        // point: a guard that 400s the response but still runs the
        // tool has answered in a dialect the client never agreed to
        // *and* mutated the graph.
        let memgine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
            None,
        )));
        let server = Arc::new(McpServer::with_memgine(memgine));
        let (addr, _task) = start_mcp(server, "127.0.0.1:0".parse().unwrap())
            .await
            .expect("start");
        tokio::time::sleep(Duration::from_millis(50)).await;

        let add = r#"{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"memory_add_fact","arguments":{"subject":"dialect","body":"wrong-revision write"}}}"#;
        let (status, body) = http_post_with_protocol(addr, add, Some("1999-01-01")).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(body["error"], "unsupported MCP-Protocol-Version");
        assert_eq!(body["requested"], "1999-01-01");
        // The supported list is in the body because the client has no
        // other way to learn what to send.
        let supported = body["supported"].as_array().expect("supported list");
        assert!(
            supported
                .iter()
                .any(|v| v.as_str() == Some(car_mcp::PROTOCOL_VERSION)),
            "the 400 must name what we do speak: {body}"
        );

        // Query with no header — allowed — and confirm the write never
        // landed.
        let query = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_query","arguments":{"query":"dialect","k":5}}}"#;
        let (_, body) = http_post(addr, query).await;
        let text = body["result"]["content"][0]["text"]
            .as_str()
            .unwrap_or_default();
        assert!(
            !text.contains("wrong-revision write"),
            "a rejected protocol version must not reach the tool: {text}"
        );
    }

    /// Same for a stale protocol version on GET: the method is not allowed
    /// before any header is worth reading.
    #[tokio::test]
    async fn mcp_get_with_unsupported_protocol_version_gets_the_same_405() {
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp", addr);
        let resp = reqwest::Client::new()
            .get(&url)
            .header("MCP-Protocol-Version", "1999-01-01")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
    }

    #[tokio::test]
    async fn health_ignores_protocol_version_header() {
        // Deliberate exemption — the probe is how a client whose
        // version we reject finds out what we speak. See
        // `handle_health`.
        let (addr, _task) = boot_test_server().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp/health", addr);
        let resp = reqwest::Client::new()
            .get(&url)
            .header("MCP-Protocol-Version", "1999-01-01")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::OK);
        let body: Value = resp.json().await.expect("json");
        assert_eq!(body["status"], "ok");
        assert_eq!(body["protocol_version"], car_mcp::PROTOCOL_VERSION);
    }
}