supercode_harness/

mcp.rs

//! P5-2 (COMPOSABLE-HARNESS-DESIGN.md §2 module 15 `mcp.client`, D7 rows
//! 1-8): full [Model Context Protocol](https://modelcontextprotocol.io)
//! client support — stdio (P5-1 baseline, GROWN not rewritten), remote
//! HTTP/SSE transports, resources + templates, prompts-as-commands,
//! server instructions, and elicitation — plus the [`handle_request`] /
//! [`serve_stdio`] harness-as-MCP-server direction (module 16).
//!
//! **Transport model.** [`McpClient`] hides three wire shapes behind one
//! `request()`/`list_tools()`/`call_tool()`/... API:
//! - [`McpClient::connect`] — stdio (newline-delimited JSON-RPC over a
//!   spawned child process's stdin/stdout). Pre-existing (P5-1 baseline).
//! - [`McpClient::connect_http`] — a single POST per request ("Streamable
//!   HTTP", non-streaming case): the response body is either a bare
//!   `application/json` object or a `text/event-stream` body carrying the
//!   one response event. A `Mcp-Session-Id` response header, if the server
//!   sends one, is captured and replayed on every subsequent request.
//! - [`McpClient::connect_sse`] — the legacy (2024-11-05) HTTP+SSE
//!   transport: a persistent `GET` stream whose first event names the POST
//!   endpoint for client→server messages; a background reader task
//!   forwards every subsequent server→client frame into an in-process
//!   channel.
//!
//! **Server-initiated requests and notifications.** A real MCP session is
//! bidirectional: while a client request is in flight, the server may push
//! a notification (`resources/updated`, …) or even issue its OWN request
//! back to the client (`elicitation/create`). [`McpClient`]'s read loop
//! (`McpClient::handle_incoming_message`) recognizes all three shapes on
//! both the stdio and SSE transports (persistent, bidirectional
//! connections) and dispatches server-initiated requests to the installed
//! [`McpElicitationHandler`] — see that trait's doc comment for the
//! HEADLESS-DENY default and the tui-deferred interactive part. The HTTP
//! transport is a single non-streaming request/response cycle with no
//! return channel for a reply; a server-initiated request arriving on it
//! is a documented, tested, fail-CLOSED error (`Error::tool("mcp", ...)`),
//! never a silent drop or a hang — see `McpClient::http_roundtrip`'s doc
//! comment.
//!
//! **Security posture (this module's own scope; see also
//! `crate::configfile`'s project-sanitization for the config-file side).**
//! MCP OAuth tokens are credentials, the same trust class as
//! `Config::api_key` (§3.2 S13) — [`crate::mcp_oauth`] handles only the
//! wire PROTOCOL (device-code grant, refresh); persistence to disk with
//! trust-grade (owner-only, user/global-directory-only) permissions is a
//! CLI-layer concern (`crates/cli/src/userconfig.rs`'s
//! `save_mcp_oauth_tokens`/`load_mcp_oauth_tokens`, mirroring
//! `save_api_key`'s existing 0600-perms precedent) — this crate never
//! writes a token to disk itself. Remote connects honor an active
//! [`crate::tools::NetworkPolicy`] (module 12's SSRF/domain-allowlist
//! floor) via `crate::tools::check_network_policy`, the exact function
//! [`crate::tools::ToolContext::check_network`] itself calls — one
//! enforcement point, not a second parallel one.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::{mpsc, Mutex};

use crate::error::{Error, Result};
use crate::tools::{NetworkPolicy, Tool, ToolContext, ToolRegistry};
#[cfg(feature = "adapter-mcp")]
use crate::{
    FrontendAttachment, FrontendResponse, HarnessSessionService, SdkError, SdkOperation,
    SdkRequest, SdkRuntime, SdkService,
};

const PROTOCOL_VERSION: &str = "2025-06-18";

/// Default per-request timeout (connect handshake + every subsequent
/// `request()`) for the network transports — stdio has no analogous
/// "hung server" risk distinct from a hung read, so it is NOT subject to
/// this timeout (a misbehaving stdio child can still be killed by the
/// caller; `kill_on_drop` already covers process cleanup).
pub const DEFAULT_MCP_TIMEOUT: Duration = Duration::from_secs(30);

/// Hardening cap (Fable-5 review, memory-DoS-from-a-hostile-configured-
/// server finding): the maximum size of a single non-streaming HTTP
/// response body (`McpClient::http_roundtrip`) this client will buffer
/// before erroring out. 16 MiB is generous for real tool-call/initialize
/// responses (the actual payloads this transport carries) while bounding
/// how much memory a misbehaving or malicious configured MCP server can
/// force this process to allocate for one response.
pub const MCP_MAX_RESPONSE_BYTES: usize = 16 * 1024 * 1024;

/// Hardening cap (same review finding as [`MCP_MAX_RESPONSE_BYTES`]): the
/// maximum size a single un-terminated SSE frame may grow to inside
/// `SseLineAccumulator` before it's treated as malformed/hostile and the
/// connection is torn down, rather than the accumulator buffer growing
/// without bound while waiting forever for a blank-line terminator that
/// never arrives.
pub const MCP_MAX_SSE_FRAME_BYTES: usize = 16 * 1024 * 1024;

/// Hardening cap (same review finding): the maximum joined size of
/// `resources/read`'s concatenated text contents [`McpClient::read_resource`]
/// will return before erroring out instead of buffering an unbounded string.
pub const MCP_MAX_RESOURCE_BYTES: usize = 16 * 1024 * 1024;

/// Hardening cap (same review finding): the SSE background reader task's
/// outbound channel capacity — bounds how many unconsumed server-pushed
/// frames (responses + notifications) can queue up before the reader task
/// blocks on `send` (applying backpressure to the socket read, never
/// growing an unbounded queue) rather than being fed by an
/// `unbounded_channel`. Large enough that ordinary notification bursts
/// don't get throttled; a `request()` in flight (or the next one issued)
/// drains it, so a reader task paused on a full channel is not a deadlock
/// — see [`sse_reader_task`]'s doc comment.
const MCP_SSE_CHANNEL_CAPACITY: usize = 256;

// ============================================================================
// ---- transport plumbing ----------------------------------------------------
// ============================================================================

/// One connected transport's read/write mechanics. Kept private —
/// [`McpClient`] is the only thing that touches this; every public method
/// (`list_tools`, `call_tool`, `list_resources`, …) is transport-agnostic.
enum Conn {
    Stdio {
        #[allow(dead_code)] // kept alive for `kill_on_drop`
        child: tokio::process::Child,
        stdin: tokio::process::ChildStdin,
        stdout: BufReader<tokio::process::ChildStdout>,
    },
    Http {
        client: reqwest::Client,
        url: String,
        headers: HeaderMap,
        /// Captured from a `Mcp-Session-Id` response header, if the server
        /// sends one, and replayed on every subsequent request — some
        /// Streamable HTTP servers require it after the first exchange.
        session_id: Option<String>,
    },
    Sse {
        client: reqwest::Client,
        post_url: String,
        headers: HeaderMap,
        inbox: mpsc::Receiver<SseInboxMsg>,
        #[allow(dead_code)] // kept alive so the background reader isn't dropped
        reader: tokio::task::JoinHandle<()>,
    },
}

/// One item the SSE background reader task ([`sse_reader_task`]) hands to
/// [`McpClient::sse_roundtrip`] over the (bounded, see
/// [`MCP_SSE_CHANNEL_CAPACITY`]) inbox channel: either a decoded JSON-RPC
/// frame, or a fatal reason the reader task is about to exit for (e.g. the
/// [`MCP_MAX_SSE_FRAME_BYTES`] cap being hit) — the latter lets a request
/// waiting on the channel fail with a NAMED error instead of the generic
/// "sse stream closed" it would otherwise see once the sender drops.
enum SseInboxMsg {
    Frame(Value),
    Error(String),
}

/// Build a [`HeaderMap`] from a plain string map — used by both
/// [`McpClient::connect_http`] and [`McpClient::connect_sse`]. An entry
/// whose key/value isn't valid header syntax is skipped rather than
/// failing the whole connect (a single malformed custom header shouldn't
/// block an otherwise-valid connection); this mirrors the "best effort,
/// never silently privilege-escalate" posture elsewhere in this crate —
/// skipping is safe here because the effect is "header absent", never
/// "wrong value sent".
fn build_header_map(headers: &BTreeMap<String, String>) -> HeaderMap {
    let mut map = HeaderMap::new();
    for (k, v) in headers {
        let (Ok(name), Ok(value)) = (
            HeaderName::from_bytes(k.as_bytes()),
            HeaderValue::from_str(v),
        ) else {
            continue;
        };
        map.insert(name, value);
    }
    map
}

/// How an [`McpClient`] was connected — kept on the client so
/// [`McpClient::reconnect`] can rebuild an equivalent connection without
/// the caller having to remember its own parameters.
#[derive(Debug, Clone)]
pub enum McpConnectParams {
    /// Spawned-process transport.
    Stdio {
        /// The command that was spawned.
        command: String,
        /// Its arguments.
        args: Vec<String>,
        /// Extra environment variables set on top of the inherited environment.
        env: BTreeMap<String, String>,
    },
    /// Streamable-HTTP (non-streaming) transport.
    Http {
        /// The server endpoint URL.
        url: String,
        /// Extra request headers.
        headers: BTreeMap<String, String>,
    },
    /// Legacy HTTP+SSE transport.
    Sse {
        /// The SSE stream URL.
        url: String,
        /// Extra request headers.
        headers: BTreeMap<String, String>,
    },
}

// ============================================================================
// ---- elicitation ------------------------------------------------------------
// ============================================================================

/// P5-2 (§2.1 dep "elicitation → `tools.question` surface"; §2 module 6's
/// own row: "⚡ headless print mode (deny-default like OC, oc§1)"): a
/// server→client `elicitation/create` request, mid-`tools/call`, asking the
/// user for structured input.
#[derive(Debug, Clone)]
pub struct ElicitationRequest {
    /// The server's human-readable prompt.
    pub message: String,
    /// JSON Schema for the requested input shape.
    pub requested_schema: Value,
}

/// The outcome an [`McpElicitationHandler`] returns — the three actions the
/// MCP elicitation spec defines.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ElicitationAction {
    /// The user supplied the requested data (see [`ElicitationResponse::content`]).
    Accept,
    /// The user was asked and declined.
    Decline,
    /// The interaction was cancelled/dismissed without a decision.
    Cancel,
}

/// What an [`McpElicitationHandler`] returns for one [`ElicitationRequest`].
#[derive(Debug, Clone)]
pub struct ElicitationResponse {
    /// Which of the three MCP elicitation outcomes this is.
    pub action: ElicitationAction,
    /// Present only when `action == Accept`.
    pub content: Option<Value>,
}

impl ElicitationResponse {
    fn decline() -> Self {
        ElicitationResponse {
            action: ElicitationAction::Decline,
            content: None,
        }
    }

    fn to_json_rpc_result(&self) -> Value {
        match self.action {
            ElicitationAction::Accept => json!({
                "action": "accept",
                "content": self.content.clone().unwrap_or(json!({})),
            }),
            ElicitationAction::Decline => json!({"action": "decline"}),
            ElicitationAction::Cancel => json!({"action": "cancel"}),
        }
    }
}

/// Handles a server-initiated `elicitation/create` request — the
/// `tools.question` surface's PROTOCOL side (§2.1 dep). The real
/// interactive prompt UI is `tui`'s job (P5 item #4, not yet built);
/// pending that, [`HeadlessElicitationHandler`] is the honest default —
/// DENY (decline), matching module 6's own "headless print mode:
/// deny-default like OC" row rather than hanging the tool call or silently
/// fabricating an answer. An embedder (or a future `tui` integration) can
/// install a real interactive handler via
/// [`McpClient::set_elicitation_handler`].
#[async_trait]
pub trait McpElicitationHandler: Send + Sync {
    /// Decide how to respond to one elicitation request.
    async fn handle(&self, request: &ElicitationRequest) -> ElicitationResponse;
}

/// The default: every elicitation request is declined. Correct for
/// non-interactive/print-mode runs (the only mode this crate's CLI
/// embedder — `crates/cli` — runs in today); a TUI-backed handler is a
/// tui-deferred follow-up, not built here.
pub struct HeadlessElicitationHandler;

#[async_trait]
impl McpElicitationHandler for HeadlessElicitationHandler {
    async fn handle(&self, _request: &ElicitationRequest) -> ElicitationResponse {
        ElicitationResponse::decline()
    }
}

fn parse_elicitation_request(params: &Value) -> ElicitationRequest {
    ElicitationRequest {
        message: params
            .get("message")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string(),
        requested_schema: params
            .get("requestedSchema")
            .cloned()
            .unwrap_or_else(|| json!({})),
    }
}

// ============================================================================
// ---- client -----------------------------------------------------------------
// ============================================================================

/// A tool exposed by a remote MCP server.
#[derive(Debug, Clone)]
pub struct McpToolDef {
    /// Tool name (unqualified — the remote server's own name for it).
    pub name: String,
    /// Human description.
    pub description: String,
    /// JSON Schema for the tool's input.
    pub input_schema: Value,
}

/// A resource exposed by a remote MCP server (`resources/list`).
#[derive(Debug, Clone, Default)]
pub struct McpResourceDef {
    /// The resource's URI.
    pub uri: String,
    /// Human-readable name.
    pub name: String,
    /// Human description.
    pub description: String,
    /// MIME type, if the server declared one.
    pub mime_type: Option<String>,
}

/// A resource TEMPLATE exposed by a remote MCP server (`resources/templates/list`).
#[derive(Debug, Clone, Default)]
pub struct McpResourceTemplateDef {
    /// The RFC 6570 URI template.
    pub uri_template: String,
    /// Human-readable name.
    pub name: String,
    /// Human description.
    pub description: String,
}

/// A prompt exposed by a remote MCP server (`prompts/list`).
#[derive(Debug, Clone, Default)]
pub struct McpPromptDef {
    /// Prompt name (unqualified — the remote server's own name for it).
    pub name: String,
    /// Human description.
    pub description: String,
    /// The arguments this prompt accepts.
    pub arguments: Vec<McpPromptArgDef>,
}

/// One argument a [`McpPromptDef`] accepts.
#[derive(Debug, Clone, Default)]
pub struct McpPromptArgDef {
    /// Argument name.
    pub name: String,
    /// Whether the server requires this argument.
    pub required: bool,
}

/// A client connected to an MCP server over stdio, HTTP, or SSE — see the
/// module doc comment for the transport model.
pub struct McpClient {
    conn: Conn,
    next_id: i64,
    params: McpConnectParams,
    /// The [`NetworkPolicy`] this client was connected under (`None` for
    /// stdio, or when no policy was passed to `connect_http`/`connect_sse`)
    /// — remembered so [`Self::reconnect`] re-applies the SAME policy to
    /// the rebuilt connection instead of silently reconnecting unchecked
    /// (Fable-5 review: `reconnect` used to pass `None` regardless of what
    /// the original connect used, reintroducing the redirect-SSRF class
    /// `connect_http`/`connect_sse` otherwise close).
    network_policy: Option<NetworkPolicy>,
    timeout: Duration,
    elicitation_handler: Arc<dyn McpElicitationHandler>,
    /// The `instructions` field from the server's `initialize` response, if
    /// any (§2 module 15 D7 row 5 "instructions"). `None` when the server
    /// didn't send one.
    pub instructions: Option<String>,
    /// Notifications this client has received but no caller has consumed
    /// yet (e.g. `notifications/resources/updated`) — a simple in-memory
    /// log, since this crate has no live-push channel to the model mid-turn
    /// (§2 module 15's resources row is request/response tool-shaped, see
    /// `McpResourceSubscribeTool`'s doc comment).
    pending_notifications: std::sync::Mutex<Vec<Value>>,
}

impl McpClient {
    /// Spawn `command args...` as an MCP server and perform the `initialize`
    /// handshake (stdio transport). `env` holds extra environment variables
    /// for the spawned process (from the server's config `env` block, e.g.
    /// an API token an MCP server needs) — they're set ON TOP OF supercode's
    /// own inherited environment, never replacing it: `tokio::process::Command`
    /// inherits the parent's environment by default (no `.env_clear()` here),
    /// and `.envs(env)` only adds/overrides the specific named vars. This
    /// matches Claude Code / Codex's own `env` semantics for MCP servers.
    pub async fn connect(
        command: &str,
        args: &[&str],
        env: &BTreeMap<String, String>,
    ) -> Result<Self> {
        let mut child = tokio::process::Command::new(command)
            .args(args)
            .envs(env)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::null())
            // Reap the server if the client is dropped, rather than relying on
            // it noticing stdin EOF — a server that ignores stdin would linger.
            .kill_on_drop(true)
            .spawn()
            .map_err(|e| Error::tool("mcp", format!("spawn {command}: {e}")))?;
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| Error::tool("mcp", "no stdin"))?;
        let stdout = BufReader::new(
            child
                .stdout
                .take()
                .ok_or_else(|| Error::tool("mcp", "no stdout"))?,
        );
        let params = McpConnectParams::Stdio {
            command: command.to_string(),
            args: args.iter().map(|s| s.to_string()).collect(),
            env: env.clone(),
        };
        let mut client = McpClient {
            conn: Conn::Stdio {
                child,
                stdin,
                stdout,
            },
            next_id: 0,
            params,
            // Stdio has no network policy to remember — nothing to reconnect
            // a stdio child process against (see `NetworkPolicy`'s doc
            // comment: it's an HTTP/SSRF floor).
            network_policy: None,
            timeout: DEFAULT_MCP_TIMEOUT,
            elicitation_handler: Arc::new(HeadlessElicitationHandler),
            instructions: None,
            pending_notifications: std::sync::Mutex::new(Vec::new()),
        };
        client.initialize().await?;
        Ok(client)
    }

    /// P5-2 (§2 module 15 D7 row 2 "remote HTTP"): connect over a single
    /// POST-per-request "Streamable HTTP" transport (the non-streaming
    /// case — see the module doc comment for what that scopes out).
    /// `network_policy`, if `Some` and enabled, is enforced against `url`
    /// BEFORE any connection is attempted (SSRF/domain-allowlist floor,
    /// same enforcement point `ToolContext::check_network` uses).
    pub async fn connect_http(
        url: &str,
        headers: &BTreeMap<String, String>,
        network_policy: Option<&NetworkPolicy>,
    ) -> Result<Self> {
        crate::tools::check_network_policy(network_policy, url)?;
        let client = reqwest::Client::builder()
            .timeout(DEFAULT_MCP_TIMEOUT)
            .redirect(crate::tools::network_checked_redirect_policy(
                network_policy.cloned(),
            ))
            .build()
            .map_err(|e| Error::tool("mcp", format!("building http client: {e}")))?;
        let params = McpConnectParams::Http {
            url: url.to_string(),
            headers: headers.clone(),
        };
        let mut mcp_client = McpClient {
            conn: Conn::Http {
                client,
                url: url.to_string(),
                headers: build_header_map(headers),
                session_id: None,
            },
            next_id: 0,
            params,
            // Remembered so `reconnect` re-enforces the SAME policy on the
            // rebuilt connection rather than reconnecting unchecked.
            network_policy: network_policy.cloned(),
            timeout: DEFAULT_MCP_TIMEOUT,
            elicitation_handler: Arc::new(HeadlessElicitationHandler),
            instructions: None,
            pending_notifications: std::sync::Mutex::new(Vec::new()),
        };
        mcp_client.initialize().await?;
        Ok(mcp_client)
    }

    /// P5-2 (§2 module 15 D7 row 2 "remote SSE"): connect over the legacy
    /// (2024-11-05) HTTP+SSE transport — a persistent `GET url` stream whose
    /// first event names the client→server POST endpoint. Same
    /// [`NetworkPolicy`] enforcement as [`Self::connect_http`].
    pub async fn connect_sse(
        url: &str,
        headers: &BTreeMap<String, String>,
        network_policy: Option<&NetworkPolicy>,
    ) -> Result<Self> {
        crate::tools::check_network_policy(network_policy, url)?;
        // No client-level `.timeout()`: the GET stream is intentionally
        // long-lived (it stays open for the connection's whole lifetime),
        // and this same client also issues the client->server POSTs — a
        // blanket per-request timeout would apply to (and could truncate)
        // the persistent GET just as much as a POST. `Self::timeout`
        // (default [`DEFAULT_MCP_TIMEOUT`]) bounds each `request()`'s WAIT
        // for its matching response instead — see `sse_roundtrip`.
        let client = reqwest::Client::builder()
            .redirect(crate::tools::network_checked_redirect_policy(
                network_policy.cloned(),
            ))
            .build()
            .map_err(|e| Error::tool("mcp", format!("building sse client: {e}")))?;
        let header_map = build_header_map(headers);
        let mut req = client.get(url);
        req = req.header(reqwest::header::ACCEPT, "text/event-stream");
        req = req.headers(header_map.clone());
        let resp = req
            .send()
            .await
            .map_err(|e| Error::tool("mcp", format!("sse connect failed: {e}")))?;
        if !resp.status().is_success() {
            return Err(Error::tool(
                "mcp",
                format!("sse connect: http status {}", resp.status()),
            ));
        }
        let base_url = url.to_string();
        let (endpoint_tx, endpoint_rx) = tokio::sync::oneshot::channel();
        // Bounded (not `unbounded_channel`): see `MCP_SSE_CHANNEL_CAPACITY`'s
        // doc comment for why a flooding server should apply backpressure to
        // the reader task rather than growing an unbounded in-memory queue.
        let (msg_tx, msg_rx) = mpsc::channel(MCP_SSE_CHANNEL_CAPACITY);
        let reader = tokio::spawn(sse_reader_task(resp, base_url, endpoint_tx, msg_tx));
        let post_url = tokio::time::timeout(DEFAULT_MCP_TIMEOUT, endpoint_rx)
            .await
            .map_err(|_| Error::tool("mcp", "timed out waiting for sse endpoint event"))?
            .map_err(|_| Error::tool("mcp", "sse stream closed before an endpoint event"))?;
        let params = McpConnectParams::Sse {
            url: url.to_string(),
            headers: headers.clone(),
        };
        let mut mcp_client = McpClient {
            conn: Conn::Sse {
                client,
                post_url,
                headers: header_map,
                inbox: msg_rx,
                reader,
            },
            next_id: 0,
            params,
            // Remembered so `reconnect` re-enforces the SAME policy on the
            // rebuilt connection rather than reconnecting unchecked.
            network_policy: network_policy.cloned(),
            timeout: DEFAULT_MCP_TIMEOUT,
            elicitation_handler: Arc::new(HeadlessElicitationHandler),
            instructions: None,
            pending_notifications: std::sync::Mutex::new(Vec::new()),
        };
        mcp_client.initialize().await?;
        Ok(mcp_client)
    }

    /// Re-establish this client's connection from its own remembered
    /// [`McpConnectParams`] AND its own remembered [`NetworkPolicy`] (see
    /// `Self::network_policy`'s field doc comment) — the "reconnect" half
    /// of "connection lifecycle, reconnect, timeouts" (§2 module 15 D7 row
    /// 1/2). Does NOT mutate `self`; the caller swaps in the returned
    /// client (and its tools/resources need re-wrapping, since a
    /// [`crate::tools::Tool`] closes over a specific
    /// `Arc<Mutex<McpClient>>`).
    ///
    /// **Security note (Fable-5 review, latent-SSRF-landmine finding):**
    /// this method has no callers today (unwired public API) — but a
    /// future caller wiring it up gets the SAME [`NetworkPolicy`]
    /// enforcement the original `connect_http`/`connect_sse` applied for
    /// free, because the http/sse arms below pass `self.network_policy`
    /// (not `None`) through to `connect_http`/`connect_sse`, which run the
    /// exact same pre-connect host check + per-hop redirect re-check as
    /// the original connect. Passing `None` here would silently reconnect
    /// with no policy at all — the exact redirect-SSRF class those two
    /// constructors otherwise close (`mcp_remote.rs`'s
    /// `reconnect_reuses_the_original_network_policy` test fails on that
    /// revert).
    pub async fn reconnect(&self) -> Result<Self> {
        match &self.params {
            McpConnectParams::Stdio { command, args, env } => {
                let args_ref: Vec<&str> = args.iter().map(String::as_str).collect();
                Self::connect(command, &args_ref, env).await
            }
            McpConnectParams::Http { url, headers } => {
                Self::connect_http(url, headers, self.network_policy.as_ref()).await
            }
            McpConnectParams::Sse { url, headers } => {
                Self::connect_sse(url, headers, self.network_policy.as_ref()).await
            }
        }
    }

    /// Install a non-default elicitation handler (e.g. a `tui` integration).
    pub fn set_elicitation_handler(&mut self, handler: Arc<dyn McpElicitationHandler>) {
        self.elicitation_handler = handler;
    }

    /// Per-request timeout for the network transports (stdio is unaffected
    /// — see [`DEFAULT_MCP_TIMEOUT`]'s doc comment). Default 30s.
    pub fn set_timeout(&mut self, timeout: Duration) {
        self.timeout = timeout;
    }

    /// Notifications received but not yet consumed by a caller (see
    /// `Self::pending_notifications`'s field doc comment). Draining
    /// (`std::mem::take`) rather than cloning — a caller that wants to peek
    /// without consuming should not call this.
    pub fn take_pending_notifications(&self) -> Vec<Value> {
        self.pending_notifications
            .lock()
            .map(|mut v| std::mem::take(&mut *v))
            .unwrap_or_default()
    }

    async fn initialize(&mut self) -> Result<()> {
        let result = self
            .request(
                "initialize",
                json!({
                    "protocolVersion": PROTOCOL_VERSION,
                    "capabilities": {
                        // Advertise elicitation support: this client CAN
                        // receive `elicitation/create` (even though the
                        // headless-default handler always declines it) — a
                        // server that gates the elicitation capability
                        // behind the client's own advertised capability
                        // still gets a real (if headless-conservative)
                        // answer instead of never being offered the chance
                        // to ask.
                        "elicitation": {}
                    },
                    "clientInfo": {"name": "supercode", "version": env!("CARGO_PKG_VERSION")}
                }),
            )
            .await?;
        self.instructions = result
            .get("instructions")
            .and_then(Value::as_str)
            .map(str::to_string);
        // Per the MCP spec, the client sends an `initialized` notification
        // once the handshake completes. Best-effort: a server that doesn't
        // require it (most don't gate on it) is unaffected either way.
        let _ = self.notify("notifications/initialized", json!({})).await;
        Ok(())
    }

    /// Send a JSON-RPC NOTIFICATION (no reply expected). Errors are the
    /// caller's to decide whether to propagate — `initialize`'s own call
    /// above deliberately ignores them (best-effort).
    async fn notify(&mut self, method: &str, params: Value) -> Result<()> {
        let msg = json!({"jsonrpc": "2.0", "method": method, "params": params});
        self.send_raw(&msg).await
    }

    /// Write one JSON-RPC message to the wire — the write half every
    /// transport needs (a client request, a reply to a server-initiated
    /// request, or a notification). The HTTP transport has no persistent
    /// connection to write an unsolicited message on; see
    /// [`Self::http_roundtrip`] for how it round-trips instead.
    async fn send_raw(&mut self, msg: &Value) -> Result<()> {
        match &mut self.conn {
            Conn::Stdio { stdin, .. } => {
                stdin
                    .write_all(format!("{msg}\n").as_bytes())
                    .await
                    .map_err(|e| Error::tool("mcp", format!("write: {e}")))?;
                stdin
                    .flush()
                    .await
                    .map_err(|e| Error::tool("mcp", format!("flush: {e}")))?;
                Ok(())
            }
            Conn::Sse {
                client,
                post_url,
                headers,
                ..
            } => {
                let resp = client
                    .post(post_url.as_str())
                    .headers(headers.clone())
                    .json(msg)
                    .send()
                    .await
                    .map_err(|e| Error::tool("mcp", format!("sse post: {e}")))?;
                if !resp.status().is_success() {
                    return Err(Error::tool(
                        "mcp",
                        format!("sse post: http status {}", resp.status()),
                    ));
                }
                Ok(())
            }
            Conn::Http { .. } => Err(Error::tool(
                "mcp",
                "cannot send an unsolicited message over the http (non-streaming) transport",
            )),
        }
    }

    /// Dispatch one incoming JSON-RPC message while waiting for `waiting_id`'s
    /// response. Returns `Some(result)` when `msg` IS that response
    /// (success or error, folded to `Result` here so the caller's loop just
    /// returns); `None` means "keep waiting" (a notification was logged, a
    /// server-initiated request was answered, or `msg` was some other
    /// stale/irrelevant frame).
    async fn handle_incoming_message(
        &mut self,
        waiting_id: i64,
        msg: Value,
    ) -> Result<Option<Value>> {
        let id = msg.get("id").and_then(Value::as_i64);
        let has_method = msg.get("method").and_then(Value::as_str);

        if id == Some(waiting_id) && has_method.is_none() {
            if let Some(err) = msg.get("error") {
                return Err(Error::tool("mcp", format!("rpc error: {err}")));
            }
            return Ok(Some(msg.get("result").cloned().unwrap_or(Value::Null)));
        }

        match (id, has_method) {
            // A server-initiated REQUEST (has both an id and a method) —
            // today only `elicitation/create` is understood; anything else
            // gets a clean JSON-RPC "method not found" reply rather than
            // silently hanging the server waiting for a response we'll
            // never send.
            (Some(req_id), Some(method)) => {
                let reply = if method == "elicitation/create" {
                    let params = msg.get("params").cloned().unwrap_or(Value::Null);
                    let request = parse_elicitation_request(&params);
                    let handler = self.elicitation_handler.clone();
                    let response = handler.handle(&request).await;
                    json!({"jsonrpc": "2.0", "id": req_id, "result": response.to_json_rpc_result()})
                } else {
                    json!({
                        "jsonrpc": "2.0", "id": req_id,
                        "error": {"code": -32601, "message": format!("supercode does not handle server-initiated `{method}`")}
                    })
                };
                self.send_raw(&reply).await?;
                Ok(None)
            }
            // A notification (method, no id) — log it and keep waiting.
            (None, Some(_)) => {
                if let Ok(mut log) = self.pending_notifications.lock() {
                    log.push(msg);
                }
                Ok(None)
            }
            // A response to some OTHER (stale) request id, or an
            // unparseable/irrelevant frame — ignore and keep waiting.
            _ => Ok(None),
        }
    }

    async fn request(&mut self, method: &str, params: Value) -> Result<Value> {
        self.next_id += 1;
        let id = self.next_id;
        let msg = json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params});
        match &self.conn {
            Conn::Stdio { .. } => self.stdio_roundtrip(id, &msg).await,
            Conn::Sse { .. } => self.sse_roundtrip(id, &msg).await,
            Conn::Http { .. } => self.http_roundtrip(id, &msg).await,
        }
    }

    async fn stdio_roundtrip(&mut self, id: i64, msg: &Value) -> Result<Value> {
        self.send_raw(msg).await?;
        loop {
            let Conn::Stdio { stdout, .. } = &mut self.conn else {
                unreachable!("stdio_roundtrip called on a non-stdio connection")
            };
            let mut buf = String::new();
            let n = stdout
                .read_line(&mut buf)
                .await
                .map_err(|e| Error::tool("mcp", format!("read: {e}")))?;
            if n == 0 {
                return Err(Error::tool("mcp", "server closed the connection"));
            }
            let Ok(incoming) = serde_json::from_str::<Value>(buf.trim()) else {
                continue;
            };
            if let Some(result) = self.handle_incoming_message(id, incoming).await? {
                return Ok(result);
            }
        }
    }

    async fn sse_roundtrip(&mut self, id: i64, msg: &Value) -> Result<Value> {
        self.send_raw(msg).await?;
        loop {
            let inbox_msg = {
                let Conn::Sse { inbox, .. } = &mut self.conn else {
                    unreachable!("sse_roundtrip called on a non-sse connection")
                };
                tokio::time::timeout(self.timeout, inbox.recv())
                    .await
                    .map_err(|_| Error::tool("mcp", "timed out waiting for an sse response"))?
                    .ok_or_else(|| Error::tool("mcp", "sse stream closed"))?
            };
            // A fatal reason the reader task sent instead of a frame (e.g.
            // MCP_MAX_SSE_FRAME_BYTES exceeded) — surface it as a named
            // error immediately rather than looping on it.
            let incoming = match inbox_msg {
                SseInboxMsg::Frame(v) => v,
                SseInboxMsg::Error(reason) => return Err(Error::tool("mcp", reason)),
            };
            if let Some(result) = self.handle_incoming_message(id, incoming).await? {
                return Ok(result);
            }
        }
    }

    /// P5-2: the http (Streamable-HTTP, non-streaming) round-trip. Named
    /// limitation (see the module doc comment): a server-initiated request
    /// embedded in the response body — e.g. an elicitation mid-call — has
    /// no channel for this client to reply on within a single POST/response
    /// cycle, so it is a clean, tested error rather than a silent drop or a
    /// hang. A `text/event-stream` response body IS still supported for the
    /// common single-event non-streaming case many Streamable HTTP servers
    /// use to answer a `tools/call`.
    async fn http_roundtrip(&mut self, id: i64, msg: &Value) -> Result<Value> {
        let (client, url, headers, session_id) = match &self.conn {
            Conn::Http {
                client,
                url,
                headers,
                session_id,
            } => (
                client.clone(),
                url.clone(),
                headers.clone(),
                session_id.clone(),
            ),
            _ => unreachable!("http_roundtrip called on a non-http connection"),
        };
        let mut req = client.post(&url).headers(headers).json(msg);
        if let Some(sid) = &session_id {
            req = req.header("Mcp-Session-Id", sid.as_str());
        }
        let resp = req
            .send()
            .await
            .map_err(|e| Error::tool("mcp", format!("http request failed: {e}")))?;
        if !resp.status().is_success() {
            return Err(Error::tool("mcp", format!("http status {}", resp.status())));
        }
        if let Some(new_sid) = resp
            .headers()
            .get("mcp-session-id")
            .and_then(|v| v.to_str().ok())
            .map(str::to_string)
        {
            if let Conn::Http { session_id, .. } = &mut self.conn {
                *session_id = Some(new_sid);
            }
        }
        let content_type = resp
            .headers()
            .get(reqwest::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("")
            .to_string();
        // Hardening (Fable-5 review, memory-DoS finding): bounded read, not
        // a bare `resp.bytes()` — see MCP_MAX_RESPONSE_BYTES's doc comment.
        let body = read_capped_body(resp, MCP_MAX_RESPONSE_BYTES, "http response body").await?;
        let frames: Vec<Value> = if content_type.starts_with("text/event-stream") {
            parse_sse_body(&body)
        } else {
            vec![serde_json::from_slice::<Value>(&body)
                .map_err(|e| Error::tool("mcp", format!("decoding http response: {e}")))?]
        };
        for frame in frames {
            let frame_id = frame.get("id").and_then(Value::as_i64);
            let has_method = frame.get("method").is_some();
            if frame_id == Some(id) && !has_method {
                if let Some(err) = frame.get("error") {
                    return Err(Error::tool("mcp", format!("rpc error: {err}")));
                }
                return Ok(frame.get("result").cloned().unwrap_or(Value::Null));
            }
            if has_method {
                // A server-initiated request/notification embedded in a
                // non-streaming http response — see this method's doc
                // comment for why this is a fail-closed error, not silently
                // dropped or hung on.
                return Err(Error::tool(
                    "mcp",
                    "server sent a server-initiated request/notification over the http \
                     (non-streaming) transport — elicitation and live notifications need \
                     stdio or sse",
                ));
            }
        }
        Err(Error::tool(
            "mcp",
            "http response never contained this request's result",
        ))
    }

    // ---- tools --------------------------------------------------------

    /// List the tools the server offers.
    pub async fn list_tools(&mut self) -> Result<Vec<McpToolDef>> {
        let result = self.request("tools/list", json!({})).await?;
        let tools = result
            .get("tools")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        Ok(tools
            .into_iter()
            .map(|t| McpToolDef {
                name: t
                    .get("name")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                description: t
                    .get("description")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_string(),
                input_schema: t
                    .get("inputSchema")
                    .cloned()
                    .unwrap_or_else(|| json!({"type": "object"})),
            })
            .collect())
    }

    /// Call a tool and return its text content.
    pub async fn call_tool(&mut self, name: &str, arguments: Value) -> Result<String> {
        let result = self
            .request("tools/call", json!({"name": name, "arguments": arguments}))
            .await?;
        Ok(extract_content_text(&result))
    }

    // ---- resources + templates (§2 module 15 D7 row 3) -----------------

    /// List the resources the server offers.
    pub async fn list_resources(&mut self) -> Result<Vec<McpResourceDef>> {
        let result = self.request("resources/list", json!({})).await?;
        Ok(result
            .get("resources")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .map(|r| McpResourceDef {
                uri: str_field(&r, "uri"),
                name: str_field(&r, "name"),
                description: str_field(&r, "description"),
                mime_type: r
                    .get("mimeType")
                    .and_then(Value::as_str)
                    .map(str::to_string),
            })
            .collect())
    }

    /// List the resource templates the server offers.
    pub async fn list_resource_templates(&mut self) -> Result<Vec<McpResourceTemplateDef>> {
        let result = self.request("resources/templates/list", json!({})).await?;
        Ok(result
            .get("resourceTemplates")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .map(|r| McpResourceTemplateDef {
                uri_template: str_field(&r, "uriTemplate"),
                name: str_field(&r, "name"),
                description: str_field(&r, "description"),
            })
            .collect())
    }

    /// Read one resource's content by URI. Errors (fail-closed, named —
    /// hardening, see [`MCP_MAX_RESOURCE_BYTES`]'s doc comment) if the
    /// joined text exceeds the cap, rather than returning/buffering an
    /// unbounded string.
    pub async fn read_resource(&mut self, uri: &str) -> Result<String> {
        let result = self.request("resources/read", json!({"uri": uri})).await?;
        let joined = result
            .get("contents")
            .and_then(Value::as_array)
            .map(|items| {
                items
                    .iter()
                    .filter_map(|i| {
                        i.get("text")
                            .and_then(Value::as_str)
                            .map(str::to_string)
                            .or_else(|| {
                                i.get("blob")
                                    .and_then(Value::as_str)
                                    .map(|b| format!("[base64 blob, {} bytes encoded]", b.len()))
                            })
                    })
                    .collect::<Vec<_>>()
                    .join("\n")
            })
            .unwrap_or_default();
        if joined.len() > MCP_MAX_RESOURCE_BYTES {
            return Err(Error::tool(
                "mcp",
                format!(
                    "resource {uri}: joined contents exceeded max {MCP_MAX_RESOURCE_BYTES} bytes"
                ),
            ));
        }
        Ok(joined)
    }

    /// Subscribe to update notifications for one resource by URI — updates
    /// arrive as `notifications/resources/updated` frames, logged in
    /// [`McpClient::take_pending_notifications`].
    pub async fn subscribe_resource(&mut self, uri: &str) -> Result<()> {
        self.request("resources/subscribe", json!({"uri": uri}))
            .await?;
        Ok(())
    }

    // ---- prompts-as-commands (§2 module 15 D7 row 4) --------------------

    /// List the prompts the server offers.
    pub async fn list_prompts(&mut self) -> Result<Vec<McpPromptDef>> {
        let result = self.request("prompts/list", json!({})).await?;
        Ok(result
            .get("prompts")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .map(|p| McpPromptDef {
                name: str_field(&p, "name"),
                description: str_field(&p, "description"),
                arguments: p
                    .get("arguments")
                    .and_then(Value::as_array)
                    .cloned()
                    .unwrap_or_default()
                    .into_iter()
                    .map(|a| McpPromptArgDef {
                        name: str_field(&a, "name"),
                        required: a.get("required").and_then(Value::as_bool).unwrap_or(false),
                    })
                    .collect(),
            })
            .collect())
    }

    /// Render a server prompt with `args` (a flat string->string map — the
    /// MCP spec's `prompts/get` `arguments` shape) into the concatenated
    /// text of every returned message — this crate's `Config.prompts`
    /// entries are likewise a single flat rendered string
    /// ([`crate::agent::Agent::expand_prompt`]'s local-template shape), so
    /// the two surfaces stay uniform to a caller.
    pub async fn get_prompt(
        &mut self,
        name: &str,
        args: BTreeMap<String, String>,
    ) -> Result<String> {
        let result = self
            .request("prompts/get", json!({"name": name, "arguments": args}))
            .await?;
        Ok(result
            .get("messages")
            .and_then(Value::as_array)
            .map(|msgs| {
                msgs.iter()
                    .filter_map(|m| {
                        m.get("content")
                            .and_then(|c| c.get("text"))
                            .and_then(Value::as_str)
                    })
                    .collect::<Vec<_>>()
                    .join("\n\n")
            })
            .unwrap_or_default())
    }
}

fn str_field(v: &Value, key: &str) -> String {
    v.get(key)
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string()
}

/// Pull the concatenated text out of an MCP `content` array.
fn extract_content_text(result: &Value) -> String {
    result
        .get("content")
        .and_then(Value::as_array)
        .map(|items| {
            items
                .iter()
                .filter_map(|i| i.get("text").and_then(Value::as_str))
                .collect::<Vec<_>>()
                .join("\n")
        })
        .unwrap_or_default()
}

/// Read `resp`'s body up to `cap` bytes, erroring (fail-closed, named error
/// naming `what`) rather than buffering further — the bounded replacement
/// for a bare `resp.bytes()` (`McpClient::http_roundtrip`'s hardening;
/// see [`MCP_MAX_RESPONSE_BYTES`]'s doc comment for why). Checks
/// `Content-Length` first as a fast reject when the server declares a
/// too-large body up front, then streams chunk-by-chunk (a hostile server
/// can lie about `Content-Length` or omit it and stream forever) so actual
/// memory use never exceeds `cap` before this errors out.
async fn read_capped_body(resp: reqwest::Response, cap: usize, what: &str) -> Result<Vec<u8>> {
    use futures::StreamExt;
    if let Some(len) = resp.content_length() {
        if len as usize > cap {
            return Err(Error::tool(
                "mcp",
                format!("{what}: declared content-length {len} bytes exceeds max {cap} bytes"),
            ));
        }
    }
    let mut buf: Vec<u8> = Vec::new();
    let mut stream = resp.bytes_stream();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk.map_err(|e| Error::tool("mcp", format!("reading {what}: {e}")))?;
        buf.extend_from_slice(&chunk);
        if buf.len() > cap {
            return Err(Error::tool(
                "mcp",
                format!("{what}: exceeded max {cap} bytes"),
            ));
        }
    }
    Ok(buf)
}

/// Parse a `text/event-stream` byte body into its JSON `data:` payloads —
/// used both by the http transport's single-response-body case
/// (`McpClient::http_roundtrip`) and by the sse reader task's per-chunk
/// incremental parser (`SseLineAccumulator`) sharing the same per-event
/// field syntax. Multiple `data:` lines within one event are joined with
/// `\n` per the SSE spec before JSON-parsing; an event whose joined data
/// doesn't parse as JSON is skipped (never fatal — matches this crate's
/// existing stdio precedent of skipping an unparseable line).
fn parse_sse_body(body: &[u8]) -> Vec<Value> {
    let text = String::from_utf8_lossy(body);
    let mut out = Vec::new();
    for event in text.split("\n\n") {
        let mut data_lines = Vec::new();
        for line in event.lines() {
            if let Some(d) = line.strip_prefix("data:") {
                data_lines.push(d.trim_start());
            }
        }
        if data_lines.is_empty() {
            continue;
        }
        if let Ok(v) = serde_json::from_str::<Value>(&data_lines.join("\n")) {
            out.push(v);
        }
    }
    out
}

/// Incremental SSE event parser for the persistent SSE reader task —
/// accumulates raw bytes across chunk boundaries (a `data:` line can be
/// split across two TCP reads) and yields one `(event_name, data)` pair per
/// complete (blank-line-terminated) event.
#[derive(Default)]
struct SseLineAccumulator {
    buf: String,
}

impl SseLineAccumulator {
    /// Feed `chunk` in and return every complete event it produced. Errors
    /// (fail-closed, hardening — see [`MCP_MAX_SSE_FRAME_BYTES`]'s doc
    /// comment) when the trailing, still-incomplete tail left after
    /// draining every complete event exceeds the cap — i.e. a single frame
    /// that never sends its terminating blank line. The buffer is cleared
    /// on that error, so a caller that (today, none do) chose to keep
    /// pushing after an error wouldn't keep growing it either.
    fn push(&mut self, chunk: &[u8]) -> Result<Vec<(Option<String>, String)>> {
        self.buf.push_str(&String::from_utf8_lossy(chunk));
        let mut out = Vec::new();
        // Process every COMPLETE event (terminated by a blank line) currently
        // in the buffer; leave any trailing partial event for the next push.
        while let Some(pos) = self.buf.find("\n\n") {
            let event_text: String = self.buf.drain(..pos + 2).collect();
            let mut event_name = None;
            let mut data_lines = Vec::new();
            for line in event_text.lines() {
                if let Some(v) = line.strip_prefix("event:") {
                    event_name = Some(v.trim_start().to_string());
                } else if let Some(v) = line.strip_prefix("data:") {
                    data_lines.push(v.trim_start().to_string());
                }
            }
            if !data_lines.is_empty() || event_name.is_some() {
                out.push((event_name, data_lines.join("\n")));
            }
        }
        if self.buf.len() > MCP_MAX_SSE_FRAME_BYTES {
            self.buf.clear();
            return Err(Error::tool(
                "mcp",
                format!(
                    "sse frame exceeded max {MCP_MAX_SSE_FRAME_BYTES} bytes without a \
                     terminating blank line"
                ),
            ));
        }
        Ok(out)
    }
}

/// Background task for [`McpClient::connect_sse`]: reads `resp`'s byte
/// stream, sends the discovered POST endpoint URL (from the first
/// `event: endpoint` frame) on `endpoint_tx` exactly once, and forwards
/// every subsequent JSON-parseable `event: message` frame's `data:` payload
/// into `msg_tx`. Exits quietly (dropping both channels) when the stream
/// ends — a `request()` waiting on `msg_tx`'s receiver then sees a closed
/// channel and reports "sse stream closed" rather than hanging forever.
///
/// `msg_tx` is bounded ([`MCP_SSE_CHANNEL_CAPACITY`]) — `.send(..).await`
/// below therefore applies backpressure (this task simply stops draining
/// the socket) when nothing has called `request()`/drained `inbox` in a
/// while, rather than this task buffering an unbounded queue of unconsumed
/// frames. That's not a deadlock: this task is the only thing that can
/// ever fill the channel, and the next `McpClient::request()` (or the one
/// already in flight) is the thing that drains it — there's no cycle where
/// this task itself needs to make progress for that drain to happen. On an
/// `SseLineAccumulator` error (an oversized, un-terminated frame — see
/// [`MCP_MAX_SSE_FRAME_BYTES`]) this task sends a named
/// [`SseInboxMsg::Error`] and exits, so a pending `request()` fails fast
/// with a clear reason instead of just seeing a closed channel.
async fn sse_reader_task(
    resp: reqwest::Response,
    base_url: String,
    endpoint_tx: tokio::sync::oneshot::Sender<String>,
    msg_tx: mpsc::Sender<SseInboxMsg>,
) {
    use futures::StreamExt;
    let mut stream = resp.bytes_stream();
    let mut acc = SseLineAccumulator::default();
    let mut endpoint_tx = Some(endpoint_tx);
    while let Some(chunk) = stream.next().await {
        let Ok(bytes) = chunk else { break };
        let events = match acc.push(&bytes) {
            Ok(events) => events,
            Err(e) => {
                let _ = msg_tx.send(SseInboxMsg::Error(e.to_string())).await;
                return;
            }
        };
        for (event_name, data) in events {
            match event_name.as_deref() {
                Some("endpoint") => {
                    if let Some(tx) = endpoint_tx.take() {
                        let resolved = resolve_endpoint_url(&base_url, data.trim());
                        let _ = tx.send(resolved);
                    }
                }
                _ => {
                    // "message" (the spec name) or an unnamed event — any
                    // frame with `data:` that isn't the endpoint discovery
                    // event is a JSON-RPC message.
                    if let Ok(v) = serde_json::from_str::<Value>(&data) {
                        if msg_tx.send(SseInboxMsg::Frame(v)).await.is_err() {
                            return; // no one is listening anymore
                        }
                    }
                }
            }
        }
    }
}

/// Resolve the `endpoint` event's `data:` payload (which the spec allows to
/// be a bare path, e.g. `/messages?session=abc`) against the SSE stream's
/// own origin — an absolute URL passes through unchanged.
fn resolve_endpoint_url(base_url: &str, endpoint: &str) -> String {
    if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
        return endpoint.to_string();
    }
    let Some(scheme_end) = base_url.find("://") else {
        return endpoint.to_string();
    };
    let after_scheme = &base_url[scheme_end + 3..];
    let origin_end = after_scheme.find('/').map(|i| scheme_end + 3 + i);
    let origin = match origin_end {
        Some(end) => &base_url[..end],
        None => base_url,
    };
    if endpoint.starts_with('/') {
        format!("{origin}{endpoint}")
    } else {
        format!("{origin}/{endpoint}")
    }
}

// ============================================================================
// ---- server-handle: shared client + everything a server attach produces ---
// ============================================================================

/// P5-2: one connected server, wrapping the `Arc<Mutex<McpClient>>` every
/// derived [`Tool`]/prompt-source shares — the single point that produces
/// tools ([`McpTool`]), resource tools, prompt names, and the server's
/// folded-in instructions, so a caller (`crates/cli`'s `attach_mcp`) only
/// has to connect once and ask this handle for everything else.
#[derive(Clone)]
pub struct McpServerHandle {
    /// This server's name (the `mcp__<server>__…` namespace prefix).
    pub server: String,
    client: Arc<Mutex<McpClient>>,
}

impl McpServerHandle {
    /// Wrap an already-connected `client` under `server`'s name.
    pub fn new(server: impl Into<String>, client: McpClient) -> Self {
        McpServerHandle {
            server: server.into(),
            client: Arc::new(Mutex::new(client)),
        }
    }

    /// The server's `initialize`-time instructions, if any.
    pub async fn instructions(&self) -> Option<String> {
        self.client.lock().await.instructions.clone()
    }

    /// This server's tools, namespaced `mcp__<server>__<tool>`.
    pub async fn tools(&self) -> Result<Vec<McpTool>> {
        let defs = self.client.lock().await.list_tools().await?;
        Ok(defs
            .into_iter()
            .map(|d| McpTool {
                name: format!("mcp__{}__{}", self.server, d.name),
                description: d.description,
                parameters: d.input_schema,
                remote_name: d.name,
                client: self.client.clone(),
            })
            .collect())
    }

    /// This server's resource-access tools (`resources_list`/`_read`/
    /// `_subscribe`), always offered regardless of whether the server
    /// actually advertised a `resources` capability — a server that
    /// doesn't support resources simply errors clearly on the underlying
    /// `resources/list` call (the same "let the remote error surface"
    /// posture [`McpTool::execute`] already has for `tools/call`), rather
    /// than this client trying to pre-negotiate capabilities perfectly.
    pub fn resource_tools(&self) -> Vec<Box<dyn Tool>> {
        vec![
            Box::new(McpResourcesListTool {
                name: format!("mcp__{}__resources_list", self.server),
                client: self.client.clone(),
            }),
            Box::new(McpResourceReadTool {
                name: format!("mcp__{}__resources_read", self.server),
                client: self.client.clone(),
            }),
            Box::new(McpResourceSubscribeTool {
                name: format!("mcp__{}__resources_subscribe", self.server),
                client: self.client.clone(),
            }),
        ]
    }

    /// This server's prompts, namespaced `mcp__<server>__<prompt>` — see
    /// [`McpPromptSource`]'s doc comment for why namespacing (not a bare
    /// name) is the mechanism that keeps an untrusted/remote server from
    /// ever being able to collide with a trusted command name.
    pub async fn prompts(&self) -> Result<Vec<(String, McpPromptSource)>> {
        let defs = self.client.lock().await.list_prompts().await?;
        Ok(defs
            .into_iter()
            .map(|d| {
                (
                    format!("mcp__{}__{}", self.server, d.name),
                    McpPromptSource {
                        client: self.client.clone(),
                        remote_name: d.name,
                        arg_names: d.arguments.into_iter().map(|a| a.name).collect(),
                    },
                )
            })
            .collect())
    }

    /// The shared client — for callers that need lower-level access (e.g.
    /// installing an elicitation handler, or OAuth token refresh wiring).
    pub fn client(&self) -> Arc<Mutex<McpClient>> {
        self.client.clone()
    }
}

/// A supercode [`Tool`] backed by a remote MCP tool. The name is namespaced
/// `mcp__<server>__<tool>` to match the convention seen in the corpus.
pub struct McpTool {
    name: String,
    description: String,
    parameters: Value,
    remote_name: String,
    client: Arc<Mutex<McpClient>>,
}

impl McpTool {
    /// Wrap every tool from `client` (already connected) under `server`
    /// prefix. Kept for API/test back-compat (P5-1 baseline signature); new
    /// callers that also want resources/prompts/instructions should use
    /// [`McpServerHandle`] directly.
    pub async fn from_client(server: &str, client: McpClient) -> Result<Vec<McpTool>> {
        McpServerHandle::new(server, client).tools().await
    }
}

#[async_trait]
impl Tool for McpTool {
    fn name(&self) -> &str {
        &self.name
    }
    fn description(&self) -> &str {
        &self.description
    }
    fn parameters(&self) -> Value {
        self.parameters.clone()
    }
    async fn execute(&self, args: Value, _ctx: &ToolContext) -> Result<String> {
        self.client
            .lock()
            .await
            .call_tool(&self.remote_name, args)
            .await
    }
}

/// A prompt this crate can render via `prompts/get` — what
/// [`McpServerHandle::prompts`] hands back for a caller to register as a
/// slash-command source (`crate::agent::Agent::register_mcp_prompt`).
///
/// **P4d-class security lesson, closed BY CONSTRUCTION (§2 module 15 D7 row
/// 4's own security note, "an MCP-provided prompt from an untrusted/
/// project-scoped server must not silently override a trusted command
/// name"):** every prompt this crate surfaces is namespaced
/// `mcp__<server>__<prompt>` — never the bare remote name. Since no
/// built-in or user-authored `[core.prompts]` command name is EVER
/// `mcp__`-prefixed (that prefix is reserved by this module), a remote
/// server — however untrusted, however maliciously named its prompts are —
/// cannot construct a colliding key: `mcp__evil__code-review` and
/// `code-review` are simply different map keys. This is the same
/// "namespace instead of trust-flag" treatment [`McpTool`] already applies
/// to tool names; a test in `crates/harness/tests/mcp_prompts.rs` pins it
/// (`untrusted_mcp_prompt_cannot_override_a_trusted_command_name`).
#[derive(Clone)]
pub struct McpPromptSource {
    client: Arc<Mutex<McpClient>>,
    remote_name: String,
    /// The server-declared argument names, in `prompts/list` order — used
    /// by `crate::agent::Agent::expand_prompt_async` to map a slash
    /// command's trailing free text onto this prompt's named arguments
    /// (single-argument prompts get the whole trailing text; multi-argument
    /// prompts expect `key=value` pairs — see that method's doc comment).
    arg_names: Vec<String>,
}

impl McpPromptSource {
    /// Render this prompt with `args` (see [`McpClient::get_prompt`]).
    pub async fn render(&self, args: BTreeMap<String, String>) -> Result<String> {
        self.client
            .lock()
            .await
            .get_prompt(&self.remote_name, args)
            .await
    }

    /// This prompt's declared argument names, in order.
    pub fn arg_names(&self) -> &[String] {
        &self.arg_names
    }
}

#[async_trait]
impl crate::sdk::SdkPromptSource for McpPromptSource {
    async fn render(&self, args: BTreeMap<String, String>) -> Result<String> {
        McpPromptSource::render(self, args).await
    }

    fn arg_names(&self) -> &[String] {
        McpPromptSource::arg_names(self)
    }
}

// ---- resource tools ---------------------------------------------------

struct McpResourcesListTool {
    name: String,
    client: Arc<Mutex<McpClient>>,
}

#[async_trait]
impl Tool for McpResourcesListTool {
    fn name(&self) -> &str {
        &self.name
    }
    fn description(&self) -> &str {
        "List this MCP server's available resources and resource templates."
    }
    fn parameters(&self) -> Value {
        json!({"type": "object", "properties": {}, "additionalProperties": false})
    }
    async fn execute(&self, _args: Value, _ctx: &ToolContext) -> Result<String> {
        let mut client = self.client.lock().await;
        let resources = client.list_resources().await?;
        let templates = client.list_resource_templates().await?;
        let mut out = String::new();
        for r in &resources {
            out.push_str(&format!("- {} ({})\n", r.uri, r.name));
        }
        for t in &templates {
            out.push_str(&format!("- template: {} ({})\n", t.uri_template, t.name));
        }
        if out.is_empty() {
            out.push_str("(no resources or templates)\n");
        }
        Ok(out)
    }
}

#[derive(serde::Deserialize)]
struct ResourceUriArgs {
    uri: String,
}

struct McpResourceReadTool {
    name: String,
    client: Arc<Mutex<McpClient>>,
}

#[async_trait]
impl Tool for McpResourceReadTool {
    fn name(&self) -> &str {
        &self.name
    }
    fn description(&self) -> &str {
        "Read one resource from this MCP server by URI."
    }
    fn parameters(&self) -> Value {
        json!({
            "type": "object",
            "properties": {"uri": {"type": "string"}},
            "required": ["uri"],
            "additionalProperties": false
        })
    }
    async fn execute(&self, args: Value, _ctx: &ToolContext) -> Result<String> {
        let a: ResourceUriArgs =
            serde_json::from_value(args).map_err(|e| Error::InvalidArguments {
                tool: self.name.clone(),
                message: e.to_string(),
            })?;
        self.client.lock().await.read_resource(&a.uri).await
    }
}

struct McpResourceSubscribeTool {
    name: String,
    client: Arc<Mutex<McpClient>>,
}

#[async_trait]
impl Tool for McpResourceSubscribeTool {
    fn name(&self) -> &str {
        &self.name
    }
    fn description(&self) -> &str {
        "Subscribe to update notifications for one resource on this MCP server by URI. \
         Updates surface as this server's pending-notifications log (no live push into the \
         conversation) — call resources_list/resources_read again to see the latest content."
    }
    fn parameters(&self) -> Value {
        json!({
            "type": "object",
            "properties": {"uri": {"type": "string"}},
            "required": ["uri"],
            "additionalProperties": false
        })
    }
    async fn execute(&self, args: Value, _ctx: &ToolContext) -> Result<String> {
        let a: ResourceUriArgs =
            serde_json::from_value(args).map_err(|e| Error::InvalidArguments {
                tool: self.name.clone(),
                message: e.to_string(),
            })?;
        self.client.lock().await.subscribe_resource(&a.uri).await?;
        Ok(format!("subscribed to {}", a.uri))
    }
}

// ============================================================================
// ---- C2 cache-invalidation signal (§2.2 C2) ---------------------------------
// ============================================================================

/// P5-2 (§2.2 C2 "connect invalidates cache prefix"; §2 module 25 `cache`
/// is the referee): the churn notice a caller (`crates/cli`'s `attach_mcp`)
/// emits when connecting a server under an active `CachePlan::ImportedPrefix`
/// — a pure, independently-testable function so the wording/threshold logic
/// isn't buried in CLI plumbing. "At minimum emit the churn signal" (P5-2
/// build brief) — this is that signal; it does not itself reset any cache
/// bookkeeping (see `Agent::register_tool`'s own C2 note for the runtime
/// half: any tool registered after the agent's first turn resets
/// `cache_established`, MCP-sourced or not).
pub fn cache_churn_notice(server: &str, tool_count: usize) -> String {
    format!(
        "mcp: connecting `{server}` added {tool_count} tool(s) to the prompt prefix — with \
         an imported-prefix cache plan active, this likely invalidates the cache hit on the \
         next turn (C2)"
    )
}

// ============================================================================
// ---- server side ------------------------------------------------------------
// ============================================================================

/// MCP tool projection of the versioned SDK facade. The MCP envelope and
/// tool-call id never enter the SDK request or its canonical session data.
#[cfg(feature = "adapter-mcp")]
pub struct SdkMcpTool {
    service: Arc<Mutex<HarnessSessionService>>,
    runtime: Option<Arc<dyn SdkRuntime>>,
    attachment: Arc<Mutex<Option<FrontendAttachment>>>,
}

#[cfg(feature = "adapter-mcp")]
impl Default for SdkMcpTool {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(feature = "adapter-mcp")]
impl SdkMcpTool {
    /// Create an independent stateful SDK projection for one MCP server.
    pub fn new() -> Self {
        Self {
            service: Arc::new(Mutex::new(HarnessSessionService::new())),
            runtime: None,
            attachment: Arc::new(Mutex::new(None)),
        }
    }

    /// Project an already-owned SDK runtime into MCP without granting MCP
    /// process-launch, persistence, or shutdown authority.
    pub async fn attached(runtime: Arc<dyn SdkRuntime>) -> std::result::Result<Self, SdkError> {
        let attachment = runtime.attach(200).await?;
        Ok(Self {
            service: Arc::new(Mutex::new(HarnessSessionService::new())),
            runtime: Some(runtime),
            attachment: Arc::new(Mutex::new(Some(attachment))),
        })
    }

    async fn execute_attached(
        &self,
        runtime: &Arc<dyn SdkRuntime>,
        operation: SdkOperation,
        params: Value,
    ) -> std::result::Result<Value, SdkError> {
        let descriptor = runtime.describe().await?;
        let session_id = descriptor.session_id;
        match operation {
            SdkOperation::Input => {
                let prompt = params
                    .get("prompt")
                    .or_else(|| params.get("text"))
                    .and_then(Value::as_str)
                    .ok_or_else(|| {
                        SdkError::new(
                            crate::SdkErrorCode::InvalidArgument,
                            operation,
                            "input requires string `prompt` or `text`",
                        )
                    })?;
                let image_urls = match params.get("image_urls") {
                    None => Vec::new(),
                    Some(Value::Array(values)) => values
                        .iter()
                        .map(|value| {
                            value.as_str().map(str::to_owned).ok_or_else(|| {
                                SdkError::new(
                                    crate::SdkErrorCode::InvalidArgument,
                                    operation,
                                    "input requires string entries in `image_urls`",
                                )
                            })
                        })
                        .collect::<std::result::Result<Vec<_>, _>>()?,
                    Some(_) => {
                        return Err(SdkError::new(
                            crate::SdkErrorCode::InvalidArgument,
                            operation,
                            "input requires array `image_urls`",
                        ))
                    }
                };
                let reply = runtime
                    .submit_with_images(prompt.to_string(), image_urls)
                    .await?;
                Ok(json!({"session_id":session_id, "reply":reply}))
            }
            SdkOperation::Events => {
                let mut attachment = self.attachment.lock().await;
                if attachment.is_none() {
                    *attachment = Some(runtime.attach(200).await?);
                }
                let event = attachment
                    .as_mut()
                    .expect("attachment initialized")
                    .next_event()
                    .await?;
                Ok(json!({"session_id":session_id, "event":event}))
            }
            SdkOperation::Interrupt => Ok(json!({
                "session_id":session_id,
                "interrupted":runtime.interrupt().await?,
            })),
            SdkOperation::Steer => {
                let prompt = params
                    .get("prompt")
                    .or_else(|| params.get("text"))
                    .and_then(Value::as_str)
                    .ok_or_else(|| {
                        SdkError::new(
                            crate::SdkErrorCode::InvalidArgument,
                            operation,
                            "steer requires string `prompt` or `text`",
                        )
                    })?;
                runtime.steer(prompt.to_string()).await?;
                Ok(json!({"session_id":session_id}))
            }
            SdkOperation::Respond => {
                let response = serde_json::from_value::<FrontendResponse>(
                    params.get("response").cloned().unwrap_or(Value::Null),
                )
                .map_err(|error| {
                    SdkError::new(
                        crate::SdkErrorCode::InvalidArgument,
                        operation,
                        error.to_string(),
                    )
                })?;
                runtime.respond(response).await?;
                Ok(json!({"session_id":session_id}))
            }
            _ => Err(SdkError::unsupported(operation)),
        }
    }
}

#[async_trait]
#[cfg(feature = "adapter-mcp")]
impl Tool for SdkMcpTool {
    fn name(&self) -> &str {
        "supercode_sdk"
    }

    fn description(&self) -> &str {
        "Invoke one operation on Supercode's versioned session/runtime SDK facade."
    }

    fn parameters(&self) -> Value {
        json!({
            "type": "object",
            "properties": {
                "operation": {
                    "type": "string",
                    "enum": ["discover", "load", "start", "resume", "input", "events", "interrupt", "steer", "respond", "export", "close"]
                },
                "params": {"type": "object"}
            },
            "required": ["operation"],
            "additionalProperties": false
        })
    }

    async fn execute(&self, args: Value, ctx: &ToolContext) -> Result<String> {
        let operation = serde_json::from_value::<SdkOperation>(
            args.get("operation").cloned().unwrap_or(Value::Null),
        )
        .map_err(|error| Error::tool(self.name(), error.to_string()))?;
        if let Some(runtime) = &self.runtime {
            return self
                .execute_attached(
                    runtime,
                    operation,
                    args.get("params").cloned().unwrap_or_else(|| json!({})),
                )
                .await
                .and_then(|value| {
                    serde_json::to_string(&value)
                        .map_err(|error| SdkError::Transport(error.to_string()))
                })
                .map_err(|error| sdk_mcp_error(self.name(), &error));
        }
        if !matches!(
            operation,
            SdkOperation::Discover | SdkOperation::Load | SdkOperation::Export
        ) {
            return Err(Error::tool(
                self.name(),
                format!(
                    "SUPERCODE_SDK_ERROR:{}",
                    json!({
                        "name":"unsupported_action",
                        "operation":operation,
                        "message":"the MCP SDK adapter is read-only; runtime control requires an owner surface",
                    })
                ),
            ));
        }
        let mut params = args.get("params").cloned().unwrap_or_else(|| json!({}));
        confine_sdk_mcp_params(operation, &mut params, ctx)?;
        let result = self
            .service
            .lock()
            .await
            .execute(SdkRequest { operation, params })
            .await
            .map_err(|error| sdk_mcp_error(self.name(), &error))?;
        serde_json::to_string(&result).map_err(|error| Error::tool(self.name(), error.to_string()))
    }
}

#[cfg(feature = "adapter-mcp")]
fn sdk_mcp_error(tool: &str, error: &SdkError) -> Error {
    Error::tool(
        tool,
        format!(
            "SUPERCODE_SDK_ERROR:{}",
            json!({
                "name":error.code(),
                "operation":error.operation(),
                "message":error.to_string(),
            })
        ),
    )
}

#[cfg(feature = "adapter-mcp")]
fn confine_sdk_mcp_params(
    operation: SdkOperation,
    params: &mut Value,
    ctx: &ToolContext,
) -> Result<()> {
    if operation == SdkOperation::Discover {
        if params.get("homes").is_some() {
            return Err(Error::tool(
                "supercode_sdk",
                "MCP discovery cannot override harness homes",
            ));
        }
        params["workspace"] = json!(ctx.cwd);
        return Ok(());
    }
    let path = params
        .pointer("/locator/storage/path")
        .and_then(Value::as_str)
        .ok_or_else(|| Error::tool("supercode_sdk", "load/export requires locator.storage.path"))?;
    let path = ctx.resolve(path);
    if !crate::safe_path::contained(&ctx.cwd, &path) {
        return Err(Error::tool(
            "supercode_sdk",
            "session locator escapes the MCP workspace",
        ));
    }
    params["locator"]["storage"]["path"] = json!(path);
    Ok(())
}

/// Register the SDK adapter alongside ordinary MCP coding tools.
#[cfg(feature = "adapter-mcp")]
pub fn register_sdk_tool(registry: &mut ToolRegistry) {
    registry.register(SdkMcpTool::new());
}

/// Handle one JSON-RPC request against a [`ToolRegistry`], returning the
/// JSON-RPC response (or `None` for notifications that need no reply).
pub async fn handle_request(
    registry: &ToolRegistry,
    ctx: &ToolContext,
    request: &Value,
) -> Option<Value> {
    let id = request.get("id").cloned();
    let method = request.get("method").and_then(Value::as_str).unwrap_or("");
    let reply = |result: Value| Some(json!({"jsonrpc": "2.0", "id": id, "result": result}));

    match method {
        "initialize" => reply(json!({
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "supercode", "version": env!("CARGO_PKG_VERSION")}
        })),
        "tools/list" => {
            let tools: Vec<Value> = registry
                .iter()
                .map(|t| {
                    json!({
                        "name": t.name(),
                        "description": t.description(),
                        "inputSchema": t.parameters(),
                    })
                })
                .collect();
            reply(json!({"tools": tools}))
        }
        "tools/call" => {
            let params = request.get("params").cloned().unwrap_or(Value::Null);
            let name = params.get("name").and_then(Value::as_str).unwrap_or("");
            let args = params.get("arguments").cloned().unwrap_or(json!({}));
            match registry.get(name) {
                None => Some(json!({
                    "jsonrpc": "2.0", "id": id,
                    "error": {"code": -32601, "message": format!("unknown tool `{name}`")}
                })),
                Some(tool) => {
                    let (text, is_error, structured) = match tool.execute(args, ctx).await {
                        Ok(t) => (t, false, None),
                        Err(e) => {
                            let text = e.to_string();
                            let structured = text
                                .split_once("SUPERCODE_SDK_ERROR:")
                                .and_then(|(_, value)| serde_json::from_str::<Value>(value).ok())
                                .map(|error| json!({"error":error}));
                            (format!("Error: {text}"), true, structured)
                        }
                    };
                    reply(json!({
                        "content": [{"type": "text", "text": text}],
                        "isError": is_error,
                        "structuredContent": structured,
                    }))
                }
            }
        }
        // Notifications (no id) and unknown methods.
        _ if id.is_none() => None,
        _ => Some(json!({
            "jsonrpc": "2.0", "id": id,
            "error": {"code": -32601, "message": format!("unknown method `{method}`")}
        })),
    }
}

/// Run a blocking stdio MCP server exposing `registry`, reading requests from
/// stdin and writing responses to stdout until EOF.
pub async fn serve_stdio(registry: &ToolRegistry, ctx: &ToolContext) -> Result<()> {
    let mut stdin = BufReader::new(tokio::io::stdin());
    let mut stdout = tokio::io::stdout();
    let mut line = String::new();
    loop {
        line.clear();
        if stdin.read_line(&mut line).await? == 0 {
            break;
        }
        let Ok(req) = serde_json::from_str::<Value>(line.trim()) else {
            continue;
        };
        if let Some(resp) = handle_request(registry, ctx, &req).await {
            stdout.write_all(format!("{resp}\n").as_bytes()).await?;
            stdout.flush().await?;
        }
    }
    Ok(())
}

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

    #[tokio::test]
    async fn mcp_sdk_tool_is_a_thin_named_error_projection() {
        let mut registry = ToolRegistry::new();
        register_sdk_tool(&mut registry);
        let ctx = ToolContext::new(std::env::temp_dir());

        let listed = handle_request(
            &registry,
            &ctx,
            &json!({"jsonrpc":"2.0", "id":1, "method":"tools/list"}),
        )
        .await
        .unwrap();
        assert_eq!(listed["result"]["tools"][0]["name"], "supercode_sdk");

        let response = handle_request(
            &registry,
            &ctx,
            &json!({
                "jsonrpc":"2.0",
                "id":2,
                "method":"tools/call",
                "params": {
                    "name":"supercode_sdk",
                    "arguments":{"operation":"steer", "params":{}}
                }
            }),
        )
        .await
        .unwrap();
        assert_eq!(response["result"]["isError"], true);
        assert_eq!(
            response["result"]["structuredContent"]["error"]["name"],
            "unsupported_action"
        );
        assert_eq!(
            response["result"]["structuredContent"]["error"]["operation"],
            "steer"
        );
    }

    #[test]
    fn cache_churn_notice_names_server_and_count() {
        let msg = cache_churn_notice("github", 12);
        assert!(msg.contains("github"));
        assert!(msg.contains("12"));
        assert!(msg.contains("C2"));
    }

    #[test]
    fn resolve_endpoint_url_passes_through_absolute_urls() {
        assert_eq!(
            resolve_endpoint_url("http://localhost:1234/sse", "https://other/msg"),
            "https://other/msg"
        );
    }

    #[test]
    fn resolve_endpoint_url_resolves_relative_path_against_origin() {
        assert_eq!(
            resolve_endpoint_url("http://localhost:1234/sse", "/messages?session=abc"),
            "http://localhost:1234/messages?session=abc"
        );
    }

    #[test]
    fn parse_sse_body_extracts_multiple_events() {
        let body = b"event: message\ndata: {\"a\":1}\n\nevent: message\ndata: {\"a\":2}\n\n";
        let out = parse_sse_body(body);
        assert_eq!(out.len(), 2);
        assert_eq!(out[0]["a"], 1);
        assert_eq!(out[1]["a"], 2);
    }

    #[test]
    fn sse_line_accumulator_handles_a_split_chunk() {
        let mut acc = SseLineAccumulator::default();
        let first = acc.push(b"event: message\ndata: {\"a\":").unwrap();
        assert!(first.is_empty(), "no complete event yet");
        let second = acc.push(b"1}\n\n").unwrap();
        assert_eq!(second.len(), 1);
        assert_eq!(second[0].0.as_deref(), Some("message"));
        assert_eq!(second[0].1, "{\"a\":1}");
    }

    #[test]
    fn sse_line_accumulator_errors_and_resets_on_an_oversized_unterminated_frame() {
        // Fable-5 review hardening: an SSE frame that never sends its
        // terminating blank line must not grow the accumulator without
        // bound — it must error (fail-closed) once it exceeds
        // MCP_MAX_SSE_FRAME_BYTES, and the buffer must be reset (not left
        // holding the oversized data) rather than growing on every push.
        let mut acc = SseLineAccumulator::default();
        let chunk = vec![b'x'; MCP_MAX_SSE_FRAME_BYTES + 1];
        let err = acc.push(&chunk).unwrap_err();
        assert!(
            err.to_string().contains("exceeded max"),
            "error should name the cap: {err}"
        );
        assert_eq!(
            acc.buf.len(),
            0,
            "buffer must be reset on overflow, not left growing"
        );
    }

    #[test]
    fn sse_line_accumulator_stays_under_cap_for_legit_small_events() {
        // No-over-block confirmation: an ordinary small event (well under
        // the cap) still parses normally.
        let mut acc = SseLineAccumulator::default();
        let events = acc
            .push(b"event: message\ndata: {\"ok\":true}\n\n")
            .unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].1, "{\"ok\":true}");
    }

    #[test]
    fn elicitation_response_decline_serializes_without_content() {
        let r = ElicitationResponse::decline();
        assert_eq!(r.to_json_rpc_result(), json!({"action": "decline"}));
    }

    #[test]
    fn elicitation_response_accept_carries_content() {
        let r = ElicitationResponse {
            action: ElicitationAction::Accept,
            content: Some(json!({"name": "value"})),
        };
        assert_eq!(
            r.to_json_rpc_result(),
            json!({"action": "accept", "content": {"name": "value"}})
        );
    }

    /// Fable-5 review, latent-SSRF-landmine finding: `reconnect` used to
    /// call `connect_http`/`connect_sse` with `None` for the network
    /// policy regardless of what the original connect used, silently
    /// skipping BOTH the pre-connect host check and the per-hop redirect
    /// re-check on every reconnect. This is a FAIL-ON-REVERT test: it
    /// builds an already-"connected" `McpClient` by hand (private-field
    /// access — this `tests` module is a child of `mcp`, so normal Rust
    /// visibility rules give it that) whose remembered `network_policy`
    /// DENIES the very host its `params` would reconnect to. A real
    /// `connect_http` call under a denying policy can never produce a
    /// connected client in the first place (see
    /// `mcp_remote.rs::network_policy_denies_a_disallowed_http_host_before_connecting`),
    /// which is why this can't be expressed as a pure public-API
    /// integration test — the point under test is specifically whether
    /// `reconnect` reuses `self.network_policy` (this test) instead of
    /// `None` (what a revert would reintroduce, and what this test would
    /// then fail to catch as an error).
    #[tokio::test]
    async fn reconnect_denies_a_disallowed_host_before_reconnecting() {
        use std::sync::atomic::{AtomicBool, Ordering};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connected = Arc::new(AtomicBool::new(false));
        let connected2 = connected.clone();
        tokio::spawn(async move {
            if let Ok((mut sock, _)) = listener.accept().await {
                connected2.store(true, Ordering::SeqCst);
                let mut buf = [0u8; 1024];
                use tokio::io::AsyncReadExt;
                let _ = sock.read(&mut buf).await;
            }
        });
        let url = format!("http://127.0.0.1:{}/mcp", addr.port());
        let deny_policy = NetworkPolicy {
            enabled: true,
            allow_domains: vec![],
            deny_domains: vec!["127.0.0.1".to_string()],
        };
        let client = McpClient {
            conn: Conn::Http {
                client: reqwest::Client::new(),
                url: url.clone(),
                headers: HeaderMap::new(),
                session_id: None,
            },
            next_id: 0,
            params: McpConnectParams::Http {
                url: url.clone(),
                headers: BTreeMap::new(),
            },
            network_policy: Some(deny_policy),
            timeout: DEFAULT_MCP_TIMEOUT,
            elicitation_handler: Arc::new(HeadlessElicitationHandler),
            instructions: None,
            pending_notifications: std::sync::Mutex::new(Vec::new()),
        };

        let result = client.reconnect().await;
        assert!(
            result.is_err(),
            "reconnect must refuse to reconnect to a host its own remembered policy denies"
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(
            !connected.load(Ordering::SeqCst),
            "the denied host must never even be contacted on reconnect"
        );
    }
}