review: a session log that cannot be read now is waited out

The worker marked a turn failed on any error reading its log. A missing log,
a damaged line or a turn not found still fail it; any other I/O error is now
waited out with the outage's doubling wait, logged once when it starts and
once when the log reads again (pointer session-log-damaged, whose runbook
entry now covers it), shown by bxctl memory sync, and the turn stays waiting.

read_records tells an unreadable file from a damaged line (read_log is
unchanged). The message helpers move from worker.rs to honcho/messages.rs,
re-exported, to keep worker.rs under 500 lines.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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<String> {
+    let chars: Vec<char> = 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<Option<(String, String)>, 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<MessageCreate> {
+    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<String> {
+    messages.iter().map(|m| m.id.clone()).collect()
+}

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;

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<Timestamp>,
     /// When `brokerd` stopped answering on its state socket, while it does not.
     pub broker_down_since: Option<Timestamp>,
+    /// 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<Timestamp>,
 }
 
 /// 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<String> {
     let mut out = Vec::new();
     if let Some(since) = status.broker_down_since {
@@ -58,6 +61,12 @@ pub fn lines(status: &Status) -> Vec<String> {
             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",

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<Vec<LogRecord>, 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<Vec<LogRecord>, 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<Vec<LogRecord>, String> {
         match serde_json::from_slice::<LogRecord>(line) {
             Ok(record) => records.push(record),
             Err(e) => {
-                return Err(format!(
-                    "{}:{}: {e}; {LOG_RUNBOOK}",
+                return Err(LogError::Damaged(format!(
+                    "{}:{}: {e}",
                     path.display(),
                     index + 1
-                ));
+                )));
             }
         }
     }

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<Duration>,
+    /// While a session log cannot be read: the next wait.
+    unreadable: Option<Duration>,
     /// Workspaces whose key was refused: nothing more is sent to them until restart.
     refused: HashSet<Workspace>,
     /// 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<String> {
-    let chars: Vec<char> = 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<MessageCreate> {
-    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<String> {
-    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<Vec<String>, 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() {