diff --git a/crates/memoryd/src/honcho/messages.rs b/crates/memoryd/src/honcho/messages.rs new file mode 100644 index 0000000..5d4a2fd --- /dev/null +++ b/crates/memoryd/src/honcho/messages.rs @@ -0,0 +1,70 @@ +//! A turn's messages for Honcho (M5c spec, sections 4 and 5): read again from `loopd`'s log, +//! neutralised, and cut into parts Honcho takes, each with the turn's metadata. + +use std::path::Path; + +use bxcore::SessionId; +use bxcore::sanitize::neutralise; +use roncho::MessageCreate; +use serde_json::{Map, json}; + +use crate::honcho::decide::{final_reply, turn_records, user_text}; +use crate::honcho::turns::{LogError, read_records}; + +/// 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; + +/// `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 is missing or has a damaged line, or has no such turn, or the turn no reply: it will not +/// read later. An error, naming the file, when the log could not be read now. +pub fn read_turn(path: &Path, turn: u64) -> Result, String> { + let records = match read_records(path) { + Ok(records) => records, + Err(LogError::Io(why)) => return Err(why), + Err(LogError::Damaged(_)) => return Ok(None), + }; + let Some(records) = turn_records(&records, turn) else { + return Ok(None); + }; + let (text, _) = user_text(records); + Ok(final_reply(records).map(|reply| (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. +pub fn ids(messages: &[roncho::Message]) -> Vec { + messages.iter().map(|m| m.id.clone()).collect() +} diff --git a/crates/memoryd/src/honcho/mod.rs b/crates/memoryd/src/honcho/mod.rs index edba558..ad2cddc 100644 --- a/crates/memoryd/src/honcho/mod.rs +++ b/crates/memoryd/src/honcho/mod.rs @@ -6,6 +6,7 @@ pub mod config; pub mod decide; pub mod edits; pub mod ledger; +pub mod messages; pub mod mirror; pub mod naming; pub mod report; diff --git a/crates/memoryd/src/honcho/status.rs b/crates/memoryd/src/honcho/status.rs index 8e7c7b1..9f96bfb 100644 --- a/crates/memoryd/src/honcho/status.rs +++ b/crates/memoryd/src/honcho/status.rs @@ -1,6 +1,6 @@ //! What the Honcho side is waiting on, beyond the ledger, for `bxctl memory sync` (M5c spec, -//! sections 10, 12 and 13): workspaces whose key Honcho refused, an outage of Honcho, and a -//! `brokerd` that does not answer. The worker and the walk write it; the report reads it. Its lock +//! sections 10, 12 and 13): workspaces whose key Honcho refused, an outage of Honcho, a session log +//! that cannot be read for now, and a `brokerd` that does not answer. The worker and the walk write it; the report reads it. Its lock //! is a leaf: nothing else is locked while it is held. use std::collections::BTreeMap; @@ -19,6 +19,9 @@ pub struct Status { pub outage_since: Option, /// When `brokerd` stopped answering on its state socket, while it does not. pub broker_down_since: Option, + /// When the worker first could not read a session log it needs, for a reason other than a + /// missing file or a damaged line, while it cannot. + pub log_unreadable_since: Option, } /// The status, shared between the worker, the walk and the report. @@ -42,8 +45,8 @@ fn role(workspace: Workspace) -> &'static str { } } -/// One line for each state that lasts, in this order: `brokerd` silent, Honcho unreachable, then -/// each refused key by workspace. None when all is well. +/// One line for each state that lasts, in this order: `brokerd` silent, Honcho unreachable, a +/// session log unreadable, then each refused key by workspace. None when all is well. pub fn lines(status: &Status) -> Vec { let mut out = Vec::new(); if let Some(since) = status.broker_down_since { @@ -58,6 +61,12 @@ pub fn lines(status: &Status) -> Vec { since.to_rfc3339() )); } + if let Some(since) = status.log_unreadable_since { + out.push(format!( + "session log unreadable since {}: sends wait and are retried", + since.to_rfc3339() + )); + } for (workspace, (name, code)) in &status.refused { out.push(format!( "key refused (HTTP {code}): {} workspace {name}: its sends wait until memoryd restarts", diff --git a/crates/memoryd/src/honcho/turns.rs b/crates/memoryd/src/honcho/turns.rs index 16f3437..c576277 100644 --- a/crates/memoryd/src/honcho/turns.rs +++ b/crates/memoryd/src/honcho/turns.rs @@ -57,14 +57,38 @@ pub fn log_path(sessions: &Path, session: &SessionId) -> PathBuf { sessions.join(session.as_str()).join("0.jsonl") } +/// Why a session log did not read. Each names the file; `Display` adds the runbook pointer. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LogError { + /// The file exists but could not be read (a permission, an I/O error): it may read later. + Io(String), + /// A complete line that does not parse, and its number: it will not read later. + Damaged(String), +} + +impl std::fmt::Display for LogError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + LogError::Io(why) | LogError::Damaged(why) => write!(f, "{why}; {LOG_RUNBOOK}"), + } + } +} + +impl std::error::Error for LogError {} + /// A session log's complete lines as records. A missing file has none. A last line without its /// newline is one `loopd` is still writing, and is left out. A complete line that does not parse /// is an error naming the file and the line, with the runbook pointer. pub fn read_log(path: &Path) -> Result, String> { + read_records(path).map_err(|e| e.to_string()) +} + +/// `read_log`, telling a file that could not be read from a damaged line. +pub fn read_records(path: &Path) -> Result, LogError> { let text = match std::fs::read(path) { Ok(bytes) => bytes, Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(e) => return Err(format!("{}: {e}; {LOG_RUNBOOK}", path.display())), + Err(e) => return Err(LogError::Io(format!("{}: {e}", path.display()))), }; let whole = match text.iter().rposition(|b| *b == b'\n') { Some(last) => last + 1, @@ -80,11 +104,11 @@ pub fn read_log(path: &Path) -> Result, String> { match serde_json::from_slice::(line) { Ok(record) => records.push(record), Err(e) => { - return Err(format!( - "{}:{}: {e}; {LOG_RUNBOOK}", + return Err(LogError::Damaged(format!( + "{}:{}: {e}", path.display(), index + 1 - )); + ))); } } } diff --git a/crates/memoryd/src/honcho/worker.rs b/crates/memoryd/src/honcho/worker.rs index 589cb92..164587c 100644 --- a/crates/memoryd/src/honcho/worker.rs +++ b/crates/memoryd/src/honcho/worker.rs @@ -6,32 +6,25 @@ //! the worker decides what `brokerd` could not answer for before, then sleeps until a turn ends. use std::collections::HashSet; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::Duration; -use bxcore::sanitize::neutralise; use bxcore::{SessionId, SessionStateReply, Timestamp}; -use roncho::MessageCreate; -use serde_json::{Map, json}; use crate::honcho::client::{Clients, HonchoError, classify}; -use crate::honcho::decide::{final_reply, turn_records, user_text}; use crate::honcho::edits; use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace}; use crate::honcho::naming::{HonchoSession, honcho_session}; use crate::honcho::status::{self, Shared}; -use crate::honcho::turns::{TurnSync, log_path, read_log}; +use crate::honcho::turns::{TurnSync, log_path}; use crate::serve::Log; +use crate::honcho::messages::ids; +pub use crate::honcho::messages::{MAX_BATCH, MAX_PART, batch, parts, read_turn}; pub use crate::honcho::start::{Opened, open, spawn}; pub use crate::honcho::wake::{Tuning, Wake}; -/// 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"; @@ -58,6 +51,8 @@ enum Unsent { Honcho(HonchoError), /// The turn cannot be read from the log any more, or is too long for one batch. Local(&'static str), + /// The turn's log could not be read now, and why: the log's path and the error. + Unreadable(String), } /// What to send for one waiting item (M5c spec, section 5). @@ -105,6 +100,8 @@ pub struct Worker { agent: String, /// In an outage: the next wait. outage: Option, + /// While a session log cannot be read: the next wait. + unreadable: Option, /// Workspaces whose key was refused: nothing more is sent to them until restart. refused: HashSet, /// What `bxctl memory sync` shows; its lock is taken alone. @@ -112,53 +109,6 @@ pub struct Worker { 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( @@ -189,6 +139,7 @@ impl Worker { tuning, agent: agent.to_string(), outage: None, + unreadable: None, refused: HashSet::new(), status, stopped: false, @@ -258,6 +209,7 @@ impl Worker { kind: e.kind().to_string(), }) } + Err(Unsent::Unreadable(why)) => self.unreadable(&next.item, why), Err(Unsent::Local(kind)) => { let pointer = if kind == "log" { LOG_DAMAGED @@ -385,8 +337,10 @@ impl Worker { speaker: &str, honcho: Option<&HonchoSession>, ) -> Result, Unsent> { - let (text, reply) = - read_turn(&log_path(&self.sessions, session), turn).ok_or(Unsent::Local("log"))?; + let path = log_path(&self.sessions, session); + let read = read_turn(&path, turn).map_err(Unsent::Unreadable)?; + self.readable(&path); + let (text, reply) = read.ok_or(Unsent::Local("log"))?; let honcho = honcho.ok_or(Unsent::Local("log"))?; let messages = batch(&text, &reply, speaker, &self.agent, session, turn); if messages.len() > MAX_BATCH { @@ -418,6 +372,37 @@ impl Worker { Step::Backoff(wait) } + /// A session log could not be read now: log it once, with the item's label and the error, + /// never the text, and wait as for an outage. The item stays waiting. + fn unreadable(&mut self, item: &Item, why: String) -> Step { + let wait = match self.unreadable { + Some(wait) => wait, + None => { + (self.log)(&format!( + "memoryd: {} cannot be read now ({why}); sends wait and are retried; {LOG_DAMAGED}", + ledger::label(item) + )); + status::update(&self.status, |s| { + s.log_unreadable_since = Some(Timestamp::now()) + }); + self.tuning.backoff + } + }; + self.unreadable = Some(wait.saturating_mul(2).min(self.tuning.backoff_max)); + Step::Backoff(wait) + } + + /// The log at `path` read: if one could not be read before, log that it reads again. + fn readable(&mut self, path: &std::path::Path) { + if self.unreadable.take().is_some() { + (self.log)(&format!( + "memoryd: {} reads again; waiting sends go on", + path.display() + )); + status::update(&self.status, |s| s.log_unreadable_since = None); + } + } + /// 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() { diff --git a/crates/memoryd/tests/honcho_status.rs b/crates/memoryd/tests/honcho_status.rs index 9a6d280..11577b0 100644 --- a/crates/memoryd/tests/honcho_status.rs +++ b/crates/memoryd/tests/honcho_status.rs @@ -42,6 +42,7 @@ fn at(minute: u32) -> Timestamp { const ALL: &str = "honcho sync: 0 decided: 0 sent, 0 waiting, 0 failed, 0 skipped brokerd unanswered since 2026-09-28T10:01:00.000Z: turns wait undecided honcho unreachable since 2026-09-28T10:02:00.000Z: sends wait and are retried +session log unreadable since 2026-09-28T10:03:00.000Z: sends wait and are retried key refused (HTTP 401): trusted workspace boxmaker0-trusted: its sends wait until memoryd restarts key refused (HTTP 403): fleet workspace fleet: its sends wait until memoryd restarts"; @@ -54,6 +55,7 @@ fn the_status_lines_are_exactly_these() { let mut status = Status { broker_down_since: Some(at(1)), outage_since: Some(at(2)), + log_unreadable_since: Some(at(3)), ..Status::default() }; status diff --git a/crates/memoryd/tests/honcho_worker_log.rs b/crates/memoryd/tests/honcho_worker_log.rs new file mode 100644 index 0000000..848cded --- /dev/null +++ b/crates/memoryd/tests/honcho_worker_log.rs @@ -0,0 +1,128 @@ +//! A session log the worker cannot read for now (review of M5c-1, finding 6): an I/O error other +//! than a missing file is waited out like an outage, with the same doubling wait, logged once when +//! it starts and once when it ends, and shown by `bxctl memory sync`; the turn stays waiting and +//! is sent once the log reads. A missing log or a damaged line still marks the turn failed. The +//! log is made unreadable by a directory in its place (EISDIR), which holds for root too. + +#[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 honcho::FakeHoncho; +use logs::assistant; +use memoryd::honcho::ledger::{Item, Record}; +use memoryd::honcho::status; +use memoryd::honcho::turns::log_path; +use memoryd::honcho::worker::Step; +use turns::{sid, spoken}; +use worker::{FAST, Home}; + +fn turn(text: &str) -> Vec { + vec![spoken(text, Some("kyle")), assistant(2, Some("ok"), &[])] +} + +fn is_sent(records: &[Record]) -> bool { + let item = Item::Turn { + session: sid("mm-a"), + turn: 1, + }; + records + .iter() + .any(|r| matches!(r, Record::Sent { item: i, .. } if *i == item)) +} + +fn has_failed(records: &[Record]) -> bool { + records.iter().any(|r| matches!(r, Record::Failed { .. })) +} + +#[test] +fn an_unreadable_log_is_waited_out() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-log-io"); + home.log("mm-a", &[turn("hello")]); + home.report("mm-a", 1); + let path = log_path(&home.dir.path().join("sessions"), &sid("mm-a")); + let kept = std::fs::read(&path).unwrap(); + std::fs::remove_file(&path).unwrap(); + std::fs::create_dir(&path).unwrap(); + + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Backoff(Duration::from_millis(1))); + assert_eq!(w.step(), Step::Backoff(Duration::from_millis(2))); + assert_eq!(w.step(), Step::Backoff(Duration::from_millis(4))); + let records = home.records(); + assert!(!has_failed(&records), "{records:?}"); + let lines = home.lines(); + assert_eq!(lines.len(), 1, "logged once: {lines:?}"); + assert!( + lines[0].starts_with("memoryd: turn 1 of session mm-a cannot be read now ("), + "{}", + lines[0] + ); + assert!( + lines[0] + .ends_with("); sends wait and are retried; see docs/runbook.md#session-log-damaged"), + "{}", + lines[0] + ); + assert!(!lines[0].contains("hello")); + let shared = home.sync.lock().unwrap().shared_status(); + assert!(status::read(&shared).log_unreadable_since.is_some()); + let shown = status::lines(&status::read(&shared)); + assert!( + shown[0].starts_with("session log unreadable since "), + "{shown:?}" + ); + assert!( + shown[0].ends_with(": sends wait and are retried"), + "{shown:?}" + ); + + std::fs::remove_dir(&path).unwrap(); + std::fs::write(&path, kept).unwrap(); + assert_eq!(w.step(), Step::Done); + assert!(is_sent(&home.records())); + let lines = home.lines(); + assert_eq!(lines.len(), 2, "{lines:?}"); + assert_eq!( + lines[1], + format!( + "memoryd: {} reads again; waiting sends go on", + path.display() + ) + ); + assert_eq!(status::read(&shared), status::Status::default()); +} + +#[test] +fn a_missing_or_damaged_log_still_fails_the_turn() { + let fake = FakeHoncho::start(); + for damage in ["missing", "damaged"] { + let home = Home::new("worker-log-gone"); + home.log("mm-a", &[turn("hello")]); + home.report("mm-a", 1); + let path = log_path(&home.dir.path().join("sessions"), &sid("mm-a")); + match damage { + "missing" => std::fs::remove_file(&path).unwrap(), + _ => std::fs::write(&path, "not a record\n").unwrap(), + } + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done, "{damage}"); + let records = home.records(); + assert!( + records + .iter() + .any(|r| matches!(r, Record::Failed { kind, .. } if kind == "log")), + "{damage}: {records:?}" + ); + } +} diff --git a/docs/runbook.md b/docs/runbook.md index 0600b50..9acc371 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -372,6 +372,21 @@ that does not parse tool calls, or whose cache does not work. logs the same line once, starting `memoryd: `, and that session's turns wait undecided for Honcho until the log reads again. +With `[honcho]`, `memoryd`'s Honcho worker also reads a decided turn again from the log to send +it. If the log is missing, has a damaged line, or no longer holds the turn, the turn is marked +failed (`memoryd: turn 3 of session mm-chat-1 cannot be sent (log); it is marked failed; …`), and +`bxctl memory sync` lists it under `failed, oldest first:` with the kind `log`. If the file is +there but cannot be read now (a permission, an I/O error, a directory in its place), the worker +logs once + +```text +memoryd: turn 3 of session mm-chat-1 cannot be read now (/srv/bx/sessions/mm-chat-1/0.jsonl: Permission denied (os error 13)); sends wait and are retried; see docs/runbook.md#session-log-damaged +``` + +and tries again with the doubling wait of `honcho-unreachable`, sending nothing else meanwhile; +`bxctl memory sync` shows `session log unreadable since