From aef6525cc78f432dfcd16c93c967c5bc14f6f8dc Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 28 Sep 2026 06:43:05 -0700 Subject: [PATCH] M5c-1 plan: task 07, the Honcho ledger in memoryd Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5c1/07-memoryd-honcho-ledger.md | 344 +++++++++++++ .../files/crates/memoryd/src/honcho/ledger.rs | 318 ++++++++++++ .../crates/memoryd/tests/honcho_ledger.rs | 451 ++++++++++++++++++ .../memoryd/tests/honcho_ledger_file.rs | 303 ++++++++++++ .../crates/memoryd/tests/support/ledger.rs | 123 +++++ .../files/runbook-honcho-ledger-damaged.md | 37 ++ 6 files changed, 1576 insertions(+) create mode 100644 docs/plans/M5c1/07-memoryd-honcho-ledger.md create mode 100644 docs/plans/M5c1/files/crates/memoryd/src/honcho/ledger.rs create mode 100644 docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger.rs create mode 100644 docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger_file.rs create mode 100644 docs/plans/M5c1/files/crates/memoryd/tests/support/ledger.rs create mode 100644 docs/plans/M5c1/files/runbook-honcho-ledger-damaged.md diff --git a/docs/plans/M5c1/07-memoryd-honcho-ledger.md b/docs/plans/M5c1/07-memoryd-honcho-ledger.md new file mode 100644 index 0000000..e764957 --- /dev/null +++ b/docs/plans/M5c1/07-memoryd-honcho-ledger.md @@ -0,0 +1,344 @@ +# M5c-1 task 07: `memoryd`'s Honcho ledger, `memory/honcho.jsonl` + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `memoryd: the Honcho ledger, decided, sent and failed, read through at start` +**Log row:** the Task cell is `M5c1/07-memoryd-honcho-ledger`. + +## Goal + +From M5c on, `memoryd` sends finished turns and memory writes to Honcho (M5c spec, section 5). +Each one is first **decided** (where it goes, or that it goes nowhere), then **sent** or +**failed** (Honcho refused it). This task adds the record of that: the ledger, an append-only file +of JSON lines, `/memory/honcho.jsonl`. What is decided and neither sent nor failed is the +**outbox**, the work still to do; there is no other queue. `memoryd` reads the ledger through when +it starts, so nothing is decided twice or sent twice. + +This task is the module `memoryd::honcho::ledger` alone. Nothing calls it yet: task 09 opens it +at start and decides turns into it, later tasks send from its outbox. Its path is passed in. + +- **Items.** A turn of a session (`Item::Turn { session, turn }`), or one part of a journalled + memory edit (`Item::Edit { id, part }`; the id is the journal's, `2026-09-28-0001`). +- **Actions** (what a `Decided` record says): the turn's messages to a workspace; a conclusion + to create or delete; a message, or its retraction, in the fleet workspace; or `Skip`, nothing + to send, and why. A turn's text is never stored here: the sender reads it from the session log. +- **The rules.** An item has exactly one `Decided` record. It has at most one `Sent` or `Failed` + record, and only while it is **waiting**: decided, not `Skip`, and neither sent nor failed yet. + `append` refuses a record that breaks a rule, and writes nothing for it. +- **Durable before believed.** `append` writes the line, syncs it to disk, and only then takes it + into the ledger's state in memory, as `loopd` does with its session log. A failed write leaves + the state as it was, so the ledger never says an item is decided or sent when the file does not. +- **Reading through.** `open` reads the whole file and replays it. A missing file is an empty + ledger. A last line with no newline was cut off by a crash in the middle of an append: it is + dropped and cut from the file, and `repaired()` says so. Any other line that does not parse, or + that breaks a rule when replayed in order, stops the ledger: `LedgerError::Damaged` with its + line number and the runbook pointer `honcho-ledger-damaged`, and the file is left as it is. + +## Interfaces + +Later tasks use these exactly as written. They are in the skeleton you copy (step 3); only 12 +private or small function bodies are yours to fill. + +In `crates/memoryd/src/honcho/ledger.rs` (the module `memoryd::honcho::ledger`): + +```rust +pub const RUNBOOK: &str = "see docs/runbook.md#honcho-ledger-damaged"; + +pub enum Workspace { Trusted, Untrusted, Fleet } +pub enum Item { Turn { session: SessionId, turn: u64 }, Edit { id: String, part: u32 } } +pub enum Action { + Turn { workspace: Workspace, speaker: String }, + Conclude { workspace: Workspace, peer: String, session: SessionId, text: String }, + Unconclude { workspace: Workspace, text: String }, + Fleet { session: SessionId, text: String }, + Retract { session: SessionId, text: String }, + Skip { why: String }, +} +pub enum Record { + Decided { time: Timestamp, item: Item, action: Action }, + Sent { time: Timestamp, item: Item, ids: Vec }, + Failed { time: Timestamp, item: Item, status: Option, kind: String }, +} +pub struct Counts { pub decided: usize, pub sent: usize, pub waiting: usize, + pub failed: usize, pub skipped: usize } +pub enum LedgerError { + Io(PathBuf, std::io::Error), + Damaged { path: PathBuf, line: usize, why: String }, + Encode(serde_json::Error), + AlreadyDecided(Item), + NotWaiting(Item), +} +pub struct Ledger { /* private */ } + +impl Ledger { + pub fn open(path: &Path) -> Result; + pub fn repaired(&self) -> bool; + pub fn append(&mut self, record: Record) -> Result<(), LedgerError>; + pub fn is_decided(&self, item: &Item) -> bool; + pub fn outbox(&self) -> Vec<(Item, Action)>; + pub fn sent_ids(&self, item: &Item) -> Option<&[String]>; + pub fn conclusion(&self, workspace: Workspace, text: &str) -> Option<(Item, String)>; + pub fn counts(&self) -> Counts; +} +``` + +`SessionId` and `Timestamp` are `bxcore`'s. The derives and serde attributes are in the skeleton; +do not change them (see "The lines"). + +## What each public function gives + +| Function | Gives | +|---|---| +| `open` | the ledger read through; see "Reading through" above | +| `repaired` | `true` when `open` dropped a cut-off last line | +| `append` | the record checked, written with its newline in one write, synced, then taken into the state; a refused or failed append changes nothing | +| `is_decided` | whether the item has its `Decided` record (a `Skip` counts) | +| `outbox` | every waiting item with its action, in the order of the `Decided` records | +| `sent_ids` | the ids of the item's `Sent` record; `None` when it has none (waiting, failed, skipped, unknown) | +| `conclusion` | the conclusion to delete for an undo: see below | +| `counts` | how many items are decided (every one, `Skip` included), sent, waiting, failed and skipped; `decided` is always the sum of the other four | + +**`conclusion(workspace, text)`.** An undo must delete the conclusion that the undone write +created. Honcho gave its id in the `Sent` record of the `Conclude` (a `Conclude`'s `Sent` has +exactly one id; one with no ids is ignored). Replay the `Sent` records in the order they were +appended: a sent `Conclude` of this workspace and text puts `(item, its id)` on top of a stack, +and a sent `Unconclude` of the same workspace and text takes the top one off. The answer is the +top of the stack, or `None`. So it is the latest one sent that is not undone yet; another +workspace's, or one decided but not sent, is never the answer. + +**The errors' text** (`Display`, given in the skeleton): + +| Error | Text | +|---|---| +| `Io(path, e)` | `: ` | +| `Damaged { path, line, why }` | `:: ; see docs/runbook.md#honcho-ledger-damaged` (the line counted from 1) | +| `Encode(e)` | `a ledger record could not be encoded: ` | +| `AlreadyDecided(item)` | `turn 3 of session chat-1: decided already`, or `part 0 of edit 2026-09-28-0001: decided already` | +| `NotWaiting(item)` | `turn 3 of session chat-1: not waiting to be sent` (the same naming) | + +When a replayed line breaks a rule, `why` is the text of `AlreadyDecided` or `NotWaiting`; when it +does not parse, `why` is serde's error text. + +## The lines + +One record per line, as `serde_json::to_string` writes it, then `\n`. Every enum except +`Workspace` is internally tagged: its variant is the field `"type"`, in snake case, first; and +every one refuses unknown fields. `Workspace` is a plain string. A `Timestamp` is RFC 3339 with +milliseconds; a `SessionId` a string. For example (each is one line in the file): + +```text +{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-1","turn":3},"action":{"type":"turn","workspace":"trusted","speaker":"kyle"}} +{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"conclude","workspace":"trusted","peer":"kyle","session":"chat-1","text":"Kyle takes his coffee black."}} +{"type":"sent","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-1","turn":3},"ids":["msg-a1","msg-a2"]} +{"type":"failed","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-2","turn":1},"status":null,"kind":"refused"} +``` + +This is our own format, so unknown fields are errors: `#[serde(tag = "type", rename_all = +"snake_case", deny_unknown_fields)]` on `Item`, `Action` and `Record`. serde honours +`deny_unknown_fields` on internally tagged enums, at every level (a given test adds a field to the +record, to its item and to its action, and expects an error each time). The skeleton has all of +this already. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/memoryd/src/honcho/mod.rs` | `pub mod ledger;` (edit H1) | +| `crates/memoryd/src/honcho/ledger.rs` | new: a skeleton you copy, then fill 12 bodies (step 4) | +| `docs/runbook.md` | the entry `honcho-ledger-damaged`, appended (step 6) | +| everything else in `memoryd` (`config.rs`, `serve.rs`, `main.rs`, `honcho/config.rs`, …) | **no change** | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - new tests: `crates/memoryd/tests/honcho_ledger.rs`, `crates/memoryd/tests/honcho_ledger_file.rs` + - new test support: `crates/memoryd/tests/support/ledger.rs` + - the skeleton `crates/memoryd/src/honcho/ledger.rs` (you fill 12 bodies in it) +- Append, from `docs/plans/M5c1/files/`, to the end of `docs/runbook.md`, with one blank line + before it: `runbook-honcho-ledger-damaged.md` (step 6 gives the command). +- Modify: `crates/memoryd/src/honcho/mod.rs`, `docs/implementer-log.md`. +- Do not touch: any other source file, any file of another crate, any `Cargo.toml`, any other + test, `crates/memoryd/tests/support/tmp.rs`, and in the skeleton anything but the 12 `todo!()` + bodies. + +## Facts you need + +- **The state in memory.** The skeleton gives it: `order` (every decided item, in the order of + its `Decided` record), `entries` (a `HashMap` from each item to its `Entry { action, state }`), + `sends` (every sent item, in the order of its `Sent` record), and `repaired`. An entry's + `State` is `Skipped` (decided `Skip`), `Waiting`, `Sent(ids)` or `Failed`. Only `apply` changes + them, and only for a record `check` accepted. +- **`open` and `append` are given**, as glue over your functions. `open` checks every complete + line before it cuts a cut-off tail, so a damaged file is never changed. `append` calls `check`, + encodes, calls `write_line`, and only then `apply`. Read both before you start. +- **One write per line.** `write_line` builds the line and its newline as one string and writes + it with one `write_all`, then `sync_data`: a crash leaves either the whole line or a cut-off + line without its newline, which `open` drops. Two writes, or no sync, would pass every test and + still be wrong; step 7 checks for them. +- **A missing file creates nothing.** `open` only reads. `write_line` creates the file on the + first append, at mode `0o600`, and never creates the directory: a missing `memory/` is an + `Io` error. The file's mode is given only when the file is created (`OpenOptionsExt::mode`, + and the process's umask can only take bits away). +- **Bytes, not text.** `open` reads the file as bytes (`std::fs::read`), because a cut-off last + line can end inside a multi-byte character. A complete line is parsed with + `serde_json::from_slice`, which accepts the trailing newline. +- **The given tests.** + - `honcho_ledger.rs` (12 tests): each kind of line exactly; `append` writes the line and a + newline; unknown fields and values refused; the outbox (what is in it, and its order); a + second decision refused and nothing written; `Sent` or `Failed` refused before the decision, + after an outcome and for a `Skip`; `sent_ids`; `conclusion` (the latest not undone, only its + workspace and text, only once sent); the counts add up. + - `honcho_ledger_file.rs` (12 tests): a missing and an empty file; reopening gives the same + state; a cut-off last line (in the middle, one byte short, inside a character, the only line) + dropped and cut; a damaged line and a line that breaks a rule, with their numbers, the text + and the file unchanged; mode 0600; a missing directory; a failed write changes nothing (the + test turns the ledger's path into a directory); an unreadable ledger. + - `support/ledger.rs`: record builders (`decided`, `sent`, `failed`, `turn`, `edit`, …) and + `Home`, a ledger path in a directory removed on drop. +- The calls the bodies use: + + | Call | Signature, and what it gives | + |---|---| + | `HashMap::contains_key` | `self.entries.contains_key(item) -> bool` | + | `HashMap::get`, `get_mut` | `self.entries.get(item) -> Option<&Entry>`; `get_mut(&item) -> Option<&mut Entry>` | + | `HashMap::insert` | `self.entries.insert(item, Entry { action, state })` (the old value, ignored) | + | `HashMap::values` | `self.entries.values()`: every `&Entry` | + | `Iterator::filter_map` | `.iter().filter_map(\|item\| { ..; Some(..) }).collect()` | + | `bool::then` | `(cond).then(\|\| value) -> Option`: `Some(value)` when `cond` | + | `slice::first` | `ids.first() -> Option<&String>` | + | `Vec::push`, `Vec::pop` | `live.push(x)`; `live.pop() -> Option`: takes the last one off | + | `Vec::as_slice` | `ids.as_slice() -> &[String]` | + | `OpenOptions` | `OpenOptions::new().append(true).create(true).mode(0o600).open(&path) -> io::Result`; `.mode` is `std::os::unix::fs::OpenOptionsExt` (imported) | + | `Write::write_all` | `file.write_all(bytes: &[u8]) -> io::Result<()>` (`std::io::Write`, imported) | + | `File::sync_data` | `file.sync_data() -> io::Result<()>`: the data on disk, not necessarily the metadata | + | `File::set_len` | `file.set_len(size: u64) -> io::Result<()>`: shrinks the file to `size` bytes; the file needs `write(true)` | + | `std::fs::read` | `std::fs::read(path) -> io::Result>` | + | `io::Error::kind` | `e.kind() == std::io::ErrorKind::NotFound` | + | `io::Error::other` | `std::io::Error::other(e)`: wraps any error as an `io::Error` | + | `u64::try_from` | `u64::try_from(keep: usize) -> Result` | + | `Iterator::rposition` | `bytes.iter().rposition(\|b\| *b == b'\n') -> Option`: the index of the last newline | + | `serde_json::from_slice` | `serde_json::from_slice::(line: &[u8]) -> serde_json::Result` | + +## The edit, exactly + +### `crates/memoryd/src/honcho/mod.rs` + +**H1.** The line + +```rust +pub mod config; +``` + +becomes + +```rust +pub mod config; +pub mod ledger; +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, three commands: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger_file.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/support/ledger.rs crates/memoryd/tests/support/ + ``` + +- [ ] **2. See them fail.** `cargo build --tests --keep-going -p memoryd`, with a shell timeout + of 600000 ms: two test targets do not compile, `honcho_ledger` and `honcho_ledger_file`, with 2 + errors each, both `E0432` (``unresolved import `memoryd::honcho::ledger` ``): one in the test + file, one in `support/ledger.rs`. Every other target compiles. +- [ ] **3. The module.** Make edit H1, then copy the skeleton: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/ledger.rs crates/memoryd/src/honcho/ + ``` + + `cargo check -p memoryd`: no error, and 20 warnings, all from the unfilled bodies: three + `unused import` (`std::fs::OpenOptions`, `std::io::Write`, `std::os::unix::fs::OpenOptionsExt`); + fourteen `unused variable` (`path` three times, `item`, `line` and `record` twice each, + `workspace`, `text`, `bytes`, `number`, `keep`); ``fields `action` and `state` are never + read``; ``fields `path`, `order`, `entries`, and `sends` are never read``; and ``variants + `Skipped`, `Waiting`, `Sent`, and `Failed` are never constructed``. They go as you fill the + bodies. + + Then `cargo test -p memoryd --test honcho_ledger`: 2 tests pass + (`each_record_is_one_line_exactly_as_given` and `an_unknown_field_or_value_is_refused`: the + types are given) and 10 fail with `not yet implemented`. +- [ ] **4. Fill the twelve bodies of `honcho/ledger.rs`, one at a time, in the order of the + file**, each from its comment, and run `cargo check -p memoryd` after each: `is_decided`, + `outbox`, `sent_ids`, `conclusion`, `counts`, `check`, `apply`, `write_line`, then the free + functions `read`, `complete`, `parse`, `cut`. When a body is filled, delete its instruction + comment (the `//` lines inside the body) and keep the function's doc comment (`///`) as it is. + Change nothing else in the file. After the twelfth, `cargo check -p memoryd` prints no warning. +- [ ] **5. The tests.** This command, and every test passes (12 and 12): + + ```sh + cargo test -p memoryd --test honcho_ledger --test honcho_ledger_file + ``` +- [ ] **6. The runbook.** From the repository root: + + ```sh + { echo; cat docs/plans/M5c1/files/runbook-honcho-ledger-damaged.md; } >> docs/runbook.md + ``` + + Then `grep -c '^## honcho-ledger-damaged$' docs/runbook.md` prints `1`. +- [ ] **7. Check.** Four commands: + - `grep -c 'sync_data' crates/memoryd/src/honcho/ledger.rs` prints `2` (in `write_line` and + in `cut`), and `grep -c 'write_all' crates/memoryd/src/honcho/ledger.rs` prints `1`. + - `grep -rn 'runbook.md#honcho-ledger-damaged' crates/memoryd/src` prints exactly one line, + the `RUNBOOK` constant in `honcho/ledger.rs`. + - `grep -n 'create_dir' crates/memoryd/src/honcho/ledger.rs` prints nothing: the ledger never + makes its directory. + - `git diff --stat -- crates/bxcore crates/loopd crates/brokerd crates/gatewayd crates/bxctl + Cargo.toml crates/memoryd/Cargo.toml` prints nothing. +- [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1212 tests: 1188 before this task, plus 24 new ones (12 + in `honcho_ledger.rs`, 12 in `honcho_ledger_file.rs`). +- [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/07-memoryd-honcho-ledger`. `git status --short` lists `crates/memoryd/src/honcho/mod.rs`, + `docs/runbook.md` and your log as modified, and `crates/memoryd/src/honcho/ledger.rs`, the two + test files and `crates/memoryd/tests/support/ledger.rs` as new. Stage with + `git add crates/memoryd docs/runbook.md docs/implementer-log.md Cargo.lock`, then commit as + `AGENTS.md` says. `Cargo.lock` does not change in this task; the `git add` is harmless. + +## Done when + +- `make gate` prints `gate: ok` with 1212 tests. +- `git diff HEAD~1 --stat` lists only `honcho/mod.rs`, `honcho/ledger.rs`, the two new test + files, `tests/support/ledger.rs`, `docs/runbook.md` and `docs/implementer-log.md`. + +## Stop and report if + +- The line of edit H1 is not in `crates/memoryd/src/honcho/mod.rs` exactly as shown, or is there + more than once, or `crates/memoryd/tests/support/ledger.rs` exists before step 1. +- Step 2 or 3 shows a different result: other test targets, errors of another kind, other + warnings, or other tests passing or failing. +- `cargo check` reports an error after a body you filled, and the comment does not tell you how + to fix it. Quote the error, and stop. +- A comment in the skeleton does not fit the code under it, or disagrees with this task's text. + Quote both and stop. +- After step 4 a test fails although each body follows its comment. Do not change a test, the + support file or the skeleton's given code (`open`, `append`, the types, `Display`): stop, and + give the failing tests' names and their messages. +- The fix seems to need a change anywhere else: another `memoryd` file, another crate, a + dependency, or a change to a derive or a serde attribute. It does not; stop and say why you + think so. +- A test outside these two files fails in the gate: check once whether it fails without your + change, then stop and report both results. +- The gate passes with a count other than 1212. + +## What the given tests pin + +Each of these changes to a finished `ledger.rs` makes at least one given test fail: the outbox +taking in a skipped, sent or failed item, or losing the decision order; `conclusion` ignoring an +`Unconclude`, the workspace, the order of the sends, or a `Sent` with no ids, or answering the +oldest; a second decision, or a `Sent` or `Failed` before the decision, after an outcome or for +a `Skip`, accepted; the state taken in before the write; the directory made; the mode; a cut-off +line kept, not cut, cut before the other lines are checked, or not reported by `repaired`; line +numbers from 0; replay without the rules; a missing file an error or an unreadable one empty; any +error text; the counts; a serde attribute removed. Two writes for one line, and a missing +`sync_data`, pass the tests; step 7 checks those. diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/ledger.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/ledger.rs new file mode 100644 index 0000000..fa163a5 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/ledger.rs @@ -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: `, 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, + }, + /// Honcho refused the send: its status, if it answered, and the kind of error. + Failed { + time: Timestamp, + item: Item, + status: Option, + 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), + 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, + entries: HashMap, + /// Every sent item, in the order of its `Sent` record. + sends: Vec, + 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 { + 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>, 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 { + // `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}"), + } +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger.rs new file mode 100644 index 0000000..790453b --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger.rs @@ -0,0 +1,451 @@ +//! The Honcho ledger, `memory/honcho.jsonl` (M5c spec, section 5): its lines, exactly, and the +//! rules of `append`, the outbox, the sent ids, the live conclusions and the counts. Reading the +//! file back is in `honcho_ledger_file.rs`. Do not edit. + +#[path = "support/ledger.rs"] +mod ledger; +#[path = "support/tmp.rs"] +mod tmp; + +use ledger::{ + Home, conclude, decided, edit, failed, sent, sid, skip, to_trusted, turn, unconclude, +}; +use memoryd::honcho::ledger::{Action, Counts, Item, LedgerError, Record, Workspace}; + +/// One record of each kind, and of each action, with its line exactly. +fn examples() -> Vec<(Record, &'static str)> { + vec![ + ( + decided(&turn("chat-1", 3), to_trusted("kyle")), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-1","turn":3},"action":{"type":"turn","workspace":"trusted","speaker":"kyle"}}"#, + ), + ( + decided( + &turn("chat-2", 1), + Action::Turn { + workspace: Workspace::Untrusted, + speaker: "ana_b".to_string(), + }, + ), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-2","turn":1},"action":{"type":"turn","workspace":"untrusted","speaker":"ana_b"}}"#, + ), + ( + decided( + &edit("2026-09-28-0001", 0), + conclude(Workspace::Trusted, "Kyle takes his coffee black."), + ), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"conclude","workspace":"trusted","peer":"kyle","session":"chat-1","text":"Kyle takes his coffee black."}}"#, + ), + ( + decided( + &edit("2026-09-28-0002", 1), + unconclude(Workspace::Trusted, "Kyle takes his coffee black."), + ), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0002","part":1},"action":{"type":"unconclude","workspace":"trusted","text":"Kyle takes his coffee black."}}"#, + ), + ( + decided( + &edit("2026-09-28-0003", 0), + Action::Fleet { + session: sid("chat-2"), + text: "The build host is \"straylight\".\nIt runs NixOS.".to_string(), + }, + ), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0003","part":0},"action":{"type":"fleet","session":"chat-2","text":"The build host is \"straylight\".\nIt runs NixOS."}}"#, + ), + ( + decided( + &edit("2026-09-28-0004", 0), + Action::Retract { + session: sid("chat-2"), + text: "The build host is titan.".to_string(), + }, + ), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0004","part":0},"action":{"type":"retract","session":"chat-2","text":"The build host is titan."}}"#, + ), + ( + decided(&turn("chat-3", 7), skip("the session is secret")), + r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-3","turn":7},"action":{"type":"skip","why":"the session is secret"}}"#, + ), + ( + sent(&turn("chat-1", 3), &["msg-a1", "msg-a2"]), + r#"{"type":"sent","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-1","turn":3},"ids":["msg-a1","msg-a2"]}"#, + ), + ( + failed(&edit("2026-09-28-0001", 0), Some(422)), + r#"{"type":"failed","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"status":422,"kind":"refused"}"#, + ), + ( + failed(&turn("chat-2", 1), None), + r#"{"type":"failed","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"chat-2","turn":1},"status":null,"kind":"refused"}"#, + ), + ] +} + +#[test] +fn each_record_is_one_line_exactly_as_given() { + for (record, text) in examples() { + assert_eq!(serde_json::to_string(&record).unwrap(), text); + assert_eq!( + serde_json::from_str::(text).unwrap(), + record, + "{text}" + ); + } + for (workspace, text) in [ + (Workspace::Trusted, "\"trusted\""), + (Workspace::Untrusted, "\"untrusted\""), + (Workspace::Fleet, "\"fleet\""), + ] { + assert_eq!(serde_json::to_string(&workspace).unwrap(), text); + assert_eq!(serde_json::from_str::(text).unwrap(), workspace); + } +} + +#[test] +fn append_writes_the_line_and_a_newline() { + let home = Home::new("ledger-append"); + let mut ledger = home.open(); + let (first, second) = (examples().remove(0), examples().remove(7)); + ledger.append(first.0).unwrap(); + assert_eq!(home.text(), format!("{}\n", first.1)); + ledger.append(second.0).unwrap(); + assert_eq!(home.text(), format!("{}\n{}\n", first.1, second.1)); +} + +#[test] +fn an_unknown_field_or_value_is_refused() { + for (_, text) in examples() { + let mut broken = vec![format!("{},\"extra\":1}}", &text[..text.len() - 1])]; + broken.push(text.replacen("\"item\":{", "\"item\":{\"extra\":1,", 1)); + if text.contains("\"action\":{") { + broken.push(text.replacen("\"action\":{", "\"action\":{\"extra\":1,", 1)); + } + for line in broken { + let err = serde_json::from_str::(&line).unwrap_err(); + assert!( + err.to_string().contains("unknown field `extra`"), + "{line}: {err}" + ); + } + } + let (_, text) = examples().remove(0); + for (from, to) in [ + ("\"trusted\"", "\"shared\""), + ("\"type\":\"decided\"", "\"type\":\"queued\""), + ( + "\"type\":\"turn\",\"session\"", + "\"type\":\"call\",\"session\"", + ), + ("\"session\":\"chat-1\"", "\"session\":\"Chat 1\""), + ( + "\"time\":\"2026-09-28T10:00:00.000Z\"", + "\"time\":\"today\"", + ), + ] { + assert_eq!(text.matches(from).count(), 1, "{from}"); + let line = text.replace(from, to); + assert!(serde_json::from_str::(&line).is_err(), "{line}"); + } +} + +#[test] +fn the_outbox_is_what_is_decided_and_not_skipped_sent_or_failed() { + let home = Home::new("ledger-outbox"); + let mut ledger = home.open(); + let (a, b, c, d, e) = ( + turn("chat-1", 1), + turn("chat-1", 2), + edit("2026-09-28-0001", 0), + edit("2026-09-28-0002", 0), + turn("chat-1", 3), + ); + ledger.append(decided(&a, to_trusted("kyle"))).unwrap(); + ledger.append(decided(&b, skip("no speaker"))).unwrap(); + ledger + .append(decided(&c, conclude(Workspace::Trusted, "one"))) + .unwrap(); + ledger + .append(decided(&d, conclude(Workspace::Trusted, "two"))) + .unwrap(); + ledger.append(decided(&e, to_trusted("ana_b"))).unwrap(); + ledger.append(sent(&c, &["c-1"])).unwrap(); + ledger.append(failed(&d, Some(400))).unwrap(); + assert_eq!( + ledger.outbox(), + vec![ + (a.clone(), to_trusted("kyle")), + (e.clone(), to_trusted("ana_b")) + ] + ); + for item in [&a, &b, &c, &d, &e] { + assert!(ledger.is_decided(item), "{item:?}"); + } + assert!(!ledger.is_decided(&turn("chat-1", 4))); + assert!(!ledger.is_decided(&edit("2026-09-28-0001", 1))); +} + +#[test] +fn the_outbox_keeps_the_order_of_decision() { + let home = Home::new("ledger-order"); + let mut ledger = home.open(); + let items = [ + turn("chat-b", 2), + turn("chat-a", 9), + edit("2026-09-28-0002", 0), + turn("chat-b", 1), + edit("2026-09-28-0001", 3), + edit("2026-09-28-0001", 0), + ]; + for item in &items { + ledger.append(decided(item, skip("placeholder"))).unwrap(); + } + assert!(ledger.outbox().is_empty()); + let home = Home::new("ledger-order-2"); + let mut ledger = home.open(); + for item in &items { + ledger.append(decided(item, to_trusted("kyle"))).unwrap(); + } + let order: Vec = ledger.outbox().into_iter().map(|(item, _)| item).collect(); + assert_eq!(order, items.to_vec()); + ledger.append(sent(&items[1], &["m"])).unwrap(); + let order: Vec = ledger.outbox().into_iter().map(|(item, _)| item).collect(); + let mut left = items.to_vec(); + left.remove(1); + assert_eq!(order, left); +} + +#[test] +fn a_second_decision_is_refused_and_writes_nothing() { + let home = Home::new("ledger-twice"); + let mut ledger = home.open(); + let item = turn("chat-1", 1); + ledger.append(decided(&item, to_trusted("kyle"))).unwrap(); + let before = home.text(); + for action in [to_trusted("kyle"), skip("again")] { + let err = ledger.append(decided(&item, action)).unwrap_err(); + assert!( + matches!(&err, LedgerError::AlreadyDecided(i) if *i == item), + "{err:?}" + ); + assert_eq!(err.to_string(), "turn 1 of session chat-1: decided already"); + } + assert_eq!(home.text(), before); + assert_eq!(ledger.outbox(), vec![(item, to_trusted("kyle"))]); + assert_eq!(ledger.counts().decided, 1); +} + +#[test] +fn sent_or_failed_before_the_decision_is_refused() { + let home = Home::new("ledger-undecided"); + let mut ledger = home.open(); + let item = edit("2026-09-28-0001", 2); + for record in [sent(&item, &["x"]), failed(&item, Some(404))] { + let err = ledger.append(record).unwrap_err(); + assert!( + matches!(&err, LedgerError::NotWaiting(i) if *i == item), + "{err:?}" + ); + assert_eq!( + err.to_string(), + "part 2 of edit 2026-09-28-0001: not waiting to be sent" + ); + } + assert!(!home.path.exists(), "a refused append wrote the file"); + assert_eq!(ledger.counts(), Counts::default()); + assert_eq!(ledger.sent_ids(&item), None); +} + +#[test] +fn an_item_is_sent_or_failed_once_and_never_when_skipped() { + let home = Home::new("ledger-once"); + let mut ledger = home.open(); + let (s, f, k) = (turn("chat-1", 1), turn("chat-1", 2), turn("chat-1", 3)); + ledger.append(decided(&s, to_trusted("kyle"))).unwrap(); + ledger.append(decided(&f, to_trusted("kyle"))).unwrap(); + ledger.append(decided(&k, skip("secret"))).unwrap(); + ledger.append(sent(&s, &["m1", "m2"])).unwrap(); + ledger.append(failed(&f, None)).unwrap(); + let before = home.text(); + for record in [ + sent(&s, &["m3"]), + failed(&s, Some(500)), + sent(&f, &["m4"]), + failed(&f, Some(400)), + sent(&k, &["m5"]), + failed(&k, None), + ] { + let err = ledger.append(record.clone()).unwrap_err(); + assert!( + matches!(err, LedgerError::NotWaiting(_)), + "{record:?}: {err:?}" + ); + } + assert_eq!(home.text(), before); + assert_eq!( + ledger.sent_ids(&s), + Some(&["m1".to_string(), "m2".to_string()][..]) + ); +} + +#[test] +fn sent_ids_are_those_of_the_sent_record() { + let home = Home::new("ledger-ids"); + let mut ledger = home.open(); + let (waiting, done, refused, skipped) = ( + turn("chat-1", 1), + turn("chat-1", 2), + turn("chat-1", 3), + turn("chat-1", 4), + ); + for item in [&waiting, &done, &refused] { + ledger.append(decided(item, to_trusted("kyle"))).unwrap(); + } + ledger + .append(decided(&skipped, skip("no speaker"))) + .unwrap(); + ledger.append(sent(&done, &["m-1", "m-2"])).unwrap(); + ledger.append(failed(&refused, Some(413))).unwrap(); + assert_eq!( + ledger.sent_ids(&done), + Some(&["m-1".to_string(), "m-2".to_string()][..]) + ); + for item in [&waiting, &refused, &skipped, &turn("chat-9", 1)] { + assert_eq!(ledger.sent_ids(item), None, "{item:?}"); + } +} + +#[test] +fn a_conclusion_is_the_latest_one_not_undone() { + let home = Home::new("ledger-latest"); + let mut ledger = home.open(); + let text = "Kyle lives in Oakland."; + let (first, second, undo1, undo2) = ( + edit("2026-09-28-0001", 0), + edit("2026-09-28-0002", 0), + edit("2026-09-28-0003", 0), + edit("2026-09-28-0004", 0), + ); + ledger + .append(decided(&first, conclude(Workspace::Trusted, text))) + .unwrap(); + assert_eq!(ledger.conclusion(Workspace::Trusted, text), None); + ledger.append(sent(&first, &["c-1"])).unwrap(); + let found = |id: &str, item: &Item| Some((item.clone(), id.to_string())); + assert_eq!( + ledger.conclusion(Workspace::Trusted, text), + found("c-1", &first) + ); + ledger + .append(decided(&second, conclude(Workspace::Trusted, text))) + .unwrap(); + ledger.append(sent(&second, &["c-2"])).unwrap(); + assert_eq!( + ledger.conclusion(Workspace::Trusted, text), + found("c-2", &second) + ); + ledger + .append(decided(&undo1, unconclude(Workspace::Trusted, text))) + .unwrap(); + assert_eq!( + ledger.conclusion(Workspace::Trusted, text), + found("c-2", &second) + ); + ledger.append(sent(&undo1, &[])).unwrap(); + assert_eq!( + ledger.conclusion(Workspace::Trusted, text), + found("c-1", &first) + ); + ledger + .append(decided(&undo2, unconclude(Workspace::Trusted, text))) + .unwrap(); + ledger.append(sent(&undo2, &[])).unwrap(); + assert_eq!(ledger.conclusion(Workspace::Trusted, text), None); + // The latest is the one sent last, not the one decided last. + let (early, late) = (edit("2026-09-28-0005", 0), edit("2026-09-28-0006", 0)); + for item in [&early, &late] { + ledger + .append(decided(item, conclude(Workspace::Trusted, "y"))) + .unwrap(); + } + ledger.append(sent(&late, &["c-late"])).unwrap(); + ledger.append(sent(&early, &["c-early"])).unwrap(); + assert_eq!( + ledger.conclusion(Workspace::Trusted, "y"), + found("c-early", &early) + ); +} + +#[test] +fn a_conclusion_is_found_only_by_its_workspace_and_text_once_sent() { + let home = Home::new("ledger-where"); + let mut ledger = home.open(); + let (sent_one, unsent, empty, other, undo_elsewhere, undo_unsent) = ( + edit("2026-09-28-0001", 0), + edit("2026-09-28-0002", 0), + edit("2026-09-28-0003", 0), + edit("2026-09-28-0004", 0), + edit("2026-09-28-0005", 0), + edit("2026-09-28-0006", 0), + ); + let actions = [ + (&sent_one, conclude(Workspace::Trusted, "a")), + (&unsent, conclude(Workspace::Trusted, "b")), + (&empty, conclude(Workspace::Trusted, "c")), + (&other, conclude(Workspace::Untrusted, "a")), + (&undo_elsewhere, unconclude(Workspace::Untrusted, "a")), + (&undo_unsent, unconclude(Workspace::Trusted, "a")), + ]; + for (item, action) in actions { + ledger.append(decided(item, action)).unwrap(); + } + ledger.append(sent(&sent_one, &["c-a"])).unwrap(); + ledger.append(sent(&empty, &[])).unwrap(); + ledger.append(sent(&other, &["c-other"])).unwrap(); + ledger.append(sent(&undo_elsewhere, &[])).unwrap(); + let a = Some((sent_one.clone(), "c-a".to_string())); + assert_eq!(ledger.conclusion(Workspace::Trusted, "a"), a); + assert_eq!(ledger.conclusion(Workspace::Trusted, "b"), None); + assert_eq!(ledger.conclusion(Workspace::Trusted, "c"), None); + assert_eq!(ledger.conclusion(Workspace::Trusted, "A"), None); + assert_eq!(ledger.conclusion(Workspace::Untrusted, "a"), None); + assert_eq!(ledger.conclusion(Workspace::Fleet, "a"), None); + ledger.append(sent(&undo_unsent, &[])).unwrap(); + assert_eq!(ledger.conclusion(Workspace::Trusted, "a"), None); +} + +#[test] +fn the_counts_add_up() { + let home = Home::new("ledger-counts"); + let mut ledger = home.open(); + assert_eq!(ledger.counts(), Counts::default()); + let items: Vec = (1..=9).map(|n| turn("chat-1", n)).collect(); + for (n, item) in items.iter().enumerate() { + let action = if n < 2 { + skip("secret") + } else { + to_trusted("kyle") + }; + ledger.append(decided(item, action)).unwrap(); + } + for item in &items[2..5] { + ledger.append(sent(item, &["m"])).unwrap(); + } + ledger.append(failed(&items[5], Some(400))).unwrap(); + let counts = ledger.counts(); + assert_eq!( + counts, + Counts { + decided: 9, + sent: 3, + waiting: 3, + failed: 1, + skipped: 2, + } + ); + assert_eq!( + counts.decided, + counts.sent + counts.waiting + counts.failed + counts.skipped + ); + assert_eq!(ledger.outbox().len(), counts.waiting); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger_file.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger_file.rs new file mode 100644 index 0000000..3c80e55 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_ledger_file.rs @@ -0,0 +1,303 @@ +//! The Honcho ledger's file (M5c spec, section 5): read through at start, a line cut off by a +//! crash dropped and cut from the file, a damaged line refused with its number and the runbook +//! pointer, the file's mode, and a failed write that changes nothing. Do not edit. + +#[path = "support/ledger.rs"] +mod ledger; +#[path = "support/tmp.rs"] +mod tmp; + +use std::os::unix::fs::PermissionsExt; + +use ledger::{ + Home, conclude, decided, edit, failed, line, sent, skip, to_trusted, turn, unconclude, +}; +use memoryd::honcho::ledger::{Counts, Item, Ledger, LedgerError, Workspace}; + +const POINTER: &str = "; see docs/runbook.md#honcho-ledger-damaged"; + +/// `open`'s error for the file of `home`, which must be `Damaged`: its line and its display. +fn damaged(home: &Home) -> (usize, String) { + let err = Ledger::open(&home.path).unwrap_err(); + let shown = err.to_string(); + match err { + LedgerError::Damaged { path, line, .. } => { + assert_eq!(path, home.path); + (line, shown) + } + other => panic!("not Damaged: {other:?}"), + } +} + +#[test] +fn a_missing_file_is_an_empty_ledger_and_nothing_is_created() { + let home = Home::new("ledger-missing"); + let ledger = home.open(); + assert_eq!(ledger.counts(), Counts::default()); + assert!(ledger.outbox().is_empty()); + assert!(!ledger.repaired()); + assert!(!home.path.exists()); + let nowhere = home.dir.path().join("memory").join("honcho.jsonl"); + let ledger = Ledger::open(&nowhere).unwrap(); + assert_eq!(ledger.counts(), Counts::default()); + assert!(!nowhere.parent().unwrap().exists()); +} + +#[test] +fn an_empty_file_is_an_empty_ledger() { + let home = Home::new("ledger-empty"); + home.write(""); + let ledger = home.open(); + assert_eq!(ledger.counts(), Counts::default()); + assert!(!ledger.repaired()); + assert_eq!(home.text(), ""); +} + +#[test] +fn reopening_reads_the_ledger_through() { + let home = Home::new("ledger-reopen"); + let mut ledger = home.open(); + let (a, b, c, d, e) = ( + turn("chat-1", 1), + turn("chat-1", 2), + edit("2026-09-28-0001", 0), + edit("2026-09-28-0002", 0), + turn("chat-2", 1), + ); + ledger.append(decided(&b, to_trusted("kyle"))).unwrap(); + ledger.append(decided(&a, skip("no speaker"))).unwrap(); + ledger + .append(decided(&c, conclude(Workspace::Trusted, "x"))) + .unwrap(); + ledger + .append(decided(&d, unconclude(Workspace::Trusted, "x"))) + .unwrap(); + ledger.append(decided(&e, to_trusted("ana_b"))).unwrap(); + ledger.append(sent(&c, &["c-1"])).unwrap(); + ledger.append(failed(&d, Some(404))).unwrap(); + let again = home.open(); + assert!(!again.repaired()); + assert_eq!(again.outbox(), ledger.outbox()); + assert_eq!(again.counts(), ledger.counts()); + assert_eq!(again.sent_ids(&c), Some(&["c-1".to_string()][..])); + assert_eq!( + again.conclusion(Workspace::Trusted, "x"), + Some((c.clone(), "c-1".to_string())) + ); + for item in [&a, &b, &c, &d, &e] { + assert!(again.is_decided(item), "{item:?}"); + } + let mut again = again; + let err = again.append(decided(&b, skip("again"))).unwrap_err(); + assert!(matches!(err, LedgerError::AlreadyDecided(_)), "{err:?}"); + let err = again.append(sent(&c, &["c-2"])).unwrap_err(); + assert!(matches!(err, LedgerError::NotWaiting(_)), "{err:?}"); + again.append(sent(&b, &["m-1", "m-2"])).unwrap(); + let third = home.open(); + assert_eq!(third.outbox(), vec![(e, to_trusted("ana_b"))]); + assert_eq!( + third.counts(), + Counts { + decided: 5, + sent: 2, + waiting: 1, + failed: 1, + skipped: 1, + } + ); +} + +#[test] +fn a_cut_off_last_line_is_dropped_and_cut_from_the_file() { + let (a, b, c) = (turn("chat-1", 1), turn("chat-1", 2), turn("chat-1", 3)); + let whole = format!( + "{}{}", + line(&decided(&a, to_trusted("kyle"))), + line(&decided(&b, to_trusted("kyle"))) + ); + let third = line(&sent(&a, &["m-1"])); + // Cut in the middle, and one byte short of the newline: a complete record without its + // newline is still a cut-off line. + for cut in [third.len() / 2, third.len() - 1] { + let home = Home::new("ledger-torn"); + home.write(&format!("{whole}{}", &third[..cut])); + let mut ledger = home.open(); + assert!(ledger.repaired()); + assert_eq!(home.text(), whole); + assert_eq!(ledger.outbox().len(), 2, "the cut-off send counted"); + assert_eq!(ledger.sent_ids(&a), None); + ledger.append(decided(&c, to_trusted("kyle"))).unwrap(); + ledger.append(sent(&a, &["m-2"])).unwrap(); + let again = home.open(); + assert!(!again.repaired()); + assert_eq!(again.counts().decided, 3); + assert_eq!(again.sent_ids(&a), Some(&["m-2".to_string()][..])); + } +} + +#[test] +fn a_cut_off_first_line_leaves_an_empty_file() { + let home = Home::new("ledger-torn-first"); + let first = line(&decided(&turn("chat-1", 1), to_trusted("kyle"))); + home.write(&first[..10]); + let ledger = home.open(); + assert!(ledger.repaired()); + assert_eq!(ledger.counts(), Counts::default()); + assert_eq!(std::fs::metadata(&home.path).unwrap().len(), 0); +} + +#[test] +fn a_line_cut_inside_a_character_is_dropped_too() { + let home = Home::new("ledger-torn-utf8"); + let a = edit("2026-09-28-0001", 0); + let first = line(&decided(&turn("chat-1", 1), to_trusted("kyle"))); + let second = line(&decided(&a, conclude(Workspace::Trusted, "Kyle's café"))); + let at = second.find('é').unwrap() + 1; + let mut bytes = first.clone().into_bytes(); + bytes.extend_from_slice(&second.as_bytes()[..at]); + std::fs::write(&home.path, &bytes).unwrap(); + let ledger = home.open(); + assert!(ledger.repaired()); + assert_eq!(home.text(), first); + assert!(!ledger.is_decided(&a)); +} + +#[test] +fn a_damaged_line_stops_the_ledger_with_its_number_and_the_pointer() { + let first = line(&decided(&turn("chat-1", 1), to_trusted("kyle"))); + let third = line(&decided(&turn("chat-1", 2), to_trusted("kyle"))); + let extra = first.replacen("\"speaker\"", "\"extra\":1,\"speaker\"", 1); + for (second, why) in [ + ("{\"type\":\"decided\"\n".to_string(), "EOF while parsing"), + ("\n".to_string(), "EOF while parsing"), + ("not json\n".to_string(), "expected"), + (extra, "unknown field `extra`"), + ] { + for tail in ["", "{\"type\":\"se"] { + let home = Home::new("ledger-damaged"); + let text = format!("{first}{second}{third}{tail}"); + home.write(&text); + let (number, shown) = damaged(&home); + assert_eq!(number, 2, "{second}"); + let start = format!("{}:2: ", home.path.display()); + assert!(shown.starts_with(&start), "{shown}"); + assert!(shown.contains(why), "{shown}"); + assert!(shown.ends_with(POINTER), "{shown}"); + assert_eq!(home.text(), text, "a damaged ledger was changed"); + } + } +} + +#[test] +fn a_line_that_breaks_the_rules_is_damage() { + let (a, b) = (turn("chat-1", 1), edit("2026-09-28-0001", 2)); + let cases = [ + ( + vec![sent(&a, &["m"])], + 1, + "turn 1 of session chat-1: not waiting to be sent", + ), + ( + vec![decided(&a, to_trusted("kyle")), decided(&a, skip("again"))], + 2, + "turn 1 of session chat-1: decided already", + ), + ( + vec![decided(&b, skip("flagged")), failed(&b, Some(400))], + 2, + "part 2 of edit 2026-09-28-0001: not waiting to be sent", + ), + ( + vec![ + decided(&a, to_trusted("kyle")), + sent(&a, &["m"]), + decided(&b, to_trusted("kyle")), + failed(&a, None), + ], + 4, + "turn 1 of session chat-1: not waiting to be sent", + ), + ]; + for (records, number, why) in cases { + let home = Home::new("ledger-rules"); + let text: String = records.iter().map(line).collect(); + home.write(&text); + let (got, shown) = damaged(&home); + assert_eq!(got, number, "{text}"); + let expected = format!("{}:{number}: {why}{POINTER}", home.path.display()); + assert_eq!(shown, expected); + assert_eq!(home.text(), text); + } +} + +#[test] +fn the_file_is_made_0600() { + let home = Home::new("ledger-mode"); + let mut ledger = home.open(); + ledger + .append(decided(&turn("chat-1", 1), to_trusted("kyle"))) + .unwrap(); + let mode = std::fs::metadata(&home.path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600, "{mode:o}"); +} + +#[test] +fn a_missing_directory_is_an_io_error_and_changes_nothing() { + let home = Home::new("ledger-nodir"); + let path = home.dir.path().join("memory").join("honcho.jsonl"); + let mut ledger = Ledger::open(&path).unwrap(); + let item = turn("chat-1", 1); + let err = ledger + .append(decided(&item, to_trusted("kyle"))) + .unwrap_err(); + assert!( + matches!(&err, LedgerError::Io(p, _) if *p == path), + "{err:?}" + ); + assert!( + err.to_string() + .starts_with(&format!("{}: ", path.display())) + ); + assert!(!ledger.is_decided(&item)); + assert_eq!(ledger.counts(), Counts::default()); + assert!(!path.parent().unwrap().exists(), "the directory was made"); +} + +#[test] +fn a_failed_write_changes_nothing_in_memory() { + let home = Home::new("ledger-failed-write"); + let mut ledger = home.open(); + let (a, b) = (turn("chat-1", 1), turn("chat-1", 2)); + ledger.append(decided(&a, to_trusted("kyle"))).unwrap(); + let kept = home.text(); + // The ledger's path turns into a directory: every write to it fails. + std::fs::remove_file(&home.path).unwrap(); + std::fs::create_dir(&home.path).unwrap(); + let records = [decided(&b, to_trusted("kyle")), sent(&a, &["m-1"])]; + for record in records.clone() { + let err = ledger.append(record).unwrap_err(); + assert!(matches!(err, LedgerError::Io(..)), "{err:?}"); + } + assert!(!ledger.is_decided(&b)); + assert_eq!(ledger.sent_ids(&a), None); + assert_eq!(ledger.outbox(), vec![(a.clone(), to_trusted("kyle"))]); + std::fs::remove_dir(&home.path).unwrap(); + home.write(&kept); + for record in records { + ledger.append(record).unwrap(); + } + let outbox: Vec = ledger.outbox().into_iter().map(|(i, _)| i).collect(); + assert_eq!(outbox, vec![b]); + assert_eq!(home.open().counts().sent, 1); +} + +#[test] +fn a_ledger_that_cannot_be_read_is_an_io_error() { + let home = Home::new("ledger-unreadable"); + std::fs::create_dir(&home.path).unwrap(); + let err = Ledger::open(&home.path).unwrap_err(); + assert!( + matches!(&err, LedgerError::Io(p, _) if *p == home.path), + "{err:?}" + ); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/support/ledger.rs b/docs/plans/M5c1/files/crates/memoryd/tests/support/ledger.rs new file mode 100644 index 0000000..7ede28d --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/ledger.rs @@ -0,0 +1,123 @@ +//! Records for the Honcho ledger's tests, and a ledger in a directory of its own. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::path::PathBuf; + +use bxcore::{SessionId, Timestamp}; +use memoryd::honcho::ledger::{Action, Item, Ledger, Record, Workspace}; + +use crate::tmp::TempDir; + +/// A ledger file that does not exist yet, in a directory removed on drop. +pub struct Home { + pub path: PathBuf, + pub dir: TempDir, +} + +impl Home { + pub fn new(tag: &str) -> Home { + let dir = TempDir::new(tag); + Home { + path: dir.path().join("honcho.jsonl"), + dir, + } + } + + pub fn open(&self) -> Ledger { + Ledger::open(&self.path).unwrap() + } + + /// The file's bytes as text; "" when there is no file. + pub fn text(&self) -> String { + std::fs::read_to_string(&self.path).unwrap_or_default() + } + + /// Writes `text` as the whole file. + pub fn write(&self, text: &str) { + std::fs::write(&self.path, text).unwrap(); + } +} + +pub const TIME: &str = "2026-09-28T10:00:00.000Z"; + +pub fn time() -> Timestamp { + Timestamp::parse(TIME).unwrap() +} + +pub fn sid(s: &str) -> SessionId { + SessionId::new(s).unwrap() +} + +pub fn turn(session: &str, turn: u64) -> Item { + Item::Turn { + session: sid(session), + turn, + } +} + +pub fn edit(id: &str, part: u32) -> Item { + Item::Edit { + id: id.to_string(), + part, + } +} + +pub fn to_trusted(speaker: &str) -> Action { + Action::Turn { + workspace: Workspace::Trusted, + speaker: speaker.to_string(), + } +} + +pub fn skip(why: &str) -> Action { + Action::Skip { + why: why.to_string(), + } +} + +pub fn conclude(workspace: Workspace, text: &str) -> Action { + Action::Conclude { + workspace, + peer: "kyle".to_string(), + session: sid("chat-1"), + text: text.to_string(), + } +} + +pub fn unconclude(workspace: Workspace, text: &str) -> Action { + Action::Unconclude { + workspace, + text: text.to_string(), + } +} + +pub fn decided(item: &Item, action: Action) -> Record { + Record::Decided { + time: time(), + item: item.clone(), + action, + } +} + +pub fn sent(item: &Item, ids: &[&str]) -> Record { + Record::Sent { + time: time(), + item: item.clone(), + ids: ids.iter().map(|s| s.to_string()).collect(), + } +} + +pub fn failed(item: &Item, status: Option) -> Record { + Record::Failed { + time: time(), + item: item.clone(), + status, + kind: "refused".to_string(), + } +} + +/// `record`'s line, as `append` writes it. +pub fn line(record: &Record) -> String { + format!("{}\n", serde_json::to_string(record).unwrap()) +} diff --git a/docs/plans/M5c1/files/runbook-honcho-ledger-damaged.md b/docs/plans/M5c1/files/runbook-honcho-ledger-damaged.md new file mode 100644 index 0000000..387db1c --- /dev/null +++ b/docs/plans/M5c1/files/runbook-honcho-ledger-damaged.md @@ -0,0 +1,37 @@ +## honcho-ledger-damaged + +**What you see.** `memoryd` logs a line naming `memory/honcho.jsonl`, a line number and a reason, +then this entry: + +```text +/srv/bx/memory/honcho.jsonl:12: turn 3 of session chat-1: decided already; see docs/runbook.md#honcho-ledger-damaged +``` + +**Why.** The ledger is `memoryd`'s record of what it sends to Honcho: one line when a turn or a +memory write is decided (where it goes, or that it goes nowhere), and one when it is sent or +refused (M5c spec, section 5). What is decided and not yet sent is the outbox. `memoryd` reads the +ledger through at start so that it sends nothing twice and nothing to a workspace it was not +decided for. A line it cannot read, or one that breaks the ledger's rules, could do either, so it +does not guess: it does not start its Honcho sync, and nothing is decided or sent to Honcho until +the ledger reads through. The rules: an item is decided once (`decided already`), and it is sent +or refused once, only after it is decided and only when it was not decided to go nowhere (`not +waiting to be sent`). A last line cut off by a crash is not damage: `memoryd` drops it and says so +in its log. + +**Confirm.** Look at the line named, and at every line of the same item: + +```sh +f="$BOXMAKER_HOME/memory/honcho.jsonl" +sed -n 'p' "$f" +grep -n '"session":"","turn":}' "$f" # a turn; for an edit: '"id":"","part":}' +``` + +**Fix.** Keep a copy first: `cp -a "$f" "$f.damaged-$(date +%s)"`. The ledger is written only by +`memoryd`, so a damaged line was changed by hand or by the disk. If you know what the line should +say, correct it. Otherwise remove it (`sed -i 'd' "$f"`), knowing what that costs: a removed +`sent` or `failed` line means the item is sent again (a turn's messages appear twice in Honcho, a +conclusion is made twice); a removed `decided` line means the item is decided again, and any +later `sent` or `failed` line of the same item then breaks the rules, so remove those too. Honcho +cannot delete a message, so a turn sent twice stays twice. + +**Check.** `memoryd` starts, and its log has no line with this entry.