liminal-cli 0.3.5

use super::wire::{Frame, Pipe, read};
use serde_json::{Value, json};
use std::path::Path;
use tokio::io::{AsyncWrite, AsyncWriteExt as _, BufReader};

async fn write<W: AsyncWrite + Unpin>(writer: &mut W, value: Value) -> anyhow::Result<()> {
    let mut bytes = serde_json::to_vec(&value)?;
    bytes.push(b'\n');
    writer.write_all(&bytes).await?;
    writer.flush().await?;
    Ok(())
}

#[tokio::test]
async fn initialize_offers_two_and_echo_one_opens_a_ticketed_session() -> anyhow::Result<()> {
    let (client, agent) = tokio::io::duplex(1024);
    let (input, output) = tokio::io::split(client);
    let mut pipe = Pipe::start(input, output, 65536, 1024, 1);
    let script = tokio::spawn(async move {
        let (input, mut output) = tokio::io::split(agent);
        let mut input = BufReader::new(input);
        let initialize = read(&mut input, 65536)
            .await?
            .ok_or_else(|| anyhow::anyhow!("initialize absent"))?;
        let Frame::Request { id, method, params } = initialize else {
            anyhow::bail!("initialize is not a request");
        };
        assert_eq!(method, "initialize");
        assert_eq!(params["protocolVersion"], 2);
        assert_eq!(params["clientCapabilities"]["terminal"], false);
        write(&mut output, json!({"jsonrpc":"2.0","id":id,"result":{"protocolVersion":1,"agentCapabilities":{},"authMethods":[]}})).await?;
        let session = read(&mut input, 65536)
            .await?
            .ok_or_else(|| anyhow::anyhow!("session/new absent"))?;
        let Frame::Request { id, method, params } = session else {
            anyhow::bail!("session/new is not a request");
        };
        assert_eq!(method, "session/new");
        assert_eq!(params["mcpServers"][0]["name"], "liminal");
        assert_eq!(params["mcpServers"][0]["args"][1], "tools");
        assert_eq!(params["mcpServers"][0]["env"], json!([]));
        write(
            &mut output,
            json!({"jsonrpc":"2.0","id":id,"result":{"sessionId":"s"}}),
        )
        .await?;
        Ok::<_, anyhow::Error>(())
    });
    let config: super::config::Harness =
        toml::from_str("id='codex'\nprogram='/bin/true'\nargs=[]")?;
    let session = super::handshake::open(
        &mut pipe,
        Path::new("/tmp"),
        Path::new("/tmp/seat.toml"),
        None,
        &config,
    )
    .await?;
    assert_eq!(session.version, super::handshake::Version::One);
    assert!(!session.resumable);
    script.await??;
    pipe.reader.abort();
    drop(pipe.output);
    pipe.writer.await??;
    Ok(())
}

#[tokio::test]
async fn seven_hundred_twenty_nine_answers_do_not_stop_stdout_drain() -> anyhow::Result<()> {
    let (client, agent) = tokio::io::duplex(1024);
    let (input, output) = tokio::io::split(client);
    let mut pipe = Pipe::start(input, output, 65536, 1024, 1);
    let script = tokio::spawn(async move {
        let (input, mut output) = tokio::io::split(agent);
        for id in 0..729 {
            write(
                &mut output,
                json!({"jsonrpc":"2.0","id":id,"method":"session/request_permission","params":{}}),
            )
            .await?;
        }
        // The peer starts reading only after all its requests have left stdout.
        let mut input = BufReader::new(input);
        for expected in 0..729 {
            let answer = read(&mut input, 65536)
                .await?
                .ok_or_else(|| anyhow::anyhow!("permission response absent"))?;
            let Frame::Response { id, result } = answer else {
                anyhow::bail!("permission response is not a response");
            };
            assert_eq!(id, json!(expected));
            let result = result.map_err(|_| anyhow::anyhow!("permission answer failed"))?;
            assert_eq!(result["outcome"]["outcome"], "cancelled");
        }
        Ok::<_, anyhow::Error>(())
    });
    for _ in 0..729 {
        let frame = pipe
            .input
            .recv()
            .await
            .ok_or_else(|| anyhow::anyhow!("ACP reader closed"))??
            .ok_or_else(|| anyhow::anyhow!("ACP stdout closed"))?;
        let Frame::Request { id, .. } = frame else {
            anyhow::bail!("peer sent a non-request");
        };
        pipe.output
            .response(id, json!({"outcome":{"outcome":"cancelled"}}))?;
    }
    script.await??;
    pipe.reader.abort();
    drop(pipe.output);
    pipe.writer.await??;
    Ok(())
}