car-server-core 0.49.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 2024-11-05 over HTTP)
//!
//! - **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.
//! - **GET `/mcp/health`** — liveness probe. Returns
//!   `{"status":"ok","protocol_version":"2024-11-05"}`. Useful for
//!   hosts that want to confirm the endpoint is up without sending
//!   a JSON-RPC initialize.
//!
//! Server-initiated requests (the SSE side of the streamable
//! transport, used for tool-execution callbacks) land in MCP-3
//! alongside the bidirectional tool plumbing.
//!
//! ## 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 both `POST /mcp`
//! and `GET /mcp`:
//!
//! - **No header → allowed.** Absent cannot be an error here.
//!   Every client in the tree sends nothing today, and
//!   [`car_mcp::SUPPORTED_VERSIONS`] holds exactly one entry, so
//!   the version a silent client negotiated is necessarily the one
//!   we would have demanded. Requiring the header would break
//!   `curl`, Claude Desktop, Cursor, and Claude Code for no gain.
//! - **A value in [`car_mcp::SUPPORTED_VERSIONS`] → allowed.**
//!   `car-connectors/src/transport.rs:126` 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`] returns a [`JoinHandle`] for the axum task. The
//! daemon's main holds it for the 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},
    response::{
        sse::{Event, KeepAlive, Sse},
        IntoResponse, Json,
    },
    routing::{get, post},
    Router,
};
use futures_util::stream::Stream;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::convert::Infallible;
use std::time::Duration;
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;

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

/// Header MCP clients use to correlate POST requests with their
/// SSE stream session — per the 2025-03-26 spec.
const SESSION_HEADER: &str = "mcp-session-id";

/// Header a client uses to name the protocol revision it negotiated
/// on `initialize` — per the 2025-06-18 spec. Lowercase to match
/// [`SESSION_HEADER`]; `HeaderMap` lookup is case-insensitive, so a
/// client sending `MCP-Protocol-Version` still matches.
const PROTOCOL_HEADER: &str = "mcp-protocol-version";

/// SSE keep-alive cadence. Browsers and some intermediaries close
/// idle HTTP connections aggressively (Cloudflare's default is
/// ~100s); 30s is the conservative number Anthropic's own MCP
/// docs suggest.
const SSE_KEEPALIVE_SECS: u64 = 30;

/// 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())
}

/// One connected SSE client. The server owns the sender; the
/// client's GET handler holds the receiver and forwards events to
/// the wire. Public so embedders can introspect (the registry is
/// exposed via [`SessionMap`] which references this type).
pub struct McpSession {
    /// Outbound channel the server pushes notifications/requests
    /// onto. The connected client's GET handler drains it.
    tx: mpsc::Sender<String>,
}

/// Process-wide registry of connected SSE clients keyed by session
/// id. Cleanup happens when the GET handler drops its receiver
/// (closed channel → next send fails → entry is removed lazily on
/// next `push_to_session`, plus eagerly via the SSE stream's
/// `Drop`).
pub type SessionMap = Mutex<HashMap<String, McpSession>>;

#[derive(Clone)]
struct McpState {
    server: Arc<McpServer>,
    sessions: Arc<SessionMap>,
}

/// Start the MCP HTTP listener bound to `addr`. Returns `Ok` with
/// `(bound_addr, join_handle, sessions)` on success — the bound
/// address may differ from the requested one when port `0` is
/// supplied (the OS picks); the `sessions` handle gives the embedder
/// access to the SSE session registry so it can call
/// [`push_to_session`] to deliver server-initiated requests to a
/// specific connected client.
///
/// Returns `Err` synchronously when binding fails so the daemon's
/// startup path can log and decide whether to continue.
pub async fn start_mcp(
    server: Arc<McpServer>,
    addr: SocketAddr,
) -> Result<(SocketAddr, JoinHandle<()>, Arc<SessionMap>), 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 sessions: Arc<SessionMap> = Arc::new(Mutex::new(HashMap::new()));
    let state = McpState {
        server,
        sessions: sessions.clone(),
    };
    let app: Router = Router::new()
        .route("/mcp", post(handle_mcp_post).get(handle_mcp_get))
        .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, sessions))
}

/// 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,
    }))
}

/// Open a server-sent-events stream the daemon can push messages on.
///
/// Per MCP 2025-03-26: the client GETs `/mcp` to receive
/// server-initiated events (notifications and requests for
/// client-side tool execution). The client SHOULD include
/// `Mcp-Session-Id` to correlate; the server uses that id to
/// route subsequent server-to-client requests.
///
/// When no session id is supplied (e.g. the canonical health
/// curl `curl http://.../mcp`), we generate one and surface it as
/// the first SSE event so the client can echo it back on
/// follow-up POSTs.
async fn handle_mcp_get(
    State(state): State<McpState>,
    headers: HeaderMap,
) -> axum::response::Response {
    // Before the registry insert, so a rejected request leaves no
    // orphan `McpSession` behind.
    if let Err(resp) = check_origin(&headers) {
        return resp;
    }
    if let Err(resp) = check_protocol_version(&headers) {
        return resp;
    }
    let session_id = headers
        .get(SESSION_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string())
        .unwrap_or_else(uuid_v4_simple);
    let (tx, rx) = mpsc::channel::<String>(64);
    {
        let mut sessions = state.sessions.lock().await;
        sessions.insert(session_id.clone(), McpSession { tx });
    }
    tracing::debug!(%session_id, "MCP SSE stream opened");

    // First event: announce the session id so clients that didn't
    // supply one can pick it up. Subsequent events are JSON-RPC
    // notifications / requests pushed via `push_to_session`.
    let init_event = serde_json::to_string(&json!({
        "jsonrpc": "2.0",
        "method": "notifications/initialized",
        "params": { "session_id": session_id.clone() },
    }))
    .unwrap_or_else(|_| "{}".to_string());

    let stream =
        async_stream::stream_init_event(init_event, rx, state.sessions.clone(), session_id.clone());

    Sse::new(stream)
        .keep_alive(
            KeepAlive::new()
                .interval(Duration::from_secs(SSE_KEEPALIVE_SECS))
                .text("ping"),
        )
        .into_response()
}

/// Push a JSON-encoded message to one connected SSE client. Returns
/// `false` when the session isn't registered or its channel is
/// closed (typically because the client disconnected) — callers can
/// use that as a cleanup signal. This is the foundation primitive
/// MCP-3b will build on for client-side tool routing: when the
/// server needs to invoke a tool only the client owns, it
/// `push_to_session(...)` a JSON-RPC request and waits for the
/// matching POST response on `/mcp`.
pub async fn push_to_session(sessions: &SessionMap, session_id: &str, payload: &Value) -> bool {
    let json = match serde_json::to_string(payload) {
        Ok(s) => s,
        Err(_) => return false,
    };
    let guard = sessions.lock().await;
    let Some(session) = guard.get(session_id) else {
        return false;
    };
    session.tx.send(json).await.is_ok()
}

/// Generate a v4-shaped UUID without pulling the `uuid` crate
/// here; same shape as the rest of car-server-core uses, so SSE
/// session ids look uniform alongside the daemon's other ids.
fn uuid_v4_simple() -> String {
    uuid::Uuid::new_v4().to_string()
}

mod async_stream {
    use super::*;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    /// SSE stream that yields the init event first, then drains the
    /// per-session channel. On `Drop` it removes the session from
    /// the registry so disconnected clients don't accumulate.
    pub fn stream_init_event(
        init: String,
        rx: mpsc::Receiver<String>,
        sessions: Arc<SessionMap>,
        session_id: String,
    ) -> McpEventStream {
        McpEventStream {
            init: Some(init),
            rx,
            cleanup: Some(SessionCleanup {
                sessions,
                session_id,
            }),
        }
    }

    pub struct McpEventStream {
        init: Option<String>,
        rx: mpsc::Receiver<String>,
        cleanup: Option<SessionCleanup>,
    }

    /// Drop guard that pulls the session entry out of the registry
    /// when the SSE stream goes away. Without this, browsers that
    /// close the connection silently would leave dangling Sender
    /// halves around forever.
    struct SessionCleanup {
        sessions: Arc<SessionMap>,
        session_id: String,
    }

    impl Drop for McpEventStream {
        fn drop(&mut self) {
            // Take the cleanup record so the spawn below owns it.
            // Using a tokio task because the registry lock is async.
            if let Some(cleanup) = self.cleanup.take() {
                tokio::spawn(async move {
                    let mut guard = cleanup.sessions.lock().await;
                    guard.remove(&cleanup.session_id);
                    tracing::debug!(session_id = %cleanup.session_id, "MCP SSE stream closed");
                });
            }
        }
    }

    impl Stream for McpEventStream {
        type Item = Result<Event, Infallible>;

        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            // Init event first; subsequent polls drain the channel.
            if let Some(init) = self.init.take() {
                return Poll::Ready(Some(Ok(Event::default().data(init))));
            }
            match self.rx.poll_recv(cx) {
                Poll::Ready(Some(payload)) => Poll::Ready(Some(Ok(Event::default().data(payload)))),
                // Channel closed — end the stream so axum drops the
                // connection cleanly. Drop impl handles registry
                // cleanup.
                Poll::Ready(None) => Poll::Ready(None),
                Poll::Pending => Poll::Pending,
            }
        }
    }
}

// `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,
    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();
        }
    };

    // Dispatch through the same handler the stdio transport uses.
    // `None` means it was a notification, and the Streamable HTTP
    // transport spells that `202 Accepted` with no body — there is
    // no JSON-RPC reply to send, and a `200` carrying `{}` invents
    // one (car#972 §3).
    match state.server.handle(req).await {
        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(),
    }
}

#[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, _sessions) = start_mcp(server, addr).await.expect("start_mcp");
        (bound, task)
    }

    /// Variant that also returns the session registry handle so
    /// SSE tests can drive `push_to_session` directly.
    async fn boot_test_server_with_sessions() -> (SocketAddr, JoinHandle<()>, Arc<SessionMap>) {
        // Re-implementing start_mcp inline here so we can hold the
        // sessions Arc — keeps `start_mcp` itself focused on the
        // production startup shape (which delegates session
        // discovery to whatever embedder owns ServerState).
        let server = Arc::new(McpServer::new());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let bound = listener.local_addr().expect("local_addr");
        let sessions: Arc<SessionMap> = Arc::new(Mutex::new(HashMap::new()));
        let state = McpState {
            server,
            sessions: sessions.clone(),
        };
        let app = Router::new()
            .route("/mcp", post(handle_mcp_post).get(handle_mcp_get))
            .route("/mcp/health", get(handle_health))
            .with_state(state);
        let task = tokio::spawn(async move {
            let _ = axum::serve(listener, app).await;
        });
        (bound, task, sessions)
    }

    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)
    }

    #[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, _sessions) = 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, _sessions) = 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}"
        );
    }

    #[tokio::test]
    async fn sse_get_emits_init_event_and_registers_session() {
        let (addr, _task, sessions) = boot_test_server_with_sessions().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp", addr);
        let client = reqwest::Client::new();
        let resp = client
            .get(&url)
            .header("mcp-session-id", "test-session-1")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::OK);
        // Verify the session showed up in the registry.
        tokio::time::sleep(Duration::from_millis(50)).await;
        {
            let guard = sessions.lock().await;
            assert!(guard.contains_key("test-session-1"));
        }
        // Read the first SSE event — should be the
        // notifications/initialized payload.
        let mut stream = resp.bytes_stream();
        use futures_util::StreamExt;
        let chunk = tokio::time::timeout(Duration::from_secs(2), stream.next())
            .await
            .expect("timeout")
            .expect("chunk")
            .expect("bytes");
        let body = String::from_utf8_lossy(&chunk).to_string();
        assert!(body.contains("notifications/initialized"));
        assert!(body.contains("test-session-1"));
    }

    #[tokio::test]
    async fn push_to_session_delivers_payload_to_connected_client() {
        let (addr, _task, sessions) = boot_test_server_with_sessions().await;
        tokio::time::sleep(Duration::from_millis(50)).await;
        let url = format!("http://{}/mcp", addr);
        let client = reqwest::Client::new();
        let resp = client
            .get(&url)
            .header("mcp-session-id", "push-session")
            .send()
            .await
            .expect("get");
        // Drain the init event so subsequent reads see the pushed
        // payload.
        let mut stream = resp.bytes_stream();
        use futures_util::StreamExt;
        let _init = tokio::time::timeout(Duration::from_secs(2), stream.next())
            .await
            .expect("timeout")
            .expect("chunk")
            .expect("bytes");

        // Wait for the registry to register the session.
        for _ in 0..20 {
            let guard = sessions.lock().await;
            if guard.contains_key("push-session") {
                break;
            }
            drop(guard);
            tokio::time::sleep(Duration::from_millis(20)).await;
        }

        // Push a request from the server side.
        let payload = json!({
            "jsonrpc": "2.0",
            "id": 99,
            "method": "tools/call",
            "params": { "name": "host_owned_tool", "arguments": {} }
        });
        let delivered = push_to_session(&sessions, "push-session", &payload).await;
        assert!(delivered, "push must succeed for connected session");

        // Client should observe the pushed payload on the SSE stream.
        let chunk = tokio::time::timeout(Duration::from_secs(2), stream.next())
            .await
            .expect("timeout")
            .expect("chunk")
            .expect("bytes");
        let body = String::from_utf8_lossy(&chunk).to_string();
        assert!(body.contains("host_owned_tool"));
        assert!(body.contains("\"id\":99"));
    }

    #[tokio::test]
    async fn push_to_session_returns_false_for_unknown_session() {
        let sessions: Arc<SessionMap> = Arc::new(Mutex::new(HashMap::new()));
        let delivered = push_to_session(&sessions, "nobody", &json!({"x":1})).await;
        assert!(!delivered);
    }

    #[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, _sessions) = 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");
    }

    #[tokio::test]
    async fn mcp_get_with_foreign_origin_is_rejected() {
        let (addr, _task, sessions) = boot_test_server_with_sessions().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", "evil-session")
            .header("Origin", "https://evil.example")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
        // And no orphan session entry: the guard runs before the insert.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let guard = sessions.lock().await;
        assert!(
            !guard.contains_key("evil-session"),
            "a rejected GET must not register a session"
        );
    }

    #[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, _sessions) = 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}"
        );
    }

    #[tokio::test]
    async fn mcp_get_with_unsupported_protocol_version_is_rejected() {
        let (addr, _task, sessions) = boot_test_server_with_sessions().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", "stale-dialect-session")
            .header("MCP-Protocol-Version", "1999-01-01")
            .send()
            .await
            .expect("get");
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        // And no orphan session entry: the guard runs before the insert.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let guard = sessions.lock().await;
        assert!(
            !guard.contains_key("stale-dialect-session"),
            "a rejected GET must not register a session"
        );
    }

    #[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);
    }
}