diff --git a/Cargo.lock b/Cargo.lock index f8d636d..9c4c117 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,6 +14,7 @@ name = "bxctl" version = "0.1.0" dependencies = [ "proto", + "serde_json", ] [[package]] diff --git a/crates/bxctl/Cargo.toml b/crates/bxctl/Cargo.toml index c75c375..60d0553 100644 --- a/crates/bxctl/Cargo.toml +++ b/crates/bxctl/Cargo.toml @@ -10,3 +10,4 @@ workspace = true [dependencies] proto.workspace = true +serde_json.workspace = true diff --git a/crates/bxctl/src/chat.rs b/crates/bxctl/src/chat.rs new file mode 100644 index 0000000..55fccf2 --- /dev/null +++ b/crates/bxctl/src/chat.rs @@ -0,0 +1,211 @@ +//! `bxctl chat`: a turn client over `loop.sock` and a printer for the events it receives. + +use std::io::Write; + +use proto::{ + Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone, TurnEvent, + WireError, read_frame, write_frame, +}; +use std::os::unix::net::UnixStream; +use std::path::Path; + +#[derive(Debug)] +pub enum ChatError { + Connect(std::io::Error), + Frame(proto::FrameError), + Refused(WireError), + Protocol(String), +} + +impl std::fmt::Display for ChatError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ChatError::Connect(e) => write!(f, "{e}"), + ChatError::Frame(e) => write!(f, "{e}"), + ChatError::Refused(w) => write!(f, "{}: {}", code_name(w.code), w.detail), + ChatError::Protocol(s) => write!(f, "{s}"), + } + } +} + +impl std::error::Error for ChatError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + ChatError::Connect(e) => Some(e), + ChatError::Frame(e) => Some(e), + _ => None, + } + } +} + +// The ErrorCode name in snake_case, spelled out in words for the owner reading a failure. +fn code_name(code: ErrorCode) -> &'static str { + match code { + ErrorCode::BadFrame => "bad frame", + ErrorCode::BadVersion => "bad version", + ErrorCode::BadMessage => "bad message", + ErrorCode::Internal => "internal", + ErrorCode::SessionFull => "session full", + ErrorCode::TurnLimit => "turn limit", + ErrorCode::SessionBusy => "session busy", + ErrorCode::NoSuchSession => "no such session", + ErrorCode::SessionExists => "session exists", + ErrorCode::Inference => "inference", + } +} + +pub fn new_session_id() -> SessionId { + // The system clock is never before the unix epoch on any machine this runs on. + let elapsed = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("the system clock is not before the unix epoch"); + // "chat--" uses only [a-z0-9-] and stays well under 64 bytes, so this cannot fail. + let candidate = format!("chat-{}-{}", elapsed.as_secs(), elapsed.subsec_nanos()); + SessionId::new(&candidate).expect("chat-- is always a valid session id") +} + +/// Sends one turn and reads the reply. `on_event` sees every event frame as it arrives. +pub fn run_turn( + socket: &Path, + session: &SessionId, + content: &str, + resume: bool, + on_event: &mut dyn FnMut(&TurnEvent), +) -> Result { + let mut stream = UnixStream::connect(socket).map_err(ChatError::Connect)?; + + let turn = Turn { + session: session.clone(), + content: content.to_string(), + resume, + }; + write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg: Message::Turn(turn), + }, + ) + .map_err(ChatError::Frame)?; + + loop { + let env = read_frame(&mut stream).map_err(ChatError::Frame)?; + if env.id != 1 { + return Err(ChatError::Protocol( + "expected a frame with id 1".to_string(), + )); + } + match (env.r#final, &env.msg) { + (false, Message::TurnEvent(e)) => on_event(e), + (true, Message::TurnDone(d)) => return Ok(d.clone()), + (true, Message::Error(w)) => return Err(ChatError::Refused(w.clone())), + _ => return Err(ChatError::Protocol("unexpected final frame".to_string())), + } + } +} + +pub struct Printer { + pub show_thinking: bool, + pub json: bool, + pub stream_content: bool, + // Whether a dimmed reasoning block is open and needs closing. + dimmed_open: bool, +} + +impl Printer { + pub fn new(show_thinking: bool, json: bool) -> Printer { + Printer { + show_thinking, + json, + stream_content: true, + dimmed_open: false, + } + } + + pub fn event(&mut self, out: &mut dyn Write, event: &TurnEvent) -> std::io::Result<()> { + if self.json { + let line = serde_json::to_string(event).map_err(std::io::Error::other)?; + out.write_all(line.as_bytes())?; + out.write_all(b"\n")?; + return out.flush(); + } + + match event { + // Reasoning opens the dimmed block on the first piece; later pieces just continue it. + TurnEvent::Reasoning { text } => { + if self.show_thinking { + if !self.dimmed_open { + out.write_all(b"\x1b[2m")?; + self.dimmed_open = true; + } + out.write_all(text.as_bytes())?; + } + } + TurnEvent::Content { text } => { + if self.dimmed_open { + out.write_all(b"\x1b[0m\n")?; + self.dimmed_open = false; + } + if self.stream_content { + out.write_all(text.as_bytes())?; + } + } + TurnEvent::ToolCallStarted { name } => { + self.close_dimmed(out)?; + writeln!(out, "[tool {name}]")?; + } + TurnEvent::ToolResult { + name, + class, + truncated, + } => { + self.close_dimmed(out)?; + if *truncated { + writeln!(out, "[{name}: {class:?}, truncated]")?; + } else { + writeln!(out, "[{name}: {class:?}]")?; + } + } + TurnEvent::Waiting { slot_busy } => { + self.close_dimmed(out)?; + let state = if *slot_busy { "busy" } else { "idle" }; + writeln!(out, "[waiting: slot {state}]")?; + } + TurnEvent::Retrying { + attempt, + after_ms, + error, + } => { + self.close_dimmed(out)?; + writeln!( + out, + "[retrying: attempt {attempt} in {after_ms} ms: {error}]" + )?; + } + TurnEvent::ThinkingCapped { tokens } => { + self.close_dimmed(out)?; + writeln!(out, "[thinking capped at {tokens} tokens]")?; + } + TurnEvent::CacheLoss { expected, got } => { + self.close_dimmed(out)?; + writeln!(out, "[cache loss: {got} of {expected}]")?; + } + TurnEvent::Queued { .. } | TurnEvent::Progress { .. } => {} + } + out.flush() + } + + pub fn end_reasoning(&mut self, out: &mut dyn Write) -> std::io::Result<()> { + self.close_dimmed(out) + } + + fn close_dimmed(&mut self, out: &mut dyn Write) -> std::io::Result<()> { + if self.dimmed_open { + out.write_all(b"\x1b[0m\n")?; + self.dimmed_open = false; + } + Ok(()) + } +} diff --git a/crates/bxctl/src/lib.rs b/crates/bxctl/src/lib.rs index 13a8c97..a56686a 100644 --- a/crates/bxctl/src/lib.rs +++ b/crates/bxctl/src/lib.rs @@ -1 +1,3 @@ //! The owner's command-line tool. + +pub mod chat; diff --git a/crates/bxctl/src/main.rs b/crates/bxctl/src/main.rs index 75459df..0639e60 100644 --- a/crates/bxctl/src/main.rs +++ b/crates/bxctl/src/main.rs @@ -1,4 +1,274 @@ -fn main() { - eprintln!("bxctl: not implemented until M2"); - std::process::exit(2); +//! `bxctl`: the owner's chat client for `loopd`. + +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; +use std::process::ExitCode; + +use bxctl::chat::{ChatError, Printer, new_session_id, run_turn}; +use proto::{ErrorCode, SessionId, TurnDone, TurnEvent}; + +const USAGE: &str = + "usage: bxctl chat [--socket ] [--session ] [--no-thinking] [--say ] [--json]"; + +struct Options { + socket: PathBuf, + session: Option, + show_thinking: bool, + say: Option, + json: bool, +} + +fn main() -> ExitCode { + let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) != Some("chat") { + eprintln!("{USAGE}"); + return ExitCode::from(2); + } + match parse_chat(&args[1..]) { + Ok(opts) => run(&opts), + Err(()) => { + eprintln!("{USAGE}"); + ExitCode::from(2) + } + } +} + +fn parse_chat(args: &[String]) -> Result { + let mut opts = Options { + socket: default_socket(), + session: None, + show_thinking: true, + say: None, + json: false, + }; + let mut i = 0; + while i < args.len() { + match args[i].as_str() { + "--socket" => { + i += 1; + match args.get(i) { + Some(p) => opts.socket = PathBuf::from(p), + None => return Err(()), + } + } + "--session" => { + i += 1; + match args.get(i) { + Some(s) => match SessionId::new(s) { + Ok(id) => opts.session = Some(id), + Err(_) => return Err(()), + }, + None => return Err(()), + } + } + "--say" => { + i += 1; + match args.get(i) { + Some(t) => opts.say = Some(t.to_string()), + None => return Err(()), + } + } + "--no-thinking" => opts.show_thinking = false, + "--json" => opts.json = true, + _ => return Err(()), + } + i += 1; + } + Ok(opts) +} + +// The base directory: $BOXMAKER_HOME if set, otherwise the system location the brief fixes. +fn default_socket() -> PathBuf { + let base = std::env::var_os("BOXMAKER_HOME") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/var/lib/boxmaker")); + base.join("run").join("loop").join("loop.sock") +} + +// A writer that records the first io error it hits, so the on_event closure (which cannot return a +// Result) does not lose a write failure. The error is checked after run_turn returns. +struct Sink<'a> { + out: &'a mut (dyn Write + 'static), + err: Option, +} + +impl Sink<'_> { + fn event(&mut self, printer: &mut Printer, event: &TurnEvent) { + if self.err.is_none() { + self.err = printer.event(self.out, event).err(); + } + } +} + +// Runs one turn, streaming events through `printer` to `out`. Returns the turn outcome, or the +// first io error that occurred while writing. +fn stream_turn( + socket: &Path, + session: &SessionId, + text: &str, + resume: bool, + sink: &mut Sink, + printer: &mut Printer, +) -> Result, std::io::Error> { + let outcome = run_turn(socket, session, text, resume, &mut |event| { + sink.event(printer, event) + }); + // Flush any dimmed block left open before reporting the turn's result. + if sink.err.is_none() { + printer.end_reasoning(sink.out)?; + } + match sink.err.take() { + Some(e) => Err(e), + None => Ok(outcome), + } +} + +fn run(opts: &Options) -> ExitCode { + match &opts.say { + Some(text) => run_say(opts, text), + None => run_interactive(opts), + } +} + +fn run_say(opts: &Options, text: &str) -> ExitCode { + let session = match &opts.session { + Some(s) => s.clone(), + None => new_session_id(), + }; + let mut printer = Printer::new(opts.show_thinking, opts.json); + let code = run_turn_twice(opts, &session, text, opts.session.is_some(), &mut printer); + match code { + Ok(()) => ExitCode::SUCCESS, + Err(RunError::Io(e)) => { + eprintln!("bxctl: {e}"); + ExitCode::from(1) + } + Err(RunError::Chat(e)) => { + eprintln!("bxctl: {e}"); + ExitCode::from(1) + } + } +} + +// Runs a --say turn, retrying once when the named session does not exist. +fn run_turn_twice( + opts: &Options, + session: &SessionId, + text: &str, + resume: bool, + printer: &mut Printer, +) -> Result<(), RunError> { + let stderr = std::io::stderr(); + let mut handle = stderr.lock(); + let mut sink = Sink { + out: &mut handle, + err: None, + }; + match stream_turn(&opts.socket, session, text, resume, &mut sink, printer) { + Err(e) => Err(RunError::Io(e)), + Ok(Ok(done)) => { + write_answer(opts, &done).map_err(RunError::Io)?; + Ok(()) + } + Ok(Err(ChatError::Refused(w))) if w.code == ErrorCode::NoSuchSession => { + // The named session does not exist: retry once, creating it (resume=false). + let mut sink = Sink { + out: &mut handle, + err: None, + }; + match stream_turn(&opts.socket, session, text, false, &mut sink, printer) { + Err(e) => Err(RunError::Io(e)), + Ok(Ok(done)) => { + write_answer(opts, &done).map_err(RunError::Io)?; + Ok(()) + } + Ok(Err(e)) => Err(RunError::Chat(e)), + } + } + Ok(Err(e)) => Err(RunError::Chat(e)), + } +} + +fn run_interactive(opts: &Options) -> ExitCode { + let mut printer = Printer::new(opts.show_thinking, opts.json); + let mut session = None; + let stdin = std::io::stdin(); + let mut reader = BufReader::new(stdin.lock()); + loop { + eprint!("> "); + let _ = std::io::stderr().flush(); + let mut line = String::new(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => {} + Err(e) => { + eprintln!("bxctl: {e}"); + return ExitCode::from(1); + } + } + let text = line.trim(); + if text == "/quit" { + break; + } + if text.is_empty() { + continue; + } + let just_created = session.is_none(); + let this_session = session.get_or_insert_with(new_session_id); + // The first turn creates the session (resume=false); later turns resume it. + let resume = !just_created; + if just_created { + println!("session {}", this_session.as_str()); + } + let stderr = std::io::stderr(); + let mut handle = stderr.lock(); + let mut sink = Sink { + out: &mut handle, + err: None, + }; + match stream_turn( + &opts.socket, + this_session, + text, + resume, + &mut sink, + &mut printer, + ) { + Err(e) => { + eprintln!("bxctl: {e}"); + return ExitCode::from(1); + } + Ok(Ok(done)) => { + if let Err(e) = write_answer(opts, &done) { + eprintln!("bxctl: {e}"); + return ExitCode::from(1); + } + } + Ok(Err(e)) => { + eprintln!("bxctl: {e}"); + return ExitCode::from(1); + } + } + } + ExitCode::SUCCESS +} + +// The plain answer goes to stdout; in --json mode the TurnDone goes to stderr after the events. +fn write_answer(opts: &Options, done: &TurnDone) -> std::io::Result<()> { + if opts.json { + let line = serde_json::to_string(done).map_err(std::io::Error::other)?; + let mut err = std::io::stderr().lock(); + err.write_all(line.as_bytes())?; + err.write_all(b"\n")?; + } + let stdout = std::io::stdout(); + let mut out = stdout.lock(); + out.write_all(done.content.as_bytes())?; + out.write_all(b"\n")?; + out.flush() +} + +enum RunError { + Io(std::io::Error), + Chat(ChatError), } diff --git a/crates/bxctl/tests/chat.rs b/crates/bxctl/tests/chat.rs new file mode 100644 index 0000000..fa3e845 --- /dev/null +++ b/crates/bxctl/tests/chat.rs @@ -0,0 +1,419 @@ +//! 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>>, +} + +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, 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 { + 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 { + 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 = + 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)); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 07ef66b..dd90d4f 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -38,6 +38,7 @@ reviewer adds findings under "Reviews" once per milestone. | M2b/06-loopd-turn | 2026-09-18 | done | 1 | pass | none | Wrote crates/loopd/src/turn.rs (299 lines) and registered `pub mod turn;` in lib.rs; copied the two given tests and three fixtures byte-identical. `TurnError` (SessionFull/TurnLimit/Infer/Session, Display + std::error::Error + From), `TurnOutcome`, `Runtime`, `is_context_full` (the one 400 whose JSON `error.type` is `exceed_context_size_error`), and `run_turn`: append User, build the ChatRequest (slot, messages, tools, thinking), capture `last_usage`, `chat_with_retry` mapping ChatEvent->TurnEvent (dropping ToolCallDelta), append Assistant then Usage, report cache loss between the two conversations, and on no tool calls return `TurnOutcome { content: completion.content.unwrap_or_default(), usage }`; otherwise iterate tool calls under the iteration cap with a repeated-call detector (first repeat returns "already called", a second repeat is TurnLimit), `cap_result`, and `dispatch` (find_tool/call_tool local, every other tool — including read_file — to the port). `run_call` maps Dispatch::Local and every ToolResponse variant to (text, Public, untrusted). Two compile fixes before the gate: `u64::try_from(*ahead).unwrap_or(u64::MAX)` (usize has no From) and `let Ok(value) = from_str(body) else { return false }` (a temporary borrow); `session.baseline()` returns a reference so it is bound inside the loop. All 6 turn and 9 limits tests pass; `make gate` prints `gate: ok`. | ? | | M2b/07-loopd-channel | 2026-09-18 | done | 3 | fail | none | Wrote crates/loopd/src/channel.rs and registered `pub mod channel;` in lib.rs. `Context` holds a private `Mutex>`; `serve` accepts forever with one thread per connection and returns on an accept error; `handle` does the seven steps (read_frame with Closed-before-anything, turn-only, mark busy, open-or-create plus run_turn streaming events, release before the final frame, error-code mapping, quiet write failure). The busy guard is a `Held` struct that borrows the context immutably and holds a clone of the id but never the lock, and it is dropped before sending turn_done or error so a client can send the next turn the moment it reads the last one — that is what keeps `a_busy_session_is_refused_at_once` and the concurrent-session test correct. Channel test reported `6 passed` ten runs in a row, all clean. Two fixes before a clean gate: `cargo fmt` import order and a clippy `question_mark` on the accept loop, re-run after each. | ? | | M2b/08-loopd-serve | 2026-09-18 | done | 2 | fail | none | Rewrote crates/loopd/src/main.rs into two commands, `selftest` and `serve`, both sharing `run_selftest_check` so the self-test lines are identical. `serve` loads config (exit 1 on failure), removes an existing socket via `channel_socket()` before the self-test, runs the self-test and exits 1 without binding on failure, then creates the socket's parent dir, binds, sets mode 0600 with `std::fs::set_permissions`, prints `serving on`, and calls `channel::serve` with a `Context` from the config, client, `Box::new(FakeTools::new())` and `Registry::m2b()`. Anything else prints both usages and exits 2. The `serve_refuses...` test's "no socket left behind" holds because the socket is removed before the self-test and binding happens only after it passes. First gate run failed on two clippy `collapsible_if` lints; collapsed the two nested `if let` into edition-2024 let-chains and re-ran, which passed. `cargo test -p loopd --test serve` reports 3 passed. | ? | +| M2b/09-bxctl-chat | 2026-09-18 | done | 5 | fail | none | Wrote crates/bxctl/src/chat.rs: `run_turn` (open socket, one `write_frame` with id 1, loop `read_frame` asserting id 1, dispatch final `TurnDone`/`Error` and non-final `TurnEvent` to `on_event`, every other frame `Protocol`); `ChatError` (Connect/Frame/Refused/Protocol) with `source()` returning the io and FrameError; `new_session_id` = `chat--` via two `expect`s (the epoch check and a private-field construction that cannot fail); `Printer` with json mode (one serde_json line per event, no skipping, no escape codes), a dimmed reasoning block opened on the first `Reasoning` and closed on the next non-reasoning event or `end_reasoning`, and every other event kind named exactly. Registered `pub mod chat;` in lib.rs. Rewrote main.rs into a `chat` subcommand: usage + exit 2 for a wrong first arg or unknown flag/missing value/invalid id, `$BOXMAKER_HOME/run/loop/loop.sock` else `/var/lib/boxmaker/...`, `--say` (events to stderr, answer to stdout, resume=true then one retry with resume=false on no_such_session), interactive (create on first turn, resume on the rest, `/quit` stops, the created session id printed once to stdout), `--json` (events to stderr, the TurnDone also to stderr after them, plain answer to stdout). A `Sink` records the first write error so the `on_event` closure (which cannot return a Result) does not lose it. All 11 chat tests pass. Four gate runs before clean: clippy `io_other_error` (switched to `Error::other`), then `redundant_closure` twice (the `other` map and `get_or_insert_with`), then a rustfmt import-order diff. | ? | Committed Cargo.lock alongside bxctl: adding serde_json to bxctl's Cargo.toml changes the workspace lock, and the gate's `--locked` deny check would otherwise fail on the checked-out tree. | ## Reviews