supercode-core 0.2.1

//! P5-2 (COMPOSABLE-HARNESS-DESIGN.md §2 module 15 D7 row 2 "remote
//! HTTP/SSE"): remote MCP transport tests. Every test spins up a minimal
//! raw-HTTP loopback server (the same idiom `provider.rs`/
//! `p4c_tool_new_smalls.rs` already use) — NEVER a real external MCP server
//! (`live-agent-test-safety`).

use std::collections::BTreeMap;

use supercode::mcp::McpClient;

async fn spawn_http_server(
    handler: impl Fn(&str) -> String + Send + Sync + 'static,
) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let handler = std::sync::Arc::new(handler);
    let handle = tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let handler = handler.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 65536];
                let n = match sock.read(&mut buf).await {
                    Ok(n) => n,
                    Err(_) => return,
                };
                let text = String::from_utf8_lossy(&buf[..n]);
                let body = text.split("\r\n\r\n").nth(1).unwrap_or("");
                let response = handler(body);
                let _ = sock.write_all(response.as_bytes()).await;
                let _ = sock.flush().await;
            });
        }
    });
    (addr, handle)
}

fn http_json_200(body: &str) -> String {
    format!(
        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        body.len(),
        body
    )
}

fn json_rpc_id(body: &str) -> i64 {
    let v: serde_json::Value = serde_json::from_str(body).unwrap_or_default();
    v.get("id").and_then(serde_json::Value::as_i64).unwrap_or(0)
}

fn json_rpc_method(body: &str) -> String {
    let v: serde_json::Value = serde_json::from_str(body).unwrap_or_default();
    v.get("method")
        .and_then(serde_json::Value::as_str)
        .unwrap_or_default()
        .to_string()
}

// ---- http (streamable, non-streaming) transport ---------------------------

#[tokio::test]
async fn http_transport_connects_lists_and_calls_a_tool() {
    let (addr, _server) = spawn_http_server(|body| {
        let id = json_rpc_id(body);
        let method = json_rpc_method(body);
        let result = match method.as_str() {
            "initialize" => {
                serde_json::json!({"protocolVersion":"2025-06-18","serverInfo":{"name":"fake-http"},"instructions":"be nice"})
            }
            "notifications/initialized" => return http_json_200(""),
            "tools/list" => serde_json::json!({"tools":[
                {"name":"echo","description":"echo back","inputSchema":{"type":"object"}}
            ]}),
            "tools/call" => serde_json::json!({"content":[{"type":"text","text":"echo: hi"}],"isError":false}),
            _ => serde_json::json!({}),
        };
        http_json_200(&serde_json::json!({"jsonrpc":"2.0","id":id,"result":result}).to_string())
    })
    .await;

    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let mut client = McpClient::connect_http(&url, &BTreeMap::new(), None)
        .await
        .unwrap();
    assert_eq!(client.instructions.as_deref(), Some("be nice"));

    let tools = client.list_tools().await.unwrap();
    assert_eq!(tools.len(), 1);
    assert_eq!(tools[0].name, "echo");

    let out = client
        .call_tool("echo", serde_json::json!({"text": "hi"}))
        .await
        .unwrap();
    assert_eq!(out, "echo: hi");
}

#[tokio::test]
async fn http_transport_sends_configured_headers() {
    let seen_auth: std::sync::Arc<std::sync::Mutex<Option<String>>> =
        std::sync::Arc::new(std::sync::Mutex::new(None));
    let seen_auth2 = seen_auth.clone();
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let _server = tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let seen_auth2 = seen_auth2.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 65536];
                let n = sock.read(&mut buf).await.unwrap_or(0);
                let text = String::from_utf8_lossy(&buf[..n]).to_string();
                for line in text.lines() {
                    // `HeaderName` always serializes lowercase on the wire
                    // regardless of the case it was constructed with — match
                    // the NAME case-insensitively, but keep the value as-is.
                    if let Some((name, value)) = line.split_once(':') {
                        if name.eq_ignore_ascii_case("authorization") {
                            *seen_auth2.lock().unwrap() = Some(value.trim().to_string());
                        }
                    }
                }
                let body = text.split("\r\n\r\n").nth(1).unwrap_or("");
                let id = json_rpc_id(body);
                let resp = http_json_200(
                    &serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"f"}}}).to_string(),
                );
                let _ = sock.write_all(resp.as_bytes()).await;
                let _ = sock.flush().await;
            });
        }
    });
    let mut headers = BTreeMap::new();
    headers.insert(
        "Authorization".to_string(),
        "Bearer secret-token".to_string(),
    );
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let _client = McpClient::connect_http(&url, &headers, None).await.unwrap();
    assert_eq!(
        seen_auth.lock().unwrap().as_deref(),
        Some("Bearer secret-token"),
        "the configured Authorization header must reach the server"
    );
}

#[tokio::test]
async fn http_transport_reports_a_clean_error_on_embedded_server_request() {
    // A server that tries to elicit mid-call over the non-streaming http
    // transport — this client must fail closed with a clear error, never
    // hang or silently drop the elicitation.
    let (addr, _server) = spawn_http_server(|body| {
        let method = json_rpc_method(body);
        if method == "initialize" {
            let id = json_rpc_id(body);
            return http_json_200(
                &serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"f"}}}).to_string(),
            );
        }
        if method == "notifications/initialized" {
            return http_json_200("");
        }
        // tools/call: respond with an embedded elicitation REQUEST instead
        // of the tool's result.
        http_json_200(
            &serde_json::json!({"jsonrpc":"2.0","id":999,"method":"elicitation/create","params":{"message":"need input"}}).to_string(),
        )
    })
    .await;
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let mut client = McpClient::connect_http(&url, &BTreeMap::new(), None)
        .await
        .unwrap();
    let err = client
        .call_tool("whatever", serde_json::json!({}))
        .await
        .unwrap_err();
    assert!(
        err.to_string().contains("stdio or sse"),
        "error should name the supported transports: {err}"
    );
}

#[tokio::test]
async fn network_policy_denies_a_disallowed_http_host_before_connecting() {
    use std::sync::atomic::{AtomicBool, Ordering};
    use tokio::io::AsyncReadExt;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let connected = std::sync::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];
            let _ = sock.read(&mut buf).await;
        }
    });
    let policy = supercode::tools::NetworkPolicy {
        enabled: true,
        allow_domains: vec![],
        deny_domains: vec!["127.0.0.1".to_string()],
    };
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let result = McpClient::connect_http(&url, &BTreeMap::new(), Some(&policy)).await;
    assert!(result.is_err(), "a denied host must not connect");
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    assert!(
        !connected.load(Ordering::SeqCst),
        "the denied host must never even be contacted"
    );
}

#[tokio::test]
async fn reconnect_reapplies_the_original_network_policy_and_still_succeeds_when_allowed() {
    // No-over-block confirmation (Fable-5 review, latent-SSRF-landmine fix):
    // `reconnect()` now re-runs `connect_http`'s pre-connect NetworkPolicy
    // check using the client's OWN remembered policy instead of `None` —
    // this proves that doesn't regress the legitimate case: a reconnect to
    // a still-allowed host must keep working exactly as before. The
    // fail-closed half (a denied host must be refused on reconnect) is
    // pinned by `mcp::tests::reconnect_denies_a_disallowed_host_before_reconnecting`
    // in mcp.rs's own unit tests (needs private-field access to construct
    // an already-"connected" client under a denying policy, since a real
    // `connect_http` under a denying policy can never succeed in the first
    // place — see that test's doc comment for why it lives there).
    let (addr, _server) = spawn_http_server(|body| {
        let id = json_rpc_id(body);
        let method = json_rpc_method(body);
        let result = match method.as_str() {
            "initialize" => {
                serde_json::json!({"protocolVersion":"2025-06-18","serverInfo":{"name":"fake-http"}})
            }
            "notifications/initialized" => return http_json_200(""),
            _ => serde_json::json!({}),
        };
        http_json_200(&serde_json::json!({"jsonrpc":"2.0","id":id,"result":result}).to_string())
    })
    .await;
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let policy = supercode::tools::NetworkPolicy {
        enabled: true,
        allow_domains: vec!["127.0.0.1".to_string()],
        deny_domains: vec![],
    };
    let client = McpClient::connect_http(&url, &BTreeMap::new(), Some(&policy))
        .await
        .unwrap();
    let _reconnected = client
        .reconnect()
        .await
        .expect("reconnect under an allowing policy must still succeed");
}

// ---- hardening: response/frame/resource size caps --------------------------

#[tokio::test]
async fn http_response_body_over_the_cap_errors_named_not_oom_or_hang() {
    // A hostile/misbehaving configured MCP server declares a Content-Length
    // far past MCP_MAX_RESPONSE_BYTES — this must be rejected by the fast
    // pre-check BEFORE this client tries to buffer anything close to that
    // size (never OOM, never hang).
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            tokio::spawn(async move {
                let mut buf = vec![0u8; 65536];
                let n = sock.read(&mut buf).await.unwrap_or(0);
                let text = String::from_utf8_lossy(&buf[..n]).to_string();
                let body = text.split("\r\n\r\n").nth(1).unwrap_or("");
                let method = json_rpc_method(body);
                if method == "initialize" {
                    let id = json_rpc_id(body);
                    let resp = http_json_200(
                        &serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"f"}}}).to_string(),
                    );
                    let _ = sock.write_all(resp.as_bytes()).await;
                    let _ = sock.flush().await;
                    return;
                }
                let oversized = supercode::mcp::MCP_MAX_RESPONSE_BYTES + 1;
                let headers = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {oversized}\r\nConnection: close\r\n\r\n"
                );
                let _ = sock.write_all(headers.as_bytes()).await;
                // Never actually send `oversized` bytes — the pre-check on
                // the declared Content-Length must reject this before this
                // client reads (or this server writes) anything close to
                // that much data.
                let _ = sock.write_all(b"{}").await;
                let _ = sock.flush().await;
            });
        }
    });
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let mut client = McpClient::connect_http(&url, &BTreeMap::new(), None)
        .await
        .unwrap();
    let result = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.call_tool("whatever", serde_json::json!({})),
    )
    .await
    .expect("must not hang");
    let err = result.expect_err("an oversized declared body must error, never succeed");
    assert!(
        err.to_string().contains("exceeds max"),
        "error should name the cap: {err}"
    );
}

#[tokio::test]
async fn http_response_streamed_body_over_the_cap_errors_named_not_oom_or_hang() {
    // Same cap, but a hostile server that OMITS Content-Length (chunked,
    // unbounded from the client's perspective) and actually streams past
    // the cap — proves the streaming enforcement (not just the
    // Content-Length pre-check) also fails closed, without this client
    // ever buffering past MCP_MAX_RESPONSE_BYTES.
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            tokio::spawn(async move {
                let mut buf = vec![0u8; 65536];
                let n = sock.read(&mut buf).await.unwrap_or(0);
                let text = String::from_utf8_lossy(&buf[..n]).to_string();
                let body = text.split("\r\n\r\n").nth(1).unwrap_or("");
                let method = json_rpc_method(body);
                if method == "initialize" {
                    let id = json_rpc_id(body);
                    let resp = http_json_200(
                        &serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"protocolVersion":"2025-06-18","serverInfo":{"name":"f"}}}).to_string(),
                    );
                    let _ = sock.write_all(resp.as_bytes()).await;
                    let _ = sock.flush().await;
                    return;
                }
                let headers =
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nTransfer-Encoding: chunked\r\n\r\n";
                let _ = sock.write_all(headers.as_bytes()).await;
                let chunk = vec![b'x'; 1024 * 1024]; // 1 MiB per chunk
                let mut framed = format!("{:x}\r\n", chunk.len()).into_bytes();
                framed.extend_from_slice(&chunk);
                framed.extend_from_slice(b"\r\n");
                let chunks_needed = supercode::mcp::MCP_MAX_RESPONSE_BYTES / (1024 * 1024) + 2;
                for _ in 0..chunks_needed {
                    if sock.write_all(&framed).await.is_err() {
                        break;
                    }
                }
                let _ = sock.write_all(b"0\r\n\r\n").await;
                let _ = sock.flush().await;
            });
        }
    });
    let url = format!("http://127.0.0.1:{}/mcp", addr.port());
    let mut client = McpClient::connect_http(&url, &BTreeMap::new(), None)
        .await
        .unwrap();
    let result = tokio::time::timeout(
        std::time::Duration::from_secs(30),
        client.call_tool("whatever", serde_json::json!({})),
    )
    .await
    .expect("must not hang");
    let err = result.expect_err("an oversized streamed body must error, never succeed");
    assert!(
        err.to_string().contains("exceeded max"),
        "error should name the cap: {err}"
    );
}

// ---- sse (legacy http+sse) transport ---------------------------------------

/// A raw SSE + message-post loopback server: the SSE `GET` connection
/// writes headers, the `endpoint` event, then relays whatever is pushed on
/// `push_rx` (chunk-framed) until the test drops the sender. Every `POST`
/// (a client->server message) is forwarded to `post_tx`. Returns the
/// listener address plus both channel handles so a test can drive a
/// request/response sequence, including injecting an out-of-band
/// notification or server-initiated request between a client request and
/// its eventual response.
async fn spawn_sse_server() -> (
    std::net::SocketAddr,
    tokio::sync::mpsc::UnboundedSender<String>,
    tokio::sync::mpsc::UnboundedReceiver<String>,
) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (push_tx, push_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
    let (post_tx, post_rx) = tokio::sync::mpsc::unbounded_channel::<String>();

    // A single accept loop that dispatches EACH connection to its own task —
    // the long-lived GET/SSE connection must never block accepting the
    // separate POST connections the client makes for its own requests.
    let post_tx = std::sync::Arc::new(post_tx);
    // `push_rx` is only ever consumed by the (single) GET/SSE connection —
    // move it into that connection's task the first time a GET arrives.
    let push_rx = std::sync::Arc::new(tokio::sync::Mutex::new(Some(push_rx)));
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let post_tx = post_tx.clone();
            let push_rx = push_rx.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 65536];
                let n = match sock.read(&mut buf).await {
                    Ok(n) => n,
                    Err(_) => return,
                };
                let text = String::from_utf8_lossy(&buf[..n]).to_string();
                let first_line = text.lines().next().unwrap_or("").to_string();
                if first_line.starts_with("GET") {
                    let Some(mut push_rx) = push_rx.lock().await.take() else {
                        return;
                    };
                    let headers = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n";
                    let _ = sock.write_all(headers.as_bytes()).await;
                    write_chunk(&mut sock, "event: endpoint\ndata: /messages\n\n").await;
                    while let Some(payload) = push_rx.recv().await {
                        write_chunk(&mut sock, &payload).await;
                    }
                } else if first_line.starts_with("POST") {
                    let body = text.split("\r\n\r\n").nth(1).unwrap_or("").to_string();
                    let _ = post_tx.send(body);
                    let resp =
                        "HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
                    let _ = sock.write_all(resp.as_bytes()).await;
                    let _ = sock.flush().await;
                }
            });
        }
    });

    (addr, push_tx, post_rx)
}

async fn write_chunk(sock: &mut tokio::net::TcpStream, data: &str) {
    use tokio::io::AsyncWriteExt;
    let framed = format!("{:x}\r\n{data}\r\n", data.len());
    let _ = sock.write_all(framed.as_bytes()).await;
    let _ = sock.flush().await;
}

/// Complete the `initialize` handshake against a [`spawn_sse_server`]
/// backend: waits for the POST, replies over the push channel, and drains
/// the client's follow-up `notifications/initialized` notification.
async fn answer_initialize(
    push_tx: &tokio::sync::mpsc::UnboundedSender<String>,
    post_rx: &mut tokio::sync::mpsc::UnboundedReceiver<String>,
    instructions: Option<&str>,
) {
    let body = post_rx.recv().await.unwrap();
    assert_eq!(json_rpc_method(&body), "initialize");
    let id = json_rpc_id(&body);
    let mut result =
        serde_json::json!({"protocolVersion":"2025-06-18","serverInfo":{"name":"sse-fake"}});
    if let Some(instr) = instructions {
        result["instructions"] = serde_json::Value::String(instr.to_string());
    }
    let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"result":result});
    push_tx
        .send(format!("event: message\ndata: {msg}\n\n"))
        .unwrap();
    let notif = post_rx.recv().await.unwrap();
    assert_eq!(json_rpc_method(&notif), "notifications/initialized");
}

#[tokio::test]
async fn sse_transport_discovers_endpoint_and_completes_initialize() {
    let (addr, push_tx, mut post_rx) = spawn_sse_server().await;
    let url = format!("http://127.0.0.1:{}/sse", addr.port());
    let connect_task =
        tokio::spawn(async move { McpClient::connect_sse(&url, &BTreeMap::new(), None).await });
    answer_initialize(&push_tx, &mut post_rx, Some("sse instructions")).await;
    let client = connect_task.await.unwrap().unwrap();
    assert_eq!(client.instructions.as_deref(), Some("sse instructions"));
}

#[tokio::test]
async fn sse_transport_lists_and_calls_a_tool() {
    let (addr, push_tx, mut post_rx) = spawn_sse_server().await;
    let url = format!("http://127.0.0.1:{}/sse", addr.port());
    let connect_task =
        tokio::spawn(async move { McpClient::connect_sse(&url, &BTreeMap::new(), None).await });
    answer_initialize(&push_tx, &mut post_rx, None).await;
    let mut client = connect_task.await.unwrap().unwrap();

    let list_task = tokio::spawn(async move {
        let tools = client.list_tools().await.unwrap();
        (client, tools)
    });
    let body = post_rx.recv().await.unwrap();
    assert_eq!(json_rpc_method(&body), "tools/list");
    let id = json_rpc_id(&body);
    let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"tools":[{"name":"echo","description":"d","inputSchema":{"type":"object"}}]}});
    push_tx
        .send(format!("event: message\ndata: {msg}\n\n"))
        .unwrap();
    let (mut client, tools) = list_task.await.unwrap();
    assert_eq!(tools.len(), 1);
    assert_eq!(tools[0].name, "echo");

    let call_task =
        tokio::spawn(async move { client.call_tool("echo", serde_json::json!({})).await });
    let body = post_rx.recv().await.unwrap();
    let id = json_rpc_id(&body);
    let msg = serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"content":[{"type":"text","text":"echoed"}],"isError":false}});
    push_tx
        .send(format!("event: message\ndata: {msg}\n\n"))
        .unwrap();
    let out = call_task.await.unwrap().unwrap();
    assert_eq!(out, "echoed");
}

#[tokio::test]
async fn sse_transport_answers_a_server_initiated_elicitation_request() {
    // The server sends an `elicitation/create` REQUEST (id + method) mid
    // `tools/call`; the client's headless-default handler must decline it
    // and keep waiting for the tool's own eventual response — proving the
    // bidirectional read loop handles a server->client request without
    // getting stuck or corrupting the pending client request's response.
    let (addr, push_tx, mut post_rx) = spawn_sse_server().await;
    let url = format!("http://127.0.0.1:{}/sse", addr.port());
    let connect_task =
        tokio::spawn(async move { McpClient::connect_sse(&url, &BTreeMap::new(), None).await });
    answer_initialize(&push_tx, &mut post_rx, None).await;
    let mut client = connect_task.await.unwrap().unwrap();

    let call_task = tokio::spawn(async move {
        let out = client.call_tool("needs_input", serde_json::json!({})).await;
        (client, out)
    });

    let body = post_rx.recv().await.unwrap();
    assert_eq!(json_rpc_method(&body), "tools/call");
    let call_id = json_rpc_id(&body);
    let elicit = serde_json::json!({"jsonrpc":"2.0","id":4242,"method":"elicitation/create","params":{"message":"need a value","requestedSchema":{}}});
    push_tx
        .send(format!("event: message\ndata: {elicit}\n\n"))
        .unwrap();

    // The client must reply to it (declining, headless-default) via a POST
    // BEFORE the original tools/call is answered.
    let reply_body = post_rx.recv().await.unwrap();
    let reply: serde_json::Value = serde_json::from_str(&reply_body).unwrap();
    assert_eq!(reply["id"], 4242);
    assert_eq!(reply["result"]["action"], "decline");

    let result = serde_json::json!({"jsonrpc":"2.0","id":call_id,"result":{"content":[{"type":"text","text":"done"}],"isError":false}});
    push_tx
        .send(format!("event: message\ndata: {result}\n\n"))
        .unwrap();
    let (_client, out) = call_task.await.unwrap();
    assert_eq!(out.unwrap(), "done");
}

#[tokio::test]
async fn sse_transport_logs_a_notification_and_keeps_waiting() {
    let (addr, push_tx, mut post_rx) = spawn_sse_server().await;
    let url = format!("http://127.0.0.1:{}/sse", addr.port());
    let connect_task =
        tokio::spawn(async move { McpClient::connect_sse(&url, &BTreeMap::new(), None).await });
    answer_initialize(&push_tx, &mut post_rx, None).await;
    let mut client = connect_task.await.unwrap().unwrap();

    let list_task = tokio::spawn(async move {
        let resources = client.list_resources().await.unwrap();
        (client, resources)
    });
    let body = post_rx.recv().await.unwrap();
    let id = json_rpc_id(&body);

    let notif = serde_json::json!({"jsonrpc":"2.0","method":"notifications/resources/updated","params":{"uri":"file:///a"}});
    push_tx
        .send(format!("event: message\ndata: {notif}\n\n"))
        .unwrap();
    let result = serde_json::json!({"jsonrpc":"2.0","id":id,"result":{"resources":[]}});
    push_tx
        .send(format!("event: message\ndata: {result}\n\n"))
        .unwrap();

    let (client, resources) = list_task.await.unwrap();
    assert!(resources.is_empty());
    let pending = client.take_pending_notifications();
    assert_eq!(pending.len(), 1);
    assert_eq!(pending[0]["method"], "notifications/resources/updated");
}

#[tokio::test]
async fn sse_stream_with_no_delimiter_over_the_cap_errors_named_not_unbounded() {
    // A hostile/misbehaving server pushes SSE bytes that never contain a
    // terminating blank line, well past MCP_MAX_SSE_FRAME_BYTES — the
    // reader task's SseLineAccumulator must error (fail-closed, named) and
    // tear the connection down instead of growing its buffer without
    // bound or hanging the pending request forever.
    let (addr, push_tx, mut post_rx) = spawn_sse_server().await;
    let url = format!("http://127.0.0.1:{}/sse", addr.port());
    let connect_task =
        tokio::spawn(async move { McpClient::connect_sse(&url, &BTreeMap::new(), None).await });
    answer_initialize(&push_tx, &mut post_rx, None).await;
    let mut client = connect_task.await.unwrap().unwrap();

    let call_task =
        tokio::spawn(async move { client.call_tool("whatever", serde_json::json!({})).await });
    let _tools_call_body = post_rx.recv().await.unwrap();

    // Never send "\n\n" — an ever-growing single un-terminated "frame".
    let chunk = "x".repeat(1024 * 1024); // 1 MiB, no blank-line terminator
    let chunks_needed = supercode::mcp::MCP_MAX_SSE_FRAME_BYTES / (1024 * 1024) + 2;
    for _ in 0..chunks_needed {
        if push_tx.send(chunk.clone()).is_err() {
            break;
        }
    }

    let result = tokio::time::timeout(std::time::Duration::from_secs(30), call_task)
        .await
        .expect("must not hang")
        .unwrap();
    let err = result.expect_err("an oversized unterminated sse frame must error, not hang");
    assert!(
        err.to_string().contains("exceeded max"),
        "error should name the cap: {err}"
    );
}