boxmaker/crates/bxctl/tests/chat_speaker.rs

//! `bxctl chat --speaker NAME` (M5c-1 task 04): the name goes on every turn of the chat, the
//! interactive ones and `--say`'s, the retried turn too; without the flag no turn has one; a bad
//! name is a usage error and nothing is sent. Every `bxctl` run has a deadline, and every path is
//! removed by a guard. Do not edit.

#[path = "support/temp.rs"]
mod temp;

use std::io::{ErrorKind, Write};
use std::os::unix::net::UnixListener;
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use bxcore::{
    Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone, Usage, WireError,
    read_frame, write_frame,
};
use bxctl::chat::{Approvals, OnPending, Printer, TurnIo, run_turn, stream_turn};

/// A fake `loopd` that records every turn it is sent and answers `ok`. A resumed turn whose text
/// is `trigger-no-such-session` is refused with `no_such_session`, as `loopd` refuses a session
/// that does not exist. Its home, which holds the socket, is removed on drop.
struct FakeLoopd {
    socket: PathBuf,
    turns: Arc<Mutex<Vec<Turn>>>,
    _home: temp::TempDir,
}

impl FakeLoopd {
    fn turns(&self) -> Vec<Turn> {
        self.turns.lock().unwrap().clone()
    }
}

fn fake_loopd() -> FakeLoopd {
    let home = temp::TempDir::new("bxctl-speaker");
    let socket = home.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 Ok(mut stream) = stream else { continue };
            // A client that never writes cannot hold the fake forever.
            let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
            let Ok(request) = read_frame(&mut stream) else {
                continue;
            };
            let Message::Turn(turn) = request.msg else {
                continue;
            };
            seen.lock().unwrap().push(turn.clone());
            let msg = if turn.resume && turn.content == "trigger-no-such-session" {
                Message::Error(WireError {
                    code: ErrorCode::NoSuchSession,
                    detail: "no such session".to_string(),
                })
            } else {
                Message::TurnDone(TurnDone {
                    content: "ok".to_string(),
                    usage: Usage {
                        cache_n: 1,
                        prompt_n: 1,
                        predicted_n: 1,
                        reasoning_tokens: 0,
                        thinking_capped: false,
                    },
                    untrusted: false,
                })
            };
            let _ = write_frame(
                &mut stream,
                &Envelope {
                    v: PROTOCOL_VERSION,
                    id: request.id,
                    r#final: true,
                    msg,
                },
            );
        }
    });
    FakeLoopd {
        socket,
        turns,
        _home: home,
    }
}

/// What a run of `bxctl` gave.
struct Run {
    code: Option<i32>,
    stdout: String,
    stderr: String,
}

/// Runs `bxctl chat --socket <fake> <args>`, writing `stdin` to its standard input, with
/// `$BOXMAKER_HOME` an empty directory removed afterwards. Kills it and fails after 20 s.
fn chat(fake: &FakeLoopd, args: &[&str], stdin: &str) -> Run {
    let home = temp::TempDir::new("bxctl-speaker-home");
    let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl"))
        .arg("chat")
        .arg("--socket")
        .arg(&fake.socket)
        .args(args)
        .env("BOXMAKER_HOME", &home)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    {
        let mut input = child.stdin.take().unwrap();
        match input.write_all(stdin.as_bytes()) {
            Ok(()) => {}
            // bxctl exited without reading it all: its exit status and output say why.
            Err(e) if e.kind() == ErrorKind::BrokenPipe => {}
            Err(e) => panic!("cannot write bxctl's standard input: {e}"),
        }
    }
    let deadline = Instant::now() + Duration::from_secs(20);
    while child.try_wait().unwrap().is_none() {
        if Instant::now() > deadline {
            let _ = child.kill();
            let _ = child.wait();
            panic!("bxctl chat {args:?} did not exit within 20 s");
        }
        thread::sleep(Duration::from_millis(10));
    }
    let output = child.wait_with_output().unwrap();
    Run {
        code: output.status.code(),
        stdout: String::from_utf8_lossy(&output.stdout).into_owned(),
        stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
    }
}

fn speakers(fake: &FakeLoopd) -> Vec<Option<String>> {
    fake.turns().into_iter().map(|turn| turn.speaker).collect()
}

fn some(name: &str) -> Option<String> {
    Some(name.to_string())
}

#[test]
fn run_turn_puts_the_speaker_on_the_turn() {
    let fake = fake_loopd();
    let session = SessionId::new("s1").unwrap();
    run_turn(
        &fake.socket,
        &session,
        "hi",
        false,
        Some("first.last"),
        &mut |_| {},
    )
    .unwrap();
    run_turn(&fake.socket, &session, "hi", true, None, &mut |_| {}).unwrap();
    assert_eq!(
        fake.turns(),
        vec![
            Turn {
                session: session.clone(),
                content: "hi".to_string(),
                resume: false,
                speaker: some("first.last"),
            },
            Turn {
                session,
                content: "hi".to_string(),
                resume: true,
                speaker: None,
            },
        ]
    );
}

#[test]
fn stream_turn_passes_the_speaker_on() {
    let fake = fake_loopd();
    let admin = fake.socket.with_file_name("admin.sock");
    let approvals = Approvals {
        admin_socket: &admin,
        on_pending: OnPending::EventOnly,
    };
    let mut printer = Printer::new(true, false);
    let mut input = std::io::Cursor::new(Vec::new());
    let mut out = Vec::new();
    let session = SessionId::new("s1").unwrap();
    for speaker in [Some("kyle"), None] {
        let mut io = TurnIo {
            printer: &mut printer,
            input: &mut input,
            out: &mut out,
        };
        let done = stream_turn(
            &fake.socket,
            &session,
            "hi",
            false,
            speaker,
            &approvals,
            &mut io,
        )
        .unwrap()
        .unwrap();
        assert_eq!(done.content, "ok");
    }
    assert_eq!(speakers(&fake), vec![some("kyle"), None]);
}

#[test]
fn say_sends_the_speaker_as_it_was_typed() {
    let fake = fake_loopd();
    let run = chat(&fake, &["--speaker", "first.last", "--say", "hi"], "");
    assert_eq!(run.code, Some(0), "{}", run.stderr);
    assert_eq!(run.stdout, "ok\n");
    assert_eq!(speakers(&fake), vec![some("first.last")]);
    assert_eq!(fake.turns()[0].content, "hi", "the text is not changed");
}

#[test]
fn without_the_flag_no_turn_has_a_speaker() {
    let fake = fake_loopd();
    let run = chat(&fake, &["--say", "hi"], "");
    assert_eq!(run.code, Some(0), "{}", run.stderr);
    let run = chat(&fake, &[], "first\nsecond\n/quit\n");
    assert_eq!(run.code, Some(0), "{}", run.stderr);
    assert_eq!(speakers(&fake), vec![None, None, None]);
}

/// `--say` on a named session that does not exist sends the turn twice: resumed, refused, then
/// created. Both carry the speaker.
#[test]
fn the_retried_turn_keeps_the_speaker() {
    let fake = fake_loopd();
    let run = chat(
        &fake,
        &[
            "--session",
            "fresh",
            "--speaker",
            "kyle",
            "--say",
            "trigger-no-such-session",
        ],
        "",
    );
    assert_eq!(run.code, Some(0), "{}", run.stderr);
    let turns = fake.turns();
    assert_eq!(turns.len(), 2, "resumed, refused, then created");
    assert!(turns[0].resume && !turns[1].resume);
    assert_eq!(speakers(&fake), vec![some("kyle"), some("kyle")]);
}

/// Every turn of an interactive chat carries the speaker, not only the one that creates the
/// session, and one that follows a failed turn too.
#[test]
fn every_interactive_turn_has_the_speaker() {
    let fake = fake_loopd();
    let run = chat(
        &fake,
        &["--speaker", "kyle"],
        "first\nsecond\ntrigger-no-such-session\nfourth\n/quit\n",
    );
    assert_eq!(run.code, Some(0), "{}", run.stderr);
    assert!(run.stderr.contains("no such session"), "{}", run.stderr);
    let turns = fake.turns();
    assert_eq!(turns.len(), 4, "{turns:?}");
    assert!(!turns[0].resume && turns[1].resume && turns[3].resume);
    assert_eq!(speakers(&fake), vec![some("kyle"); 4]);
}

/// A bad name stops `bxctl` before it connects: the usage text on stderr, status 2, nothing on
/// stdout, and no turn sent.
#[test]
fn a_bad_speaker_is_a_usage_error_and_sends_nothing() {
    let fake = fake_loopd();
    for args in [
        vec!["--speaker", "", "--say", "hi"],
        vec!["--speaker", "first last", "--say", "hi"],
        vec!["--speaker", "kyle\t", "--say", "hi"],
        vec!["--speaker", "ky\u{1b}le", "--say", "hi"],
        vec!["--speaker", "kyle", "--speaker", "eve", "--say", "hi"],
        vec!["--say", "hi", "--speaker"],
        vec!["--speaker", "first last"],
    ] {
        let run = chat(&fake, &args, "hello\n/quit\n");
        assert_eq!(run.code, Some(2), "{args:?}");
        assert_eq!(run.stdout, "", "{args:?}");
        assert!(run.stderr.contains("[--speaker NAME]"), "{args:?}");
        assert!(run.stderr.contains("bxctl approve <id>"), "{args:?}");
    }
    assert!(fake.turns().is_empty(), "{:?}", fake.turns());
}