huskies/server/src/http/gateway/mcp.rs

//! MCP JSON-RPC POST/GET handlers and gateway tool dispatch.

use super::jsonrpc::{JsonRpcRequest, JsonRpcResponse, to_json_response};
use crate::service::gateway::{self, GatewayState};
use poem::handler;
use poem::http::StatusCode;
use poem::web::Data;
use poem::web::sse::{Event, SSE};
use poem::{Body, IntoResponse, Request, Response};
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

// ── MCP tool definitions ─────────────────────────────────────────────────────

/// Gateway-specific MCP tools exposed alongside the proxied tools.
const GATEWAY_TOOLS: &[&str] = &[
    "switch_project",
    "gateway_status",
    "gateway_health",
    "init_project",
    "adopt_project",
    "aggregate_pipeline_status",
    "agents.list",
    // Handled at the gateway so the Matrix bot's perm_rx listener is used
    // rather than the container's (which has no interactive session attached).
    "prompt_permission",
    // Binary self-update: gateway serves its own binary and triggers upgrade on sleds.
    "upgrade_sled",
];

/// Gateway tool definitions.
pub(crate) fn gateway_tool_definitions() -> Vec<Value> {
    vec![
        json!({
            "name": "switch_project",
            "description": "Switch the active project. All subsequent MCP tool calls will be proxied to this project's container.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "project": {
                        "type": "string",
                        "description": "Name of the project to switch to (must exist in projects.toml)"
                    }
                },
                "required": ["project"]
            }
        }),
        json!({
            "name": "gateway_status",
            "description": "Show pipeline status for the active project by proxying the get_pipeline_status tool call.",
            "inputSchema": {
                "type": "object",
                "properties": {}
            }
        }),
        json!({
            "name": "gateway_health",
            "description": "Health check aggregation across all registered projects. Returns the health status of every project container.",
            "inputSchema": {
                "type": "object",
                "properties": {}
            }
        }),
        json!({
            "name": "init_project",
            "description": "Initialize a new huskies project at the given path by scaffolding .huskies/ and related files — the same as running `huskies init <path>`. Prefer this tool over asking the user to run the CLI. If `name` and `url` are supplied the project is also registered in projects.toml so switch_project can reach it immediately.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Absolute filesystem path to the project directory to initialise. The directory is created if it does not exist."
                    },
                    "name": {
                        "type": "string",
                        "description": "Optional: short name to register the project under in projects.toml (e.g. 'my-app'). Requires `url`."
                    },
                    "url": {
                        "type": "string",
                        "description": "Optional: base URL of the huskies container that will serve this project (e.g. 'http://my-app:3001'). Required when `name` is given."
                    }
                },
                "required": ["path"]
            }
        }),
        json!({
            "name": "adopt_project",
            "description": "Wrap a Docker container around an existing host checkout — the same as `new project <name> --adopt <path>`. No git clone or git init is performed; the directory is bind-mounted at /workspace. Launches the appropriate stack-specific image, generates an SSH keypair, and registers the project in projects.toml. Returns the SSH connection command and detected stack.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "Short project name (letters, digits, hyphens, underscores). Must be unique across registered projects."
                    },
                    "path": {
                        "type": "string",
                        "description": "Absolute host filesystem path to the existing checkout to adopt. Must be an existing directory."
                    },
                    "stack": {
                        "type": "string",
                        "description": "Optional: override stack detection (e.g. 'rust', 'node', 'python'). Auto-detected from directory contents when omitted."
                    }
                },
                "required": ["name", "path"]
            }
        }),
        json!({
            "name": "aggregate_pipeline_status",
            "description": "Fetch pipeline status from ALL registered projects in parallel and return an aggregated report. For each project: stage counts (backlog/current/qa/merge/done) and a list of blocked or failing items with triage detail. Unreachable projects are included with an error state rather than failing the whole call.",
            "inputSchema": {
                "type": "object",
                "properties": {}
            }
        }),
        json!({
            "name": "agents.list",
            "description": "List all alive build agents currently registered with this gateway. Returns an array of agent objects with id, label, address, registered_at, last_seen, and assigned_project fields.",
            "inputSchema": {
                "type": "object",
                "properties": {}
            }
        }),
        json!({
            "name": "upgrade_sled",
            "description": "Trigger a binary self-update on a project sled. The sled downloads the new binary from `source_url` (defaults to this gateway's /api/huskies-binary endpoint), atomically replaces its own executable, drains CRDT persistence so no ops are lost, and re-execs. Without `project`, upgrades the active project.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "project": {
                        "type": "string",
                        "description": "Name of the project sled to upgrade. Defaults to the currently active project."
                    },
                    "source_url": {
                        "type": "string",
                        "description": "HTTP URL of the binary to install (e.g. 'http://gateway:3000/api/huskies-binary'). Defaults to this gateway's own binary endpoint."
                    }
                }
            }
        }),
    ]
}

// ── MCP POST handler ─────────────────────────────────────────────────────────

/// Main MCP POST handler for the gateway. Intercepts gateway-specific tools and
/// proxies everything else to the active project's container.
#[handler]
pub async fn gateway_mcp_post_handler(
    req: &Request,
    body: Body,
    state: Data<&Arc<GatewayState>>,
) -> Response {
    let content_type = req.header("content-type").unwrap_or("");
    if !content_type.is_empty() && !content_type.contains("application/json") {
        return to_json_response(JsonRpcResponse::error(
            None,
            -32700,
            "Unsupported Content-Type; expected application/json".into(),
        ));
    }

    let bytes = match body.into_bytes().await {
        Ok(b) => b,
        Err(_) => {
            return to_json_response(JsonRpcResponse::error(None, -32700, "Parse error".into()));
        }
    };

    let rpc: JsonRpcRequest = match serde_json::from_slice(&bytes) {
        Ok(r) => r,
        Err(_) => {
            return to_json_response(JsonRpcResponse::error(None, -32700, "Parse error".into()));
        }
    };

    if rpc.jsonrpc != "2.0" {
        return to_json_response(JsonRpcResponse::error(
            rpc.id,
            -32600,
            "Invalid JSON-RPC version".into(),
        ));
    }

    if rpc.id.is_none() || rpc.id.as_ref() == Some(&Value::Null) {
        if rpc.method.starts_with("notifications/") {
            return Response::builder()
                .status(StatusCode::ACCEPTED)
                .body(Body::empty());
        }
        return to_json_response(JsonRpcResponse::error(None, -32600, "Missing id".into()));
    }

    // SSE proxy: tools/call with Accept: text/event-stream + progressToken for
    // non-gateway tools is forwarded to the sled's SSE endpoint so progress
    // notifications flow through to the gateway client unchanged.
    if rpc.method == "tools/call" {
        let accepts_sse = req
            .header("accept")
            .map(|h| h.contains("text/event-stream"))
            .unwrap_or(false);
        let has_progress_token = rpc
            .params
            .get("_meta")
            .and_then(|m| m.get("progressToken"))
            .is_some();
        if accepts_sse && has_progress_token {
            let tool_name = rpc
                .params
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            if !GATEWAY_TOOLS.contains(&tool_name) {
                return proxy_and_respond_sse(&state, &bytes, rpc.id).await;
            }
        }
    }

    match rpc.method.as_str() {
        "initialize" => to_json_response(handle_initialize(rpc.id)),
        "tools/list" => match handle_tools_list(&state, rpc.id.clone()).await {
            Ok(resp) => to_json_response(resp),
            Err(e) => to_json_response(JsonRpcResponse::error(rpc.id, -32603, e)),
        },
        "pipeline.get" => to_json_response(handle_pipeline_get(&state, rpc.id).await),
        "tools/call" => {
            let tool_name = rpc
                .params
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("");

            if GATEWAY_TOOLS.contains(&tool_name) {
                to_json_response(
                    handle_gateway_tool(tool_name, &rpc.params, &state, rpc.id.clone()).await,
                )
            } else {
                proxy_and_respond(&state, &bytes, rpc.id).await
            }
        }
        _ => proxy_and_respond(&state, &bytes, rpc.id).await,
    }
}

/// Proxy a request to the active project and format the response.
///
/// Prefers the live sled-uplink WebSocket when one is attached (story 899
/// AC 2); falls back to the legacy HTTP proxy otherwise.
async fn proxy_and_respond(state: &GatewayState, bytes: &[u8], id: Option<Value>) -> Response {
    match state.proxy_active_mcp(bytes).await {
        Ok(resp_body) => Response::builder()
            .status(StatusCode::OK)
            .header("Content-Type", "application/json")
            .body(Body::from(resp_body)),
        Err(e) => to_json_response(JsonRpcResponse::error(
            id,
            -32603,
            format!("proxy error: {e}"),
        )),
    }
}

/// Stream an MCP tool call to the active sled as SSE, re-emitting each `data:`
/// event from the sled to the originating gateway client without buffering.
///
/// On sled disconnect mid-stream a JSON-RPC error event is emitted so the
/// client does not hang forever.
#[allow(clippy::string_slice)] // pos from buf.find('\n'); '\n' is ASCII so pos and pos+1 are valid boundaries
async fn proxy_and_respond_sse(state: &GatewayState, bytes: &[u8], id: Option<Value>) -> Response {
    let url = match state.active_url().await {
        Ok(u) => u,
        Err(e) => return sse_error_response(id, -32603, e.to_string()),
    };

    let resp = match gateway::io::proxy_mcp_call_sse(&state.client, &url, bytes).await {
        Ok(r) => r,
        Err(e) => return sse_error_response(id, -32603, format!("proxy error: {e}")),
    };

    let id_for_error = id;
    let stream = async_stream::stream! {
        use futures::StreamExt as _;
        let mut buf = String::new();
        let byte_stream = resp.bytes_stream();
        tokio::pin!(byte_stream);

        while let Some(chunk) = byte_stream.next().await {
            match chunk {
                Ok(bytes) => {
                    if let Ok(text) = std::str::from_utf8(&bytes) {
                        buf.push_str(text);
                        // Emit a gateway SSE event for each complete `data:` line.
                        while let Some(pos) = buf.find('\n') {
                            let line = buf[..pos].trim_end_matches('\r').to_string();
                            buf = buf[pos + 1..].to_string();
                            if let Some(data) = line.strip_prefix("data: ") {
                                yield Event::message(data.to_string());
                            }
                        }
                    }
                }
                Err(e) => {
                    let err = JsonRpcResponse::error(
                        id_for_error.clone(),
                        -32603,
                        format!("upstream disconnected: {e}"),
                    );
                    let data = serde_json::to_string(&err).unwrap_or_default();
                    yield Event::message(data);
                    break;
                }
            }
        }
    };

    SSE::new(stream)
        .keep_alive(Duration::from_secs(15))
        .into_response()
}

/// Build a minimal SSE response containing a single JSON-RPC error event.
fn sse_error_response(id: Option<Value>, code: i64, msg: String) -> Response {
    let err = JsonRpcResponse::error(id, code, msg);
    let data = serde_json::to_string(&err).unwrap_or_default();
    let stream = async_stream::stream! {
        yield Event::message(data);
    };
    SSE::new(stream).into_response()
}

/// GET handler — method not allowed.
#[handler]
pub async fn gateway_mcp_get_handler() -> Response {
    Response::builder()
        .status(StatusCode::METHOD_NOT_ALLOWED)
        .body(Body::empty())
}

// ── Protocol handlers ────────────────────────────────────────────────────────

fn handle_initialize(id: Option<Value>) -> JsonRpcResponse {
    JsonRpcResponse::success(
        id,
        json!({
            "protocolVersion": "2025-03-26",
            "capabilities": { "tools": {} },
            "serverInfo": {
                "name": "huskies-gateway",
                "version": "1.0.0"
            }
        }),
    )
}

/// Fetch tools/list from the active project and merge in gateway tools.
///
/// Routes via the sled-uplink WS when one is attached (story 899 AC 2);
/// falls back to HTTP otherwise.
async fn handle_tools_list(
    state: &GatewayState,
    id: Option<Value>,
) -> Result<JsonRpcResponse, String> {
    let rpc_body = json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/list",
        "params": {}
    });
    let bytes = serde_json::to_vec(&rpc_body).map_err(|e| e.to_string())?;
    let resp_bytes = state.proxy_active_mcp(&bytes).await?;
    let resp_json: Value =
        serde_json::from_slice(&resp_bytes).map_err(|e| format!("invalid tools/list JSON: {e}"))?;

    let mut tools: Vec<Value> = resp_json
        .get("result")
        .and_then(|r| r.get("tools"))
        .and_then(|t| t.as_array())
        .cloned()
        .unwrap_or_default();

    let mut all_tools = gateway_tool_definitions();
    all_tools.append(&mut tools);

    Ok(JsonRpcResponse::success(id, json!({ "tools": all_tools })))
}

// ── Gateway tool dispatch ────────────────────────────────────────────────────

/// Dispatch a gateway-specific tool call.
async fn handle_gateway_tool(
    tool_name: &str,
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    match tool_name {
        "switch_project" => handle_switch_project_tool(params, state, id).await,
        "gateway_status" => handle_gateway_status_tool(state, id).await,
        "gateway_health" => handle_gateway_health_tool(state, id).await,
        "init_project" => handle_init_project_tool(params, state, id).await,
        "adopt_project" => handle_adopt_project_tool(params, state, id).await,
        "aggregate_pipeline_status" => handle_aggregate_pipeline_status_tool(state, id).await,
        "agents.list" => handle_agents_list_tool(id),
        "prompt_permission" => handle_prompt_permission_tool(params, state, id).await,
        "upgrade_sled" => handle_upgrade_sled_tool(params, state, id).await,
        _ => JsonRpcResponse::error(id, -32601, format!("Unknown gateway tool: {tool_name}")),
    }
}

async fn handle_switch_project_tool(
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    let project = params
        .get("arguments")
        .and_then(|a| a.get("project"))
        .or_else(|| params.get("project"))
        .and_then(|v| v.as_str())
        .unwrap_or("");

    match gateway::switch_project(state, project).await {
        Ok(url) => JsonRpcResponse::success(
            id,
            json!({
                "content": [{
                    "type": "text",
                    "text": format!("Switched to project '{project}' ({url})")
                }]
            }),
        ),
        Err(e) => JsonRpcResponse::error(id, -32602, e.to_string()),
    }
}

async fn handle_gateway_status_tool(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
    let active = state.active_project.read().await.clone();
    let url = match state.active_url().await {
        Ok(u) => u,
        Err(e) => return JsonRpcResponse::error(id.clone(), -32603, e.to_string()),
    };

    match gateway::io::fetch_pipeline_status_for_project(&state.client, &url).await {
        Ok(upstream) => {
            let pipeline = upstream.get("result").cloned().unwrap_or(json!(null));
            JsonRpcResponse::success(
                id,
                json!({
                    "content": [{
                        "type": "text",
                        "text": format!(
                            "Pipeline status for '{active}':\n{}",
                            serde_json::to_string_pretty(&pipeline).unwrap_or_default()
                        )
                    }]
                }),
            )
        }
        Err(e) => JsonRpcResponse::error(id, -32603, e),
    }
}

async fn handle_gateway_health_tool(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
    let mut results = BTreeMap::new();

    // Build the project list, preferring the WS-uplink heartbeat as the
    // source of truth for liveness (story 899 AC 3). HTTP polls are used
    // only as a fallback when no live sled is connected.
    let project_names: Vec<(String, Option<String>)> = state
        .projects
        .read()
        .await
        .iter()
        .map(|(n, e)| (n.clone(), e.url.clone()))
        .collect();
    let sled_conns = state.sled_connections.read().await;
    for (name, url_opt) in &project_names {
        let status = if let Some(conn) = sled_conns.get(name) {
            if conn.is_alive(crate::service::gateway::HEARTBEAT_MAX_AGE_MS) {
                "healthy (ws)".to_string()
            } else {
                "stale (ws heartbeat overdue)".to_string()
            }
        } else if let Some(url) = url_opt {
            match gateway::io::check_project_health(&state.client, url).await {
                Ok(true) => "healthy".to_string(),
                Ok(false) => "unhealthy".to_string(),
                Err(e) => e,
            }
        } else {
            "no uplink and no url configured".to_string()
        };
        results.insert(name.clone(), status);
    }
    drop(sled_conns);

    let active = state.active_project.read().await.clone();
    JsonRpcResponse::success(
        id,
        json!({
            "content": [{
                "type": "text",
                "text": format!(
                    "Health check (active: '{active}'):\n{}",
                    results.iter()
                        .map(|(name, status)| format!("  {name}: {status}"))
                        .collect::<Vec<_>>()
                        .join("\n")
                )
            }]
        }),
    )
}

async fn handle_init_project_tool(
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    let args = params.get("arguments").unwrap_or(params);

    let path_str = args.get("path").and_then(|v| v.as_str()).unwrap_or("");
    let name = args.get("name").and_then(|v| v.as_str());
    let url = args.get("url").and_then(|v| v.as_str());

    match gateway::init_project(state, path_str, name, url).await {
        Ok(registered_name) => {
            let next_steps = if let Some(ref n) = registered_name {
                format!(
                    "Project registered as '{n}' in projects.toml.\n\
                     Next steps:\n\
                     1. Start a huskies server at '{path_str}' \
                        (e.g. `huskies {path_str}` or via Docker).\n\
                     2. Call switch_project with name='{n}' to make it active.\n\
                     3. Call wizard_status to begin the setup wizard."
                )
            } else {
                format!(
                    "Next steps:\n\
                     1. Start a huskies server at '{path_str}' \
                        (e.g. `huskies {path_str}` or via Docker).\n\
                     2. Register the project: call init_project again with name and url \
                        parameters, or add it to projects.toml manually.\n\
                     3. Call switch_project and then wizard_status to begin the setup wizard.\n\n\
                     Note: wizard_* MCP tools require a running huskies server for the project."
                )
            };

            JsonRpcResponse::success(
                id,
                json!({
                    "content": [{
                        "type": "text",
                        "text": format!("Successfully initialised huskies project at '{path_str}'.\n\n{next_steps}")
                    }]
                }),
            )
        }
        Err(e) => {
            let code = match &e {
                gateway::Error::Config(_) => -32602,
                gateway::Error::DuplicateToken(_) => -32602,
                _ => -32603,
            };
            JsonRpcResponse::error(id, code, e.to_string())
        }
    }
}

/// Handle the `adopt_project` gateway tool.
///
/// Wraps a Docker container around an existing host checkout — the MCP
/// equivalent of the `new project <name> --adopt <path>` chat command.
/// Validates that `path` exists and is a directory before delegating to
/// `handle_new_project`, which performs stack detection, container launch,
/// SSH keypair generation, and project registration.
async fn handle_adopt_project_tool(
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    use crate::chat::transport::matrix::new_project::handle_new_project;

    let args = params.get("arguments").unwrap_or(params);
    let name = args
        .get("name")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim();
    let path_str = args
        .get("path")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim();
    let stack = args.get("stack").and_then(|v| v.as_str());

    if name.is_empty() {
        return JsonRpcResponse::error(id, -32602, "missing required parameter: name".into());
    }
    if path_str.is_empty() {
        return JsonRpcResponse::error(id, -32602, "missing required parameter: path".into());
    }

    let path = std::path::Path::new(path_str);
    if !path.exists() {
        return JsonRpcResponse::error(
            id,
            -32602,
            format!(
                "Adopt path `{path_str}` does not exist — specify the path to an existing checkout."
            ),
        );
    }
    if !path.is_dir() {
        return JsonRpcResponse::error(
            id,
            -32602,
            format!("Adopt path `{path_str}` is not a directory."),
        );
    }

    let result = handle_new_project(
        name,
        stack,
        None,
        None,
        None,
        Some(path_str),
        false,
        &state.projects,
        &state.config_dir,
    )
    .await;

    JsonRpcResponse::success(
        id,
        json!({
            "content": [{
                "type": "text",
                "text": result
            }]
        }),
    )
}

async fn handle_aggregate_pipeline_status_tool(
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    let project_urls: BTreeMap<String, String> = state
        .projects
        .read()
        .await
        .iter()
        .filter_map(|(name, entry)| entry.url.as_ref().map(|u| (name.clone(), u.clone())))
        .collect();

    let statuses =
        gateway::io::fetch_all_project_pipeline_statuses(&project_urls, &state.client).await;
    let active = state.active_project.read().await.clone();

    JsonRpcResponse::success(
        id,
        json!({
            "content": [{
                "type": "text",
                "text": format!(
                    "Aggregate pipeline status (active: '{active}'):\n{}",
                    serde_json::to_string_pretty(&statuses).unwrap_or_default()
                )
            }],
            "projects": statuses,
            "active": active,
        }),
    )
}

/// Handle the `prompt_permission` tool at the gateway level.
///
/// Mirrors `tool_prompt_permission` in `http/mcp/diagnostics/permission.rs` but
/// uses the gateway's `perm_tx`/`perm_rx` so requests reach the Matrix bot that
/// is listening on the gateway, not the proxied container (which has no
/// interactive session and would auto-deny immediately).
async fn handle_prompt_permission_tool(
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    use crate::http::context::PermissionDecision;
    use crate::http::context::PermissionForward;

    let args = params.get("arguments").unwrap_or(params);
    let tool_name = args
        .get("tool_name")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown")
        .to_string();
    let tool_input = args.get("input").cloned().unwrap_or(json!({}));

    // Auto-approve huskies MCP tools — mirrors the standard server's allowlist.
    if tool_name.starts_with("mcp__huskies__") {
        crate::slog!(
            "[gateway/permission] Auto-approved '{tool_name}' (matches mcp__huskies__* allowlist)"
        );
        let text = json!({"behavior": "allow", "updatedInput": tool_input}).to_string();
        return JsonRpcResponse::success(id, json!({"content": [{"type": "text", "text": text}]}));
    }

    // Auto-deny when no interactive session holds perm_rx (i.e. no Matrix bot
    // listener is running — try_lock succeeds when nobody else holds the lock).
    if state.perm_rx.try_lock().is_ok() {
        crate::slog!("[gateway/permission] Auto-denied '{tool_name}' (no interactive session)");
        let text = json!({
            "behavior": "deny",
            "message": format!("Permission denied for '{tool_name}'. No interactive session active.")
        })
        .to_string();
        return JsonRpcResponse::success(id, json!({"content": [{"type": "text", "text": text}]}));
    }

    let request_id = uuid::Uuid::new_v4().to_string();
    let (response_tx, response_rx) = tokio::sync::oneshot::channel();

    if state
        .perm_tx
        .send(PermissionForward {
            request_id,
            tool_name: tool_name.clone(),
            tool_input: tool_input.clone(),
            response_tx,
        })
        .is_err()
    {
        crate::slog!("[gateway/permission] Auto-denied '{tool_name}' (perm_tx send failed)");
        let text =
            json!({"behavior": "deny", "message": format!("Permission denied for '{tool_name}'.")})
                .to_string();
        return JsonRpcResponse::success(id, json!({"content": [{"type": "text", "text": text}]}));
    }

    let decision =
        match tokio::time::timeout(std::time::Duration::from_secs(300), response_rx).await {
            Ok(Ok(d)) => d,
            Ok(Err(_)) => {
                return JsonRpcResponse::error(
                    id,
                    -32603,
                    "Permission response channel closed unexpectedly".into(),
                );
            }
            Err(_) => {
                return JsonRpcResponse::error(
                    id,
                    -32603,
                    format!("Permission request for '{tool_name}' timed out after 5 minutes"),
                );
            }
        };

    let text = if matches!(
        decision,
        PermissionDecision::Approve | PermissionDecision::AlwaysAllow
    ) {
        json!({"behavior": "allow", "updatedInput": tool_input}).to_string()
    } else {
        crate::slog_warn!("[gateway/permission] User denied permission for '{tool_name}'");
        json!({"behavior": "deny", "message": format!("User denied permission for '{tool_name}'")})
            .to_string()
    };

    JsonRpcResponse::success(id, json!({"content": [{"type": "text", "text": text}]}))
}

/// Handle the `agents.list` gateway tool — returns all alive build agents from the CRDT.
fn handle_agents_list_tool(id: Option<Value>) -> JsonRpcResponse {
    let agents = gateway::list_agents();
    let agents_json = serde_json::to_value(&agents).unwrap_or(json!([]));
    JsonRpcResponse::success(
        id,
        json!({
            "content": [{
                "type": "text",
                "text": serde_json::to_string_pretty(&agents).unwrap_or_default()
            }],
            "agents": agents_json,
        }),
    )
}

/// Handle the `upgrade_sled` gateway tool.
///
/// Posts `{"source_url": "<url>"}` to the target sled's `/api/upgrade` endpoint,
/// which triggers the sled to download the new binary, drain CRDT persistence,
/// and re-exec.  Returns 202 text immediately — the sled connection will drop
/// shortly after as `exec()` replaces the process.
async fn handle_upgrade_sled_tool(
    params: &Value,
    state: &GatewayState,
    id: Option<Value>,
) -> JsonRpcResponse {
    let args = params.get("arguments").unwrap_or(params);

    // Resolve target project URL (explicit project arg or active project).
    let project_name = args.get("project").and_then(|v| v.as_str());
    let sled_url = if let Some(name) = project_name {
        let projects = state.projects.read().await;
        match projects.get(name).and_then(|e| e.url.clone()) {
            Some(u) => u,
            None => {
                return JsonRpcResponse::error(
                    id,
                    -32602,
                    format!("Project '{name}' not found or has no URL configured"),
                );
            }
        }
    } else {
        match state.active_url().await {
            Ok(u) => u,
            Err(e) => return JsonRpcResponse::error(id, -32603, e.to_string()),
        }
    };

    // Build the binary source URL: caller-supplied or this gateway's own endpoint.
    let source_url = args
        .get("source_url")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .unwrap_or_else(|| {
            // Default: the gateway serves its own binary at /api/huskies-binary.
            // Use the same host/port the gateway is bound to.
            std::env::var("HUSKIES_GATEWAY_BINARY_URL")
                .unwrap_or_else(|_| format!("http://gateway:{}/api/huskies-binary", state.port))
        });

    let upgrade_url = format!("{sled_url}/api/upgrade");
    let body = serde_json::json!({ "source_url": source_url });

    let active_name = project_name.map(|s| s.to_string()).unwrap_or_else(|| {
        state
            .active_project
            .try_read()
            .map(|g| g.clone())
            .unwrap_or_default()
    });

    match state.client.post(&upgrade_url).json(&body).send().await {
        Ok(resp) if resp.status().is_success() || resp.status().as_u16() == 202 => {
            JsonRpcResponse::success(
                id,
                json!({
                    "content": [{
                        "type": "text",
                        "text": format!(
                            "Upgrade triggered on '{active_name}'. The sled is downloading the new binary from {source_url} and will re-exec momentarily."
                        )
                    }]
                }),
            )
        }
        Ok(resp) => JsonRpcResponse::error(
            id,
            -32603,
            format!(
                "Sled returned HTTP {} for upgrade request to {upgrade_url}",
                resp.status()
            ),
        ),
        Err(e) => JsonRpcResponse::error(
            id,
            -32603,
            format!("Failed to send upgrade request to {upgrade_url}: {e}"),
        ),
    }
}

/// Handle the `pipeline.get` read-RPC — returns per-project item lists in the
/// shape expected by the gateway web UI:
/// `{ "active": "...", "projects": { "name": { "active": [...], "backlog_count": N } } }`.
async fn handle_pipeline_get(state: &GatewayState, id: Option<Value>) -> JsonRpcResponse {
    let project_urls: BTreeMap<String, String> = state
        .projects
        .read()
        .await
        .iter()
        .filter_map(|(n, e)| e.url.as_ref().map(|u| (n.clone(), u.clone())))
        .collect();

    let results = gateway::io::fetch_all_project_pipeline_items(&project_urls, &state.client).await;
    let active = state.active_project.read().await.clone();

    JsonRpcResponse::success(id, json!({ "active": active, "projects": results }))
}

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::service::gateway::config::{GatewayConfig, ProjectEntry};
    use std::collections::BTreeMap;
    use std::sync::Arc;

    fn make_test_state(config_dir: &std::path::Path) -> Arc<GatewayState> {
        let mut projects = BTreeMap::new();
        projects.insert(
            "test-project".to_string(),
            ProjectEntry::with_url("http://127.0.0.1:3001"),
        );
        let config = GatewayConfig {
            projects,
            sled_tokens: BTreeMap::new(),
        };
        Arc::new(GatewayState::new(config, config_dir.to_path_buf(), 3000).unwrap())
    }

    #[tokio::test]
    async fn adopt_project_tool_missing_name_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let state = make_test_state(dir.path());
        let params = json!({ "arguments": { "path": "/some/path" } });
        let resp = handle_adopt_project_tool(&params, &state, Some(json!(1))).await;
        assert!(resp.error.is_some(), "expected error for missing name");
        let msg = resp.error.unwrap().message;
        assert!(msg.contains("name"), "expected 'name' in error, got: {msg}");
    }

    #[tokio::test]
    async fn adopt_project_tool_missing_path_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let state = make_test_state(dir.path());
        let params = json!({ "arguments": { "name": "myapp" } });
        let resp = handle_adopt_project_tool(&params, &state, Some(json!(1))).await;
        assert!(resp.error.is_some(), "expected error for missing path");
        let msg = resp.error.unwrap().message;
        assert!(msg.contains("path"), "expected 'path' in error, got: {msg}");
    }

    #[tokio::test]
    async fn adopt_project_tool_nonexistent_path_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let state = make_test_state(dir.path());
        let params = json!({ "arguments": { "name": "myapp", "path": "/nonexistent/xyz/abc123" } });
        let resp = handle_adopt_project_tool(&params, &state, Some(json!(1))).await;
        assert!(resp.error.is_some(), "expected error for nonexistent path");
        let msg = resp.error.unwrap().message;
        assert!(
            msg.contains("does not exist"),
            "expected 'does not exist' in error, got: {msg}"
        );
    }

    #[tokio::test]
    async fn adopt_project_tool_file_path_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("not_a_dir.txt");
        std::fs::write(&file, "content").unwrap();
        let state = make_test_state(dir.path());
        let params = json!({ "arguments": { "name": "myapp", "path": file.to_str().unwrap() } });
        let resp = handle_adopt_project_tool(&params, &state, Some(json!(1))).await;
        assert!(resp.error.is_some(), "expected error for file path");
        let msg = resp.error.unwrap().message;
        assert!(
            msg.contains("not a directory"),
            "expected 'not a directory' in error, got: {msg}"
        );
    }

    /// The MCP entry point produces the same validation outcome as the chat-routed call.
    ///
    /// Both paths ultimately run the same checks: path-doesn't-exist and
    /// path-is-file are tested here to verify the MCP layer is consistent
    /// with `handle_new_project` in `new_project.rs`.
    #[tokio::test]
    async fn adopt_project_tool_matches_chat_routed_call() {
        use crate::chat::transport::matrix::new_project::handle_new_project;
        use tokio::sync::RwLock;

        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("a_file.txt");
        std::fs::write(&file, "not a dir").unwrap();
        let file_path = file.to_str().unwrap();

        // Chat-routed: handle_new_project returns a text string with the error.
        let store = Arc::new(RwLock::new(BTreeMap::new()));
        let chat_result = handle_new_project(
            "myapp",
            None,
            None,
            None,
            None,
            Some(file_path),
            false,
            &store,
            dir.path(),
        )
        .await;
        assert!(
            chat_result.contains("not a directory"),
            "chat path should report 'not a directory', got: {chat_result}"
        );

        // MCP-routed: handle_adopt_project_tool returns a JSON-RPC error.
        let state = make_test_state(dir.path());
        let params = json!({ "arguments": { "name": "myapp2", "path": file_path } });
        let mcp_resp = handle_adopt_project_tool(&params, &state, Some(json!(1))).await;
        assert!(mcp_resp.error.is_some(), "MCP path should return an error");
        let mcp_msg = mcp_resp.error.unwrap().message;
        assert!(
            mcp_msg.contains("not a directory"),
            "MCP path should report 'not a directory', got: {mcp_msg}"
        );
    }
}