//! When sync began, as `memoryd` serves (M5c spec, section 5): the `began` line written once, at //! the first start, before any socket is bound; a second start leaves it as it is; and a turn //! logged before it, reported after, is skipped. Every wait has a deadline. 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::PathBuf; use std::sync::Arc; use std::time::{Duration, Instant}; use brokerd::{Answer, FakeBrokerd}; use bxcore::{Envelope, Message, SessionId, SessionUpdated, Timestamp, read_frame, write_frame}; use logs::assistant; use memoryd::config::Config; use memoryd::honcho::decide::BEFORE_SYNC; use memoryd::honcho::ledger::{Action, Ledger, Record}; use memoryd::honcho::start; use memoryd::serve; use tmp::TempDir; use turns::{clean, spoken_at, 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\" "; fn config(dir: &TempDir) -> Config { let soul = dir.write("etc/soul.md", "You are Boxmaker.\n"); let text = format!( "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n{SECTION}", dir.path().display(), soul.display() ); Config::parse(&text).unwrap() } fn ledger(dir: &TempDir) -> PathBuf { dir.path().join("memory/honcho.jsonl") } fn records(dir: &TempDir) -> Vec { std::fs::read_to_string(ledger(dir)) .unwrap_or_default() .lines() .map(|l| serde_json::from_str(l).unwrap()) .collect() } #[test] fn the_first_start_writes_when_sync_began_once() { let dir = TempDir::new("began-start"); let config = config(&dir); let before = Timestamp::now(); let log: serve::Log = Arc::new(|_: &str| {}); let running = serve::start(config.clone(), log).unwrap(); // `start` has returned: the sockets are bound, and the line is already on disk. assert!(config.loop_socket().exists()); let began = match records(&dir).as_slice() { [Record::Began { time }] => *time, other => panic!("{other:?}"), }; assert!(began >= before && began <= Timestamp::now()); drop(running); // A start again (the ledger opened and begun as at start) leaves it as it is. let honcho = config.honcho_config().unwrap().unwrap(); let log: serve::Log = Arc::new(|_: &str| {}); let opened = start::open(&config, &honcho, &log).unwrap(); opened.begin().unwrap(); assert_eq!(records(&dir), vec![Record::Began { time: began }]); assert_eq!(Ledger::open(&ledger(&dir)).unwrap().began(), Some(began)); } #[test] fn a_turn_from_before_sync_began_is_skipped() { let dir = TempDir::new("began-old"); let config = config(&dir); let old = Timestamp::parse("2026-09-01T10:00:00.000Z").unwrap(); write_turns( dir.path(), "mm-a", &[vec![ spoken_at(old, "hello", Some("kyle")), assistant(2, Some("hi"), &[]), ]], ); let _brokerd = FakeBrokerd::start(&config.state_socket(), Answer::State(clean())); let log: serve::Log = Arc::new(|_: &str| {}); let running = serve::start(config.clone(), log).unwrap(); std::thread::spawn(move || running.run()); let mut stream = UnixStream::connect(config.loop_socket()).unwrap(); stream .set_read_timeout(Some(Duration::from_secs(10))) .unwrap(); let report = Envelope { v: 1, id: 1, r#final: true, msg: Message::SessionUpdated(SessionUpdated { session: SessionId::new("mm-a").unwrap(), turn: 1, }), }; write_frame(&mut stream, &report).unwrap(); assert!(matches!( read_frame(&mut stream).unwrap().msg, Message::Ok(_) )); let deadline = Instant::now() + Duration::from_secs(10); let skipped = Action::Skip { why: BEFORE_SYNC.to_string(), }; let decided = || { records(&dir) .into_iter() .any(|r| matches!(&r, Record::Decided { action, .. } if *action == skipped)) }; while Instant::now() < deadline && !decided() { std::thread::sleep(Duration::from_millis(10)); } assert!(decided(), "{:?}", records(&dir)); }