basemind 0.23.0

//! The daemon's second MCP front-end: a stateless streamable-HTTP transport (rmcp 3.0, SEP-2567).
//!
//! The comms daemon already hosts the rmcp code-map router per-connection over its Unix-socket relay
//! (see [`Broker::serve_relay_connection`](super::daemon::Broker::serve_relay_connection)). This adds
//! a SECOND front-end to the SAME daemon: a loopback-TCP HTTP listener that routes each request to
//! the right workspace's shared read stack and serves it statelessly. One daemon, sole fjall writer,
//! now also the MCP HTTP server.
//!
//! ## URL scheme
//!
//! `http://127.0.0.1:<port>/mcp?root=<percent-encoded-abs-repo-path>&agent=<agent-id>`
//!
//! Each POST is self-describing: the router parses `root` + `agent` from the query, resolves the
//! workspace's [`SharedReadStack`](crate::mcp::SharedReadStack) (building it on first touch, shared
//! by `Arc` with any relay client on the same workspace), and constructs a fresh
//! [`StreamableHttpService`] bound to that stack for the one request. There is no session registry
//! to maintain — `NeverSessionManager` keeps the transport stateless per SEP-2567.
//!
//! ## Bind-as-lock + portfile
//!
//! The listener binds a fixed loopback address (default `127.0.0.1:51786`, override
//! [`HTTP_ADDR_ENV`]). Binding IS the lock: only one daemon holds it. The actually-bound `host:port`
//! is written to `<comms_dir>/http.addr` (mode 0600 on Unix) so tooling — `basemind daemon ensure`
//! and any launcher — can discover it without guessing the port.
//!
//! ## Idle lifecycle
//!
//! Stateless HTTP holds no persistent connection, so the daemon's UDS-link-based idle reaper would
//! see it as idle between requests and self-terminate mid-use. Every request is wrapped in a
//! [`Broker::begin_http_request`](super::daemon::Broker::begin_http_request) guard that both pins the
//! daemon while the request is in flight and stamps HTTP recency — see
//! [`Broker::is_idle_for`](super::daemon::Broker::is_idle_for).

use std::convert::Infallible;
use std::future::Future;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use bytes::Bytes;
use http::{Request, Response, StatusCode};
use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::service::Service;
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto;
use rmcp::transport::streamable_http_server::session::never::NeverSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;

use super::daemon::Broker;
use super::identity;
use super::ids::AgentId;
use crate::mcp::BasemindServer;

/// Env var overriding the bound loopback address. `host:port`, e.g. `127.0.0.1:0` to let the OS pick
/// a free port (the actual port is discoverable via the portfile). Empty/unset uses
/// [`DEFAULT_HTTP_ADDR`].
pub const HTTP_ADDR_ENV: &str = "BASEMIND_HTTP_ADDR";
/// Default fixed loopback address for the streamable-HTTP MCP transport.
const DEFAULT_HTTP_ADDR: &str = "127.0.0.1:51786";
/// Portfile name under `comms_dir` holding the actually-bound `host:port`, so tooling reads the
/// address instead of assuming the default port.
const PORTFILE_NAME: &str = "http.addr";
/// The one path the transport serves. Anything else is a 404.
const MCP_PATH: &str = "/mcp";
/// Owner-only mode for the portfile, matching the socket's `0600`.
#[cfg(unix)]
const PORTFILE_MODE: u32 = 0o600;
/// Poll cadence while [`await_http_ready`] waits for the listener to answer.
const HTTP_READY_POLL: Duration = Duration::from_millis(50);

/// Response body type produced by every path here — the shape [`StreamableHttpService::handle`]
/// returns, so the router and the service agree.
type HttpBody = BoxBody<Bytes, Infallible>;

/// The portfile path for a resolved comms dir.
pub fn portfile_path(comms_dir: &Path) -> PathBuf {
    comms_dir.join(PORTFILE_NAME)
}

/// Resolve the address to bind: [`HTTP_ADDR_ENV`] when set and non-blank, else [`DEFAULT_HTTP_ADDR`].
fn resolve_addr() -> Result<SocketAddr> {
    let raw = std::env::var(HTTP_ADDR_ENV)
        .ok()
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| DEFAULT_HTTP_ADDR.to_string());
    raw.parse::<SocketAddr>()
        .with_context(|| format!("parse {HTTP_ADDR_ENV}={raw:?} as a host:port socket address"))
}

/// Write the bound `host:port` to the portfile (mode 0600 on Unix).
fn write_portfile(comms_dir: &Path, addr: &SocketAddr) -> std::io::Result<()> {
    let path = portfile_path(comms_dir);
    std::fs::write(&path, addr.to_string())?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(PORTFILE_MODE))?;
    }
    Ok(())
}

/// Read the bound address back from the portfile, if present and parseable.
fn read_portfile(comms_dir: &Path) -> Option<SocketAddr> {
    std::fs::read_to_string(portfile_path(comms_dir))
        .ok()?
        .trim()
        .parse()
        .ok()
}

/// Parse `root` (required) and `agent` (optional) from the request query. A blank `agent` is treated
/// as absent so the router falls back to the workspace's stable default identity.
fn parse_target(query: Option<&str>) -> Option<(PathBuf, Option<String>)> {
    let query = query?;
    let mut root: Option<String> = None;
    let mut agent: Option<String> = None;
    for (key, value) in form_urlencoded::parse(query.as_bytes()) {
        match key.as_ref() {
            "root" => root = Some(value.into_owned()),
            "agent" => agent = Some(value.into_owned()),
            _ => {}
        }
    }
    let root = root.filter(|value| !value.trim().is_empty())?;
    let agent = agent.filter(|value| !value.trim().is_empty());
    Some((PathBuf::from(root), agent))
}

/// Build a plain-text HTTP response with a boxed body matching [`HttpBody`].
fn text_response(status: StatusCode, message: &str) -> Response<HttpBody> {
    Response::builder()
        .status(status)
        .header(http::header::CONTENT_TYPE, "text/plain; charset=utf-8")
        // ~keep A response built from constant parts is infallible; mirrors rmcp's own handlers.
        .body(Full::new(Bytes::from(message.to_string())).boxed())
        .expect("text response builds from constant parts")
}

/// The per-request router: shared across every accepted connection, cheap to clone.
struct HttpRouter {
    broker: Arc<Broker>,
    /// One stateless session manager shared by every request; it rejects all session ops so the
    /// transport is stateless per SEP-2567.
    session_manager: Arc<NeverSessionManager>,
    /// Parent cancellation token; each request's [`StreamableHttpServerConfig`] gets a child, so a
    /// daemon drain cancels in-flight handlers.
    cancel: CancellationToken,
}

impl HttpRouter {
    /// Route and serve one HTTP request. Always returns a response (errors are HTTP status codes,
    /// never a torn connection), mirroring the relay path's never-brick-a-client contract.
    async fn handle(&self, request: Request<Incoming>) -> Response<HttpBody> {
        // Pin the daemon (and stamp HTTP recency) for the whole request so the idle reaper cannot
        // tear the process down mid-request. See `Broker::begin_http_request`.
        let _activity = self.broker.begin_http_request();

        if request.uri().path() != MCP_PATH {
            return text_response(StatusCode::NOT_FOUND, "not found: this server serves POST /mcp only");
        }
        let Some((raw_root, agent)) = parse_target(request.uri().query()) else {
            return text_response(StatusCode::NOT_FOUND, "not found: missing ?root=<abs-repo-path>");
        };
        let Ok(root) = std::fs::canonicalize(&raw_root) else {
            return text_response(
                StatusCode::NOT_FOUND,
                "not found: root does not resolve to an existing path",
            );
        };

        let shared = match self.broker.host_read_stack(&root).await {
            Ok(shared) => shared,
            Err(error) => {
                tracing::warn!(%error, root = %root.display(), "http: hosting read stack failed");
                return text_response(StatusCode::NOT_FOUND, "not found: workspace could not be hosted");
            }
        };
        // Keep the workspace pinned against eviction for the request's lifetime.
        let _conn = match self.broker.begin_workspace_conn(&root) {
            Ok(guard) => guard,
            Err(error) => {
                tracing::warn!(%error, root = %root.display(), "http: workspace connection accounting failed");
                return text_response(StatusCode::NOT_FOUND, "not found: workspace could not be hosted");
            }
        };

        // Enforce the same `AgentId` invariant the UDS relay does (non-empty, <=128 bytes, charset ~keep
        // `[A-Za-z0-9._:-]`): the agent id becomes the memory-owner / comms-sender key, so a present ~keep
        // but malformed value must be rejected (400) rather than poisoning scoped keys. A missing ~keep
        // agent falls back to the workspace's stable default identity. ~keep
        let agent_id = match agent {
            Some(raw) => match AgentId::parse(raw) {
                Ok(id) => id.into_string(),
                Err(error) => {
                    return text_response(
                        StatusCode::BAD_REQUEST,
                        &format!("bad request: invalid ?agent= ({error})"),
                    );
                }
            },
            None => identity::cli_agent_id(&root).into_string(),
        };
        tracing::debug!(agent = %agent_id, root = %root.display(), "http: serving stateless mcp request");

        // The factory is called per request by rmcp; it has no access to the HTTP request, so the
        // root+agent binding is captured here and a fresh server is minted over the shared stack.
        let factory = move || Ok(BasemindServer::from_shared(shared.clone(), agent_id.clone()));
        let config = StreamableHttpServerConfig::default()
            .with_legacy_session_mode(false)
            .with_json_response(true)
            .with_cancellation_token(self.cancel.child_token());
        let service = StreamableHttpService::new(factory, self.session_manager.clone(), config);
        service.handle(request).await
    }
}

/// A [`hyper::service::Service`] wrapper over the shared router. A newtype is required because the
/// orphan rules forbid implementing the foreign `Service` trait directly for `Arc<HttpRouter>`.
#[derive(Clone)]
struct HyperSvc(Arc<HttpRouter>);

impl Service<Request<Incoming>> for HyperSvc {
    type Response = Response<HttpBody>;
    type Error = Infallible;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn call(&self, request: Request<Incoming>) -> Self::Future {
        let router = self.0.clone();
        Box::pin(async move { Ok(router.handle(request).await) })
    }
}

/// Serve the streamable-HTTP MCP transport until `shutdown` flips true (or its sender drops). Binds
/// the loopback listener (bind-as-lock), writes the portfile, and serves each connection with a
/// hyper auto (h1/h2) server over the shared router. A bind failure is returned to the caller, which
/// logs it and leaves the rest of the daemon (the UDS relay) running — HTTP is additive, never a
/// precondition for comms.
pub async fn serve_http(broker: Arc<Broker>, comms_dir: PathBuf, mut shutdown: watch::Receiver<bool>) -> Result<()> {
    let addr = resolve_addr()?;
    if !addr.ip().is_loopback() {
        tracing::warn!(
            %addr,
            "BASEMIND_HTTP_ADDR binds a NON-loopback address: the MCP transport becomes reachable \
             off-host. rmcp Host-header (DNS-rebinding) validation still applies, but for genuine \
             remote access you must configure allowed_hosts/origins for the real hostnames."
        );
    }
    let listener = match TcpListener::bind(addr).await {
        Ok(listener) => listener,
        Err(error) => {
            // A stale portfile from a prior crashed daemon would otherwise misdirect
            // `await_http_ready` / launchers / hard-coded manifests to whatever now holds the port.
            // Clear it before bailing so discovery fails cleanly instead of pointing at a foreign
            // process. (The portfile lives in this user's comms dir, so it can only be our own stale
            // write — the per-user singleton lock means we never race another basemind daemon here.)
            let path = portfile_path(&comms_dir);
            if let Err(remove_error) = std::fs::remove_file(&path)
                && remove_error.kind() != std::io::ErrorKind::NotFound
            {
                tracing::debug!(error = %remove_error, path = %path.display(),
                    "http: clearing stale portfile after bind failure");
            }
            return Err(anyhow::Error::new(error)).with_context(|| {
                format!("bind streamable-HTTP MCP listener on {addr} (is another process holding it?)")
            });
        }
    };
    let local = listener.local_addr().context("read the bound HTTP address")?;
    write_portfile(&comms_dir, &local).with_context(|| format!("write HTTP portfile under {}", comms_dir.display()))?;
    tracing::info!(addr = %local, "comms: streamable-HTTP MCP transport listening");

    let cancel = CancellationToken::new();
    let router = Arc::new(HttpRouter {
        broker,
        session_manager: Arc::new(NeverSessionManager::default()),
        cancel: cancel.clone(),
    });

    loop {
        tokio::select! {
            changed = shutdown.changed() => {
                // A drain sends `true`; a dropped sender ends the daemon too. Either stops accepts.
                if changed.is_err() || *shutdown.borrow() {
                    break;
                }
            }
            accepted = listener.accept() => {
                let (stream, _peer) = match accepted {
                    Ok(pair) => pair,
                    Err(error) => {
                        tracing::warn!(%error, "http: accept failed");
                        continue;
                    }
                };
                let io = TokioIo::new(stream);
                let service = HyperSvc(router.clone());
                let conn_cancel = cancel.clone();
                // Pin the daemon for the connection's whole lifetime, incremented HERE — before the
                // task is spawned — so there is no accept→handler gap the idle reaper can slip
                // through and tear down the process on the first request after idle (the exact
                // wake-from-idle case the pin exists to protect). `handle` also pins per request to
                // refresh HTTP recency; nested counting on the atomic is fine.
                let conn_activity = router.broker.begin_http_request();
                tokio::spawn(async move {
                    let _conn_activity = conn_activity;
                    let builder = auto::Builder::new(TokioExecutor::new());
                    tokio::select! {
                        result = builder.serve_connection(io, service) => {
                            if let Err(error) = result {
                                tracing::debug!(error = %error, "http: connection ended with error");
                            }
                        }
                        _ = conn_cancel.cancelled() => {}
                    }
                });
            }
        }
    }

    cancel.cancel();
    let path = portfile_path(&comms_dir);
    match std::fs::remove_file(&path) {
        Ok(()) => {}
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => tracing::debug!(%error, path = %path.display(), "http: removing portfile failed"),
    }
    tracing::info!("comms: streamable-HTTP MCP transport stopped");
    Ok(())
}

/// Poll until the transport answers a TCP connect (or `timeout` elapses), returning the bound
/// `host:port`. Used by `basemind daemon ensure` to confirm readiness before printing the base URL.
pub async fn await_http_ready(comms_dir: &Path, timeout: Duration) -> Result<String> {
    let deadline = std::time::Instant::now() + timeout;
    loop {
        if let Some(addr) = read_portfile(comms_dir)
            && tokio::net::TcpStream::connect(addr).await.is_ok()
        {
            return Ok(addr.to_string());
        }
        if std::time::Instant::now() >= deadline {
            anyhow::bail!("streamable-HTTP MCP transport did not become ready within {timeout:?}");
        }
        tokio::time::sleep(HTTP_READY_POLL).await;
    }
}

/// The MCP base URL for a bound `host:port`.
pub fn base_url(addr: &str) -> String {
    format!("http://{addr}{MCP_PATH}")
}

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

    #[test]
    fn parse_target_reads_root_and_agent() {
        let (root, agent) = parse_target(Some("root=%2Ftmp%2Fmy%20repo&agent=alice")).expect("parses");
        assert_eq!(root, PathBuf::from("/tmp/my repo"));
        assert_eq!(agent.as_deref(), Some("alice"));
    }

    #[test]
    fn parse_target_treats_blank_agent_as_absent() {
        let (root, agent) = parse_target(Some("root=%2Ftmp%2Fr&agent=%20%20")).expect("parses");
        assert_eq!(root, PathBuf::from("/tmp/r"));
        assert_eq!(agent, None);
    }

    #[test]
    fn parse_target_requires_root() {
        assert!(parse_target(Some("agent=alice")).is_none());
        assert!(parse_target(Some("root=")).is_none());
        assert!(parse_target(None).is_none());
    }

    #[test]
    fn resolve_addr_defaults_and_honors_env() {
        // Default when unset.
        // SAFETY: single-threaded test; no other thread reads the env concurrently.
        unsafe { std::env::remove_var(HTTP_ADDR_ENV) };
        assert_eq!(resolve_addr().expect("default parses").to_string(), DEFAULT_HTTP_ADDR);
        unsafe { std::env::set_var(HTTP_ADDR_ENV, "127.0.0.1:0") };
        assert_eq!(
            resolve_addr().expect("env parses"),
            "127.0.0.1:0".parse::<SocketAddr>().unwrap()
        );
        unsafe { std::env::remove_var(HTTP_ADDR_ENV) };
    }

    #[test]
    fn base_url_appends_mcp_path() {
        assert_eq!(base_url("127.0.0.1:51786"), "http://127.0.0.1:51786/mcp");
    }
}