Add bxctl chat

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

Cargo.lock

@@ -14,6 +14,7 @@ name = "bxctl"
 version = "0.1.0"
 dependencies = [
  "proto",
+ "serde_json",
 ]
 
 [[package]]

crates/bxctl/Cargo.toml

@@ -10,3 +10,4 @@ workspace = true
 
 [dependencies]
 proto.workspace = true
+serde_json.workspace = true

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-<digits>-<digits>" 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-<digits>-<digits> 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<TurnDone, ChatError> {
+    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(())
+    }
+}

crates/bxctl/src/lib.rs

@@ -1 +1,3 @@
 //! The owner's command-line tool.
+
+pub mod chat;

crates/bxctl/src/main.rs

@@ -1,4 +1,274 @@
-fn main() {
+//! `bxctl`: the owner's chat client for `loopd`.
-    eprintln!("bxctl: not implemented until M2");
+
-    std::process::exit(2);
+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 <path>] [--session <id>] [--no-thinking] [--say <text>] [--json]";
+
+struct Options {
+    socket: PathBuf,
+    session: Option<SessionId>,
+    show_thinking: bool,
+    say: Option<String>,
+    json: bool,
+}
+
+fn main() -> ExitCode {
+    let args: Vec<String> = 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<Options, ()> {
+    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<std::io::Error>,
+}
+
+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<Result<TurnDone, ChatError>, 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),
 }

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<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 {
+                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));
+}

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<SessionError>), `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<u64>) 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<HashSet<SessionId>>`; `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-<secs>-<nanos>` 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