car-server-core 0.55.0

//! Per-request MCP bridge into the real daemon dispatcher.
//!
//! Credentials come only from the current HTTP request, never from daemon
//! state, process environment, or disk. Each invocation authenticates a fresh
//! connection; the ordinary dispatch path owns authorization and cleanup.

use std::{sync::Arc, time::Duration};

use axum::http::HeaderMap;
use futures::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio_tungstenite::tungstenite::{Error as WsError, Message};

use crate::ServerState;

tokio::task_local! {
    static REQUEST_AUTH: Result<Value, String>;
}

/// Register reads only on the production daemon endpoint. Mutation exposure
/// remains an explicit embedder decision through car_mcp::DaemonToolOptions.
pub fn register_daemon_tools(
    server: &mut car_mcp::Server,
    state: Arc<ServerState>,
) -> Result<(), car_mcp::RegisterError> {
    car_mcp::register_daemon_tools(
        server,
        Arc::new(DispatcherClient { state }),
        &car_mcp::DaemonToolOptions::default(),
    )
}

pub(crate) async fn handle_request(
    server: &car_mcp::Server,
    request: car_mcp::Request,
    headers: &HeaderMap,
) -> Option<car_mcp::Response> {
    // An agent auth attaches/replaces the lifecycle connection. A short-lived
    // MCP request must never take over that routing claim or ignore identity.
    let auth = if headers.contains_key("x-car-agent-id") {
        Err("supervised-agent credentials are not supported by daemon MCP tools; use the agent's existing WebSocket connection".to_owned())
    } else {
        headers
            .get("authorization")
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.strip_prefix("Bearer "))
            .filter(|token| !token.is_empty())
            .map(|token| {
                let mut auth = json!({"token": token});
                if let Some(tenant) = headers.get("x-car-tenant-id").and_then(|v| v.to_str().ok()) {
                    auth["tenant_id"] = Value::String(tenant.to_owned());
                }
                auth
            })
            .ok_or_else(|| {
                "daemon MCP tools require the caller's Authorization: Bearer token".to_owned()
            })
    };
    // Scope the actual awaited tool future, not just construction of it.
    REQUEST_AUTH.scope(auth, server.handle(request)).await
}

struct DispatcherClient {
    state: Arc<ServerState>,
}

#[async_trait::async_trait]
impl car_mcp::DaemonClient for DispatcherClient {
    async fn invoke(&self, method: &str, params: Value) -> Result<Value, String> {
        let auth = REQUEST_AUTH.try_with(Clone::clone).unwrap_or_else(|_| {
            Err("daemon MCP tools require an HTTP request credential scope".to_owned())
        })?;
        invoke_authenticated(
            self.state.clone(),
            auth,
            method,
            params,
            Duration::from_secs(30),
        )
        .await
    }
}

async fn invoke_authenticated(
    state: Arc<ServerState>,
    auth: Value,
    method: &str,
    params: Value,
    deadline: Duration,
) -> Result<Value, String> {
    let (send, read) = futures::channel::mpsc::unbounded::<Message>();
    let (write, mut receive) = futures::channel::mpsc::unbounded::<Message>();
    // Dropping send on success, error, timeout, or cancellation closes input.
    // Never abort this task: run_dispatch's disconnect tail removes the session
    // and cancels its connection-owned handlers/subscriptions normally.
    let dispatch = tokio::spawn(async move {
        let sink = Box::pin(write.sink_map_err(|_| WsError::ConnectionClosed));
        let _ = crate::run_dispatch(read.map(Ok), sink, "mcp:request".into(), state).await;
    });
    let result = tokio::time::timeout(deadline, async {
        for (id, rpc, arguments) in [
            (1u64, "session.auth", auth),
            (
                2,
                "server.handshake",
                json!({"protocol_version": car_proto::PROTOCOL_VERSION}),
            ),
            (3, method, params),
        ] {
            send.unbounded_send(Message::Text(
                json!({
                    "jsonrpc":"2.0", "id":id, "method":rpc, "params":arguments,
                })
                .to_string()
                .into(),
            ))
            .map_err(|_| "daemon dispatcher closed".to_owned())?;
            let response = loop {
                let message = receive
                    .next()
                    .await
                    .ok_or_else(|| "daemon dispatcher closed".to_owned())?;
                let Message::Text(text) = message else {
                    continue;
                };
                let value: Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
                if value.get("id").and_then(Value::as_u64) == Some(id) {
                    break value;
                }
            };
            if let Some(error) = response.get("error") {
                return Err(error
                    .get("message")
                    .and_then(Value::as_str)
                    .unwrap_or("daemon RPC refused")
                    .to_owned());
            }
            if id == 3 {
                return response
                    .get("result")
                    .cloned()
                    .ok_or_else(|| "daemon RPC omitted result".to_owned());
            }
        }
        unreachable!("the final RPC returns its result")
    })
    .await
    .unwrap_or_else(|_| Err("daemon MCP request timed out".into()));
    drop(send);
    drop(receive);
    // Cleanup normally finishes immediately. If delayed, retain the running
    // task rather than aborting it or returning an authenticated connection to
    // a pool. Its input is already closed and cannot be reused by another call.
    let _ = tokio::time::timeout(Duration::from_secs(5), dispatch).await;
    result
}

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

    async fn post(
        address: std::net::SocketAddr,
        token: Option<&str>,
        method: &str,
        params: Value,
    ) -> Value {
        let mut request = reqwest::Client::new()
            .post(format!("http://{address}/mcp"))
            .json(&json!({"jsonrpc":"2.0", "id":1, "method":method, "params":params}));
        if let Some(token) = token {
            request = request.bearer_auth(token);
        }
        request.send().await.unwrap().json().await.unwrap()
    }

    #[tokio::test]
    async fn production_registration_dispatches_reads_with_per_request_authority() {
        let temp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(ServerState::standalone(temp.path().into()));
        state.install_auth_token("caller-a".into()).unwrap();
        let mut server = car_mcp::Server::new();
        register_daemon_tools(&mut server, state.clone()).unwrap();
        let (address, listener) =
            crate::mcp::start_mcp(Arc::new(server), "127.0.0.1:0".parse().unwrap())
                .await
                .unwrap();
        let listed = post(address, None, "tools/list", json!({})).await;
        let names: Vec<_> = listed["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .map(|tool| tool["name"].as_str().unwrap())
            .collect();
        for name in [
            "events_query",
            "events_stats",
            "events_cost_by_agent",
            "workflow_verify",
            "workflow_list_paused",
            "scheduler_list",
        ] {
            assert!(names.contains(&name), "missing {name}");
        }
        for name in [
            "workflow_run",
            "workflow_resume",
            "scheduler_schedule",
            "scheduler_unschedule",
        ] {
            assert!(
                !names.contains(&name),
                "mutation unexpectedly exposed: {name}"
            );
        }
        let arguments = json!({"name":"events_stats", "arguments":{}});
        let (allowed, invalid, missing) = tokio::join!(
            post(address, Some("caller-a"), "tools/call", arguments.clone()),
            post(address, Some("caller-b"), "tools/call", arguments.clone()),
            post(address, None, "tools/call", arguments),
        );
        assert_ne!(allowed["result"]["isError"], true, "{allowed}");
        assert!(
            allowed["result"]["structuredContent"].is_object(),
            "{allowed}"
        );
        assert_eq!(invalid["result"]["isError"], true, "{invalid}");
        assert_eq!(missing["result"]["isError"], true, "{missing}");
        assert!(
            state.sessions.lock().await.is_empty(),
            "per-call sessions must be removed"
        );
        listener.abort();
    }

    #[tokio::test]
    async fn opted_in_mutation_still_requires_real_daemon_authentication() {
        let temp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(ServerState::standalone(temp.path().into()));
        state.install_auth_token("caller-a".into()).unwrap();
        let mut server = car_mcp::Server::new();
        car_mcp::register_daemon_tools(
            &mut server,
            Arc::new(DispatcherClient {
                state: state.clone(),
            }),
            &car_mcp::DaemonToolOptions {
                enable_mutations: true,
            },
        )
        .unwrap();
        let (address, listener) =
            crate::mcp::start_mcp(Arc::new(server), "127.0.0.1:0".parse().unwrap())
                .await
                .unwrap();
        let result = post(
            address,
            Some("invalid"),
            "tools/call",
            json!({"name":"scheduler_schedule", "arguments":{}}),
        )
        .await;
        assert_eq!(result["result"]["isError"], true, "{result}");
        assert!(
            result["result"]["content"][0]["text"]
                .as_str()
                .unwrap()
                .contains("auth"),
            "{result}"
        );
        assert!(state.sessions.lock().await.is_empty());
        listener.abort();
    }

    #[tokio::test]
    async fn valid_agent_credentials_cannot_replace_live_routing_or_elevate_host_tokens() {
        use car_registry::supervisor::{AgentSpec, RestartPolicy, Supervisor};
        let temp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(ServerState::standalone(temp.path().into()));
        state.install_auth_token("generic-token".into()).unwrap();
        state.install_host_token("host-token".into()).unwrap();
        let supervisor = Arc::new(
            Supervisor::with_paths(
                temp.path().join("agents.json"),
                temp.path().join("agent-logs"),
            )
            .unwrap(),
        );
        supervisor
            .upsert(AgentSpec {
                id: "agent-a".into(),
                name: "Agent A".into(),
                command: std::env::current_exe()
                    .unwrap()
                    .to_string_lossy()
                    .into_owned(),
                args: vec![],
                cwd: None,
                env: Default::default(),
                restart: RestartPolicy::Never,
                max_restarts: 0,
                backoff_secs: 1,
                auto_start: false,
                token: "valid-agent-token".into(),
                method_allowlist: None,
                capabilities: vec![],
            })
            .await
            .unwrap();
        assert!(
            supervisor
                .validate_agent_token("agent-a", "valid-agent-token")
                .await
        );
        state
            .install_supervisor(supervisor)
            .map_err(|_| ())
            .unwrap();
        state
            .attached_agents
            .lock()
            .await
            .insert("agent-a".into(), "live-connection".into());
        let mut server = car_mcp::Server::new();
        register_daemon_tools(&mut server, state.clone()).unwrap();
        let (address, listener) =
            crate::mcp::start_mcp(Arc::new(server), "127.0.0.1:0".parse().unwrap())
                .await
                .unwrap();
        for token in ["valid-agent-token", "wrong-token", "generic-token"] {
            let result: Value = reqwest::Client::new().post(format!("http://{address}/mcp"))
                .bearer_auth(token).header("x-car-agent-id", "agent-a")
                .json(&json!({"jsonrpc":"2.0", "id":1, "method":"tools/call", "params":{"name":"events_stats", "arguments":{}}}))
                .send().await.unwrap().json().await.unwrap();
            assert_eq!(result["result"]["isError"], true, "{result}");
            assert!(
                result["result"]["content"][0]["text"]
                    .as_str()
                    .unwrap()
                    .contains("supervised-agent credentials are not supported"),
                "{result}"
            );
            assert!(state.sessions.lock().await.is_empty());
            assert_eq!(
                state
                    .attached_agents
                    .lock()
                    .await
                    .get("agent-a")
                    .map(String::as_str),
                Some("live-connection")
            );
        }
        let host = post(
            address,
            Some("host-token"),
            "tools/call",
            json!({"name":"events_stats", "arguments":{}}),
        )
        .await;
        assert_eq!(
            host["result"]["isError"], true,
            "host token must not become a generic token: {host}"
        );
        assert!(state.sessions.lock().await.is_empty());
        assert_eq!(
            state
                .attached_agents
                .lock()
                .await
                .get("agent-a")
                .map(String::as_str),
            Some("live-connection")
        );
        listener.abort();
    }

    #[tokio::test]
    async fn timed_out_and_cancelled_requests_leave_no_dispatcher_session() {
        let temp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(ServerState::standalone(temp.path().into()));
        state.install_auth_token("caller-a".into()).unwrap();
        // Stall session creation, so the real dispatcher cannot answer auth.
        let sessions = state.sessions.lock().await;
        let call = tokio::spawn(invoke_authenticated(
            state.clone(),
            json!({"token":"caller-a"}),
            "events.stats",
            json!({}),
            Duration::from_millis(10),
        ));
        tokio::time::sleep(Duration::from_millis(30)).await;
        drop(sessions);
        assert!(call.await.unwrap().unwrap_err().contains("timed out"));
        assert!(state.sessions.lock().await.is_empty());

        let sessions = state.sessions.lock().await;
        let call = tokio::spawn(invoke_authenticated(
            state.clone(),
            json!({"token":"caller-a"}),
            "events.stats",
            json!({}),
            Duration::from_secs(30),
        ));
        tokio::task::yield_now().await;
        call.abort(); // Cancel only the caller; the dispatcher must finish normally.
        assert!(call.await.unwrap_err().is_cancelled());
        drop(sessions);
        // Cancellation returns before the detached dispatcher finishes normal
        // teardown. Observe completion with a bound, not a scheduler-yield count.
        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if state.sessions.lock().await.is_empty() {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("cancelled request dispatcher must remove its session");
    }
}