boxmaker/docs/plans/M2b/files/crates/bxctl/tests/chat.rs

//! Tests for `bxctl chat`, against a fake `loopd` that speaks the channel protocol. Do not edit.

use bxctl::chat::{ChatError, Printer, new_session_id, run_turn};
use proto::{
    DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone,
    TurnEvent, Usage, WireError, read_frame, write_frame,
};
use std::os::unix::net::UnixListener;
use std::path::PathBuf;
use std::process::Command;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;

static NEXT: AtomicU32 = AtomicU32::new(0);

/// What the fake `loopd` sends back for one turn, after the events: done or an error.
#[derive(Clone)]
enum End {
    Done(TurnDone),
    Error(ErrorCode, &'static str),
}

struct FakeLoopd {
    socket: PathBuf,
    turns: Arc<Mutex<Vec<Turn>>>,
}

fn usage() -> Usage {
    Usage {
        cache_n: 10,
        prompt_n: 5,
        predicted_n: 7,
        reasoning_tokens: 3,
        thinking_capped: false,
    }
}

/// Serves every connection with the same script: the given events, then `end`. Records the
/// turns it received.
fn fake_loopd(events: Vec<TurnEvent>, end: End) -> FakeLoopd {
    let n = NEXT.fetch_add(1, Ordering::SeqCst);
    let dir = std::env::temp_dir().join(format!("bxctl-test-{}-{n}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let socket = dir.join("loop.sock");
    let listener = UnixListener::bind(&socket).unwrap();
    let turns = Arc::new(Mutex::new(Vec::new()));
    let seen = Arc::clone(&turns);
    thread::spawn(move || {
        for stream in listener.incoming() {
            let mut stream = stream.unwrap();
            let request = read_frame(&mut stream).unwrap();
            let Message::Turn(turn) = request.msg else {
                panic!("not a turn")
            };
            seen.lock().unwrap().push(turn.clone());
            let id = request.id;
            let end = if turn.resume && turn.content == "trigger-no-such-session" {
                End::Error(ErrorCode::NoSuchSession, "session x does not exist")
            } else if turn.content == "trigger-turn-limit" {
                End::Error(ErrorCode::TurnLimit, "the turn hit a limit")
            } else {
                end.clone()
            };
            for e in &events {
                write_frame(
                    &mut stream,
                    &Envelope {
                        v: PROTOCOL_VERSION,
                        id,
                        r#final: false,
                        msg: Message::TurnEvent(e.clone()),
                    },
                )
                .unwrap();
            }
            let last = match end {
                End::Done(done) => Message::TurnDone(done),
                End::Error(code, detail) => Message::Error(WireError {
                    code,
                    detail: detail.to_string(),
                }),
            };
            write_frame(
                &mut stream,
                &Envelope {
                    v: PROTOCOL_VERSION,
                    id,
                    r#final: true,
                    msg: last,
                },
            )
            .unwrap();
        }
    });
    FakeLoopd { socket, turns }
}

fn events() -> Vec<TurnEvent> {
    vec![
        TurnEvent::Queued { ahead: 1 },
        TurnEvent::Waiting { slot_busy: true },
        TurnEvent::Progress {
            total: 100,
            cache: 50,
            processed: 75,
        },
        TurnEvent::Reasoning {
            text: "let me ".to_string(),
        },
        TurnEvent::Reasoning {
            text: "think".to_string(),
        },
        TurnEvent::ToolCallStarted {
            name: "clock".to_string(),
        },
        TurnEvent::ToolResult {
            name: "clock".to_string(),
            class: DataClass::Public,
            truncated: true,
        },
        TurnEvent::ThinkingCapped { tokens: 4096 },
        TurnEvent::Retrying {
            attempt: 2,
            after_ms: 1500,
            error: "the server went silent".to_string(),
        },
        TurnEvent::CacheLoss {
            expected: 500,
            got: 20,
        },
        TurnEvent::Content {
            text: "It is ".to_string(),
        },
        TurnEvent::Content {
            text: "noon.".to_string(),
        },
    ]
}

fn done() -> End {
    End::Done(TurnDone {
        content: "It is noon.".to_string(),
        usage: usage(),
    })
}

fn id(s: &str) -> SessionId {
    SessionId::new(s).unwrap()
}

#[test]
fn run_turn_delivers_every_event_in_order_then_the_answer() {
    let fake = fake_loopd(events(), done());
    let mut seen = Vec::new();
    let done = run_turn(
        &fake.socket,
        &id("s1"),
        "what time is it?",
        false,
        &mut |e| seen.push(e.clone()),
    )
    .unwrap();
    assert_eq!(seen, events());
    assert_eq!(done.content, "It is noon.");
    assert_eq!(done.usage, usage());
    let turns = fake.turns.lock().unwrap();
    assert_eq!(turns.len(), 1);
    assert_eq!(
        turns[0],
        Turn {
            session: id("s1"),
            content: "what time is it?".to_string(),
            resume: false
        }
    );
}

#[test]
fn an_error_frame_is_refused_with_its_code_and_detail() {
    let fake = fake_loopd(
        vec![TurnEvent::Content {
            text: "partial".to_string(),
        }],
        End::Error(ErrorCode::SessionFull, "this conversation is full"),
    );
    let mut seen = 0;
    let err = run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| seen += 1).unwrap_err();
    assert_eq!(seen, 1, "events before the error are still delivered");
    match err {
        ChatError::Refused(w) => {
            assert_eq!(w.code, ErrorCode::SessionFull);
            assert_eq!(w.detail, "this conversation is full");
        }
        other => panic!("{other:?}"),
    }
    let e: Box<dyn std::error::Error> =
        Box::new(run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| {}).unwrap_err());
    assert!(e.to_string().contains("session full"), "{e}");
}

#[test]
fn no_loopd_is_a_connect_error() {
    let missing = std::env::temp_dir().join("bxctl-no-such-loopd.sock");
    assert!(matches!(
        run_turn(&missing, &id("s1"), "x", false, &mut |_| {}),
        Err(ChatError::Connect(_))
    ));
}

#[test]
fn new_session_ids_are_valid_and_distinct() {
    let a = new_session_id();
    let b = new_session_id();
    assert!(a.as_str().starts_with("chat-"));
    assert_ne!(a, b);
}

#[test]
fn the_printer_formats_each_event_kind() {
    let mut out = Vec::new();
    let mut p = Printer::new(true, false);
    for e in events() {
        p.event(&mut out, &e).unwrap();
    }
    p.end_reasoning(&mut out).unwrap();
    let text = String::from_utf8(out).unwrap();
    assert!(
        text.contains("\x1b[2mlet me think\x1b[0m\n"),
        "reasoning dimmed, joined, and ended once: {text:?}"
    );
    assert!(text.contains("[tool clock]\n"), "{text:?}");
    assert!(text.contains("[clock: Public, truncated]\n"), "{text:?}");
    assert!(text.contains("[waiting: slot busy]\n"), "{text:?}");
    assert!(
        text.contains("[retrying: attempt 2 in 1500 ms: the server went silent]\n"),
        "{text:?}"
    );
    assert!(
        text.contains("[thinking capped at 4096 tokens]\n"),
        "{text:?}"
    );
    assert!(text.contains("[cache loss: 20 of 500]\n"), "{text:?}");
    assert!(
        text.ends_with("It is noon."),
        "content streams as it is: {text:?}"
    );
    assert!(
        !text.contains("Queued") && !text.contains("Progress"),
        "queued and progress are silent: {text:?}"
    );

    let mut out = Vec::new();
    let mut p = Printer::new(false, false);
    for e in events() {
        p.event(&mut out, &e).unwrap();
    }
    let text = String::from_utf8(out).unwrap();
    assert!(
        !text.contains("let me"),
        "--no-thinking hides reasoning: {text:?}"
    );
    assert!(
        !text.contains("\x1b["),
        "and no escape codes are left: {text:?}"
    );

    let mut out = Vec::new();
    let mut p = Printer::new(true, true);
    for e in events() {
        p.event(&mut out, &e).unwrap();
    }
    let text = String::from_utf8(out).unwrap();
    let lines: Vec<&str> = text.lines().collect();
    assert_eq!(
        lines.len(),
        events().len(),
        "--json: one line per event, none skipped"
    );
    let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
    assert_eq!(first, serde_json::json!({"event": "queued", "ahead": 1}));
}

#[test]
fn say_prints_only_the_answer_on_stdout() {
    let fake = fake_loopd(events(), done());
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .args(["--session", "scripted-1", "--say", "what time is it?"])
        .output()
        .unwrap();
    assert!(output.status.success());
    assert_eq!(String::from_utf8_lossy(&output.stdout), "It is noon.\n");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("[tool clock]"),
        "events go to stderr: {stderr}"
    );
    let turns = fake.turns.lock().unwrap();
    assert_eq!(turns.len(), 1);
    assert_eq!(turns[0].session, id("scripted-1"));
    assert!(
        turns[0].resume,
        "a session given on the command line is resumed"
    );
}

#[test]
fn say_creates_a_named_session_that_does_not_exist_yet() {
    let fake = fake_loopd(vec![], done());
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .args(["--session", "fresh", "--say", "trigger-no-such-session"])
        .output()
        .unwrap();
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    let turns = fake.turns.lock().unwrap();
    assert_eq!(turns.len(), 2, "resume was refused, so it was created");
    assert!(turns[0].resume && !turns[1].resume);
}

#[test]
fn say_reports_an_error_on_stderr_with_status_1() {
    let fake = fake_loopd(
        vec![],
        End::Error(ErrorCode::TurnLimit, "the turn hit a limit"),
    );
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .args(["--say", "x"])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(1));
    assert_eq!(String::from_utf8_lossy(&output.stdout), "");
    assert!(String::from_utf8_lossy(&output.stderr).contains("turn limit: the turn hit a limit"));
}

#[test]
fn json_mode_prints_frames_as_json_lines() {
    let fake = fake_loopd(events(), done());
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .args(["--session", "j", "--json", "--say", "x"])
        .output()
        .unwrap();
    assert!(output.status.success());
    let stderr = String::from_utf8_lossy(&output.stderr);
    let lines: Vec<&str> = stderr.lines().collect();
    assert_eq!(
        lines.len(),
        events().len() + 1,
        "every event, then the done frame: {stderr}"
    );
    let last: serde_json::Value = serde_json::from_str(lines[lines.len() - 1]).unwrap();
    assert_eq!(last["content"], "It is noon.");
    assert_eq!(last["usage"]["cache_n"], 10);
}

#[test]
fn interactive_mode_reads_lines_until_quit() {
    let fake = fake_loopd(
        vec![TurnEvent::Content {
            text: "ok".to_string(),
        }],
        End::Done(TurnDone {
            content: "ok".to_string(),
            usage: usage(),
        }),
    );
    let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .unwrap();
    {
        let mut stdin = child.stdin.take().unwrap();
        std::io::Write::write_all(&mut stdin, b"first\n\nsecond\n/quit\nnever sent\n").unwrap();
    }
    let output = child.wait_with_output().unwrap();
    assert!(output.status.success());
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.starts_with("session chat-"), "{stdout}");
    assert_eq!(stdout.matches("ok").count(), 2, "{stdout}");
    let turns = fake.turns.lock().unwrap();
    assert_eq!(
        turns.len(),
        2,
        "a blank line sends nothing, and /quit stops"
    );
    assert!(
        !turns[0].resume && turns[1].resume,
        "the first turn creates, the second resumes"
    );
    assert_eq!(turns[0].session, turns[1].session);
}

#[test]
fn bad_arguments_print_usage() {
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--session", "Not Valid!"])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(2));
    assert!(String::from_utf8_lossy(&output.stderr).contains("usage"));
    let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["dance"])
        .output()
        .unwrap();
    assert_eq!(output.status.code(), Some(2));
}

/// A failed turn is reported, and the conversation goes on: the session still exists and the
/// next line is a new turn on it.
#[test]
fn interactive_mode_survives_a_failed_turn() {
    let fake = fake_loopd(
        vec![TurnEvent::Content {
            text: "ok".to_string(),
        }],
        End::Done(TurnDone {
            content: "ok".to_string(),
            usage: usage(),
        }),
    );
    let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .args(["chat", "--socket"])
        .arg(&fake.socket)
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .unwrap();
    {
        let mut stdin = child.stdin.take().unwrap();
        std::io::Write::write_all(&mut stdin, b"first\ntrigger-turn-limit\nthird\n/quit\n")
            .unwrap();
    }
    let output = child.wait_with_output().unwrap();
    assert!(
        output.status.success(),
        "a failed turn does not end the chat"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("turn limit: the turn hit a limit"),
        "{stderr}"
    );
    let turns = fake.turns.lock().unwrap();
    assert_eq!(turns.len(), 3, "the turn after the failure was sent");
    assert!(turns[2].resume, "and it resumed the same session");
    assert_eq!(turns[2].session, turns[0].session);
}