memoryd: memory writes decided into the Honcho ledger as they are journalled

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

crates/memoryd/src/honcho/mirror.rs

@@ -0,0 +1,216 @@
+//! Memory writes mirrored in Honcho (M5c spec, section 4): each journalled edit is decided into
+//! the ledger the moment it is journalled, one item per Honcho action, in the order they must be
+//! sent. Only decided here; the worker sends (task 13). The text is kept as journalled.
+
+use std::sync::{Arc, Mutex};
+
+use bxcore::{SessionId, Timestamp};
+
+use crate::files::Store;
+use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace};
+use crate::journal;
+use crate::notes::{self, Kind};
+
+/// The session the journal gives an owner's edit (`bxctl memory undo`).
+pub const OWNER: &str = "owner";
+
+/// The part of an edit that has no Boxmaker session to stamp its conclusion with.
+pub const NO_SESSION: &str = "the edit has no Boxmaker session";
+
+/// `before` and `after` as multisets of entry text: what `before` has more of (dropped, in
+/// `before`'s order), and what `after` has more of (added, in `after`'s order).
+pub fn diff(before: &[String], after: &[String]) -> (Vec<String>, Vec<String>) {
+    (minus(before, after), minus(after, before))
+}
+
+/// `a` without one occurrence of each text of `b`, in `a`'s order.
+fn minus(a: &[String], b: &[String]) -> Vec<String> {
+    let mut left: Vec<&String> = b.iter().collect();
+    let mut out: Vec<String> = Vec::new();
+    for text in a {
+        match left.iter().position(|t| *t == text) {
+            Some(at) => {
+                left.swap_remove(at);
+            }
+            None => out.push(text.clone()),
+        }
+    }
+    out
+}
+
+/// The journalled notes of `record`, read back.
+fn notes_of(record: &journal::Record) -> Vec<notes::Note> {
+    record.notes.iter().flat_map(|n| notes::parse(n)).collect()
+}
+
+/// The session an edit's conclusions carry: its own, or for an owner's undo the undone edit's,
+/// following `undoes` back through undos of undos. `None` when there is none.
+pub fn session_of(record: &journal::Record, records: &[journal::Record]) -> Option<SessionId> {
+    let mut current = record;
+    for _ in 0..=records.len() {
+        if current.session != OWNER {
+            return SessionId::new(&current.session).ok();
+        }
+        let undone = current.undoes.as_deref()?;
+        current = records.iter().find(|r| r.id == undone)?;
+    }
+    None
+}
+
+/// A conclusion about `peer` in the trusted workspace, stamped with `session`; without a session
+/// it cannot be stamped, and is a `Skip`.
+fn conclusion(peer: &str, session: Option<&SessionId>, text: &str) -> Action {
+    match session {
+        Some(session) => Action::Conclude {
+            workspace: Workspace::Trusted,
+            peer: peer.to_string(),
+            session: session.clone(),
+            text: text.to_string(),
+        },
+        None => Action::Skip {
+            why: NO_SESSION.to_string(),
+        },
+    }
+}
+
+/// An identity edit (an undo included), `file` its name: each entry dropped, a conclusion
+/// deleted, then each entry added, a conclusion made, about the owner for `user.md` and the agent
+/// for `memory.md`. Entries `consolidate` moved to notes are not dropped: they are still held.
+fn identity_actions(
+    record: &journal::Record,
+    file: &str,
+    notes: &[notes::Note],
+    session: Option<&SessionId>,
+    agent: &str,
+    owner: &str,
+) -> Vec<Action> {
+    let moved: Vec<String> = notes
+        .iter()
+        .filter(|n| matches!(n.kind, Kind::Moved(_)))
+        .map(|n| n.text.clone())
+        .collect();
+    let (dropped, added) = diff(&record.before, &record.after);
+    let peer = if file == "user.md" { owner } else { agent };
+    let mut out: Vec<Action> = minus(&dropped, &moved)
+        .into_iter()
+        .map(|text| Action::Unconclude {
+            workspace: Workspace::Trusted,
+            text,
+        })
+        .collect();
+    out.extend(added.iter().map(|text| conclusion(peer, session, text)));
+    out
+}
+
+/// A write saved as notes: when flagged, one `Skip` naming the rule; otherwise each `noted` note,
+/// a fleet message when the write was untrusted, else a conclusion about the agent.
+fn note_actions(
+    record: &journal::Record,
+    notes: &[notes::Note],
+    session: Option<&SessionId>,
+    agent: &str,
+) -> Vec<Action> {
+    if let Some(rule) = &record.flag {
+        return vec![Action::Skip {
+            why: format!("flagged: {rule}"),
+        }];
+    }
+    notes
+        .iter()
+        .filter(|n| matches!(n.kind, Kind::Noted { .. }))
+        .map(|n| {
+            if record.untrusted {
+                match session {
+                    Some(session) => Action::Fleet {
+                        session: session.clone(),
+                        text: n.text.clone(),
+                    },
+                    None => Action::Skip {
+                        why: NO_SESSION.to_string(),
+                    },
+                }
+            } else {
+                conclusion(agent, session, &n.text)
+            }
+        })
+        .collect()
+}
+
+/// What `record` asks of Honcho, in the order to send it: an identity edit's, or a write saved as
+/// notes. Anything else (a rejection) asks nothing.
+pub fn actions(
+    record: &journal::Record,
+    session: Option<&SessionId>,
+    agent: &str,
+    owner: &str,
+) -> Vec<Action> {
+    let notes = notes_of(record);
+    match &record.file {
+        Some(file) => identity_actions(record, file, &notes, session, agent, owner),
+        None => note_actions(record, &notes, session, agent),
+    }
+}
+
+/// With `[honcho]`: decides each journalled edit into the ledger.
+pub struct Mirror {
+    ledger: Arc<Mutex<Ledger>>,
+    agent: String,
+    owner: String,
+}
+
+impl Mirror {
+    pub fn new(ledger: Arc<Mutex<Ledger>>, agent: &str, owner: &str) -> Mirror {
+        Mirror {
+            ledger,
+            agent: agent.to_string(),
+            owner: owner.to_string(),
+        }
+    }
+
+    /// Decide the journalled edit `id` (the latest when `None`), called under the index lock
+    /// right after it was journalled, so the ledger has the journal's order. Its parts are
+    /// `Edit { id, part }` from 0. An edit already decided, or asking nothing, adds no line. A
+    /// failure is logged; the edit itself stands.
+    pub fn journalled(&self, store: &Store, id: Option<&str>, now: Timestamp, log: &dyn Fn(&str)) {
+        let records = match journal::records(store) {
+            Ok(records) => records,
+            Err(e) => {
+                log(&format!(
+                    "memoryd: {e}; the edit is not mirrored in Honcho; {}",
+                    ledger::RUNBOOK
+                ));
+                return;
+            }
+        };
+        let found = match id {
+            Some(id) => records.iter().find(|r| r.id == id),
+            None => records.last(),
+        };
+        let Some(record) = found else {
+            return;
+        };
+        let session = session_of(record, &records);
+        let actions = actions(record, session.as_ref(), &self.agent, &self.owner);
+        let mut book = self.ledger.lock().unwrap_or_else(|p| p.into_inner());
+        for (part, action) in actions.into_iter().enumerate() {
+            let Ok(part) = u32::try_from(part) else {
+                return;
+            };
+            let item = Item::Edit {
+                id: record.id.clone(),
+                part,
+            };
+            if book.is_decided(&item) {
+                continue;
+            }
+            if let Err(e) = book.append(Record::Decided {
+                time: now,
+                item,
+                action,
+            }) {
+                log(&format!("memoryd: {e}; {}", ledger::RUNBOOK));
+                return;
+            }
+        }
+    }
+}

crates/memoryd/src/honcho/mod.rs

@@ -4,6 +4,7 @@ pub mod client;
 pub mod config;
 pub mod decide;
 pub mod ledger;
+pub mod mirror;
 pub mod naming;
 pub mod state;
 pub mod turns;

crates/memoryd/src/honcho/turns.rs

@@ -5,6 +5,7 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, MutexGuard};
 
 use bxcore::{LogRecord, SessionId, SessionStateReply, Timestamp};
 
@@ -32,9 +33,11 @@ enum Walk {
     Stop,
 }
 
-/// The turns reported and not yet decided, and the ledger their decisions go to.
+/// The turns reported and not yet decided, and the ledger their decisions go to. The ledger has a
+/// lock of its own, held only to read or append: a walk holds `TurnSync`'s lock while it asks
+/// `brokerd`, and a memory write, decided under the index lock, takes only the ledger's.
 pub struct TurnSync {
-    ledger: Ledger,
+    ledger: Arc<Mutex<Ledger>>,
     agent: String,
     sessions: PathBuf,
     /// Each session with undecided turns, by its id, and the highest turn `loopd` reported for
@@ -99,7 +102,7 @@ impl TurnSync {
     /// `loopd`'s session logs.
     pub fn new(ledger: Ledger, agent: &str, sessions: &Path) -> TurnSync {
         TurnSync {
-            ledger,
+            ledger: Arc::new(Mutex::new(ledger)),
             agent: agent.to_string(),
             sessions: sessions.to_path_buf(),
             reported: BTreeMap::new(),
@@ -108,8 +111,14 @@ impl TurnSync {
         }
     }
 
-    pub fn ledger(&self) -> &Ledger {
-        &self.ledger
+    /// The ledger, locked. Never hold it while taking another lock.
+    pub fn ledger(&self) -> MutexGuard<'_, Ledger> {
+        self.ledger.lock().unwrap_or_else(|p| p.into_inner())
+    }
+
+    /// The ledger's lock itself, for the worker and for memory writes.
+    pub fn shared_ledger(&self) -> Arc<Mutex<Ledger>> {
+        Arc::clone(&self.ledger)
     }
 
     /// The directory of `loopd`'s session logs.
@@ -128,7 +137,7 @@ impl TurnSync {
     /// 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() {
+        for (item, _) in self.ledger().decisions() {
             if let Item::Turn { session, .. } = item
                 && !sessions.contains(session)
             {
@@ -147,7 +156,7 @@ impl TurnSync {
 
     /// 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)
+        self.ledger().append(record)
     }
 
     /// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk.
@@ -200,7 +209,7 @@ impl TurnSync {
             turn,
         };
         let undecided: Vec<u64> = (1..=limit)
-            .filter(|turn| !self.ledger.is_decided(&item(*turn)))
+            .filter(|turn| !self.ledger().is_decided(&item(*turn)))
             .collect();
         if undecided.is_empty() {
             return if held { Walk::Waiting } else { Walk::Done };
@@ -217,7 +226,8 @@ impl TurnSync {
                 item: item(turn),
                 action,
             };
-            if let Err(e) = self.ledger.append(record) {
+            let appended = self.ledger().append(record);
+            if let Err(e) = appended {
                 log(&format!("memoryd: {e}; {}", ledger::RUNBOOK));
                 return Walk::Stop;
             }

crates/memoryd/src/honcho/worker.rs

@@ -6,7 +6,7 @@
 //! 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::path::{Path, PathBuf};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
@@ -20,6 +20,7 @@ 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::mirror::Mirror;
 use crate::honcho::naming::{HonchoSession, honcho_session};
 use crate::honcho::state;
 use crate::honcho::turns::{LEDGER, TurnSync, log_path, read_log};
@@ -114,6 +115,9 @@ struct Next {
 
 pub struct Worker {
     sync: Arc<Mutex<TurnSync>>,
+    /// The ledger's own lock (task 12): taken alone, briefly, never with `sync`'s.
+    ledger: Arc<Mutex<Ledger>>,
+    sessions: PathBuf,
     clients: Clients,
     wake: Arc<Wake>,
     ask: AskBox,
@@ -185,8 +189,14 @@ impl Worker {
         tuning: Tuning,
         agent: &str,
     ) -> Worker {
+        let (ledger, sessions) = {
+            let sync = sync.lock().unwrap_or_else(|p| p.into_inner());
+            (sync.shared_ledger(), sync.sessions().to_path_buf())
+        };
         Worker {
             sync,
+            ledger,
+            sessions,
             clients,
             wake,
             ask,
@@ -279,8 +289,7 @@ impl Worker {
 
     /// 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();
+        let ledger = self.ledger.lock().unwrap_or_else(|p| p.into_inner());
         ledger.outbox().into_iter().find_map(|(item, action)| {
             let Item::Turn { session, turn } = &item else {
                 return None;
@@ -304,13 +313,7 @@ impl Worker {
 
     /// 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)
+        let (text, reply) = read_turn(&log_path(&self.sessions, &next.session), next.turn)
             .ok_or(Unsent::Local("log"))?;
         let honcho = next.honcho.as_ref().ok_or(Unsent::Local("log"))?;
         let messages = batch(
@@ -367,7 +370,7 @@ impl Worker {
     /// would be repeated at every step.
     fn record(&mut self, record: Record) -> Step {
         let result = self
-            .sync
+            .ledger
             .lock()
             .unwrap_or_else(|p| p.into_inner())
             .append(record);
@@ -394,6 +397,19 @@ pub struct Opened {
     pub sync: Arc<Mutex<TurnSync>>,
     clients: Clients,
     agent: String,
+    owner: String,
+}
+
+impl Opened {
+    /// The memory writes' side of the ledger (task 12).
+    pub fn mirror(&self) -> Mirror {
+        let ledger = self
+            .sync
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .shared_ledger();
+        Mirror::new(ledger, &self.agent, &self.owner)
+    }
 }
 
 /// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the
@@ -419,6 +435,7 @@ pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result<Opened,
         sync: Arc::new(Mutex::new(sync)),
         clients,
         agent: honcho.agent.clone(),
+        owner: honcho.owner.clone(),
     })
 }
 

crates/memoryd/src/serve.rs

@@ -17,6 +17,7 @@ use crate::config::Config;
 use crate::edit::{self, Context};
 use crate::files::Store;
 use crate::honcho::ledger::{self, LedgerError};
+use crate::honcho::mirror::Mirror;
 use crate::honcho::state;
 use crate::honcho::turns::TurnSync;
 use crate::honcho::worker::{self, Wake};
@@ -43,6 +44,8 @@ pub struct Memoryd {
     honcho: Option<Arc<Mutex<TurnSync>>>,
     /// Wakes the Honcho worker when a turn has been decided.
     wake: Arc<Wake>,
+    /// With `[honcho]`: decides memory writes into the ledger, under the ledger's own lock.
+    mirror: Option<Mirror>,
     log: Log,
 }
 
@@ -112,13 +115,14 @@ pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
         (admin_listener, Which::Admin),
     ];
     let wake = Arc::new(Wake::default());
-    let honcho = match opened {
+    let (honcho, mirror) = match opened {
         Some(opened) => {
             let sync = Arc::clone(&opened.sync);
+            let mirror = opened.mirror();
             worker::spawn(opened, &config, Arc::clone(&wake), Arc::clone(&log))?;
-            Some(sync)
+            (Some(sync), Some(mirror))
         }
-        None => None,
+        None => (None, None),
     };
     Ok(Running {
         memoryd: Arc::new(Memoryd {
@@ -127,6 +131,7 @@ pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
             state: Mutex::new(State { index }),
             honcho,
             wake,
+            mirror,
             log,
         }),
         listeners,
@@ -217,6 +222,14 @@ impl Memoryd {
         }
     }
 
+    /// With `[honcho]`: decide journalled edit `id` (the latest when `None`) into the ledger. Called
+    /// under the index lock, so edits are decided in the journal's order (M5c spec, section 4).
+    fn mirror(&self, id: Option<&str>) {
+        if let Some(mirror) = &self.mirror {
+            mirror.journalled(&self.store, id, Timestamp::now(), &*self.log);
+        }
+    }
+
     /// With `[honcho]`: decide the reported turn, and every other that waits (`TurnSync`).
     fn decide_turns(&self, update: &SessionUpdated) {
         let Some(honcho) = &self.honcho else {
@@ -359,6 +372,7 @@ impl Memoryd {
         } else {
             let outcome = edit::apply(&self.store, &args, &ctx)?;
             let id = journal::record(&self.store, &ctx, &outcome)?;
+            self.mirror(Some(&id));
             if let Some(rule) = outcome.flag {
                 (self.log)(
                     format!(
@@ -441,6 +455,9 @@ impl Memoryd {
             MemoryCommand::Accept => review::accept(&self.store, arg, now),
             MemoryCommand::Reject => review::reject(&self.store, arg, now),
         };
+        if result.is_ok() && matches!(command, MemoryCommand::Undo | MemoryCommand::Accept) {
+            self.mirror(None);
+        }
         self.index_notes(&state);
         match result {
             Ok(text) => Message::MemoryAdminReply(MemoryAdminReply { text }),