diff --git a/crates/memoryd/src/honcho/decide.rs b/crates/memoryd/src/honcho/decide.rs index 75062c7..b9af94f 100644 --- a/crates/memoryd/src/honcho/decide.rs +++ b/crates/memoryd/src/honcho/decide.rs @@ -95,7 +95,7 @@ pub fn has_secret(text: &str) -> bool { } /// The turn's user text and its speaker, from its first record, the `User` one. -fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) { +pub fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) { match turn.first() { Some(LogRecord::User { content, speaker, .. @@ -106,7 +106,7 @@ fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) { /// The turn's final reply: the content of its last `Assistant` record, when that record asks for /// no tool. A turn that stopped in a tool round, or before any answer, has none. -fn final_reply(turn: &[LogRecord]) -> Option<&str> { +pub fn final_reply(turn: &[LogRecord]) -> Option<&str> { let last = turn .iter() .rev() diff --git a/crates/memoryd/src/honcho/mod.rs b/crates/memoryd/src/honcho/mod.rs index d0845c9..04b9a03 100644 --- a/crates/memoryd/src/honcho/mod.rs +++ b/crates/memoryd/src/honcho/mod.rs @@ -7,3 +7,4 @@ pub mod ledger; pub mod naming; pub mod state; pub mod turns; +pub mod worker; diff --git a/crates/memoryd/src/honcho/turns.rs b/crates/memoryd/src/honcho/turns.rs index 32b3b50..e4cf716 100644 --- a/crates/memoryd/src/honcho/turns.rs +++ b/crates/memoryd/src/honcho/turns.rs @@ -112,13 +112,45 @@ impl TurnSync { &self.ledger } + /// The directory of `loopd`'s session logs. + pub fn sessions(&self) -> &Path { + &self.sessions + } + /// The sessions that still have undecided turns, with the highest turn reported for each. pub fn waiting(&self) -> Vec<(SessionId, u64)> { self.reported.values().cloned().collect() } - /// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk every session - /// with undecided turns, in the order of their ids, until one stops the walk. + /// After a restart: each session with a decided turn in the ledger waits again, at the turns + /// its log holds, so that a turn left undecided before the restart is decided at the next + /// walk (`decidable` holds back one still running). A session with no decided turn is left + /// to its next report: sessions from before Honcho was turned on are never walked. + pub fn resume(&mut self) { + let mut sessions: Vec = Vec::new(); + for (item, _) in self.ledger.decisions() { + if let Item::Turn { session, .. } = item + && !sessions.contains(session) + { + sessions.push(session.clone()); + } + } + for session in sessions { + if let Ok(records) = read_log(&log_path(&self.sessions, &session)) { + self.reported.insert( + session.as_str().to_string(), + (session.clone(), turns(&records)), + ); + } + } + } + + /// Append `record` to the ledger (the worker's `Sent` and `Failed`). + pub fn append(&mut self, record: Record) -> Result<(), ledger::LedgerError> { + self.ledger.append(record) + } + + /// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk. pub fn updated( &mut self, session: &SessionId, @@ -132,6 +164,12 @@ impl TurnSync { .entry(session.as_str().to_string()) .or_insert((session.clone(), 0)); entry.1 = entry.1.max(turn); + self.retry(ask, now, log); + } + + /// Walk every session with undecided turns, in the order of their ids, until one stops the + /// walk: at a report, and from the worker when it has nothing to send (M5c spec, section 5). + pub fn retry(&mut self, ask: Ask<'_>, now: Timestamp, log: &dyn Fn(&str)) { for (session, turn) in self.waiting() { match self.walk(&session, turn, ask, now, log) { Walk::Done => { diff --git a/crates/memoryd/src/honcho/worker.rs b/crates/memoryd/src/honcho/worker.rs new file mode 100644 index 0000000..8420442 --- /dev/null +++ b/crates/memoryd/src/honcho/worker.rs @@ -0,0 +1,444 @@ +//! The worker: one thread that sends what the ledger says is waiting, oldest first (M5c spec, +//! section 5). A turn is read again from `loopd`'s log, neutralised, cut into parts Honcho takes, +//! and added to its Honcho session as one batch; the ids Honcho gives go in the ledger. An outage +//! is waited out with a doubling wait and logged when it starts and when it ends; a refused send is +//! marked failed; a refused key stops that workspace until `memoryd` restarts. When nothing waits, +//! the worker decides what `brokerd` could not answer for before, then sleeps until a turn ends. + +use std::collections::HashSet; +use std::path::Path; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +use bxcore::sanitize::neutralise; +use bxcore::{SessionId, SessionStateReply, Timestamp}; +use roncho::MessageCreate; +use serde_json::{Map, json}; + +use crate::config::Config; +use crate::honcho::client::{Clients, HonchoError, Timeouts, classify}; +use crate::honcho::config::HonchoConfig; +use crate::honcho::decide::{final_reply, turn_records, user_text}; +use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace}; +use crate::honcho::naming::{HonchoSession, honcho_session}; +use crate::honcho::state; +use crate::honcho::turns::{LEDGER, TurnSync, log_path, read_log}; +use crate::serve::{Log, StartError}; + +/// The most characters Honcho takes in one message (recorded: 25,000 is taken, 25,001 refused). +pub const MAX_PART: usize = 25_000; +/// The most messages the worker puts in one batch. A turn that needs more is marked failed. +pub const MAX_BATCH: usize = 100; + +pub const UNREACHABLE: &str = "see docs/runbook.md#honcho-unreachable"; +pub const SEND_REFUSED: &str = "see docs/runbook.md#honcho-send-refused"; +pub const KEY_REFUSED: &str = "see docs/runbook.md#honcho-key-refused"; +pub const LOG_DAMAGED: &str = "see docs/runbook.md#session-log-damaged"; + +/// The worker's waits. Tests shorten them. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Tuning { + /// The first wait in an outage; each next one doubles, up to `backoff_max`. + pub backoff: Duration, + pub backoff_max: Duration, + /// How long the worker sleeps with nothing to send, unless a turn ends first. + pub idle: Duration, +} + +impl Tuning { + pub const DEFAULT: Tuning = Tuning { + backoff: Duration::from_secs(5), + backoff_max: Duration::from_secs(600), + idle: Duration::from_secs(30), + }; +} + +/// Wakes the worker from its idle sleep when a turn ends. +#[derive(Debug, Default)] +pub struct Wake { + woken: Mutex, + ready: Condvar, +} + +impl Wake { + /// Wake the worker if it sleeps, or make its next sleep return at once. + pub fn notify(&self) { + let mut woken = self.woken.lock().unwrap_or_else(|p| p.into_inner()); + *woken = true; + self.ready.notify_one(); + } + + /// Sleep until `notify`, or at most `timeout`. + pub fn wait(&self, timeout: Duration) { + let woken = self.woken.lock().unwrap_or_else(|p| p.into_inner()); + let (mut woken, _) = self + .ready + .wait_timeout_while(woken, timeout, |woken| !*woken) + .unwrap_or_else(|p| p.into_inner()); + *woken = false; + } +} + +/// What one step of the worker did, and so what the loop does next. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Step { + /// A turn was sent, or marked failed: go on at once. + Done, + /// Honcho is unreachable: wait this long, then try the same turn again. + Backoff(Duration), + /// Nothing to send, and the retry decided nothing new: sleep. + Idle, + /// The ledger cannot be written: the worker stops. + Stopped, +} + +/// `brokerd`'s answer for a session, for the retry. +pub type AskBox = Box Result + Send>; + +/// Why a turn could not be sent. +enum Unsent { + Honcho(HonchoError), + /// The turn cannot be read from the log any more, or is too long for one batch. + Local(&'static str), +} + +/// A turn to send: its item, workspace, speaker and Honcho session. +struct Next { + item: Item, + session: SessionId, + turn: u64, + workspace: Workspace, + speaker: String, + honcho: Option, +} + +pub struct Worker { + sync: Arc>, + clients: Clients, + wake: Arc, + ask: AskBox, + log: Log, + tuning: Tuning, + agent: String, + /// In an outage: the next wait. + outage: Option, + /// Workspaces whose key was refused: nothing more is sent to them until restart. + refused: HashSet, + stopped: bool, +} + +/// `text` in parts of at most `MAX_PART` characters, in order. Empty text has no part. +pub fn parts(text: &str) -> Vec { + let chars: Vec = text.chars().collect(); + chars + .chunks(MAX_PART) + .map(|chunk| chunk.iter().collect()) + .collect() +} + +/// Turn `turn`'s user text and final reply, read again from the log at `path`. `None` when the log +/// does not read, or has no such turn, or the turn no reply. +pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)> { + let records = read_log(path).ok()?; + let records = turn_records(&records, turn)?; + let (text, _) = user_text(records); + let reply = final_reply(records)?; + Some((text.to_string(), reply.to_string())) +} + +/// A turn's messages, in one batch: the speaker's text, then the agent's reply, each neutralised +/// and in parts of at most `MAX_PART` characters, each part with the turn's metadata. +pub fn batch( + text: &str, + reply: &str, + speaker: &str, + agent: &str, + session: &SessionId, + turn: u64, +) -> Vec { + let mut metadata = Map::new(); + metadata.insert("agent".to_string(), json!(agent)); + metadata.insert("boxmaker_session".to_string(), json!(session.as_str())); + metadata.insert("turn".to_string(), json!(turn)); + let mut messages = Vec::new(); + for (content, peer) in [(text, speaker), (reply, agent)] { + for part in parts(&neutralise(content)) { + messages.push(MessageCreate::new(part, peer).with_metadata(metadata.clone())); + } + } + messages +} + +/// The ids of the messages Honcho made, in order. +fn ids(messages: &[roncho::Message]) -> Vec { + messages.iter().map(|m| m.id.clone()).collect() +} + +impl Worker { + /// `agent` is `[honcho] agent`. `sync` and `wake` are shared with the socket threads. + pub fn new( + sync: Arc>, + clients: Clients, + wake: Arc, + ask: AskBox, + log: Log, + tuning: Tuning, + agent: &str, + ) -> Worker { + Worker { + sync, + clients, + wake, + ask, + log, + tuning, + agent: agent.to_string(), + outage: None, + refused: HashSet::new(), + stopped: false, + } + } + + /// Step until the ledger cannot be written; wait as each step says. + pub fn run(mut self) { + loop { + match self.step() { + Step::Done => {} + Step::Backoff(wait) => std::thread::sleep(wait), + Step::Idle => self.wake.wait(self.tuning.idle), + Step::Stopped => return, + } + } + } + + /// Send the oldest waiting turn, or, with none, retry the undecided ones. + pub fn step(&mut self) -> Step { + if self.stopped { + return Step::Stopped; + } + let Some(next) = self.next() else { + self.retry(); + return match self.next() { + Some(_) => Step::Done, + None => Step::Idle, + }; + }; + match self.send(&next) { + Ok(sent) => { + self.recovered(); + self.record(Record::Sent { + time: Timestamp::now(), + item: next.item, + ids: sent, + }) + } + Err(Unsent::Honcho(e @ HonchoError::Unreachable { .. })) => self.unreachable(&e), + Err(Unsent::Honcho(HonchoError::KeyRefused { status })) => { + self.recovered(); + self.refused.insert(next.workspace); + (self.log)(&format!( + "memoryd: Honcho refused the key of workspace {} (HTTP {status}); nothing more is sent to it until memoryd restarts; {KEY_REFUSED}", + self.clients.client(next.workspace).workspace_id() + )); + Step::Done + } + Err(Unsent::Honcho(e)) => { + self.recovered(); + (self.log)(&format!( + "memoryd: turn {} of session {} was not sent: {e}; it is marked failed; {SEND_REFUSED}", + next.turn, + next.session.as_str() + )); + self.record(Record::Failed { + time: Timestamp::now(), + item: next.item, + status: e.status(), + kind: e.kind().to_string(), + }) + } + Err(Unsent::Local(kind)) => { + let pointer = if kind == "log" { + LOG_DAMAGED + } else { + SEND_REFUSED + }; + (self.log)(&format!( + "memoryd: turn {} of session {} cannot be sent ({kind}); it is marked failed; {pointer}", + next.turn, + next.session.as_str() + )); + self.record(Record::Failed { + time: Timestamp::now(), + item: next.item, + status: None, + kind: kind.to_string(), + }) + } + } + } + + /// The oldest waiting turn whose workspace's key was not refused. Other actions wait. + fn next(&self) -> Option { + let sync = self.sync.lock().unwrap_or_else(|p| p.into_inner()); + let ledger = sync.ledger(); + ledger.outbox().into_iter().find_map(|(item, action)| { + let Item::Turn { session, turn } = &item else { + return None; + }; + let Action::Turn { workspace, speaker } = action else { + return None; + }; + if self.refused.contains(&workspace) { + return None; + } + Some(Next { + honcho: honcho_session(ledger.decisions(), session, *turn), + session: session.clone(), + turn: *turn, + item: item.clone(), + workspace, + speaker, + }) + }) + } + + /// Read the turn again, make sure of its Honcho session, and add its messages as one batch. + fn send(&mut self, next: &Next) -> Result, Unsent> { + let sessions = self + .sync + .lock() + .unwrap_or_else(|p| p.into_inner()) + .sessions() + .to_path_buf(); + let (text, reply) = read_turn(&log_path(&sessions, &next.session), next.turn) + .ok_or(Unsent::Local("log"))?; + let honcho = next.honcho.as_ref().ok_or(Unsent::Local("log"))?; + let messages = batch( + &text, + &reply, + &next.speaker, + &self.agent, + &next.session, + next.turn, + ); + if messages.len() > MAX_BATCH { + return Err(Unsent::Local("too-long")); + } + self.clients + .ensure_session(next.workspace, honcho) + .map_err(Unsent::Honcho)?; + let client = self.clients.client(next.workspace); + match client.add_messages(&honcho.name, &messages) { + Ok(made) => Ok(ids(&made)), + Err(e) => Err(Unsent::Honcho(classify(&e))), + } + } + + /// Honcho did not answer: log the outage's start once, and wait longer each time. + fn unreachable(&mut self, error: &HonchoError) -> Step { + let wait = match self.outage { + Some(wait) => wait, + None => { + (self.log)( + format!("memoryd: {error}; sends wait and are retried; {UNREACHABLE}").as_str(), + ); + self.tuning.backoff + } + }; + self.outage = Some(wait.saturating_mul(2).min(self.tuning.backoff_max)); + Step::Backoff(wait) + } + + /// Honcho answered: if that ends an outage, log it once, and decide what waited meanwhile. + fn recovered(&mut self) { + if self.outage.take().is_some() { + (self.log)("memoryd: Honcho answers again; waiting sends go on"); + self.retry(); + } + } + + /// Task 09's walk, for the turns `brokerd` could not answer for. + fn retry(&mut self) { + let mut sync = self.sync.lock().unwrap_or_else(|p| p.into_inner()); + sync.retry(&*self.ask, Timestamp::now(), &*self.log); + } + + /// Append the worker's record. A ledger that cannot be written stops the worker: the send + /// would be repeated at every step. + fn record(&mut self, record: Record) -> Step { + let result = self + .sync + .lock() + .unwrap_or_else(|p| p.into_inner()) + .append(record); + match result { + Ok(()) => Step::Done, + Err(e) => { + (self.log)( + format!( + "memoryd: {e}; the Honcho worker stops until memoryd restarts; {}", + ledger::RUNBOOK + ) + .as_str(), + ); + self.stopped = true; + Step::Stopped + } + } + } +} + +/// With `[honcho]`, at start: the ledger read through, the sessions that wait again, and the +/// clients with their keys read. Nothing is bound or started yet. +pub struct Opened { + pub sync: Arc>, + clients: Clients, + agent: String, +} + +/// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the +/// clients (their key files' warnings are logged). A key that cannot be read is `Secret`. +pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result { + let path = config.memory_dir().join(LEDGER); + let ledger = Ledger::open(&path).map_err(StartError::Ledger)?; + if ledger.repaired() { + (log)(&format!( + "memoryd: {}: a last line cut off by a crash was dropped", + path.display() + )); + } + let mut sync = TurnSync::new(ledger, &honcho.agent, &config.sessions_dir()); + sync.resume(); + let env = |name: &str| std::env::var_os(name); + let (clients, warnings) = + Clients::new(honcho, &env, Timeouts::WRITES).map_err(StartError::Secret)?; + for warning in warnings { + (log)(&warning); + } + Ok(Opened { + sync: Arc::new(Mutex::new(sync)), + clients, + agent: honcho.agent.clone(), + }) +} + +/// Start the worker's thread, asking `brokerd` on `config`'s state socket. +pub fn spawn(opened: Opened, config: &Config, wake: Arc, log: Log) -> Result<(), String> { + let socket = config.state_socket(); + let ask: AskBox = + Box::new(move |session: &SessionId| state::ask(&socket, session, state::WAIT)); + let worker = Worker::new( + opened.sync, + opened.clients, + wake, + ask, + log, + Tuning::DEFAULT, + &opened.agent, + ); + std::thread::Builder::new() + .name("honcho".to_string()) + .spawn(move || worker.run()) + .map(|_| ()) + .map_err(|e| format!("cannot start the Honcho worker: {e}")) +} diff --git a/crates/memoryd/src/main.rs b/crates/memoryd/src/main.rs index 30d5b44..e4bdfc3 100644 --- a/crates/memoryd/src/main.rs +++ b/crates/memoryd/src/main.rs @@ -44,8 +44,8 @@ fn main() -> ExitCode { let log: serve::Log = Arc::new(|line: &str| eprintln!("{line}")); let running = match serve::start(config, log) { Ok(running) => running, - // The ledger's line names its own pointer, `honcho-ledger-damaged`. - Err(e @ serve::StartError::Ledger(_)) => { + // The ledger's and a key's lines name their own pointers. + Err(e @ (serve::StartError::Ledger(_) | serve::StartError::Secret(_))) => { eprintln!("memoryd: {e}"); return ExitCode::from(1); } diff --git a/crates/memoryd/src/serve.rs b/crates/memoryd/src/serve.rs index e2054fc..2370d28 100644 --- a/crates/memoryd/src/serve.rs +++ b/crates/memoryd/src/serve.rs @@ -16,9 +16,10 @@ use bxcore::{ use crate::config::Config; use crate::edit::{self, Context}; use crate::files::Store; -use crate::honcho::ledger::{self, Ledger, LedgerError}; +use crate::honcho::ledger::{self, LedgerError}; use crate::honcho::state; -use crate::honcho::turns::{self, TurnSync}; +use crate::honcho::turns::TurnSync; +use crate::honcho::worker::{self, Wake}; use crate::index::{self, Index}; use crate::recall::Recalled; use crate::{journal, recall, review}; @@ -39,7 +40,9 @@ pub struct Memoryd { state: Mutex, /// With `[honcho]`: the turns reported and not yet decided, and the ledger. Never taken /// while `state` is held, and never held by a request that waits for its answer. - honcho: Option>, + honcho: Option>>, + /// Wakes the Honcho worker when a turn has been decided. + wake: Arc, log: Log, } @@ -49,12 +52,14 @@ pub struct Memoryd { pub enum StartError { Failed(String), Ledger(LedgerError), + /// A Honcho key that cannot be read: `bxcore::secrets`' line, with its own pointer. + Secret(String), } impl std::fmt::Display for StartError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - StartError::Failed(why) => write!(f, "{why}"), + StartError::Failed(why) | StartError::Secret(why) => write!(f, "{why}"), // `Damaged` names its pointer itself. StartError::Ledger(e @ LedgerError::Damaged { .. }) => write!(f, "{e}"), StartError::Ledger(e) => write!(f, "{e}; {}", ledger::RUNBOOK), @@ -83,28 +88,12 @@ enum Which { Admin, } -/// With `[honcho]`, read the ledger through first. Then create what is missing under `memory/`, -/// open the index and catch it up, and bind the three sockets. The error names what failed. +/// With `[honcho]`, read the ledger through and the keys first. Then create what is missing +/// under `memory/`, open the index and catch it up, bind the three sockets, and start the Honcho +/// worker. The error names what failed. pub fn start(config: Config, log: Log) -> Result { - let honcho = match config.honcho_config()? { - Some(honcho) => { - let path = config.memory_dir().join(turns::LEDGER); - let ledger = Ledger::open(&path).map_err(StartError::Ledger)?; - if ledger.repaired() { - (log)( - format!( - "memoryd: {}: a last line cut off by a crash was dropped", - path.display() - ) - .as_str(), - ); - } - Some(Mutex::new(TurnSync::new( - ledger, - &honcho.agent, - &config.sessions_dir(), - ))) - } + let opened = match config.honcho_config()? { + Some(honcho) => Some(worker::open(&config, &honcho, &log)?), None => None, }; let store = Store::new(&config.memory_dir()); @@ -122,12 +111,22 @@ pub fn start(config: Config, log: Log) -> Result { (broker_listener, Which::Broker), (admin_listener, Which::Admin), ]; + let wake = Arc::new(Wake::default()); + let honcho = match opened { + Some(opened) => { + let sync = Arc::clone(&opened.sync); + worker::spawn(opened, &config, Arc::clone(&wake), Arc::clone(&log))?; + Some(sync) + } + None => None, + }; Ok(Running { memoryd: Arc::new(Memoryd { config, store, state: Mutex::new(State { index }), honcho, + wake, log, }), listeners, @@ -233,6 +232,8 @@ impl Memoryd { Timestamp::now(), &*self.log, ); + drop(sync); + self.wake.notify(); } /// The answer to one request on one socket. A message sent to the wrong socket is refused. diff --git a/crates/memoryd/tests/honcho_serve.rs b/crates/memoryd/tests/honcho_serve.rs index 2392f75..34e0b99 100644 --- a/crates/memoryd/tests/honcho_serve.rs +++ b/crates/memoryd/tests/honcho_serve.rs @@ -2,7 +2,8 @@ //! `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. Do not edit. +//! 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; @@ -41,6 +42,9 @@ 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"}}"#; +/// 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, @@ -263,7 +267,7 @@ fn a_damaged_ledger_stops_the_start() { 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!("{DECIDED}\n{{\"type\":\"se")).unwrap(); + std::fs::write(home.ledger(), format!("{SKIPPED}\n{{\"type\":\"se")).unwrap(); let lines = home.serve(); assert_eq!( *lines.lock().unwrap(), @@ -272,7 +276,7 @@ fn a_cut_off_last_line_is_dropped_and_logged() { home.ledger().display() )] ); - assert_eq!(home.ledger_text(), format!("{DECIDED}\n")); + assert_eq!(home.ledger_text(), format!("{SKIPPED}\n")); } /// A `memoryd` process, killed and waited for when dropped: declared after its `Home`, so it diff --git a/crates/memoryd/tests/honcho_worker.rs b/crates/memoryd/tests/honcho_worker.rs new file mode 100644 index 0000000..c2e5c44 --- /dev/null +++ b/crates/memoryd/tests/honcho_worker.rs @@ -0,0 +1,380 @@ +//! The Honcho worker, stepped by hand against the fake Honcho (M5c spec, section 5): a turn sent +//! once as one batch, the speaker's parts then the agent's, neutralised, in parts of 25,000 +//! characters, to the `-2` session for a new speaker; an outage waited out with a doubling wait +//! and logged twice in all; a refused send marked failed; a refused key stopping its workspace +//! only; a turn the log no longer holds marked failed; other actions left waiting. The waits and +//! a restart are in `honcho_worker_more.rs`. Do not edit. + +#[path = "support/honcho.rs"] +mod honcho; +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; +#[path = "support/worker.rs"] +mod worker; + +use std::time::Duration; + +use bxcore::sanitize::TAGS; +use bxcore::{DataClass, SessionStateReply}; +use honcho::{FakeHoncho, Mode, Request}; +use logs::assistant; +use memoryd::honcho::ledger::{Item, Record}; +use memoryd::honcho::worker::{MAX_BATCH, MAX_PART, Step, parts}; +use serde_json::{Value, json}; +use turns::{sid, spoken}; +use worker::{FAST, Home}; + +/// The ids fixture 08 answers with. +const IDS: [&str; 2] = ["7HqDWzXcKAdw9tnElCNFg", "mjsyhwuGsXTh1_zG_tWRx"]; + +fn message_posts(fake: &FakeHoncho) -> Vec { + fake.requests() + .into_iter() + .filter(|r| r.path.ends_with("/messages")) + .collect() +} + +fn turn(text: &str, speaker: &str, reply: &str) -> Vec { + vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])] +} + +fn meta(session: &str, turn: u64) -> Value { + json!({ "agent": "boxmaker0", "boxmaker_session": session, "turn": turn }) +} + +fn sent(session: &str, turn: u64) -> impl Fn(&Record) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + move |r| matches!(r, Record::Sent { item: i, .. } if *i == item) +} + +#[test] +fn the_limits() { + assert_eq!(MAX_PART, 25_000); + assert_eq!(MAX_BATCH, 100); + assert!(parts("").is_empty()); + assert_eq!(parts("abc"), vec!["abc".to_string()]); + let long = "é".repeat(MAX_PART); + assert_eq!(parts(&long), vec![long.clone()], "characters, not bytes"); + let over = format!("{long}xy"); + assert_eq!(parts(&over), vec![long, "xy".to_string()]); +} + +#[test] +fn a_turn_is_sent_once_as_one_batch() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-once"); + home.log("mm-a", &[turn("hello", "kyle", "hi")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let posts = message_posts(&fake); + assert_eq!(posts.len(), 1); + assert_eq!( + posts[0].path, + "/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages" + ); + assert_eq!( + posts[0].body, + Some(json!({ "messages": [ + { "content": "hello", "peer_id": "kyle", "metadata": meta("mm-a", 1) }, + { "content": "hi", "peer_id": "boxmaker0", "metadata": meta("mm-a", 1) }, + ] })) + ); + // The session was made first, with both peers. + assert_eq!(fake.requests().len(), 5); + assert!(fake.requests()[3].path.ends_with("/sessions")); + let records = home.records(); + assert_eq!(records.len(), 2); + assert!(matches!(&records[1], Record::Sent { ids, .. } if *ids == IDS)); + fake.clear(); + assert_eq!(w.step(), Step::Idle); + assert!(fake.requests().is_empty()); + assert!(home.lines().is_empty(), "{:?}", home.lines()); +} + +#[test] +fn a_text_over_the_limit_is_sent_in_parts() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-parts"); + let text = "a".repeat(MAX_PART + 1); + let reply = "é".repeat(MAX_PART); + home.log("mm-a", &[turn(&text, "kyle", &reply)]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let body = message_posts(&fake)[0].body.clone().unwrap(); + let got: Vec<(usize, String)> = body["messages"] + .as_array() + .unwrap() + .iter() + .map(|m| { + ( + m["content"].as_str().unwrap().chars().count(), + m["peer_id"].as_str().unwrap().to_string(), + ) + }) + .collect(); + assert_eq!( + got, + vec![ + (MAX_PART, "kyle".to_string()), + (1, "kyle".to_string()), + (MAX_PART, "boxmaker0".to_string()), + ] + ); +} + +#[test] +fn both_texts_are_neutralised() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-neutral"); + let text = format!("a{}b", TAGS[0].text); + let reply = format!("c{}d", TAGS[1].text); + home.log("mm-a", &[turn(&text, "kyle", &reply)]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let body = message_posts(&fake)[0].body.clone().unwrap(); + let contents: Vec = body["messages"] + .as_array() + .unwrap() + .iter() + .map(|m| m["content"].as_str().unwrap().to_string()) + .collect(); + assert_eq!(contents.len(), 2); + for (content, tag) in contents.iter().zip([TAGS[0].text, TAGS[1].text]) { + assert!(!content.contains(tag)); + assert!(content.contains("<"), "not neutralised"); + } +} + +#[test] +fn a_new_speaker_goes_to_the_next_session() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-second"); + home.log( + "mm-a", + &[turn("one", "kyle", "a"), turn("two", "ana_b", "b")], + ); + home.report("mm-a", 2); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert_eq!(w.step(), Step::Done); + let posts = message_posts(&fake); + assert_eq!(posts.len(), 2); + assert!(posts[0].path.ends_with("/sessions/mm-a/messages")); + assert!(posts[1].path.ends_with("/sessions/mm-a-2/messages")); + let second = fake + .requests() + .into_iter() + .find(|r| r.path.ends_with("/sessions") && r.body.as_ref().unwrap()["id"] == "mm-a-2") + .unwrap(); + let peers = second.body.unwrap()["peers"].clone(); + assert_eq!(peers["kyle"]["observe_me"], true); + assert_eq!(peers["ana_b"]["observe_me"], true); + assert_eq!(peers["boxmaker0"]["observe_me"], false); +} + +#[test] +fn an_outage_is_waited_out_and_logged_twice_in_all() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-outage"); + home.log("mm-a", &[turn("private-alpha", "kyle", "a")]); + home.log("mm-b", &[turn("private-beta", "kyle", "b")]); + home.report("mm-a", 1); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(503)); + let waits: Vec = (0..6).map(|_| w.step()).collect(); + let ms = |n| Step::Backoff(Duration::from_millis(n)); + assert_eq!(waits, vec![ms(1), ms(2), ms(4), ms(8), ms(8), ms(8)]); + assert_eq!(home.lines().len(), 1, "{:?}", home.lines()); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + assert_eq!(w.step(), Step::Done); + let lines = home.lines(); + assert_eq!(lines.len(), 2, "{lines:?}"); + assert_eq!( + lines[0], + "memoryd: Honcho is unreachable (server, HTTP 503); sends wait and are retried; see docs/runbook.md#honcho-unreachable" + ); + assert_eq!( + lines[1], + "memoryd: Honcho answers again; waiting sends go on" + ); + // In order: mm-a, then mm-b. + let records = home.records(); + let a = records.iter().position(sent("mm-a", 1)).unwrap(); + let b = records.iter().position(sent("mm-b", 1)).unwrap(); + assert!(a < b); + // A new outage starts from the first wait again. + home.log("mm-c", &[turn("c", "kyle", "c")]); + home.report("mm-c", 1); + fake.set(Mode::Hang); + assert_eq!(w.step(), ms(1)); + assert!(home.lines()[2].contains("(timeout)"), "{:?}", home.lines()); +} + +#[test] +fn a_refused_send_is_marked_failed_and_the_next_goes_on() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-refused"); + home.log("mm-a", &[turn("private-alpha", "kyle", "a")]); + home.log("mm-b", &[turn("private-beta", "kyle", "b")]); + home.report("mm-a", 1); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(422)); + assert_eq!(w.step(), Step::Done); + let failed = home.records().into_iter().find_map(|r| match r { + Record::Failed { + item, status, kind, .. + } => Some((item, status, kind)), + _ => None, + }); + assert_eq!( + failed, + Some(( + Item::Turn { + session: sid("mm-a"), + turn: 1 + }, + Some(422), + "status".to_string() + )) + ); + let lines = home.lines(); + assert_eq!( + lines, + vec![ + "memoryd: turn 1 of session mm-a was not sent: Honcho refused the request (status, HTTP 422); it is marked failed; see docs/runbook.md#honcho-send-refused" + .to_string() + ] + ); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + assert!(home.records().iter().any(sent("mm-b", 1))); + assert_eq!(w.step(), Step::Idle, "the failed turn is not tried again"); +} + +#[test] +fn a_refused_key_stops_its_workspace_only() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-key"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.log("mm-b", &[turn("b", "kyle", "b")]); + home.report("mm-a", 1); + *home.answer.lock().unwrap() = Ok(SessionStateReply { + class: DataClass::Private, + untrusted: true, + }); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(401)); + assert_eq!(w.step(), Step::Done); + assert_eq!( + home.lines(), + vec![ + "memoryd: Honcho refused the key of workspace boxmaker0-trusted (HTTP 401); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused" + .to_string() + ] + ); + fake.set(Mode::Fixtures); + fake.clear(); + assert_eq!(w.step(), Step::Done); + assert!(home.records().iter().any(sent("mm-b", 1))); + assert!( + fake.requests() + .iter() + .all(|r| !r.path.contains("boxmaker0-trusted")) + ); + assert_eq!(w.step(), Step::Idle); + // The trusted turn waits: neither sent nor failed. + let sync = home.sync.lock().unwrap(); + assert_eq!(sync.ledger().outbox().len(), 1); + assert_eq!(sync.ledger().counts().failed, 0); +} + +#[test] +fn a_turn_the_log_no_longer_holds_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-gone"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let path = home.dir.path().join("sessions/mm-a/0.jsonl"); + std::fs::write(&path, "not json\n").unwrap(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "log" + )); + assert_eq!( + home.lines(), + vec![ + "memoryd: turn 1 of session mm-a cannot be sent (log); it is marked failed; see docs/runbook.md#session-log-damaged" + .to_string() + ] + ); + assert!(message_posts(&fake).is_empty()); +} + +#[test] +fn a_turn_without_its_reply_any_more_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-noreply"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + // The log now ends in a tool round: the reply is gone. + home.log("mm-a", &[turns::running(Some("kyle"))]); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "log" + )); + assert!(message_posts(&fake).is_empty()); +} + +#[test] +fn a_refused_key_ends_an_outage() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-key-outage"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(503)); + assert!(matches!(w.step(), Step::Backoff(_))); + fake.set(Mode::Status(403)); + assert_eq!(w.step(), Step::Done); + let lines = home.lines(); + assert_eq!(lines.len(), 3, "{lines:?}"); + assert_eq!( + lines[1], + "memoryd: Honcho answers again; waiting sends go on" + ); + assert!(lines[2].ends_with("see docs/runbook.md#honcho-key-refused")); +} + +#[test] +fn a_turn_too_long_for_one_batch_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-long"); + let text = "a".repeat(MAX_PART * MAX_BATCH); + home.log("mm-a", &[turn(&text, "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "too-long" + )); + assert!(fake.requests().is_empty()); +} diff --git a/crates/memoryd/tests/honcho_worker_more.rs b/crates/memoryd/tests/honcho_worker_more.rs new file mode 100644 index 0000000..95a0414 --- /dev/null +++ b/crates/memoryd/tests/honcho_worker_more.rs @@ -0,0 +1,327 @@ +//! The Honcho worker's other duties (M5c spec, section 5): with nothing to send it decides what +//! `brokerd` could not answer for; after a restart nothing is sent again, and a session with a +//! decided turn waits again (a session without one does not); a running turn is never decided; +//! actions of later tasks wait; a ledger that cannot be written stops it; no log line holds a +//! turn's text. Then `memoryd` itself: a reported turn reaches Honcho, woken at once, and a key +//! that cannot be read stops the start. Do not edit. + +#[path = "support/brokerd.rs"] +mod brokerd; +#[path = "support/honcho.rs"] +mod honcho; +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; +#[path = "support/worker.rs"] +mod worker; + +use std::process::{Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use brokerd::{Answer, FakeBrokerd}; +use bxcore::{Envelope, Message, SessionId, SessionUpdated, read_frame, write_frame}; +use honcho::{FakeHoncho, Mode}; +use logs::assistant; +use memoryd::config::Config; +use memoryd::honcho::ledger::{Item, Record}; +use memoryd::honcho::worker::{Step, Tuning}; +use memoryd::serve::{self, StartError}; +use tmp::TempDir; +use turns::{clean, running, sid, spoken}; +use worker::{FAST, Home}; + +fn turn(text: &str, speaker: &str, reply: &str) -> Vec { + vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])] +} + +fn messages(fake: &FakeHoncho) -> usize { + fake.requests() + .iter() + .filter(|r| r.path.ends_with("/messages")) + .count() +} + +fn is_sent(records: &[Record], session: &str, turn: u64) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + records + .iter() + .any(|r| matches!(r, Record::Sent { item: i, .. } if *i == item)) +} + +fn decided(records: &[Record], session: &str, turn: u64) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + records + .iter() + .any(|r| matches!(r, Record::Decided { item: i, .. } if *i == item)) +} + +#[test] +fn the_default_waits() { + assert_eq!( + Tuning::DEFAULT, + Tuning { + backoff: Duration::from_secs(5), + backoff_max: Duration::from_secs(600), + idle: Duration::from_secs(30), + } + ); +} + +#[test] +fn with_nothing_to_send_undecided_turns_are_decided() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-retry"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + *home.answer.lock().unwrap() = Err("brokerd is down".to_string()); + home.report("mm-a", 1); + assert!(!decided(&home.records(), "mm-a", 1)); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle, "brokerd still down"); + *home.answer.lock().unwrap() = Ok(clean()); + assert_eq!(w.step(), Step::Done, "decided now"); + assert!(decided(&home.records(), "mm-a", 1)); + assert_eq!(w.step(), Step::Done, "sent"); + assert!(is_sent(&home.records(), "mm-a", 1)); +} + +#[test] +fn an_outage_that_ends_is_followed_by_the_retry() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-after-outage"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.log("mm-b", &[turn("b", "kyle", "b")]); + home.report("mm-a", 1); + *home.answer.lock().unwrap() = Err("brokerd is down".to_string()); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(502)); + assert!(matches!(w.step(), Step::Backoff(_))); + *home.answer.lock().unwrap() = Ok(clean()); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + // mm-a is sent, and the outage's end decided mm-b before any idle step. + assert!(decided(&home.records(), "mm-b", 1)); +} + +#[test] +fn a_restart_sends_nothing_again_and_resumes_the_sessions_it_knows() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-restart"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + home.worker(&fake, FAST).step(); + assert_eq!(messages(&fake), 1); + // While memoryd is down: mm-a has a second turn, never reported; mm-z, never seen, one. + home.log("mm-a", &[turn("a", "kyle", "a"), turn("b", "kyle", "b")]); + home.log("mm-z", &[turn("z", "kyle", "z")]); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done, "mm-a's second turn decided"); + assert_eq!(w.step(), Step::Done, "and sent"); + assert_eq!(w.step(), Step::Idle); + let records = home.records(); + assert!(is_sent(&records, "mm-a", 2)); + assert!(!decided(&records, "mm-z", 1), "a session never reported"); + assert_eq!(messages(&fake), 2, "turn 1 was sent again"); +} + +#[test] +fn a_restart_never_decides_a_running_turn() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-running"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + home.worker(&fake, FAST).step(); + home.log("mm-a", &[turn("a", "kyle", "a"), running(Some("kyle"))]); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle); + assert!(!decided(&home.records(), "mm-a", 2)); +} + +#[test] +fn actions_of_later_tasks_wait() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-later"); + let conclude = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"conclude","workspace":"trusted","peer":"kyle","session":"mm-a","text":"a fact"}}"#; + std::fs::write(home.ledger_path(), format!("{conclude}\n")).unwrap(); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle); + assert_eq!(home.records().len(), 1, "the conclusion was failed or sent"); + assert!(fake.requests().is_empty()); +} + +#[test] +fn a_ledger_that_cannot_be_written_stops_the_worker() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-ledger"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + let ledger = home.ledger_path(); + let kept = std::fs::read_to_string(&ledger).unwrap(); + std::fs::remove_file(&ledger).unwrap(); + std::fs::create_dir(&ledger).unwrap(); + assert_eq!(w.step(), Step::Stopped); + assert_eq!(w.step(), Step::Stopped); + assert_eq!( + messages(&fake), + 1, + "sent once; its sent line could not be written" + ); + let lines = home.lines(); + assert_eq!(lines.len(), 1, "{lines:?}"); + assert!(lines[0].ends_with( + "; the Honcho worker stops until memoryd restarts; see docs/runbook.md#honcho-ledger-damaged" + )); + std::fs::remove_dir(&ledger).unwrap(); + std::fs::write(&ledger, kept).unwrap(); +} + +#[test] +fn no_log_line_holds_a_turns_text() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-quiet"); + for (session, status) in [("mm-a", 503), ("mm-b", 422), ("mm-c", 401)] { + home.log( + session, + &[turn("private-question", "kyle", "private-answer")], + ); + home.report(session, 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(status)); + w.step(); + } + let lines = home.lines(); + assert_eq!(lines.len(), 3, "{lines:?}"); + for line in lines { + assert!(!line.contains("private"), "{line}"); + } +} + +/// `memoryd serve`'s config for `home`, with `[honcho]` at `url`. +fn serve_config(home: &TempDir, url: &str, extra: &str) -> String { + let soul = home.write("etc/soul.md", "You are Boxmaker.\n"); + format!( + "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n[honcho]\nurl = \"{url}\"\nagent = \"boxmaker0\"\nowner = \"kyle\"\ntrusted = \"boxmaker0-trusted\"\nuntrusted = \"boxmaker0-untrusted\"\nfleet = \"fleet\"\n{extra}", + home.path().display(), + soul.display() + ) +} + +#[test] +fn memoryd_sends_a_reported_turn_at_once() { + let fake = FakeHoncho::start(); + let dir = TempDir::new("worker-serve"); + turns::write_turns(dir.path(), "mm-a", &[turn("a", "kyle", "a")]); + let config = Config::parse(&serve_config(&dir, &fake.url, "")).unwrap(); + let _brokerd = FakeBrokerd::start(&config.state_socket(), Answer::State(clean())); + 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(config.clone(), log).unwrap(); + std::thread::spawn(move || running.run()); + // The worker has slept since start (30 s when idle): the report wakes it. + std::thread::sleep(Duration::from_millis(100)); + let mut stream = std::os::unix::net::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); + while Instant::now() < deadline && messages(&fake) == 0 { + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(messages(&fake), 1, "{:?}", lines.lock().unwrap()); +} + +#[test] +fn a_key_that_cannot_be_read_stops_the_start() { + let fake = FakeHoncho::start(); + let dir = TempDir::new("worker-key-start"); + let text = serve_config( + &dir, + &fake.url, + "[secrets.honcho_trusted]\nenv = \"MEMORYD_TEST_UNSET_KEY\"\n", + ); + let config = Config::parse(&text).unwrap(); + let log: serve::Log = Arc::new(|_: &str| {}); + let err = match serve::start(config.clone(), log) { + Err(err) => err, + Ok(_) => panic!("memoryd started"), + }; + assert!(matches!(err, StartError::Secret(_)), "{err:?}"); + assert!(!config.loop_socket().exists(), "a socket was bound"); + // The process: exit 1 with the secret's own lines and pointer, not the start pointer. + let path = dir.write("memoryd.toml", &text); + let mut child = Daemon( + Command::new(env!("CARGO_BIN_EXE_memoryd")) + .args(["serve", "--config"]) + .arg(&path) + .env_remove("MEMORYD_TEST_UNSET_KEY") + .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}"); + assert!( + stderr.starts_with("memoryd: secret honcho_trusted: "), + "{stderr}" + ); + assert!( + stderr.ends_with("\nsee docs/runbook.md#secret-unavailable\n"), + "{stderr}" + ); + assert!(!stderr.contains("memoryd-start-failed"), "{stderr}"); +} + +/// A `memoryd` process, killed and waited for when dropped. +struct Daemon(std::process::Child); + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} diff --git a/crates/memoryd/tests/support/worker.rs b/crates/memoryd/tests/support/worker.rs new file mode 100644 index 0000000..d39869c --- /dev/null +++ b/crates/memoryd/tests/support/worker.rs @@ -0,0 +1,133 @@ +//! A Honcho worker for tests, stepped by hand: a home with session logs and a ledger, the fake +//! Honcho, `brokerd` as a closure whose answer the test sets, and the log lines collected. Do not +//! edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::collections::BTreeMap; +use std::ffi::OsString; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bxcore::{LogRecord, SessionStateReply, Timestamp}; +use memoryd::honcho::client::{Clients, Timeouts}; +use memoryd::honcho::config::HonchoConfig; +use memoryd::honcho::ledger::{Ledger, Record}; +use memoryd::honcho::turns::{LEDGER, TurnSync}; +use memoryd::honcho::worker::{Tuning, Wake, Worker}; +use memoryd::serve::Log; + +use crate::honcho::FakeHoncho; +use crate::tmp::TempDir; +use crate::turns::{clean, sid, write_turns}; + +pub const FAST: Tuning = Tuning { + backoff: Duration::from_millis(1), + backoff_max: Duration::from_millis(8), + idle: Duration::from_millis(1), +}; + +pub fn config(url: &str) -> HonchoConfig { + HonchoConfig { + url: url.to_string(), + agent: "boxmaker0".to_string(), + owner: "kyle".to_string(), + trusted: "boxmaker0-trusted".to_string(), + untrusted: "boxmaker0-untrusted".to_string(), + fleet: "fleet".to_string(), + read: Vec::new(), + keys: BTreeMap::new(), + } +} + +pub struct Home { + pub sync: Arc>, + /// What `brokerd` answers. + pub answer: Arc>>, + pub lines: Arc>>, + pub dir: TempDir, +} + +impl Home { + pub fn new(tag: &str) -> Home { + let dir = TempDir::new(tag); + std::fs::create_dir_all(dir.path().join("memory")).unwrap(); + Home { + sync: Arc::new(Mutex::new(TurnSync::new( + Ledger::open(&dir.path().join("memory").join(LEDGER)).unwrap(), + "boxmaker0", + &dir.path().join("sessions"), + ))), + answer: Arc::new(Mutex::new(Ok(clean()))), + lines: Arc::new(Mutex::new(Vec::new())), + dir, + } + } + + pub fn ledger_path(&self) -> PathBuf { + self.dir.path().join("memory").join(LEDGER) + } + + /// A new `TurnSync` over the ledger as it is on disk, resumed: what a restart makes. + pub fn restart(&mut self) { + let mut sync = TurnSync::new( + Ledger::open(&self.ledger_path()).unwrap(), + "boxmaker0", + &self.dir.path().join("sessions"), + ); + sync.resume(); + self.sync = Arc::new(Mutex::new(sync)); + } + + pub fn log(&self, session: &str, turns: &[Vec]) { + write_turns(self.dir.path(), session, turns); + } + + /// `loopd` reports turn `turn` of `session`: decided now, with `answer`. + pub fn report(&self, session: &str, turn: u64) { + let answer = Arc::clone(&self.answer); + let ask = move |_: &bxcore::SessionId| answer.lock().unwrap().clone(); + let lines = Arc::clone(&self.lines); + let log = move |line: &str| lines.lock().unwrap().push(line.to_string()); + let now = Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap(); + self.sync + .lock() + .unwrap() + .updated(&sid(session), turn, &ask, now, &log); + } + + pub fn worker(&self, fake: &FakeHoncho, tuning: Tuning) -> Worker { + let no_env = |_: &str| -> Option { None }; + let timeouts = Timeouts { + connect: Duration::from_secs(5), + read: Duration::from_millis(300), + }; + let (clients, _) = Clients::new(&config(&fake.url), &no_env, timeouts).unwrap(); + let answer = Arc::clone(&self.answer); + let lines = Arc::clone(&self.lines); + let log: Log = Arc::new(move |line: &str| lines.lock().unwrap().push(line.to_string())); + Worker::new( + Arc::clone(&self.sync), + clients, + Arc::new(Wake::default()), + Box::new(move |_: &bxcore::SessionId| answer.lock().unwrap().clone()), + log, + tuning, + "boxmaker0", + ) + } + + /// Every record of the ledger on disk, in order. + pub fn records(&self) -> Vec { + std::fs::read_to_string(self.ledger_path()) + .unwrap_or_default() + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect() + } + + pub fn lines(&self) -> Vec { + self.lines.lock().unwrap().clone() + } +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index ec4c5be..0de7c23 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M5c1/11-memoryd-honcho-worker | 2026-09-28 | done | 1 | pass | none | Copied the four given files from `docs/plans/M5c1/files/`: the tests `crates/memoryd/tests/honcho_worker.rs` and `honcho_worker_more.rs` (from `honcho_worker_more.rs-task11`), a new version of `crates/memoryd/tests/honcho_serve.rs` (from `honcho_serve.rs-task11`, replacing the tree's) and the support `crates/memoryd/tests/support/worker.rs` (from `worker.rs-task11`), then the skeleton `crates/memoryd/src/honcho/worker.rs`. Step 2 showed exactly the named failures: `honcho_worker` with three (E0432 for `memoryd::honcho::worker` in the test and in `support/worker.rs`, E0599 for the missing `resume` on `TurnSync`); `honcho_worker_more` with four (the same three plus E0599 for the missing `Secret` on `StartError`); every other target compiled and `honcho_serve` passed its 6 already. Made D1/D2 (`user_text`/`final_reply` `pub` in `decide.rs`), H1 (`pub mod worker;` after `pub mod turns;` in `honcho/mod.rs`), T1 (`sessions` in `impl TurnSync`), T2 (`resume` walking `decisions` deduped by id then reading each log, and `append` delegating to `ledger.append` before `updated`'s doc), T3 (`updated` calling `self.retry` then the new `retry`), S1 (imports: drop `Ledger`, add `worker::{self, Wake}`), S2 (`Memoryd` holds `Arc>` and a `wake`), S3 (`StartError::Secret(String)`), S4 (the `Display` arm), S5 (`start` calling `worker::open`), S6 (`wake` built and `worker::spawn` bound after the sockets), S7 (`wake` in the `Memoryd`), S8 (`decide_turns` dropping the lock then `wake.notify()`), M1 (`main.rs` matching both `Ledger` and `Secret`). Filled its eleven bodies in the order of the file, deleting each instruction comment and keeping the doc comments, each from its comment and `cargo check` after: `notify` (lock, set flag, `notify_one`); `wait` (lock, `wait_timeout_while(..., |woken| !*woken)`, clear the flag); `parts` (collect `chars`, `chunks(MAX_PART)`); `read_turn` (`read_log` ok, `turn_records`, `user_text`, `final_reply`, both `to_string`); `batch` (a `Map` with `agent`/`boxmaker_session`/`turn` in order, then for each of `(text,speaker)`/`(reply,agent)` the neutralised `parts`, each a `MessageCreate` with the metadata); `ids` (`m.id.clone()`); `next` (lock, `outbox().find_map`, `Item::Turn` and `Action::Turn` by `let .. else`, skip when `refused.contains(workspace)`, build `Next` with `honcho_session`); `unreachable` (the `outage` match, log the start once, double-and-cap the wait, `Backoff`); `recovered` (`outage.take()` ends it, log once, `retry`); `retry` (lock, `sync.retry`); `record` (`append`, `Ok` -> `Done`, `Err` logs with `ledger::RUNBOOK`, sets `stopped`, `Stopped`). After the last body `cargo check -p memoryd` is clean with no warning. Step 5 all pass: `honcho_worker` 12, `honcho_worker_more` 10, `honcho_serve` 6, and every unchanged suite. Step 6 all pass: the `thread::spawn`/`Builder::new` grep prints one line in `spawn`, the three runbook-pointer greps print the three constants, and the other-crates `git diff --stat` is empty. Appended the three runbook fragments (`runbook-honcho-unreachable`, `-send-refused`, `-key-refused`) with `{ echo; cat ... } >>`; `grep -c '^## honcho-'` printed 5. `make gate` printed `gate: ok` with 1300 tests (1299 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5c1/10-memoryd-honcho-client | 2026-09-28 | done | 1 | pass | none | Copied the given files from `docs/plans/M5c1/files/`: the tests `crates/memoryd/tests/honcho_client.rs`, `honcho_keys.rs` and `honcho_naming.rs` (from `honcho_naming.rs-task10`) and the support `crates/memoryd/tests/support/honcho.rs` (from `honcho.rs-task10`), and the directory `crates/memoryd/tests/fixtures/honcho/` (30 files, 29 JSON and a README). Step 2 showed exactly the named failures: three test targets (`honcho_client`, `honcho_keys`, `honcho_naming`) with E0432 for the `memoryd::honcho::client` and `...::naming` imports and E0599 for the missing `decisions` method on `Ledger`; every other target compiled. Made H1 (`pub mod client;` and `pub mod naming;` in `honcho/mod.rs`, the former first so the imports resolve and the latter last to keep the alphabetical order) and L1 (the `decisions` method before `outbox`, walking `order` with `.iter().filter_map` returning `(item, &self.entries.get(item)?.action)` for every decided item, `Skip` included), then copied the two skeletons; `cargo check` gave 26 warnings (unused `json`; unused `PeerCreate` and `WorkspaceCreate`; unused variables `workspace` five, `session` twice, the rest once each; the three `Clients` fields never read; the `Ensured::Workspace` and `Peer` variants never constructed). Filled its thirteen bodies in the order of each file, deleting each instruction comment and keeping the doc comments, each from its comment and `cargo check` after: in `client.rs` `kind` (one arm for the two `{ kind, .. }` variants, `"key-refused"` for `KeyRefused`), `status` (the two `{ status, .. }` arms, `Some(*status)` for `KeyRefused`), `fleet_configuration` (`Map::new`, insert each of `reasoning`, `summary`, `dream` -> `{ "enabled": false }`), `session_peers` (agent `observe_me(false)`, each speaker `observe_me(true)`), `created` (`Ok` and a `Refused` 409 are success), the free `client` (chain `.without_config_file()`, `.base_url`, `.workspace_id`, the two timeouts, `.api_key` only when there is a key, `.build()`), `key` (`config.keys.get(name)` else `None`, `secrets::load`, extend the warnings, return the secret), `Clients::client` (match the workspace to its field), `ensure_workspace` (the fleet one with `fleet_configuration`, the other two with defaults, remembered), `ensure_peer` (the agent's peer with `observe_me(false)`, remembered), `delete_conclusion` (Ok and a 404 are done); in `naming.rs` `session_name` (the id for `count <= 1`, `-` otherwise) and `honcho_session` (walk `decisions`, keep each `Turn` of the session's speakers each counted once in decision order, return when the turn-th is reached, else `None`). After the last `cargo check` is clean with no warning; `cargo test -p memoryd --test honcho_client --test honcho_keys --test honcho_naming` gives 13, 3 (one ignored, run by the third) and 5 pass. Step 6 all pass: `without_config_file` counts once, `expose()` appears once in the free `client`, the other-crates `git diff --stat` is empty. `make gate` printed `gate: ok` with 1278 tests (1257 before this task, plus the 21 new: 13, 3 and 5). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5c1/09-memoryd-honcho-turns | 2026-09-28 | done | 1 | pass | none | Copied the five given files from `docs/plans/M5c1/files/`: the tests `crates/memoryd/tests/honcho_state.rs`, `honcho_turns.rs` (from `honcho_turns.rs-task09`) and `honcho_serve.rs` (from `honcho_serve.rs-task09`) and the support `crates/memoryd/tests/support/brokerd.rs` and `support/turns.rs` (from `turns.rs-task09`). Step 2 showed exactly the named failures: three test targets (`honcho_state`, `honcho_turns`, `honcho_serve`) with one E0432 each (`unresolved import memoryd::honcho::state`, `...::turns` and `memoryd::serve::StartError`); every other target compiled. Made H1 (`pub mod state;` and `pub mod turns;` after `pub mod ledger;` in `honcho/mod.rs`) and copied the two skeletons; `cargo check` gave 19 warnings (four unused imports: `UnixStream`; the five `bxcore` frame items plus `crate::honcho::state` and `turn_records`/`turns`; fourteen unused variables: `session` four, `log` twice, the rest once each; three fields never read: `sessions`, `broker_down`, `unreadable`). Filled its seven bodies in the order of the file, each from its comment and `cargo check` after: `ask` (`state.rs`) — connect, set read and write timeout to `wait`, write an `Envelope { v: PROTOCOL_VERSION, id: 1, r#final: true, msg: Message::SessionState(SessionState { session }) }`, match `read_frame`'s msg: `SessionStateReply` -> Ok, `Error` -> `Err(format!("brokerd: {}", e.detail))`, anything else the wrong-message error, every io error through `socket: `; `log_path` (`sessions.join(id).join("0.jsonl")`); `read_log` (`fs::read`, NotFound -> empty, else Io with the runbook, walk the complete lines up to the last newline so a partial last line is left out, each line parsed and a bad line named with its number and the runbook); `finished` (last `Assistant` with empty `tool_calls`); `records` (`read_log`, remove from `unreadable` on Ok, log a damaged log once on Err); `state` (`ask`, log the outage and its return once each); `decidable` (`reported.min(count)`, hold the log's last turn when it is not finished). After the seventh `cargo check` is clean with no warning; `cargo test -p memoryd --test honcho_state --test honcho_turns` gives all 21 pass. Step 6 all pass: `honcho_serve` 6, `honcho_turns` 16, `honcho_state` 5, `serve` 8, `honcho_start` 4, the existing suites unchanged; the three step-8 greps print as expected and the other-crates `git diff --stat` is empty. `make gate` printed `gate: ok` with 1257 tests (1256 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately with `BOXMAKER_BROKERD`). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5c1/08-memoryd-honcho-decide | 2026-09-28 | done | 1 | pass | none | Copied the given test `crates/memoryd/tests/honcho_decide.rs` and the skeleton `crates/memoryd/src/honcho/decide.rs` from `docs/plans/M5c1/files/`. Step 2 showed exactly the named failure: `honcho_decide` with one E0432 (`unresolved import memoryd::honcho::decide`); every other target compiled. Made H1 (`pub mod decide;` after `pub mod config;` in `honcho/mod.rs`) and copied the skeleton; `cargo check` gave 11 warnings (one unused import `bxcore::sanitize::refuse`, ten unused variables: `records` four, `turn` four, `speaker` and `text` once each). Step 3 `cargo test -p memoryd --test honcho_decide` gave 1 pass (`the_skip_reasons_are_these_texts`) and 17 fail with `not yet implemented`. Filled its eight bodies in the order of the file, deleting each instruction comment as it went and keeping the doc comments, each from its comment and `cargo check` after: `turns` (filter `User`, `u64::try_from(count).unwrap_or(u64::MAX)`); `bounds` (walk `enumerate`, when a second `User` follows the tracked turn-th one return `(start, index)`, else count and set `start` at `seen == turn`, then `start.map(|s| (s, len))`); `turn_records` (`bounds` then `get`); `peer_name` (`replace('.', "_")`); `has_secret` (map each char `refuse` refuses on its own to a space via a 4-byte `encode_utf8` buffer, then keep only a `secret-` rule); `user_text` (first record's `User`, `speaker.as_deref()`); `final_reply` (last `Assistant` with empty `tool_calls`, `content.as_deref()`); `labelled_untrusted` (any `ToolResult { untrusted: true }`). After the eighth `cargo check` is clean with no warning; `cargo test` gives all 18 pass. Step 6 all pass: the grep for `fs::`/`std::io`/`Timestamp::now` prints 0, the two `refuse(` lines (87, 94) are both in `has_secret`, and the other-crates `git diff --stat` is empty. `make gate` printed `gate: ok` with 1230 tests (1229 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately with `BOXMAKER_BROKERD` set). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | diff --git a/docs/runbook.md b/docs/runbook.md index aa1f84f..2a4a4a0 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -1010,3 +1010,91 @@ log names the problem, usually `broker-state-damaged`. **Check.** At the next turn's end `memoryd` logs that `brokerd` answers again, and `memory/honcho.jsonl` gains a `decided` line for each turn that waited. + +## honcho-unreachable + +**What you see.** `memoryd` logs one line when it cannot reach Honcho, then this entry: + +```text +memoryd: Honcho is unreachable (connect); sends wait and are retried; see docs/runbook.md#honcho-unreachable +``` + +The word in parentheses is the kind: `connect` (nothing answers at `[honcho] url`), `timeout` (5 +seconds to connect, or 30 without a byte), `closed`, `io`, or `server, HTTP 5xx` (Honcho answered +with a server error). When a send goes through again, `memoryd` logs `memoryd: Honcho answers +again; waiting sends go on`. Conversations are not affected. + +**Why.** Every finished turn that is decided for Honcho waits in the ledger, +`memory/honcho.jsonl`, until it is sent (M5c spec, section 5). While Honcho does not answer, +`memoryd` tries the oldest waiting turn again, waiting 5 seconds, then 10, 20 and so on up to 10 +minutes between tries, and logs the outage once, not at every try. Nothing is lost: the turns go, +in order, once Honcho answers. A restart of `memoryd` starts the waits again from 5 seconds. + +**Confirm.** + +```sh +grep -n '^url' /etc/boxmaker/memoryd.toml +curl -sS -m 5 "/health" +``` + +**Fix.** Start Honcho, or make the network path to it work (on straylight, rift over the +tailnet: `tailscale status`). If `[honcho] url` is wrong, correct it and restart `memoryd`. + +**Check.** `memoryd` logs `Honcho answers again`, and `memory/honcho.jsonl` gains `sent` lines. + +## honcho-send-refused + +**What you see.** `memoryd` logs one line for a turn that Honcho refused, or that could not be +sent at all, then this entry: + +```text +memoryd: turn 3 of session mm-chat-1 was not sent: Honcho refused the request (status, HTTP 422); it is marked failed; see docs/runbook.md#honcho-send-refused +``` + +or `memoryd: turn 3 of session mm-chat-1 cannot be sent (too-long); it is marked failed; …` for a +turn whose text needs more than 100 messages of 25,000 characters. + +**Why.** Honcho answered with a 4xx other than 401 or 403, a redirect, or a body `roncho` could +not read; sending the same again would get the same answer, or could store it twice. So the turn +is marked `failed` in `memory/honcho.jsonl`, with the kind and the status, and the next turn goes +on. The line never holds the turn's text or Honcho's answer (a 422 echoes the whole request). + +**Confirm.** + +```sh +grep -n '"type":"failed"' "$BOXMAKER_HOME/memory/honcho.jsonl" +``` + +A 422 usually means a name Honcho does not accept, or text over its limit; a 404, a workspace or +session removed on the Honcho side. + +**Fix.** A failed turn is not sent again by itself. To send it again after fixing the cause, +stop `memoryd`, keep a copy of the ledger, remove the turn's `failed` line (see +`honcho-ledger-damaged` for what removing a line means), and start `memoryd`. + +**Check.** The next turns are sent (`sent` lines follow). + +## honcho-key-refused + +**What you see.** `memoryd` logs one line when Honcho refuses a workspace's key, then this entry: + +```text +memoryd: Honcho refused the key of workspace boxmaker0-trusted (HTTP 401); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused +``` + +**Why.** Each workspace is sent to with its own key, `[secrets.honcho_trusted]`, +`honcho_untrusted` or `honcho_fleet` (M5c spec, section 10), or none. A 401 or 403 means the key +is wrong, expired, or missing where Honcho now wants one. Trying again would be refused again, so +`memoryd` stops sending to that workspace until it restarts; its turns stay waiting in +`memory/honcho.jsonl`, and the other workspaces go on. The key is never printed. + +**Confirm.** Which workspace is named, and where its key comes from: + +```sh +grep -n -A2 '^\[secrets\.honcho_' /etc/boxmaker/memoryd.toml +``` + +**Fix.** Put the right key where `[secrets.]` points (see `secret-unavailable` for the +sources), or remove the table if the workspace needs no key, then restart `memoryd`. + +**Check.** After the restart the waiting turns are sent, and no line names this entry.