majordomus_cli/http/

mcp.rs

//! MCP over HTTP at `/mcp`: the Streamable HTTP transport's request half. A client
//! `POST`s one JSON-RPC message (or a batch) and gets the response in the body; the
//! server answers `initialize` with an `Mcp-Session-Id` header, requires it on every later
//! request, and forgets the session on `DELETE` or after a silence longer than
//! [`SESSION_IDLE_TIMEOUT`]. There is no server-initiated stream (`GET` is 405): this
//! server sends no notifications, so a stream would carry nothing.
//!
//! Every session is one [`Server`] over the shared surface, seen from its own peer, so the
//! stdio bridges of other `majordomus mcp` processes and clients speaking this transport
//! directly are peers of one board, and `peers.announce` knows who spoke.

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};

use serde_json::Value;

use crate::mcp::{Server, Surface};
use crate::peers::{PeerId, Transport};

use super::router::{Request, Response};

/// The path of the endpoint.
pub const PATH: &str = "/mcp";

/// The session header, as the transport specifies it (compared case-insensitively).
pub const SESSION_HEADER: &str = "Mcp-Session-Id";

/// A session that has been silent this long is forgotten. A `majordomus mcp` bridge
/// pings every [`crate::mcp::bridge::HEARTBEAT`], well inside it; a client that speaks
/// the transport directly and falls silent longer re-initialises on the 404 it gets, as
/// the transport prescribes.
pub const SESSION_IDLE_TIMEOUT: Duration = Duration::from_secs(90);

struct Session {
    peer: PeerId,
    server: Mutex<Server>,
    last_seen: Mutex<Instant>,
}

/// The endpoint: the sessions it holds and the surface they share.
pub struct McpEndpoint {
    surface: Surface,
    version: &'static str,
    url: String,
    sessions: Mutex<BTreeMap<String, Arc<Session>>>,
}

impl std::fmt::Debug for McpEndpoint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("McpEndpoint")
            .field("url", &self.url)
            .field("sessions", &self.active())
            .finish()
    }
}

impl McpEndpoint {
    /// An endpoint over a context, announcing `version`, reachable at `url`.
    pub fn new(live: impl crate::live::IntoLive, version: &'static str, url: String) -> Self {
        McpEndpoint {
            surface: Surface::new(live),
            version,
            url,
            sessions: Mutex::new(BTreeMap::new()),
        }
    }

    /// The base URL of the server this endpoint belongs to.
    pub fn url(&self) -> &str {
        &self.url
    }

    /// How many sessions are open.
    pub fn active(&self) -> usize {
        self.sessions().len()
    }

    /// Answer one request to the endpoint.
    pub fn handle(&self, req: &Request) -> Response {
        match req.method.as_str() {
            "POST" => self.post(req),
            "DELETE" => self.delete(req),
            other => Response::error(
                405,
                "method_not_allowed",
                &format!("{other} {PATH} is not served: POST a JSON-RPC message, DELETE to end the session; this server opens no stream"),
            ),
        }
    }

    fn post(&self, req: &Request) -> Response {
        // a client names the protocol version it negotiated on every request after
        // `initialize`; one this server does not speak is refused rather than guessed at
        if let Some(version) = req.header("mcp-protocol-version") {
            if !crate::mcp::protocol::PROTOCOL_VERSIONS.contains(&version) {
                return Response::error(
                    400,
                    "unsupported_protocol_version",
                    &format!(
                        "MCP-Protocol-Version {version} is not one this server speaks ({})",
                        crate::mcp::protocol::PROTOCOL_VERSIONS.join(", ")
                    ),
                );
            }
        }
        let message: Value = match serde_json::from_slice(&req.body) {
            Ok(v) => v,
            Err(e) => {
                return Response::error(400, "invalid_json", &format!("the body is not JSON: {e}"))
            }
        };
        let (id, session) = match req.header(SESSION_HEADER) {
            Some(id) => match self.sessions().get(id) {
                Some(s) => (id.to_string(), Arc::clone(s)),
                None => {
                    return Response::error(
                        404,
                        "session_not_found",
                        "unknown or expired Mcp-Session-Id; send initialize again to open a new session",
                    )
                }
            },
            // A server whose lease was taken over serves the peers it already has and ends
            // with them. Opening a *new* session would be the opposite of that: the client
            // would get a peer board, a generation of the layer and an execution history
            // that no other client of this checkout can see, and nothing in its answers
            // would say so. It is told instead, and its launcher's election finds the
            // process that does hold the lease.
            None if is_initialize(&message) && crate::lease::was_lost() => {
                return Response::error(
                    409,
                    "lease_lost",
                    "this server no longer holds this checkout's lease: another process took it \
                     over and is the one to attach to. This one serves the sessions it already \
                     had and ends with them; start again through the launcher, whose election \
                     finds the current server.",
                )
            }
            None if is_initialize(&message) => self.open(),
            None => {
                return Response::error(
                    400,
                    "session_required",
                    &format!("the {SESSION_HEADER} header is required on every request after initialize"),
                )
            }
        };
        // A server that lost the checkout's lease answers its sessions for the busy grace and
        // then hands each one on: the session is closed and answered 404, the client's bridge
        // opens a new one, is refused 409 lease_lost, elects, and attaches to the current
        // server. Without this, the bridges' pings kept the superseded server alive forever,
        // and its clients on the code it was started from (I2127).
        if crate::lease::lost_for().is_some_and(|age| age >= crate::lease::timings().busy_grace) {
            // the session this request holds: one that a concurrent request handed on first is
            // detached again, which changes nothing
            self.sessions().remove(&id);
            self.surface.context().episodes.detach(&session.peer);
            self.surface.context().peers.detach(&session.peer);
            tracing::info!(session = %id, peer = %session.peer, "session handed on: this server lost the lease");
            return Response::error(
                404,
                "session_not_found",
                "this server lost the checkout's lease and has handed the session on; open a new session through the launcher, whose election finds the current server",
            );
        }
        *lock(&session.last_seen) = Instant::now();
        let response = {
            let mut server = lock(&session.server);
            server.set_remote(req.is_remote());
            server.handle(message)
        };
        // stamped again once answered: a request that ran longer than the reap interval must
        // not find its session reaped behind it (I2147)
        *lock(&session.last_seen) = Instant::now();
        let mut out = match response {
            None => Response::new(202, "application/json", String::new()),
            Some(reply) => Response::new(
                200,
                "application/json",
                serde_json::to_string(&reply).unwrap_or_default(),
            ),
        };
        out.headers.push((SESSION_HEADER.to_string(), id));
        out
    }

    fn open(&self) -> (String, Arc<Session>) {
        let peer = self.surface.context().peers.attach(Transport::Http);
        // the session id is the client's only credential to its session: 128 bits from the
        // operating system's random source, never a pid, a counter and a clock (I2147)
        let id = crate::mesh::link::fresh_token();
        let server = Server::new(self.surface.for_peer(peer.clone()), self.version)
            .with_endpoint(Some(self.url.clone()));
        let session = Arc::new(Session {
            peer: peer.clone(),
            server: Mutex::new(server),
            last_seen: Mutex::new(Instant::now()),
        });
        self.sessions().insert(id.clone(), Arc::clone(&session));
        tracing::info!(session = %id, peer = %peer, "http session opened");
        (id, session)
    }

    fn delete(&self, req: &Request) -> Response {
        let Some(id) = req.header(SESSION_HEADER) else {
            return Response::error(
                400,
                "session_required",
                &format!("DELETE {PATH} names the session to end in {SESSION_HEADER}"),
            );
        };
        match self.sessions().remove(id) {
            Some(s) => {
                // The episode is detached, never closed here: a client that ends its
                // transport session has not necessarily ended its work, and it may come
                // back inside the reattach grace to the very same episode. What closes one
                // is a deliberate `episodes.detach`, the reaper, or the server stopping.
                self.surface.context().episodes.detach(&s.peer);
                self.surface.context().peers.detach(&s.peer);
                tracing::info!(session = %id, peer = %s.peer, "http session closed by the client");
                Response::new(204, "application/json", String::new())
            }
            None => Response::error(404, "session_not_found", "no such session"),
        }
    }

    /// Forget every session silent longer than [`SESSION_IDLE_TIMEOUT`], and close every
    /// episode detached longer than [`crate::episodes::REATTACH_GRACE`]; the peers it
    /// returns are gone.
    pub fn reap(&self) -> Vec<PeerId> {
        self.reap_idle(SESSION_IDLE_TIMEOUT, crate::episodes::REATTACH_GRACE)
    }

    /// The same, with both bounds given.
    ///
    /// Two parameters and not one, because there are two clocks and a suite that could only
    /// move the first could never see the second fire. The episode grace is fifteen minutes
    /// and no test may wait it out, so a reaper wired to a constant would be a reaper
    /// nothing proves — the shape this repository keeps calling a half-verdict. Production
    /// passes the constants; `tests/shared_units.rs` passes zero and watches an episode close.
    pub fn reap_idle(&self, timeout: Duration, episode_grace: Duration) -> Vec<PeerId> {
        let mut gone = Vec::new();
        let mut sessions = self.sessions();
        sessions.retain(|id, s| {
            let idle = lock(&s.last_seen).elapsed();
            if idle > timeout {
                tracing::info!(session = %id, peer = %s.peer, idle_seconds = idle.as_secs(), "http session expired");
                gone.push(s.peer.clone());
                false
            } else {
                true
            }
        });
        drop(sessions);
        for p in &gone {
            self.surface.context().episodes.detach(p);
            self.surface.context().peers.detach(p);
        }
        // And the other reaper, over a different clock. A connection is forgotten after
        // ninety seconds of silence; the episode it was carrying waits out the reattach
        // grace for its client to come back, and is closed as interrupted only when it does
        // not. Two timeouts because they answer two questions: is this socket alive, and is
        // this work over.
        for episode in self.surface.context().episodes.reap(episode_grace) {
            tracing::info!(
                episode = %episode.external_id,
                "episode expired: no connection came back inside the reattach grace"
            );
        }
        gone
    }

    /// Forget every session: the server is stopping.
    pub fn close_all(&self) {
        let mut sessions = self.sessions();
        for (_, s) in std::mem::take(&mut *sessions) {
            self.surface.context().peers.detach(&s.peer);
        }
        drop(sessions);
        // An episode left open by a server that has gone is an episode nothing will ever
        // close, and the repository would carry it as open for ever. The server is the last
        // thing that can close them, so it does, before it stops.
        for episode in self.surface.context().episodes.close_all() {
            tracing::info!(episode = %episode.external_id, "episode closed: the shared server is stopping");
        }
    }

    fn sessions(&self) -> MutexGuard<'_, BTreeMap<String, Arc<Session>>> {
        lock(&self.sessions)
    }
}

fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
    m.lock().unwrap_or_else(|e| e.into_inner())
}

/// Is this message, or any message of this batch, an `initialize` request?
///
/// ```
/// use majordomus_cli::http::mcp::is_initialize;
/// use serde_json::json;
/// assert!(is_initialize(&json!({ "jsonrpc": "2.0", "id": 1, "method": "initialize" })));
/// assert!(is_initialize(&json!([{ "method": "ping" }, { "method": "initialize" }])));
/// assert!(!is_initialize(&json!({ "method": "ping" })));
/// ```
pub fn is_initialize(message: &Value) -> bool {
    match message {
        Value::Array(batch) => batch.iter().any(is_initialize),
        Value::Object(m) => m.get("method").and_then(Value::as_str) == Some("initialize"),
        _ => false,
    }
}