memoryd: the Honcho worker sends decided turns, waits out outages, retries undecided turns

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

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()

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;

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<SessionId> = 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 => {

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<bool>,
+    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<dyn Fn(&SessionId) -> Result<SessionStateReply, String> + 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<HonchoSession>,
+}
+
+pub struct Worker {
+    sync: Arc<Mutex<TurnSync>>,
+    clients: Clients,
+    wake: Arc<Wake>,
+    ask: AskBox,
+    log: Log,
+    tuning: Tuning,
+    agent: String,
+    /// In an outage: the next wait.
+    outage: Option<Duration>,
+    /// Workspaces whose key was refused: nothing more is sent to them until restart.
+    refused: HashSet<Workspace>,
+    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(
+        sync: Arc<Mutex<TurnSync>>,
+        clients: Clients,
+        wake: Arc<Wake>,
+        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<Next> {
+        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<Vec<String>, 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<Mutex<TurnSync>>,
+    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<Opened, StartError> {
+    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<Wake>, 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}"))
+}