memoryd: a reported turn decided after loopd has its answer, from brokerd and the log

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/memoryd/src/honcho/mod.rs

@@ -3,3 +3,5 @@
 pub mod config;
 pub mod decide;
 pub mod ledger;
+pub mod state;
+pub mod turns;

crates/memoryd/src/honcho/state.rs

@@ -0,0 +1,45 @@
+//! Asking `brokerd` for a session's state on its `state.sock` (M5c spec, section 7): one question
+//! per connection, one final answer, as on every socket between the roles.
+
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+use std::time::Duration;
+
+use bxcore::{
+    Envelope, Message, PROTOCOL_VERSION, SessionId, SessionState, SessionStateReply, read_frame,
+    write_frame,
+};
+
+/// How long `brokerd` may leave the connection without a byte, sending or answering.
+pub const WAIT: Duration = Duration::from_secs(5);
+
+/// The runbook entry for a `brokerd` that does not answer.
+pub const RUNBOOK: &str = "see docs/runbook.md#broker-state-unavailable";
+
+/// `brokerd`'s answer for `session`, asked on `socket`. `wait` is the read and the write timeout;
+/// the connect has none, and fails at once when nothing listens. Any error, an error frame
+/// included, is `Err` with one line that names the socket or `brokerd`.
+pub fn ask(
+    socket: &Path,
+    session: &SessionId,
+    wait: Duration,
+) -> Result<SessionStateReply, String> {
+    let line = |e: &dyn std::fmt::Display| format!("{}: {e}", socket.display());
+    let mut stream = UnixStream::connect(socket).map_err(|e| line(&e))?;
+    stream.set_read_timeout(Some(wait)).map_err(|e| line(&e))?;
+    stream.set_write_timeout(Some(wait)).map_err(|e| line(&e))?;
+    let question = Envelope {
+        v: PROTOCOL_VERSION,
+        id: 1,
+        r#final: true,
+        msg: Message::SessionState(SessionState {
+            session: session.clone(),
+        }),
+    };
+    write_frame(&mut stream, &question).map_err(|e| line(&e))?;
+    match read_frame(&mut stream).map_err(|e| line(&e))?.msg {
+        Message::SessionStateReply(reply) => Ok(reply),
+        Message::Error(e) => Err(format!("brokerd: {}", e.detail)),
+        _ => Err("brokerd answered with the wrong message".to_string()),
+    }
+}

crates/memoryd/src/honcho/turns.rs

@@ -0,0 +1,247 @@
+//! Deciding finished turns when `loopd` reports one (M5c spec, section 5). At each report every
+//! session with undecided turns is walked: its log is read, `brokerd` is asked once for its state,
+//! and each undecided turn is decided and appended to the ledger at once. A turn that cannot be
+//! decided now (`brokerd` silent, the log unreadable) waits for the next report of any session.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::path::{Path, PathBuf};
+
+use bxcore::{LogRecord, SessionId, SessionStateReply, Timestamp};
+
+use crate::honcho::decide::{decide, turn_records, turns};
+use crate::honcho::ledger::{self, Item, Ledger, Record};
+use crate::honcho::state;
+
+/// The ledger's file name, in `memory/`.
+pub const LEDGER: &str = "honcho.jsonl";
+
+/// The runbook entry for a session log with a damaged line.
+pub const LOG_RUNBOOK: &str = "see docs/runbook.md#session-log-damaged";
+
+/// `brokerd`'s answer for a session, or why there is none.
+pub type Ask<'a> = &'a dyn Fn(&SessionId) -> Result<SessionStateReply, String>;
+
+/// How one session's walk ended.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Walk {
+    /// Every turn up to the reported one is decided: the session leaves the map.
+    Done,
+    /// Some turn waits: the session stays in the map.
+    Waiting,
+    /// `brokerd` did not answer, or the ledger could not be written: the walk stops.
+    Stop,
+}
+
+/// The turns reported and not yet decided, and the ledger their decisions go to.
+pub struct TurnSync {
+    ledger: Ledger,
+    agent: String,
+    sessions: PathBuf,
+    /// Each session with undecided turns, by its id, and the highest turn `loopd` reported for
+    /// it. `SessionId` has no order, so the map is keyed by its text.
+    reported: BTreeMap<String, (SessionId, u64)>,
+    /// Whether the last question to `brokerd` went unanswered: an outage is logged once.
+    broker_down: bool,
+    /// Sessions whose log last had a damaged line: logged once each.
+    unreadable: BTreeSet<String>,
+}
+
+/// The path of `session`'s log under `sessions`: epoch 0, the only one `loopd` writes.
+pub fn log_path(sessions: &Path, session: &SessionId) -> PathBuf {
+    sessions.join(session.as_str()).join("0.jsonl")
+}
+
+/// 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> {
+    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())),
+    };
+    let whole = match text.iter().rposition(|b| *b == b'\n') {
+        Some(last) => last + 1,
+        None => 0,
+    };
+    let mut records = Vec::new();
+    for (index, line) in text
+        .get(..whole)
+        .unwrap_or_default()
+        .split_inclusive(|b| *b == b'\n')
+        .enumerate()
+    {
+        match serde_json::from_slice::<LogRecord>(line) {
+            Ok(record) => records.push(record),
+            Err(e) => {
+                return Err(format!(
+                    "{}:{}: {e}; {LOG_RUNBOOK}",
+                    path.display(),
+                    index + 1
+                ));
+            }
+        }
+    }
+    Ok(records)
+}
+
+/// Whether a turn has ended with an answer: its last `Assistant` record asks for no tool.
+pub fn finished(turn: &[LogRecord]) -> bool {
+    let last = turn
+        .iter()
+        .rev()
+        .find(|r| matches!(r, LogRecord::Assistant { .. }));
+    matches!(last, Some(LogRecord::Assistant { tool_calls, .. }) if tool_calls.is_empty())
+}
+
+impl TurnSync {
+    /// `ledger` read through at start; `agent` is `[honcho] agent`; `sessions` the directory of
+    /// `loopd`'s session logs.
+    pub fn new(ledger: Ledger, agent: &str, sessions: &Path) -> TurnSync {
+        TurnSync {
+            ledger,
+            agent: agent.to_string(),
+            sessions: sessions.to_path_buf(),
+            reported: BTreeMap::new(),
+            broker_down: false,
+            unreadable: BTreeSet::new(),
+        }
+    }
+
+    pub fn ledger(&self) -> &Ledger {
+        &self.ledger
+    }
+
+    /// 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.
+    pub fn updated(
+        &mut self,
+        session: &SessionId,
+        turn: u64,
+        ask: Ask<'_>,
+        now: Timestamp,
+        log: &dyn Fn(&str),
+    ) {
+        let entry = self
+            .reported
+            .entry(session.as_str().to_string())
+            .or_insert((session.clone(), 0));
+        entry.1 = entry.1.max(turn);
+        for (session, turn) in self.waiting() {
+            match self.walk(&session, turn, ask, now, log) {
+                Walk::Done => {
+                    self.reported.remove(session.as_str());
+                }
+                Walk::Waiting => {}
+                Walk::Stop => return,
+            }
+        }
+    }
+
+    /// Decide `session`'s undecided turns from 1 to `reported`, as far as its log goes. The log's
+    /// last turn waits until it has an answer: it may still be running.
+    fn walk(
+        &mut self,
+        session: &SessionId,
+        reported: u64,
+        ask: Ask<'_>,
+        now: Timestamp,
+        log: &dyn Fn(&str),
+    ) -> Walk {
+        let Some(records) = self.records(session, log) else {
+            return Walk::Waiting;
+        };
+        let (limit, held) = decidable(&records, reported);
+        let item = |turn: u64| Item::Turn {
+            session: session.clone(),
+            turn,
+        };
+        let undecided: Vec<u64> = (1..=limit)
+            .filter(|turn| !self.ledger.is_decided(&item(*turn)))
+            .collect();
+        if undecided.is_empty() {
+            return if held { Walk::Waiting } else { Walk::Done };
+        }
+        let Some(state) = self.state(session, ask, log) else {
+            return Walk::Stop;
+        };
+        for turn in undecided {
+            let Ok(action) = decide(&state, &records, turn, &self.agent) else {
+                continue;
+            };
+            let record = Record::Decided {
+                time: now,
+                item: item(turn),
+                action,
+            };
+            if let Err(e) = self.ledger.append(record) {
+                log(&format!("memoryd: {e}; {}", ledger::RUNBOOK));
+                return Walk::Stop;
+            }
+        }
+        if held { Walk::Waiting } else { Walk::Done }
+    }
+
+    /// `session`'s log, read. A log that does not read is logged once, until it reads again.
+    fn records(&mut self, session: &SessionId, log: &dyn Fn(&str)) -> Option<Vec<LogRecord>> {
+        match read_log(&log_path(&self.sessions, session)) {
+            Ok(records) => {
+                self.unreadable.remove(session.as_str());
+                Some(records)
+            }
+            Err(e) => {
+                if self.unreadable.insert(session.as_str().to_string()) {
+                    log(&format!("memoryd: {e}"));
+                }
+                None
+            }
+        }
+    }
+
+    /// `brokerd`'s answer for `session`. The first failure after an answer is logged, and so is
+    /// the first answer after a failure.
+    fn state(
+        &mut self,
+        session: &SessionId,
+        ask: Ask<'_>,
+        log: &dyn Fn(&str),
+    ) -> Option<SessionStateReply> {
+        match ask(session) {
+            Ok(state) => {
+                if self.broker_down {
+                    self.broker_down = false;
+                    log("memoryd: brokerd answers on its state socket again; turns are decided");
+                }
+                Some(state)
+            }
+            Err(e) => {
+                if !self.broker_down {
+                    self.broker_down = true;
+                    log(&format!(
+                        "memoryd: {e}; turns wait undecided; {}",
+                        state::RUNBOOK
+                    ));
+                }
+                None
+            }
+        }
+    }
+}
+
+/// How far a log's turns may be decided for a report of turn `reported`: up to the smaller of
+/// `reported` and the log's turns, but for the log's last turn when it has no answer yet. The
+/// limit, and whether that last turn was held back.
+pub fn decidable(records: &[LogRecord], reported: u64) -> (u64, bool) {
+    let count = turns(records);
+    let limit = reported.min(count);
+    if limit > 0 && limit == count && !turn_records(records, limit).is_some_and(finished) {
+        (limit - 1, true)
+    } else {
+        (limit, false)
+    }
+}

crates/memoryd/src/main.rs

@@ -44,6 +44,11 @@ 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(_)) => {
+            eprintln!("memoryd: {e}");
+            return ExitCode::from(1);
+        }
         Err(e) => {
             eprintln!("memoryd: {e}\n{START_FAILED}");
             return ExitCode::from(1);

crates/memoryd/src/serve.rs

@@ -1,5 +1,7 @@
 //! Serving: three sockets, one request per connection, each answered by one frame (M5a spec,
-//! section 7). One lock covers the files and the index, so writes never interleave.
+//! section 7). One lock covers the files and the index, so writes never interleave. With
+//! `[honcho]`, a reported turn is decided after `loopd` has its answer, under a lock of its own
+//! (M5c spec, section 5).
 
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::sync::{Arc, Mutex, MutexGuard};
@@ -7,13 +9,16 @@ use std::time::Duration;
 
 use bxcore::{
     DataClass, Empty, Envelope, ErrorCode, MemoryAdminReply, MemoryArgs, MemoryBlockReply,
-    MemoryCall, MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, Timestamp,
-    ToolResponse, WireError, read_frame, write_frame,
+    MemoryCall, MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, SessionId,
+    SessionUpdated, Timestamp, ToolResponse, WireError, read_frame, write_frame,
 };
 
 use crate::config::Config;
 use crate::edit::{self, Context};
 use crate::files::Store;
+use crate::honcho::ledger::{self, Ledger, LedgerError};
+use crate::honcho::state;
+use crate::honcho::turns::{self, TurnSync};
 use crate::index::{self, Index};
 use crate::recall::Recalled;
 use crate::{journal, recall, review};
@@ -32,9 +37,39 @@ pub struct Memoryd {
     config: Config,
     store: Store,
     state: Mutex<State>,
+    /// 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<Mutex<TurnSync>>,
     log: Log,
 }
 
+/// Why `start` failed. `main` prints a start failure with the pointer `memoryd-start-failed`, and
+/// a ledger that does not read through with its own.
+#[derive(Debug)]
+pub enum StartError {
+    Failed(String),
+    Ledger(LedgerError),
+}
+
+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}"),
+            // `Damaged` names its pointer itself.
+            StartError::Ledger(e @ LedgerError::Damaged { .. }) => write!(f, "{e}"),
+            StartError::Ledger(e) => write!(f, "{e}; {}", ledger::RUNBOOK),
+        }
+    }
+}
+
+impl std::error::Error for StartError {}
+
+impl From<String> for StartError {
+    fn from(why: String) -> StartError {
+        StartError::Failed(why)
+    }
+}
+
 /// A started `memoryd`: its three listeners, bound and waiting.
 pub struct Running {
     memoryd: Arc<Memoryd>,
@@ -48,9 +83,30 @@ enum Which {
     Admin,
 }
 
-/// Create what is missing under `memory/`, open the index and catch it up, then bind the three
-/// sockets. The error names what failed; the caller points at `memoryd-start-failed`.
-pub fn start(config: Config, log: Log) -> Result<Running, String> {
+/// 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.
+pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
+    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(),
+            )))
+        }
+        None => None,
+    };
     let store = Store::new(&config.memory_dir());
     store.init(&config.memory.soul_starter)?;
     let index = Index::open(&config.index_path(), config.index.max_class)?;
@@ -71,6 +127,7 @@ pub fn start(config: Config, log: Log) -> Result<Running, String> {
             config,
             store,
             state: Mutex::new(State { index }),
+            honcho,
             log,
         }),
         listeners,
@@ -141,6 +198,10 @@ impl Memoryd {
             Ok(request) => request,
             Err(_) => return,
         };
+        let update = match (which, &request.msg) {
+            (Which::Loop, Message::SessionUpdated(update)) => Some(update.clone()),
+            _ => None,
+        };
         let reply = self.handle(request.msg, which);
         let env = Envelope {
             v: PROTOCOL_VERSION,
@@ -149,6 +210,29 @@ impl Memoryd {
             msg: reply,
         };
         let _ = write_frame(&mut stream, &env);
+        // `loopd` has its answer before `brokerd` is asked: it sends the turn's reply only after
+        // this frame, and `brokerd` may take seconds.
+        drop(stream);
+        if let Some(update) = update {
+            self.decide_turns(&update);
+        }
+    }
+
+    /// With `[honcho]`: decide the reported turn, and every other that waits (`TurnSync`).
+    fn decide_turns(&self, update: &SessionUpdated) {
+        let Some(honcho) = &self.honcho else {
+            return;
+        };
+        let socket = self.config.state_socket();
+        let ask = |session: &SessionId| state::ask(&socket, session, state::WAIT);
+        let mut sync = honcho.lock().unwrap_or_else(|p| p.into_inner());
+        sync.updated(
+            &update.session,
+            update.turn,
+            &ask,
+            Timestamp::now(),
+            &*self.log,
+        );
     }
 
     /// The answer to one request on one socket. A message sent to the wrong socket is refused.