M5c-1 plan: task 07, the Honcho ledger in memoryd

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 06:43:05 -07:00
parent 0e1e070fc3
commit aef6525cc7
6 changed files with 1576 additions and 0 deletions
@@ -0,0 +1,318 @@
+//! The Honcho ledger, `memory/honcho.jsonl`: one JSON line per event, appended and never rewritten
+//! (M5c spec, section 5). A turn or a memory write is `decided` once, then `sent` or `failed` once.
+//! What is decided and neither sent nor failed is the outbox; there is no other queue. `memoryd`
+//! reads the ledger through at start, so nothing is decided or sent twice.
+
+use std::collections::HashMap;
+use std::fs::OpenOptions;
+use std::io::Write;
+use std::os::unix::fs::OpenOptionsExt;
+use std::path::{Path, PathBuf};
+
+use bxcore::{SessionId, Timestamp};
+use serde::{Deserialize, Serialize};
+
+/// The runbook entry for a ledger that does not read through.
+pub const RUNBOOK: &str = "see docs/runbook.md#honcho-ledger-damaged";
+
+/// One of this agent's three workspaces.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Workspace {
+    Trusted,
+    Untrusted,
+    Fleet,
+}
+
+/// What is sent: a finished turn of a session, or one part of a journalled memory edit.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
+pub enum Item {
+    Turn { session: SessionId, turn: u64 },
+    Edit { id: String, part: u32 },
+}
+
+/// Where an item goes, decided once. A turn's text is read from the session log when it is sent,
+/// never stored here.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
+pub enum Action {
+    /// The turn's two messages, the speaker's and the agent's reply.
+    Turn {
+        workspace: Workspace,
+        speaker: String,
+    },
+    /// A conclusion about `peer`, stamped with the Honcho session of `session`.
+    Conclude {
+        workspace: Workspace,
+        peer: String,
+        session: SessionId,
+        text: String,
+    },
+    /// Delete the conclusion that a sent `Conclude` with this text created.
+    Unconclude { workspace: Workspace, text: String },
+    /// A message from the agent's peer in the fleet workspace.
+    Fleet { session: SessionId, text: String },
+    /// `retracted: <text>`, from the same peer in the same workspace.
+    Retract { session: SessionId, text: String },
+    /// Nothing is sent, and why.
+    Skip { why: String },
+}
+
+/// One line of the ledger.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
+pub enum Record {
+    Decided {
+        time: Timestamp,
+        item: Item,
+        action: Action,
+    },
+    /// Honcho's ids for what the send created: a `Conclude` has exactly one.
+    Sent {
+        time: Timestamp,
+        item: Item,
+        ids: Vec<String>,
+    },
+    /// Honcho refused the send: its status, if it answered, and the kind of error.
+    Failed {
+        time: Timestamp,
+        item: Item,
+        status: Option<u16>,
+        kind: String,
+    },
+}
+
+/// How many items are in each state. `decided` counts every item, `Skip` included, so it is
+/// always `sent + waiting + failed + skipped`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct Counts {
+    pub decided: usize,
+    pub sent: usize,
+    pub waiting: usize,
+    pub failed: usize,
+    pub skipped: usize,
+}
+
+#[derive(Debug)]
+pub enum LedgerError {
+    Io(PathBuf, std::io::Error),
+    /// A complete line (numbered from 1) that does not parse, or breaks the ledger's rules.
+    Damaged {
+        path: PathBuf,
+        line: usize,
+        why: String,
+    },
+    Encode(serde_json::Error),
+    AlreadyDecided(Item),
+    NotWaiting(Item),
+}
+
+impl std::fmt::Display for LedgerError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LedgerError::Io(path, err) => write!(f, "{}: {err}", path.display()),
+            LedgerError::Damaged { path, line, why } => {
+                write!(f, "{}:{line}: {why}; {RUNBOOK}", path.display())
+            }
+            LedgerError::Encode(err) => write!(f, "a ledger record could not be encoded: {err}"),
+            LedgerError::AlreadyDecided(item) => write!(f, "{}: decided already", describe(item)),
+            LedgerError::NotWaiting(item) => {
+                write!(f, "{}: not waiting to be sent", describe(item))
+            }
+        }
+    }
+}
+
+impl std::error::Error for LedgerError {}
+
+/// Where a decided item stands.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum State {
+    Skipped,
+    Waiting,
+    Sent(Vec<String>),
+    Failed,
+}
+
+#[derive(Debug, Clone)]
+struct Entry {
+    action: Action,
+    state: State,
+}
+
+/// The ledger, read through: the file's path and the state its records add up to.
+#[derive(Debug)]
+pub struct Ledger {
+    path: PathBuf,
+    /// Every decided item, in the order of its `Decided` record.
+    order: Vec<Item>,
+    entries: HashMap<Item, Entry>,
+    /// Every sent item, in the order of its `Sent` record.
+    sends: Vec<Item>,
+    repaired: bool,
+}
+
+impl Ledger {
+    /// Read the ledger at `path` through. A missing file is an empty ledger, and nothing is
+    /// created. A last line without its newline, cut off by a crash, is dropped and cut from the
+    /// file; any other line that does not read is `Damaged`, and the file is left as it is.
+    pub fn open(path: &Path) -> Result<Ledger, LedgerError> {
+        let mut ledger = Ledger {
+            path: path.to_path_buf(),
+            order: Vec::new(),
+            entries: HashMap::new(),
+            sends: Vec::new(),
+            repaired: false,
+        };
+        let Some(bytes) = read(path)? else {
+            return Ok(ledger);
+        };
+        let whole = complete(&bytes);
+        let lines = bytes.get(..whole).unwrap_or_default();
+        for (index, line) in lines.split_inclusive(|b| *b == b'\n').enumerate() {
+            let record = parse(path, index + 1, line)?;
+            if let Err(e) = ledger.check(&record) {
+                return Err(LedgerError::Damaged {
+                    path: path.to_path_buf(),
+                    line: index + 1,
+                    why: e.to_string(),
+                });
+            }
+            ledger.apply(record);
+        }
+        if whole < bytes.len() {
+            cut(path, whole)?;
+            ledger.repaired = true;
+        }
+        Ok(ledger)
+    }
+
+    /// Whether `open` dropped a cut-off last line.
+    pub fn repaired(&self) -> bool {
+        self.repaired
+    }
+
+    /// Append `record` to the file and sync it, and only then take it into the state. A record
+    /// the rules refuse, or a failed write, changes nothing in memory.
+    pub fn append(&mut self, record: Record) -> Result<(), LedgerError> {
+        self.check(&record)?;
+        let line = serde_json::to_string(&record).map_err(LedgerError::Encode)?;
+        self.write_line(&line)?;
+        self.apply(record);
+        Ok(())
+    }
+
+    /// Whether `item` has its `Decided` record.
+    pub fn is_decided(&self, item: &Item) -> bool {
+        // `self.entries.contains_key(item)`.
+        todo!()
+    }
+
+    /// The items waiting to be sent, with their actions, in the order they were decided.
+    pub fn outbox(&self) -> Vec<(Item, Action)> {
+        // Walk `self.order`, in order, with `.iter().filter_map(..)`: for each item,
+        // `let entry = self.entries.get(item)?;`, then keep
+        // `(item.clone(), entry.action.clone())` only when `entry.state == State::Waiting`:
+        // `(entry.state == State::Waiting).then(|| ..)`. `.collect()` the result.
+        todo!()
+    }
+
+    /// The ids of `item`'s `Sent` record, if it has one.
+    pub fn sent_ids(&self, item: &Item) -> Option<&[String]> {
+        // `match &self.entries.get(item)?.state`: `State::Sent(ids)` -> `Some(ids.as_slice())`; any
+        // other state -> `None`.
+        todo!()
+    }
+
+    /// The latest sent `Conclude` of `text` in `workspace` that no sent `Unconclude` has undone:
+    /// its item and the conclusion's id.
+    pub fn conclusion(&self, workspace: Workspace, text: &str) -> Option<(Item, String)> {
+        // `let mut live: Vec<(Item, String)> = Vec::new();`. For each `item` in `&self.sends`
+        // (the order of the `Sent` records):
+        // `let Some(entry) = self.entries.get(item) else { continue; };` and
+        // `let State::Sent(ids) = &entry.state else { continue; };`. Then `match &entry.action`:
+        // `Action::Conclude { workspace: w, text: t, .. } if *w == workspace && t == text` pushes
+        // `(item.clone(), id.clone())` when `ids.first()` is `Some(id)` (no ids: nothing);
+        // `Action::Unconclude { workspace: w, text: t }` with the same guard does `live.pop();`;
+        // `_ => {}`. At the end, `live.pop()`.
+        todo!()
+    }
+
+    /// How many items are in each state.
+    pub fn counts(&self) -> Counts {
+        // Start from `Counts::default()`. For each entry of `self.entries.values()`, add 1 to
+        // `decided`, and 1 to the field of its state: `Skipped` -> `skipped`, `Waiting` ->
+        // `waiting`, `Sent(_)` -> `sent`, `Failed` -> `failed`. Return the counts.
+        todo!()
+    }
+
+    /// The ledger's rules: an item is decided once; it is sent or failed once, and only while it
+    /// is waiting.
+    fn check(&self, record: &Record) -> Result<(), LedgerError> {
+        // `match record`. `Record::Decided { item, .. }`: if `self.entries.contains_key(item)`,
+        // `Err(LedgerError::AlreadyDecided(item.clone()))`, else `Ok(())`.
+        // `Record::Sent { item, .. } | Record::Failed { item, .. }`: a block `{ .. }` holding
+        // `match self.entries.get(item)`: `Some(entry) if entry.state == State::Waiting => Ok(())`,
+        // and otherwise (no entry, or another state) `Err(LedgerError::NotWaiting(item.clone()))`.
+        todo!()
+    }
+
+    /// Take a checked record into the state.
+    fn apply(&mut self, record: Record) {
+        // `match record`. `Record::Decided { item, action, .. }`: `let state = match action {
+        // Action::Skip { .. } => State::Skipped, _ => State::Waiting };`, push `item.clone()`
+        // onto `self.order`, then `self.entries.insert(item, Entry { action, state })`.
+        // `Record::Sent { item, ids, .. }`: `if let Some(entry) = self.entries.get_mut(&item)`,
+        // set `entry.state = State::Sent(ids)`; then push `item` onto `self.sends`.
+        // `Record::Failed { item, .. }`: the same `if let`, setting `entry.state = State::Failed`.
+        todo!()
+    }
+
+    /// Append `line` and a newline to the file in one write, then sync the data.
+    fn write_line(&self, line: &str) -> Result<(), LedgerError> {
+        // `let io = |e| LedgerError::Io(self.path.clone(), e);`. Open `self.path` with
+        // `OpenOptions::new().append(true).create(true).mode(0o600)`, then ONE
+        // `write_all(format!("{line}\n").as_bytes())`, then `sync_data()`. Each error goes through
+        // `.map_err(io)`; the last line is `file.sync_data().map_err(io)`.
+        todo!()
+    }
+}
+
+/// The file's bytes, or `None` when there is no file.
+fn read(path: &Path) -> Result<Option<Vec<u8>>, LedgerError> {
+    // `match std::fs::read(path)`: `Ok(bytes)` -> `Ok(Some(bytes))`; an error `e` whose `e.kind()`
+    // is `std::io::ErrorKind::NotFound` -> `Ok(None)`; any other error `e` ->
+    // `Err(LedgerError::Io(path.to_path_buf(), e))`.
+    todo!()
+}
+
+/// How many bytes the complete lines take: up to and including the last newline.
+fn complete(bytes: &[u8]) -> usize {
+    // `match bytes.iter().rposition(|b| *b == b'\n')`: `Some(last)` -> `last + 1`; `None` -> `0`.
+    todo!()
+}
+
+/// Line `number` (from 1), with its newline, as a record.
+fn parse(path: &Path, number: usize, line: &[u8]) -> Result<Record, LedgerError> {
+    // `serde_json::from_slice(line)`, its error `e` mapped to `LedgerError::Damaged` with `path:
+    // path.to_path_buf()`, `line: number` and `why: e.to_string()`.
+    todo!()
+}
+
+/// Cut the file to its first `keep` bytes, and sync it.
+fn cut(path: &Path, keep: usize) -> Result<(), LedgerError> {
+    // `let io = |e| LedgerError::Io(path.to_path_buf(), e);` and `let len =
+    // u64::try_from(keep).map_err(|e| io(std::io::Error::other(e)))?;`. Open `path` with
+    // `OpenOptions::new().write(true)` (no `create`), then `set_len(len)`, then `sync_data()`. Each
+    // error goes through `.map_err(io)`; the last line is `file.sync_data().map_err(io)`.
+    todo!()
+}
+
+/// An item as the error messages name it.
+fn describe(item: &Item) -> String {
+    match item {
+        Item::Turn { session, turn } => format!("turn {turn} of session {}", session.as_str()),
+        Item::Edit { id, part } => format!("part {part} of edit {id}"),
+    }
+}