github-copilot-sdk 1.0.29

#![cfg(feature = "test-support")]
#![allow(clippy::unwrap_used)]

use github_copilot_sdk::test_support::{JsonRpcClient, JsonRpcNotification, JsonRpcRequest};
use github_copilot_sdk::{ErrorKind, ProtocolErrorKind};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, duplex};
use tokio::sync::{broadcast, mpsc};

/// Write a Content-Length framed JSON-RPC message to a writer.
async fn write_framed(writer: &mut (impl AsyncWrite + Unpin), body: &[u8]) {
    let header = format!("Content-Length: {}\r\n\r\n", body.len());
    writer.write_all(header.as_bytes()).await.unwrap();
    writer.write_all(body).await.unwrap();
    writer.flush().await.unwrap();
}

async fn read_framed(reader: &mut (impl AsyncRead + Unpin)) -> Vec<u8> {
    let mut header = String::new();
    loop {
        let mut byte = [0u8; 1];
        reader.read_exact(&mut byte).await.unwrap();
        header.push(byte[0] as char);
        if header.ends_with("\r\n\r\n") {
            break;
        }
    }

    let length = header
        .trim()
        .strip_prefix("Content-Length: ")
        .unwrap()
        .parse()
        .unwrap();
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body).await.unwrap();
    body
}

#[tokio::test]
async fn request_response_round_trip() {
    // duplex: client_write → server_read, server_write → client_read
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);

    let (notification_tx, _) = broadcast::channel(16);
    let (_request_tx, _request_rx) = mpsc::unbounded_channel();
    let request_tx = _request_tx;

    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    // Spawn a task that reads the request from the server side and sends a response.
    let server_handle = tokio::spawn(async move {
        let mut buf = Vec::new();
        // Read the Content-Length header
        let mut header = String::new();
        loop {
            let mut byte = [0u8; 1];
            tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut byte)
                .await
                .unwrap();
            header.push(byte[0] as char);
            if header.ends_with("\r\n\r\n") {
                break;
            }
        }
        let length: usize = header
            .trim()
            .strip_prefix("Content-Length: ")
            .unwrap()
            .parse()
            .unwrap();
        buf.resize(length, 0);
        tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut buf)
            .await
            .unwrap();

        let request: JsonRpcRequest = serde_json::from_slice(&buf).unwrap();
        assert_eq!(request.method, "test.echo");
        assert_eq!(request.jsonrpc, "2.0");

        // Send response
        let response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "result": { "echoed": true }
        });
        write_framed(&mut server_write, &serde_json::to_vec(&response).unwrap()).await;

        request.id
    });

    let response = client
        .send_request("test.echo", Some(serde_json::json!({"hello": "world"})))
        .await
        .unwrap();

    let request_id = server_handle.await.unwrap();
    assert_eq!(response.id, request_id);
    assert!(!response.is_error());
    assert_eq!(response.result.unwrap()["echoed"], serde_json::json!(true));
}

#[tokio::test]
async fn unreadable_error_fails_request_and_keeps_reading() {
    for (error_json, extension_json) in [
        "null",
        r#""not an object""#,
        r#"{"code":-32603,"message":42}"#,
        r#"{"code":9223372036854775808,"message":"overflow"}"#,
        r#"{"code":true,"message":"\ud83d"}"#,
        r#"{"code":-32603,"message":"readable","data":{"detail":true}}"#,
    ]
    .into_iter()
    .flat_map(|error| ["", r#","peerExtension":true"#].map(|extension| (error, extension)))
    {
        let (client_write, mut server_read) = duplex(4096);
        let (mut server_write, client_read) = duplex(4096);
        let (notification_tx, _) = broadcast::channel(16);
        let (request_tx, _request_rx) = mpsc::unbounded_channel();
        let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

        let peer = async {
            for i in 0..2 {
                let request: JsonRpcRequest =
                    serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
                let payload = if i == 0 {
                    format!(r#""error":{error_json}"#)
                } else {
                    r#""result":{"recovered":true}"#.to_string()
                };
                let body = format!(
                    r#"{{"jsonrpc":"2.0","id":{}{extension_json},{payload}}}"#,
                    request.id
                );
                write_framed(&mut server_write, body.as_bytes()).await;
            }
        };
        let requests = async {
            let first = client.send_request("bad", None).await;
            if error_json.contains("readable") {
                let error = first.unwrap().error.unwrap();
                assert_eq!(error.message, "readable");
                assert_eq!(error.data, Some(serde_json::json!({"detail": true})));
            } else {
                let error = first.expect_err("malformed error must fail its request");
                assert!(matches!(error.kind(), ErrorKind::Json), "{error:?}");
            }
            let next = client.send_request("next", None).await.unwrap();
            assert_eq!(next.result, Some(serde_json::json!({"recovered": true})));
        };
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            tokio::join!(peer, requests);
        })
        .await
        .expect("requests did not settle while the peer remained connected");
    }
}

#[tokio::test]
async fn notification_broadcasting() {
    let (_client_write, _discard) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);

    let (notification_tx, mut notification_rx) = broadcast::channel(16);
    let (request_tx, _request_rx) = mpsc::unbounded_channel();

    let _client = JsonRpcClient::new(_client_write, client_read, notification_tx, request_tx);

    // Server sends a notification (no id field).
    let notification = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "session.event",
        "params": { "session_id": "s1", "event": "started" }
    });
    write_framed(
        &mut server_write,
        &serde_json::to_vec(&notification).unwrap(),
    )
    .await;

    let received: std::sync::Arc<JsonRpcNotification> =
        tokio::time::timeout(std::time::Duration::from_secs(2), notification_rx.recv())
            .await
            .expect("timed out waiting for notification")
            .unwrap();

    assert_eq!(received.method, "session.event");
    assert_eq!(received.params.as_ref().unwrap()["session_id"], "s1");
}

#[tokio::test]
async fn server_request_forwarding() {
    let (_client_write, _discard) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);

    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, mut request_rx) = mpsc::unbounded_channel();

    let _client = JsonRpcClient::new(_client_write, client_read, notification_tx, request_tx);

    // Server sends a request (has both id and method).
    let request = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 42,
        "method": "permission.request",
        "params": { "kind": "shell" }
    });
    write_framed(&mut server_write, &serde_json::to_vec(&request).unwrap()).await;

    let received: JsonRpcRequest =
        tokio::time::timeout(std::time::Duration::from_secs(2), request_rx.recv())
            .await
            .expect("timed out waiting for request")
            .unwrap();

    assert_eq!(received.method, "permission.request");
    assert_eq!(received.id, 42);
}

#[tokio::test]
async fn error_response_round_trip() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);

    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();

    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_handle = tokio::spawn(async move {
        // Read request
        let mut header = String::new();
        loop {
            let mut byte = [0u8; 1];
            tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut byte)
                .await
                .unwrap();
            header.push(byte[0] as char);
            if header.ends_with("\r\n\r\n") {
                break;
            }
        }
        let length: usize = header
            .trim()
            .strip_prefix("Content-Length: ")
            .unwrap()
            .parse()
            .unwrap();
        let mut buf = vec![0u8; length];
        tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut buf)
            .await
            .unwrap();
        let request: JsonRpcRequest = serde_json::from_slice(&buf).unwrap();

        // Send error response
        let error_response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "error": { "code": -32600, "message": "Invalid Request" }
        });
        write_framed(
            &mut server_write,
            &serde_json::to_vec(&error_response).unwrap(),
        )
        .await;
    });

    let response = client.send_request("bad.method", None).await.unwrap();
    server_handle.await.unwrap();

    assert!(response.is_error());
    let error = response.error.unwrap();
    assert_eq!(error.code, -32600);
    assert_eq!(error.message, "Invalid Request");
}

/// Cancel-safety regression: dropping a `write()` future after the actor has
/// committed to writing must NOT produce a partial frame on the wire.
///
/// Strategy: read a prefix to acknowledge that writing has begun, then
/// cancel the caller while the rest of the frame cannot fit in the wire.
/// Drain the wire and verify both frames land on it intact.
///
/// Closes RFD-400 finding #1: `JsonRpcClient::write` was holding a Tokio
/// mutex across `write_all` + `flush`, so caller cancellation mid-frame
/// could desync the transport. The writer-actor refactor moves the I/O
/// onto a dedicated task that owns the writer; caller cancellation drops
/// the ack receiver but does not interrupt the in-flight write.
#[tokio::test]
async fn write_actor_completes_on_caller_cancel() {
    use std::sync::Arc;

    let (client_write, mut server_read) = duplex(8);
    let (_server_write, client_read) = duplex(8);

    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let frame_a = JsonRpcRequest {
        jsonrpc: "2.0".to_string(),
        id: 100,
        method: "first.write".to_string(),
        params: None,
    };
    let frame_b = JsonRpcRequest {
        jsonrpc: "2.0".to_string(),
        id: 101,
        method: "second.write".to_string(),
        params: None,
    };

    let mut prefix = [0; 8];
    {
        let writing = client.write(&frame_a);
        tokio::pin!(writing);
        tokio::select! {
            _ = &mut writing => panic!("write completed before the wire was drained"),
            result = server_read.read_exact(&mut prefix) => { result.unwrap(); }
        }
        // The entire frame cannot fit in the eight-byte wire even after this
        // prefix is read, so cancellation is inside a committed, incomplete write.
    }

    let reader_task = tokio::spawn(async move {
        let mut server_read = std::io::Cursor::new(prefix).chain(server_read);
        let mut frames = Vec::new();
        for _ in 0..2 {
            let mut header = String::new();
            loop {
                let mut byte = [0u8; 1];
                tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut byte)
                    .await
                    .unwrap();
                header.push(byte[0] as char);
                if header.ends_with("\r\n\r\n") {
                    break;
                }
            }
            let length: usize = header
                .trim()
                .strip_prefix("Content-Length: ")
                .unwrap()
                .parse()
                .unwrap();
            let mut body = vec![0u8; length];
            tokio::io::AsyncReadExt::read_exact(&mut server_read, &mut body)
                .await
                .unwrap();
            let req: JsonRpcRequest = serde_json::from_slice(&body).unwrap();
            frames.push(req);
        }
        frames
    });

    // The actor must finish the cancelled caller's frame before this next one.
    let second_handle = tokio::spawn({
        let frame_b = frame_b.clone();
        let client_arc = Arc::new(client);
        let client_clone = client_arc.clone();
        async move { client_clone.write(&frame_b).await }
    });

    let frames = reader_task.await.unwrap();
    second_handle.await.unwrap().unwrap();

    assert_eq!(frames.len(), 2);
    assert_eq!(frames[0].method, "first.write");
    assert_eq!(frames[0].id, 100);
    assert_eq!(frames[1].method, "second.write");
    assert_eq!(frames[1].id, 101);
}

/// Cancel-safety regression: cancelling a `send_request` future before the
/// response arrives must NOT leak the pending-requests entry. The RAII
/// `PendingGuard` removes the entry on drop.
///
/// Strategy: spawn `send_request`, drop the JoinHandle immediately so the
/// future is cancelled. The CLI eventually sends a response for the
/// cancelled request id; the read loop logs a warning and discards it
/// (the pending entry was already removed by the guard). The next
/// `send_request` should work normally and not collide with the orphan.
///
/// Closes RFD-400 finding #4.
#[tokio::test]
async fn send_request_cancellation_does_not_leak_pending() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);

    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);
    let client = std::sync::Arc::new(client);

    // First request: cancel before the server replies.
    let cancelled = tokio::spawn({
        let client = client.clone();
        async move {
            // Will await the response oneshot; the JoinHandle abort
            // below cancels this future.
            let _ = client.send_request("first", None).await;
        }
    });

    // Read the first request off the wire so we know it was sent.
    async fn read_one_method(reader: &mut tokio::io::DuplexStream) -> (u64, String) {
        let mut header = String::new();
        loop {
            let mut byte = [0u8; 1];
            tokio::io::AsyncReadExt::read_exact(reader, &mut byte)
                .await
                .unwrap();
            header.push(byte[0] as char);
            if header.ends_with("\r\n\r\n") {
                break;
            }
        }
        let length: usize = header
            .trim()
            .strip_prefix("Content-Length: ")
            .unwrap()
            .parse()
            .unwrap();
        let mut body = vec![0u8; length];
        tokio::io::AsyncReadExt::read_exact(reader, &mut body)
            .await
            .unwrap();
        let req: JsonRpcRequest = serde_json::from_slice(&body).unwrap();
        (req.id, req.method)
    }

    let (first_id, first_method) = read_one_method(&mut server_read).await;
    assert_eq!(first_method, "first");

    // Now cancel the in-flight request.
    cancelled.abort();
    let _ = cancelled.await;

    // Send a (late) response for the cancelled id. The read loop should
    // log a warning and not blow up.
    let stale_resp = serde_json::json!({
        "jsonrpc": "2.0",
        "id": first_id,
        "result": {"echo": "ignored"}
    });
    write_framed(&mut server_write, &serde_json::to_vec(&stale_resp).unwrap()).await;

    // Second request: should succeed normally without collision.
    let server_task = tokio::spawn(async move {
        let (id, method) = read_one_method(&mut server_read).await;
        assert_eq!(method, "second");
        let resp = serde_json::json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": {"ok": true}
        });
        write_framed(&mut server_write, &serde_json::to_vec(&resp).unwrap()).await;
    });

    let response = client.send_request("second", None).await.unwrap();
    assert_eq!(response.result.unwrap()["ok"], true);
    server_task.await.unwrap();
}

#[tokio::test]
async fn lone_surrogate_frame_is_recovered_without_closing_connection() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_task = tokio::spawn(async move {
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = format!(
            r#"{{"jsonrpc":"2.0","id":{},"result":{{"name":"invalid \ud83d value"}}}}"#,
            request.id
        );
        write_framed(&mut server_write, response.as_bytes()).await;

        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "result": {"name": "still connected"}
        });
        write_framed(&mut server_write, &serde_json::to_vec(&response).unwrap()).await;
    });

    let response = client.send_request("models.list", None).await.unwrap();
    assert_eq!(
        response.result.unwrap()["name"],
        serde_json::json!("invalid \u{FFFD} value")
    );

    let response = client.send_request("account.getQuota", None).await.unwrap();
    assert_eq!(
        response.result.unwrap()["name"],
        serde_json::json!("still connected")
    );
    server_task.await.unwrap();
}

#[tokio::test]
async fn unrepairable_frame_remains_fatal() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_task = tokio::spawn(async move {
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = format!(
            r#"{{"jsonrpc":"2.0","id":{},"result":{{"surrogate":"\ud83d","escape":"\q"}}}}"#,
            request.id
        );
        write_framed(&mut server_write, response.as_bytes()).await;
    });

    let error = tokio::time::timeout(
        std::time::Duration::from_secs(2),
        client.send_request("models.list", None),
    )
    .await
    .expect("unrepairable frame did not terminate the pending request")
    .unwrap_err();

    assert_eq!(error.to_string(), "request cancelled");
    assert!(error.is_transport_failure());
    server_task.await.unwrap();
}

/// The CLI answers a request it could not read, for example one its parser
/// refused for its nesting depth, with an error whose `id` is null (#24583).
/// That request's caller gets the CLI's rejection instead of waiting forever,
/// and the connection keeps serving the next request.
#[tokio::test]
async fn null_id_error_fails_the_pending_request_and_keeps_the_connection() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_task = tokio::spawn(async move {
        // The CLI reads the request but cannot identify it.
        read_framed(&mut server_read).await;
        write_framed(
            &mut server_write,
            br#"{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"recursion limit exceeded at line 1 column 513"}}"#,
        )
        .await;
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "result": {"name": "still connected"}
        });
        write_framed(&mut server_write, &serde_json::to_vec(&response).unwrap()).await;
    });

    let error = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("session.send", None),
    )
    .await
    .expect("the rejected request completes")
    .unwrap_err();
    assert_eq!(
        error.kind(),
        &ErrorKind::Protocol(ProtocolErrorKind::UncorrelatedRejection {
            code: -32700,
            message: "recursion limit exceeded at line 1 column 513".to_string(),
        })
    );
    assert!(!error.is_transport_failure());

    let response = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("models.list", None),
    )
    .await
    .expect("the next request is answered")
    .unwrap();
    assert_eq!(
        response.result.unwrap()["name"],
        serde_json::json!("still connected")
    );
    server_task.await.unwrap();
}

/// With several requests pending, a null-id error cannot say which one the CLI
/// rejected, so every pending request fails with the rejection and none
/// waits. The rejection repeats the CLI's message without its quoted values.
#[tokio::test]
async fn null_id_error_fails_every_pending_request() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_task = tokio::spawn(async move {
        read_framed(&mut server_read).await;
        read_framed(&mut server_read).await;
        write_framed(
            &mut server_write,
            br#"{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"invalid type: string \"ghp_secret\", expected u64"}}"#,
        )
        .await;
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "result": {"name": "still connected"}
        });
        write_framed(&mut server_write, &serde_json::to_vec(&response).unwrap()).await;
    });

    let (first, second) = tokio::time::timeout(std::time::Duration::from_secs(5), async {
        tokio::join!(
            client.send_request("session.send", None),
            client.send_request("models.list", None)
        )
    })
    .await
    .expect("both pending requests complete");
    for result in [first, second] {
        let error = result.unwrap_err();
        assert_eq!(
            error.kind(),
            &ErrorKind::Protocol(ProtocolErrorKind::UncorrelatedRejection {
                code: -32600,
                message: "invalid type: string \"…\", expected u64".to_string(),
            })
        );
        assert!(!error.to_string().contains("ghp_secret"), "{error}");
    }

    let response = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("account.getQuota", None),
    )
    .await
    .expect("the next request is answered")
    .unwrap();
    assert_eq!(
        response.result.unwrap()["name"],
        serde_json::json!("still connected")
    );
    server_task.await.unwrap();
}

/// A null-id error that arrives with nothing pending fails nothing, and the
/// connection keeps serving.
#[tokio::test]
async fn null_id_error_with_nothing_pending_keeps_the_connection() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, mut notifications) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    write_framed(
        &mut server_write,
        br#"{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"Parse error"}}"#,
    )
    .await;
    // The read loop handles frames in order: once this notification arrives,
    // the error before it has been handled.
    write_framed(
        &mut server_write,
        br#"{"jsonrpc":"2.0","method":"session.event","params":{}}"#,
    )
    .await;
    let notification: std::sync::Arc<JsonRpcNotification> =
        tokio::time::timeout(std::time::Duration::from_secs(5), notifications.recv())
            .await
            .expect("the notification after the error arrives")
            .unwrap();
    assert_eq!(notification.method, "session.event");

    let server_task = tokio::spawn(async move {
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = serde_json::json!({
            "jsonrpc": "2.0",
            "id": request.id,
            "result": {"name": "still connected"}
        });
        write_framed(&mut server_write, &serde_json::to_vec(&response).unwrap()).await;
    });
    let response = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("models.list", None),
    )
    .await
    .expect("the request after the null-id error is answered")
    .unwrap();
    assert_eq!(
        response.result.unwrap()["name"],
        serde_json::json!("still connected")
    );
    server_task.await.unwrap();
}

#[tokio::test]
async fn valid_pairs_and_escaped_backslashes_are_untouched() {
    let (client_write, mut server_read) = duplex(4096);
    let (mut server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _) = mpsc::unbounded_channel();
    let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);

    let server_task = tokio::spawn(async move {
        let request: JsonRpcRequest =
            serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
        let response = format!(
            r#"{{"jsonrpc":"2.0","id":{},"result":{{"emoji":"\ud83d\ude00","path":"C:\\ud83d","invalid":"\ud83d"}}}}"#,
            request.id
        );
        write_framed(&mut server_write, response.as_bytes()).await;
    });

    let result = client
        .send_request("models.list", None)
        .await
        .unwrap()
        .result
        .unwrap();

    assert_eq!(result["emoji"], serde_json::json!("😀"));
    assert_eq!(result["path"], serde_json::json!(r"C:\ud83d"));
    assert_eq!(result["invalid"], serde_json::json!("\u{FFFD}"));
    server_task.await.unwrap();
}

/// Once the peer has closed its side, the read loop has ended and failed the
/// requests waiting for a response. A request sent after that fails at once and
/// is never written, though the transport may still take writes, as a TCP
/// socket does after the peer's FIN.
#[tokio::test]
async fn request_after_the_peer_closed_fails_at_once() {
    let (client_write, mut server_read) = duplex(4096);
    let (server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _request_rx) = mpsc::unbounded_channel();
    let client = std::sync::Arc::new(JsonRpcClient::new(
        client_write,
        client_read,
        notification_tx,
        request_tx,
    ));

    let waiting = tokio::spawn({
        let client = client.clone();
        async move { client.send_request("test.wait", None).await }
    });
    let request: JsonRpcRequest =
        serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
    assert_eq!(request.method, "test.wait");

    // The peer closes its side: the read loop ends and fails the waiting request.
    drop(server_write);
    let drained = tokio::time::timeout(std::time::Duration::from_secs(5), waiting)
        .await
        .expect("the waiting request was not failed")
        .unwrap()
        .unwrap_err();
    assert_eq!(drained.to_string(), "request cancelled");

    let late = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("test.late", None),
    )
    .await
    .expect("the request after the close was not refused")
    .unwrap_err();
    assert_eq!(late.kind(), &ErrorKind::Io);
    assert!(late.is_transport_failure());

    // Nothing was written for it: once the client is gone, the pipe holds no frame.
    drop(client);
    let mut written = Vec::new();
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        server_read.read_to_end(&mut written),
    )
    .await
    .expect("the writer was not released")
    .unwrap();
    assert!(written.is_empty(), "frames written: {written:?}");
}

/// A `session.sendAndWait` refused because the connection is closed is never
/// written, and dropping its request does not queue a `$/cancelRequest` for the
/// id that was never sent: the pipe holds no frame at all.
#[tokio::test]
async fn send_and_wait_after_the_peer_closed_writes_no_cancel_frame() {
    let (client_write, mut server_read) = duplex(4096);
    let (server_write, client_read) = duplex(4096);
    let (notification_tx, _) = broadcast::channel(16);
    let (request_tx, _request_rx) = mpsc::unbounded_channel();
    let client = std::sync::Arc::new(JsonRpcClient::new(
        client_write,
        client_read,
        notification_tx,
        request_tx,
    ));

    let waiting = tokio::spawn({
        let client = client.clone();
        async move { client.send_request("test.wait", None).await }
    });
    let request: JsonRpcRequest =
        serde_json::from_slice(&read_framed(&mut server_read).await).unwrap();
    assert_eq!(request.method, "test.wait");

    // The peer closes its side: the read loop ends and fails the waiting request.
    drop(server_write);
    tokio::time::timeout(std::time::Duration::from_secs(5), waiting)
        .await
        .expect("the waiting request was not failed")
        .unwrap()
        .unwrap_err();

    let late = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client.send_request("session.sendAndWait", None),
    )
    .await
    .expect("the request after the close was not refused")
    .unwrap_err();
    assert_eq!(late.kind(), &ErrorKind::Io);

    // A frame queued for the writer arrives within moments; none may.
    if let Ok(frame) = tokio::time::timeout(
        std::time::Duration::from_millis(300),
        read_framed(&mut server_read),
    )
    .await
    {
        panic!("a frame was written: {}", String::from_utf8_lossy(&frame));
    }

    drop(client);
    let mut written = Vec::new();
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        server_read.read_to_end(&mut written),
    )
    .await
    .expect("the writer was not released")
    .unwrap();
    assert!(written.is_empty(), "frames written: {written:?}");
}