car-server-core 0.8.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 (or `{}` for notifications). The HTTP
//!   layer never invents protocol semantics; everything routes
//!   through [`car_mcp::Server::handle`].
//! - **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.
//!
//! ## 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";

/// 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;

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

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,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    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"),
    )
}

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

async fn handle_mcp_post(
    State(state): State<McpState>,
    body: String,
) -> impl IntoResponse {
    // 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));
        }
    };

    // Dispatch through the same handler the stdio transport uses.
    // `None` means it was a notification — return an empty 200
    // (clients ignore the body for notifications).
    match state.server.handle(req).await {
        Some(resp) => match serde_json::to_value(&resp) {
            Ok(v) => (StatusCode::OK, Json(v)),
            Err(e) => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({
                    "jsonrpc": "2.0",
                    "id": Value::Null,
                    "error": {
                        "code": -32603,
                        "message": format!("response serialization failed: {e}"),
                    },
                })),
            ),
        },
        None => (StatusCode::OK, Json(json!({}))),
    }
}

#[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) {
        let url = format!("http://{}/mcp", addr);
        let client = reqwest::Client::new();
        let resp = client
            .post(&url)
            .header("Content-Type", "application/json")
            .body(body.to_string())
            .send()
            .await
            .expect("post");
        let status = resp.status();
        let value: Value = resp.json().await.expect("json");
        (status, value)
    }

    #[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 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");
        assert_eq!(tools.len(), 6);
    }

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