use std::future::Future;
use std::time::Duration;

use anyhow::Context;
use anyhow::Result;
use anyhow::bail;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCRequest;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_exec_server::ExecServerClient;
use codex_exec_server::HttpHeader;
use codex_exec_server::HttpRequestBodyDeltaNotification;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::HttpRequestResponse;
use codex_exec_server::InitializeParams;
use codex_exec_server::InitializeResponse;
use codex_exec_server::RemoteExecServerConnectArgs;
use futures::SinkExt;
use futures::StreamExt;
use pretty_assertions::assert_eq;
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_json::from_slice;
use serde_json::from_str;
use serde_json::from_value;
use serde_json::to_string;
use serde_json::to_value;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;

const CLIENT_NAME: &str = "test-exec-server-client";
const HTTP_REQUEST_METHOD: &str = "http/request";
const HTTP_REQUEST_BODY_DELTA_METHOD: &str = "http/request/bodyDelta";
const INITIALIZE_METHOD: &str = "initialize";
const INITIALIZED_METHOD: &str = "initialized";
const TEST_TIMEOUT: Duration = Duration::from_secs(5);
const HTTP_BODY_DELTA_CHANNEL_CAPACITY: u64 = 256;
const OVERFLOWING_BODY_DELTA_FRAMES: u64 = 1_024;

/// What this tests: the buffered HTTP helper always sends a buffered
/// `http/request`, even when a caller accidentally provides streaming flags.
#[tokio::test]
async fn http_request_forces_buffered_request_params() -> Result<()> {
    // Phase 1: start a fake WebSocket exec-server so the test covers the
    // public client connection path without depending on the HTTP runner.
    let server = spawn_scripted_exec_server(|mut peer| async move {
        // Phase 2: verify the buffered helper forces buffered mode before it
        // sends the JSON-RPC call.
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/buffered".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "ignored-stream-id".to_string(),
                stream_response: false,
            }
        );

        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: b"buffered".to_vec().into(),
            },
        )
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 3: call the buffered helper with streaming-only fields populated
    // and assert callers still receive the buffered response body.
    let response = timeout(
        TEST_TIMEOUT,
        client.http_request(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/buffered".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "ignored-stream-id".to_string(),
            stream_response: true,
        }),
    )
    .await
    .context("buffered http/request should complete")??;
    assert_eq!(
        response,
        HttpRequestResponse {
            status: 200,
            headers: Vec::new(),
            body: b"buffered".to_vec().into(),
        }
    );

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: streamed executor HTTP response frames are routed by the
/// client's generated request id, delivered in sequence, and concatenated by
/// the caller.
#[tokio::test]
async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_deltas() -> Result<()> {
    // Phase 1: script two requests. The caller supplies reusable ids, but the
    // client replaces them with connection-local ids on the wire.
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp".to_string(),
                headers: vec![HttpHeader {
                    name: "accept".to_string(),
                    value: "text/event-stream".to_string(),
                }],
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );

        // Phase 2: return headers first, then body notifications in the order
        // the public body stream should expose them.
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: vec![HttpHeader {
                    name: "content-type".to_string(),
                    value: "text/event-stream".to_string(),
                }],
                body: Vec::new().into(),
            },
        )
        .await?;
        for delta in [
            HttpRequestBodyDeltaNotification {
                request_id: "http-1".to_string(),
                seq: 1,
                delta: b"hello ".to_vec().into(),
                done: false,
                error: None,
            },
            HttpRequestBodyDeltaNotification {
                request_id: "http-1".to_string(),
                seq: 2,
                delta: b"world".to_vec().into(),
                done: false,
                error: None,
            },
            HttpRequestBodyDeltaNotification {
                request_id: "http-1".to_string(),
                seq: 3,
                delta: b"!".to_vec().into(),
                done: true,
                error: None,
            },
        ] {
            peer.write_body_delta(delta).await?;
        }

        // Phase 3: accept the next generated request id after EOF.
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/reuse".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-2".to_string(),
                stream_response: true,
            }
        );
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 204,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 4: start a streaming HTTP request through the public client API.
    let (response, mut body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp".to_string(),
            headers: vec![HttpHeader {
                name: "accept".to_string(),
                value: "text/event-stream".to_string(),
            }],
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;
    assert_eq!(
        response,
        HttpRequestResponse {
            status: 200,
            headers: vec![HttpHeader {
                name: "content-type".to_string(),
                value: "text/event-stream".to_string(),
            }],
            body: Vec::new().into(),
        }
    );

    // Phase 5: drain the body stream and verify the caller-visible byte order.
    let mut body = Vec::new();
    while let Some(chunk) = timeout(TEST_TIMEOUT, body_stream.recv())
        .await
        .context("http response body delta should arrive")??
    {
        body.extend_from_slice(&chunk);
    }
    assert_eq!(body, b"hello world!".to_vec());

    // Phase 6: start another stream through the public API to validate cleanup
    // after EOF without reaching into the client routing table.
    let (reuse_response, _reuse_body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/reuse".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("second streamed http/request should return headers")??;
    assert_eq!(
        reuse_response,
        HttpRequestResponse {
            status: 204,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: dropping a body stream with a queued terminal frame removes
/// the old route while the next stream gets a fresh generated id.
#[tokio::test]
async fn http_response_body_stream_drops_queued_terminal_before_next_generated_id() -> Result<()> {
    // Phase 1: send terminal EOF before the header response so the public body
    // stream starts with EOF already queued but unread.
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/queued-terminal".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        peer.write_body_delta(HttpRequestBodyDeltaNotification {
            request_id: "http-1".to_string(),
            seq: 1,
            delta: Vec::new().into(),
            done: true,
            error: None,
        })
        .await?;
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await?;

        // Phase 2: accept another stream after the client drops the unread
        // body. The second request receives a distinct generated id.
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/retry-queued-terminal".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-2".to_string(),
                stream_response: true,
            }
        );
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 204,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 3: drop the body stream without reading the queued EOF frame.
    let (response, body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/queued-terminal".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;
    assert_eq!(
        response,
        HttpRequestResponse {
            status: 200,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );
    drop(body_stream);

    // Phase 4: start another stream through the public API. The caller-provided
    // id is ignored, so the request uses the next generated route id.
    let params = HttpRequestParams {
        method: "GET".to_string(),
        url: "https://example.test/mcp/retry-queued-terminal".to_string(),
        headers: Vec::new(),
        body: None,
        timeout_ms: None,
        request_id: "caller-stream-id".to_string(),
        stream_response: false,
    };
    let (reuse_response, _reuse_body_stream) =
        timeout(TEST_TIMEOUT, client.http_request_stream(params))
            .await
            .context("second streamed http/request should return headers")??;
    assert_eq!(
        reuse_response,
        HttpRequestResponse {
            status: 204,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: cancelling a streaming HTTP request while it is waiting for
/// headers drops its route, and a later stream gets a fresh generated id.
#[tokio::test]
async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() -> Result<()> {
    // Phase 1: coordinate cancellation after the fake server observes the
    // first request but before it returns headers. The server later sends a
    // stale delta for the cancelled id before serving the fresh stream.
    let (request_seen_tx, request_seen_rx) = oneshot::channel();
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (_request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/cancel".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        request_seen_tx
            .send(())
            .expect("test should wait for the first request");

        // Phase 2: the next stream uses a new generated id. A late body delta
        // for the cancelled id is ignored by the client-side router.
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/retry-cancelled".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-2".to_string(),
                stream_response: true,
            }
        );
        peer.write_body_delta(HttpRequestBodyDeltaNotification {
            request_id: "http-1".to_string(),
            seq: 1,
            delta: b"stale".to_vec().into(),
            done: false,
            error: None,
        })
        .await?;
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await?;
        peer.write_body_delta(HttpRequestBodyDeltaNotification {
            request_id: "http-2".to_string(),
            seq: 1,
            delta: b"fresh".to_vec().into(),
            done: true,
            error: None,
        })
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 3: start a streaming request and abort the caller future while it
    // is blocked waiting for response headers.
    let client_for_request = client.clone();
    let stream_task = tokio::spawn(async move {
        let _ = client_for_request
            .http_request_stream(HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/cancel".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "caller-stream-id".to_string(),
                stream_response: false,
            })
            .await;
    });
    request_seen_rx
        .await
        .expect("server should observe the first http/request");
    stream_task.abort();
    let _ = stream_task.await;

    // Phase 4: start a new stream immediately. It receives only the fresh body
    // bytes for its generated id.
    let (response, mut body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/retry-cancelled".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("second streamed http/request should return headers")??;
    assert_eq!(
        response,
        HttpRequestResponse {
            status: 200,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );
    let mut body = Vec::new();
    while let Some(chunk) = timeout(TEST_TIMEOUT, body_stream.recv())
        .await
        .context("fresh http response body delta should arrive")??
    {
        body.extend_from_slice(&chunk);
    }
    assert_eq!(body, b"fresh".to_vec());

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: dropping a returned body stream before EOF removes its
/// route and prevents stale body deltas from reaching the next stream.
#[tokio::test]
async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()> {
    // Phase 1: script two requests. The first returns only headers; after the
    // client drops its body receiver, the server sends a stale body delta.
    let (body_dropped_tx, body_dropped_rx) = oneshot::channel();
    let (stale_delta_sent_tx, stale_delta_sent_rx) = oneshot::channel();
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/drop".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await?;
        body_dropped_rx
            .await
            .expect("test should drop the first body stream");
        peer.write_body_delta(HttpRequestBodyDeltaNotification {
            request_id: "http-1".to_string(),
            seq: 1,
            delta: b"stale".to_vec().into(),
            done: false,
            error: None,
        })
        .await?;
        stale_delta_sent_tx
            .send(())
            .expect("test should wait for the stale delta");

        // Phase 2: accept the next request with a new generated id. The new
        // stream must receive only fresh body bytes.
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/retry-dropped".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-2".to_string(),
                stream_response: true,
            }
        );
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await?;
        peer.write_body_delta(HttpRequestBodyDeltaNotification {
            request_id: "http-2".to_string(),
            seq: 1,
            delta: b"fresh".to_vec().into(),
            done: true,
            error: None,
        })
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 3: receive headers for the first stream, then drop the body stream
    // without reading any body frames.
    let (response, body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/drop".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;
    assert_eq!(
        response,
        HttpRequestResponse {
            status: 200,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );
    drop(body_stream);
    body_dropped_tx
        .send(())
        .expect("server should wait for the body stream drop");
    stale_delta_sent_rx
        .await
        .expect("server should send one stale nonterminal delta");

    // Phase 4: start the next stream immediately. The caller-provided id is
    // ignored, and the fresh generated id isolates it from stale bytes.
    let (reuse_response, mut reuse_body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/retry-dropped".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("second streamed http/request should return headers")??;
    assert_eq!(
        reuse_response,
        HttpRequestResponse {
            status: 200,
            headers: Vec::new(),
            body: Vec::new().into(),
        }
    );
    let mut body = Vec::new();
    while let Some(chunk) = timeout(TEST_TIMEOUT, reuse_body_stream.recv())
        .await
        .context("fresh http response body delta should arrive")??
    {
        body.extend_from_slice(&chunk);
    }
    assert_eq!(body, b"fresh".to_vec());

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: an in-flight streamed HTTP body is failed when the shared
/// JSON-RPC transport disconnects before a terminal body frame.
#[tokio::test]
async fn http_response_body_stream_fails_when_transport_disconnects() -> Result<()> {
    // Phase 1: return response headers for a streaming request, then drop the
    // fake server transport without sending EOF.
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/disconnect".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 2: start a streaming HTTP request and receive headers.
    let (_response, mut body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/disconnect".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;

    // Phase 3: assert transport disconnect wakes the body stream with a
    // terminal error instead of hanging.
    let error = timeout(TEST_TIMEOUT, body_stream.recv())
        .await
        .context("disconnect should wake http body stream")?
        .expect_err("disconnect should fail the http body stream");
    let error_message = error.to_string();
    assert_eq!(
        error_message.starts_with(
            "exec-server protocol error: http response stream `http-1` failed: exec-server transport disconnected"
        ),
        true
    );

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: transport disconnect still records a terminal stream
/// failure even when the client-side body-delta queue is already full.
#[tokio::test]
async fn http_response_body_stream_reports_disconnect_when_queue_is_full() -> Result<()> {
    // Phase 1: fill the queued body-delta route exactly to capacity before the
    // response headers arrive, then drop the transport without sending EOF.
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/disconnect-full-queue".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        for seq in 1..=HTTP_BODY_DELTA_CHANNEL_CAPACITY {
            peer.write_body_delta(HttpRequestBodyDeltaNotification {
                request_id: "http-1".to_string(),
                seq,
                delta: b"x".to_vec().into(),
                done: false,
                error: None,
            })
            .await?;
        }
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 2: start the streaming request and receive headers while the
    // queue is already full.
    let (_response, mut body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/disconnect-full-queue".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;

    // Phase 3: drain the queued chunks and assert the transport disconnect is
    // still reported as an error rather than a clean EOF.
    let mut chunks = 0;
    let error = loop {
        match timeout(TEST_TIMEOUT, body_stream.recv())
            .await
            .context("disconnect should wake the full queued body stream")?
        {
            Ok(Some(_chunk)) => {
                chunks += 1;
            }
            Ok(None) => bail!("disconnect with a full queue should not look like clean EOF"),
            Err(error) => break error,
        }
    };
    assert_eq!(
        (
            chunks,
            error
                .to_string()
                .starts_with(
                    "exec-server protocol error: http response stream `http-1` failed: exec-server transport disconnected",
                ),
        ),
        (HTTP_BODY_DELTA_CHANNEL_CAPACITY as usize, true)
    );

    drop(client);
    server.finish().await?;
    Ok(())
}

/// What this tests: body-delta backpressure closes the public body stream as
/// an error rather than letting callers accept a truncated body as clean EOF.
#[tokio::test]
async fn http_response_body_stream_reports_backpressure_truncation() -> Result<()> {
    // Phase 1: send enough body frames before headers to overflow the bounded
    // client-side route while the public request future is still pending.
    let (finish_tx, finish_rx) = oneshot::channel();
    let server = spawn_scripted_exec_server(|mut peer| async move {
        let (request_id, params) = peer.read_http_request().await?;
        assert_eq!(
            params,
            HttpRequestParams {
                method: "GET".to_string(),
                url: "https://example.test/mcp/backpressure".to_string(),
                headers: Vec::new(),
                body: None,
                timeout_ms: None,
                request_id: "http-1".to_string(),
                stream_response: true,
            }
        );
        for seq in 1..=OVERFLOWING_BODY_DELTA_FRAMES {
            peer.write_body_delta(HttpRequestBodyDeltaNotification {
                request_id: "http-1".to_string(),
                seq,
                delta: b"x".to_vec().into(),
                done: false,
                error: None,
            })
            .await?;
        }
        peer.write_response(
            request_id,
            HttpRequestResponse {
                status: 200,
                headers: Vec::new(),
                body: Vec::new().into(),
            },
        )
        .await?;

        // Phase 2: keep the transport connected so the body stream reports the
        // backpressure failure rather than a disconnect.
        finish_rx.await.expect("test should finish server task");
        Ok(())
    })
    .await?;
    let client = server.connect_client().await?;

    // Phase 3: start the streaming request; the server overfills the route
    // before returning the body stream to this consumer.
    let (_response, mut body_stream) = timeout(
        TEST_TIMEOUT,
        client.http_request_stream(HttpRequestParams {
            method: "GET".to_string(),
            url: "https://example.test/mcp/backpressure".to_string(),
            headers: Vec::new(),
            body: None,
            timeout_ms: None,
            request_id: "caller-stream-id".to_string(),
            stream_response: false,
        }),
    )
    .await
    .context("streamed http/request should return headers")??;

    // Phase 4: drain queued chunks and assert the truncated stream ends in an
    // explicit error, not a clean EOF.
    let mut chunks = 0;
    let error = loop {
        match timeout(TEST_TIMEOUT, body_stream.recv())
            .await
            .context("backpressure should close http body stream")?
        {
            Ok(Some(_chunk)) => {
                chunks += 1;
            }
            Ok(None) => bail!("backpressure truncation should not look like clean EOF"),
            Err(error) => break error,
        }
    };
    assert_eq!(
        (
            chunks < OVERFLOWING_BODY_DELTA_FRAMES as usize,
            error.to_string(),
        ),
        (
            true,
            "exec-server protocol error: http response stream `http-1` failed: body delta channel filled before delivery".to_string(),
        )
    );

    finish_tx
        .send(())
        .expect("server task should wait for test completion");
    drop(client);
    server.finish().await?;
    Ok(())
}

/// Fake WebSocket exec-server used by the integration tests.
///
/// The helper exercises `ExecServerClient::connect_websocket`, including the
/// initialize handshake, while each test controls the exact JSON-RPC traffic
/// that follows.
struct ScriptedExecServer {
    websocket_url: String,
    task: JoinHandle<Result<()>>,
}

impl ScriptedExecServer {
    /// Connects the public exec-server client to this fake WebSocket endpoint.
    async fn connect_client(&self) -> Result<ExecServerClient> {
        ExecServerClient::connect_websocket(RemoteExecServerConnectArgs::new(
            self.websocket_url.clone(),
            CLIENT_NAME.to_string(),
        ))
        .await
        .context("client should connect to fake exec-server")
    }

    /// Waits for the scripted fake server to finish.
    async fn finish(self) -> Result<()> {
        self.task
            .await
            .context("fake exec-server task should join")??;
        Ok(())
    }
}

/// Starts a fake exec-server that accepts one WebSocket client.
async fn spawn_scripted_exec_server<F, Fut>(script: F) -> Result<ScriptedExecServer>
where
    F: FnOnce(JsonRpcPeer) -> Fut + Send + 'static,
    Fut: Future<Output = Result<()>> + Send + 'static,
{
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .context("fake exec-server should bind")?;
    let websocket_url = format!("ws://{}", listener.local_addr()?);
    let task = tokio::spawn(async move {
        let (stream, _) = timeout(TEST_TIMEOUT, listener.accept())
            .await
            .context("fake exec-server should accept a client")??;
        let websocket = accept_async(stream)
            .await
            .context("fake exec-server websocket handshake should complete")?;
        let mut peer = JsonRpcPeer { websocket };
        peer.complete_initialize().await?;
        script(peer).await
    });
    Ok(ScriptedExecServer {
        websocket_url,
        task,
    })
}

/// JSON-RPC peer for the fake exec-server WebSocket.
struct JsonRpcPeer {
    websocket: WebSocketStream<TcpStream>,
}

impl JsonRpcPeer {
    /// Completes and validates the client initialize handshake.
    async fn complete_initialize(&mut self) -> Result<()> {
        let request = self.read_request(INITIALIZE_METHOD).await?;
        let params: InitializeParams = decode_request_params(&request)?;
        assert_eq!(
            params,
            InitializeParams {
                client_name: CLIENT_NAME.to_string(),
                resume_session_id: None,
            }
        );
        self.write_response(
            request.id,
            InitializeResponse {
                session_id: "session-1".to_string(),
            },
        )
        .await?;
        self.read_notification(INITIALIZED_METHOD).await?;
        Ok(())
    }

    /// Reads one typed `http/request` call from the client.
    async fn read_http_request(&mut self) -> Result<(RequestId, HttpRequestParams)> {
        let request = self.read_request(HTTP_REQUEST_METHOD).await?;
        let params = decode_request_params(&request)?;
        Ok((request.id, params))
    }

    /// Reads a JSON-RPC request and validates its method.
    async fn read_request(&mut self, expected_method: &str) -> Result<JSONRPCRequest> {
        let message = self.read_message().await?;
        let JSONRPCMessage::Request(request) = message else {
            bail!("expected JSON-RPC request `{expected_method}`, got {message:?}");
        };
        if request.method != expected_method {
            bail!(
                "expected JSON-RPC request `{expected_method}`, got `{}`",
                request.method
            );
        }
        Ok(request)
    }

    /// Reads a JSON-RPC notification and validates its method.
    async fn read_notification(&mut self, expected_method: &str) -> Result<JSONRPCNotification> {
        let message = self.read_message().await?;
        let JSONRPCMessage::Notification(notification) = message else {
            bail!("expected JSON-RPC notification `{expected_method}`, got {message:?}");
        };
        if notification.method != expected_method {
            bail!(
                "expected JSON-RPC notification `{expected_method}`, got `{}`",
                notification.method
            );
        }
        Ok(notification)
    }

    /// Sends a successful JSON-RPC response.
    async fn write_response<T>(&mut self, id: RequestId, result: T) -> Result<()>
    where
        T: Serialize,
    {
        self.write_message(JSONRPCMessage::Response(JSONRPCResponse {
            id,
            result: to_value(result)?,
        }))
        .await
    }

    /// Sends one streamed HTTP body notification.
    async fn write_body_delta(&mut self, delta: HttpRequestBodyDeltaNotification) -> Result<()> {
        self.write_message(JSONRPCMessage::Notification(JSONRPCNotification {
            method: HTTP_REQUEST_BODY_DELTA_METHOD.to_string(),
            params: Some(to_value(delta)?),
        }))
        .await
    }

    /// Reads one WebSocket JSON-RPC message.
    async fn read_message(&mut self) -> Result<JSONRPCMessage> {
        let message = timeout(TEST_TIMEOUT, self.websocket.next())
            .await
            .context("timed out waiting for JSON-RPC message")?
            .context("client websocket closed before JSON-RPC message arrived")?
            .context("failed to read websocket message")?;
        match message {
            Message::Text(text) => from_str(text.as_ref()).context("text JSON-RPC"),
            Message::Binary(bytes) => from_slice(bytes.as_ref()).context("binary JSON-RPC"),
            Message::Close(frame) => bail!("client websocket closed: {frame:?}"),
            other => bail!("expected text or binary JSON-RPC message, got {other:?}"),
        }
    }

    /// Writes one WebSocket JSON-RPC message.
    async fn write_message(&mut self, message: JSONRPCMessage) -> Result<()> {
        let encoded = to_string(&message)?;
        timeout(
            TEST_TIMEOUT,
            self.websocket.send(Message::Text(encoded.into())),
        )
        .await
        .context("timed out writing JSON-RPC message")?
        .context("failed to write JSON-RPC message")
    }
}

/// Decodes a request params object into its typed protocol payload.
fn decode_request_params<T>(request: &JSONRPCRequest) -> Result<T>
where
    T: DeserializeOwned,
{
    let params = request
        .params
        .clone()
        .context("JSON-RPC request should include params")?;
    from_value(params).context("JSON-RPC request params should decode")
}