boxmaker/crates/memoryd/tests/honcho_serve.rs

//! `memoryd` deciding turns as it serves (M5c spec, section 5): `loopd` has its answer to
//! `session_updated` before `brokerd` is asked, and a silent `brokerd` holds up neither `loopd`
//! nor a memory call; without `[honcho]` nothing is asked or written; a damaged ledger stops the
//! start with its pointer, and a cut-off last line is dropped and logged. Every wait has a
//! deadline. (From M5c-1 task 11 a Honcho worker runs beside it; nothing here waits for it, and the
//! cut-off line is a skipped turn, so that it has nothing to send.) Do not edit.

#[path = "support/brokerd.rs"]
mod brokerd;
#[path = "support/logs.rs"]
mod logs;
#[path = "support/tmp.rs"]
mod tmp;
#[path = "support/turns.rs"]
mod turns;

use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use brokerd::{Answer, FakeBrokerd};
use bxcore::{
    CallId, DataClass, Envelope, MemoryCall, Message, SessionId, SessionUpdated, ToolRequest,
    read_frame, write_frame,
};
use memoryd::config::Config;
use memoryd::honcho::ledger::LedgerError;
use memoryd::serve::{self, StartError};
use tmp::TempDir;
use turns::{clean, write_turns};

const SECTION: &str = "[honcho]
url = \"http://127.0.0.1:9\"
agent = \"boxmaker0\"
owner = \"kyle\"
trusted = \"boxmaker0-trusted\"
untrusted = \"boxmaker0-untrusted\"
fleet = \"fleet\"
";

const DECIDED: &str = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"mm-a","turn":1},"action":{"type":"turn","workspace":"trusted","speaker":"kyle"}}"#;

/// When sync began: every ledger's first line (M5c-1 task 15).
const BEGAN: &str = r#"{"type":"began","time":"2026-09-28T09:00:00.000Z"}"#;

/// A turn decided to go nowhere: the worker has nothing to send for it.
const SKIPPED: &str = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"mm-a","turn":1},"action":{"type":"skip","why":"the turn has no single speaker"}}"#;

struct Home {
    config: Config,
    text: String,
    dir: TempDir,
}

impl Home {
    fn new(tag: &str, honcho: bool) -> Home {
        let dir = TempDir::new(tag);
        let soul = dir.write("etc/soul.md", "You are Boxmaker.\n");
        let text = format!(
            "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n{}",
            dir.path().display(),
            soul.display(),
            if honcho { SECTION } else { "" }
        );
        Home {
            config: Config::parse(&text).unwrap(),
            text,
            dir,
        }
    }

    fn ledger(&self) -> PathBuf {
        self.dir.path().join("memory/honcho.jsonl")
    }

    fn ledger_text(&self) -> String {
        std::fs::read_to_string(self.ledger()).unwrap_or_default()
    }

    /// `serve::start`, run on a thread; its log lines are collected.
    fn serve(&self) -> Arc<Mutex<Vec<String>>> {
        let lines = Arc::new(Mutex::new(Vec::new()));
        let sink = Arc::clone(&lines);
        let log: serve::Log = Arc::new(move |l: &str| sink.lock().unwrap().push(l.to_string()));
        let running = serve::start(self.config.clone(), log).unwrap();
        std::thread::spawn(move || running.run());
        lines
    }
}

/// One request and its answer, with a 10 s limit; how long the answer took.
fn ask(socket: &Path, msg: Message) -> (Message, Duration) {
    let started = Instant::now();
    let mut stream = UnixStream::connect(socket).unwrap();
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .unwrap();
    let env = Envelope {
        v: 1,
        id: 7,
        r#final: true,
        msg,
    };
    write_frame(&mut stream, &env).unwrap();
    let reply = read_frame(&mut stream).unwrap();
    (reply.msg, started.elapsed())
}

fn updated(home: &Home, session: &str, turn: u64) -> (Message, Duration) {
    let msg = Message::SessionUpdated(SessionUpdated {
        session: SessionId::new(session).unwrap(),
        turn,
    });
    ask(&home.config.loop_socket(), msg)
}

/// Waits up to 10 s for `done`; true if it came.
fn within_10s(done: impl Fn() -> bool) -> bool {
    let deadline = Instant::now() + Duration::from_secs(10);
    while Instant::now() < deadline {
        if done() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    false
}

#[test]
fn a_reported_turn_is_decided_after_loopd_has_its_answer() {
    let home = Home::new("serve-decide", true);
    write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]);
    write_turns(home.dir.path(), "mm-z", &[turns::finished(Some("kyle"))]);
    let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::State(clean()));
    let _lines = home.serve();
    // A report on another socket is refused, and decides nothing.
    let wrong = Message::SessionUpdated(SessionUpdated {
        session: SessionId::new("mm-z").unwrap(),
        turn: 1,
    });
    let (answer, _) = ask(&home.config.admin_socket(), wrong);
    assert!(matches!(answer, Message::Error(_)), "{answer:?}");
    let (answer, _) = updated(&home, "mm-a", 1);
    assert!(matches!(answer, Message::Ok(_)), "{answer:?}");
    assert!(within_10s(|| home.ledger_text().contains("\"turn\":1}")));
    // The line after the began line (M5c-1 task 15): as `DECIDED`, but for its time, which is the
    // decision's.
    let text = home.ledger_text();
    assert_eq!(text.lines().count(), 2, "{text}");
    assert!(text.starts_with(r#"{"type":"began","#), "{text}");
    let mut got: serde_json::Value = serde_json::from_str(text.lines().nth(1).unwrap()).unwrap();
    let mut want: serde_json::Value = serde_json::from_str(DECIDED).unwrap();
    got["time"] = serde_json::Value::Null;
    want["time"] = serde_json::Value::Null;
    assert_eq!(got, want);
    assert_eq!(fake.seen().len(), 1);
}

#[test]
fn a_silent_brokerd_holds_up_neither_loopd_nor_memory() {
    let home = Home::new("serve-silent", true);
    write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]);
    write_turns(home.dir.path(), "mm-b", &[turns::finished(Some("kyle"))]);
    let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::Hang);
    let lines = home.serve();
    let (answer, took) = updated(&home, "mm-a", 1);
    assert!(matches!(answer, Message::Ok(_)), "{answer:?}");
    assert!(took < Duration::from_secs(1), "loopd waited {took:?}");
    assert!(fake.wait_for(1), "brokerd was never asked");
    // While the question hangs: another report, and a memory call.
    let (answer, took) = updated(&home, "mm-b", 1);
    assert!(matches!(answer, Message::Ok(_)), "{answer:?}");
    assert!(took < Duration::from_secs(1), "loopd waited {took:?}");
    let call = Message::MemoryCall(MemoryCall {
        request: ToolRequest {
            session: SessionId::new("mm-b").unwrap(),
            call: CallId(1),
            tool: "recall".to_string(),
            arguments: r#"{"query":"hello"}"#.to_string(),
        },
        taint: DataClass::Private,
        untrusted: false,
    });
    let (answer, took) = ask(&home.config.broker_socket(), call);
    assert!(matches!(answer, Message::ToolResponse(_)), "{answer:?}");
    assert!(
        took < Duration::from_secs(1),
        "the memory call waited {took:?}"
    );
    // After 5 s without a byte, the outage is logged once, and nothing is decided.
    let pointer = "turns wait undecided; see docs/runbook.md#broker-state-unavailable";
    assert!(within_10s(|| lines
        .lock()
        .unwrap()
        .iter()
        .any(|l| l.contains(pointer))));
    let logged = lines.lock().unwrap().clone();
    let outage: Vec<&String> = logged.iter().filter(|l| l.contains(pointer)).collect();
    assert_eq!(outage.len(), 1, "{logged:?}");
    let start = format!("memoryd: {}: ", home.config.state_socket().display());
    assert!(outage[0].starts_with(&start), "{}", outage[0]);
    assert!(
        !home.ledger_text().contains("decided"),
        "{}",
        home.ledger_text()
    );
}

#[test]
fn without_honcho_nothing_is_asked_or_written() {
    let home = Home::new("serve-off", false);
    write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]);
    let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::State(clean()));
    let lines = home.serve();
    let (answer, _) = updated(&home, "mm-a", 1);
    assert!(matches!(answer, Message::Ok(_)), "{answer:?}");
    // A negative: nothing may happen, so there is nothing to wait for but a while.
    std::thread::sleep(Duration::from_millis(300));
    assert!(fake.seen().is_empty());
    assert!(!home.ledger().exists());
    assert!(
        lines.lock().unwrap().is_empty(),
        "{:?}",
        lines.lock().unwrap()
    );
}

#[test]
fn a_damaged_ledger_stops_the_start() {
    let home = Home::new("serve-damaged", true);
    std::fs::create_dir_all(home.dir.path().join("memory")).unwrap();
    std::fs::write(home.ledger(), "not json\n").unwrap();
    let log: serve::Log = Arc::new(|_: &str| {});
    let err = match serve::start(home.config.clone(), log) {
        Err(err) => err,
        Ok(_) => panic!("memoryd started"),
    };
    assert!(
        matches!(
            err,
            StartError::Ledger(LedgerError::Damaged { line: 1, .. })
        ),
        "{err:?}"
    );
    let shown = err.to_string();
    assert!(shown.starts_with(&format!("{}:1: ", home.ledger().display())));
    assert!(
        shown.ends_with("; see docs/runbook.md#honcho-ledger-damaged"),
        "{shown}"
    );
    assert_eq!(shown.matches("see docs/runbook.md").count(), 1, "{shown}");
    assert!(!home.config.loop_socket().exists(), "a socket was bound");
    // A ledger that cannot be read is named with the same pointer.
    std::fs::remove_file(home.ledger()).unwrap();
    std::fs::create_dir(home.ledger()).unwrap();
    let log: serve::Log = Arc::new(|_: &str| {});
    let err = match serve::start(home.config.clone(), log) {
        Err(err) => err,
        Ok(_) => panic!("memoryd started"),
    };
    assert!(
        matches!(err, StartError::Ledger(LedgerError::Io(..))),
        "{err:?}"
    );
    let shown = err.to_string();
    assert!(shown.starts_with(&format!("{}: ", home.ledger().display())));
    assert!(
        shown.ends_with("; see docs/runbook.md#honcho-ledger-damaged"),
        "{shown}"
    );
}

#[test]
fn a_cut_off_last_line_is_dropped_and_logged() {
    let home = Home::new("serve-repaired", true);
    std::fs::create_dir_all(home.dir.path().join("memory")).unwrap();
    std::fs::write(
        home.ledger(),
        format!("{BEGAN}\n{SKIPPED}\n{{\"type\":\"se"),
    )
    .unwrap();
    let lines = home.serve();
    assert_eq!(
        *lines.lock().unwrap(),
        vec![format!(
            "memoryd: {}: a last line cut off by a crash was dropped",
            home.ledger().display()
        )]
    );
    assert_eq!(home.ledger_text(), format!("{BEGAN}\n{SKIPPED}\n"));
}

/// A `memoryd` process, killed and waited for when dropped: declared after its `Home`, so it
/// drops first.
struct Daemon(std::process::Child);

impl Drop for Daemon {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

#[test]
fn memoryd_exits_with_the_ledgers_line() {
    let home = Home::new("serve-exit", true);
    let config = home.dir.write("memoryd.toml", &home.text);
    std::fs::create_dir_all(home.dir.path().join("memory")).unwrap();
    std::fs::write(home.ledger(), "not json\n").unwrap();
    let mut child = Daemon(
        Command::new(env!("CARGO_BIN_EXE_memoryd"))
            .args(["serve", "--config"])
            .arg(&config)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap(),
    );
    let deadline = Instant::now() + Duration::from_secs(10);
    let mut code = None;
    while Instant::now() < deadline {
        if let Ok(Some(status)) = child.0.try_wait() {
            code = status.code();
            break;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    let _ = child.0.kill();
    let _ = child.0.wait();
    let mut stderr = String::new();
    if let Some(mut pipe) = child.0.stderr.take() {
        let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr);
    }
    assert_eq!(code, Some(1), "{stderr}");
    let start = format!("memoryd: {}:1: ", home.ledger().display());
    assert!(stderr.starts_with(&start), "{stderr}");
    assert!(
        stderr.ends_with("; see docs/runbook.md#honcho-ledger-damaged\n"),
        "{stderr}"
    );
    assert_eq!(stderr.lines().count(), 1, "{stderr}");
    assert!(!home.config.loop_socket().exists(), "a socket was bound");
}