diff --git a/docs/plans/M5c1/09-memoryd-honcho-turns.md b/docs/plans/M5c1/09-memoryd-honcho-turns.md new file mode 100644 index 0000000..4c52ed3 --- /dev/null +++ b/docs/plans/M5c1/09-memoryd-honcho-turns.md @@ -0,0 +1,649 @@ +# M5c-1 task 09: `memoryd` decides a reported turn, after `loopd` has its answer + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `memoryd: a reported turn decided after loopd has its answer, from brokerd and the log` +**Log row:** the Task cell is `M5c1/09-memoryd-honcho-turns`. + +## Goal + +When a turn ends, `loopd` sends `memoryd` a `SessionUpdated { session, turn }` (task 02). From +this task on, with a `[honcho]` section, `memoryd` then decides where that turn goes in Honcho +(task 08's `decide`) and appends the decision to the ledger (task 07) at once (M5c spec, section +5). Nothing is sent to Honcho yet: that is task 11. + +- **Answer first.** `loopd` waits for `memoryd`'s answer to `SessionUpdated` before it sends the + turn's reply to the person, and a question to `brokerd` may take seconds. So `memoryd` answers + `loopd` exactly as today, closes the connection, and only then, on the same connection's + thread, decides. The Honcho state has a lock of its own, never the index lock, so a slow + `brokerd` holds up neither `loopd` nor a memory call. +- **Every waiting turn.** `memoryd` keeps, in memory, each session with undecided turns and the + highest turn `loopd` reported for it. At each report it walks every such session, in the order + of their ids: it reads the session's log, and decides every undecided turn from 1 to the + smaller of the reported turn and the log's turns. So a turn left undecided while `brokerd` was + down is decided at the next report of **any** session (spec 5, step 3). +- **The log's last turn waits for its reply.** A log's last turn without a final answer may still + be running: it is not decided (and never skipped) until it has one. This only matters when a + report names a turn past the log's end; `loopd` never sends one, but the number comes over a + socket. +- **One question to `brokerd` per session per walk**, on its `state.sock` (task 05), and none for + a session with nothing undecided. The answer is the session's class and trust; an error frame, + no listener or 5 s without a byte means no answer. +- **`brokerd` does not answer:** nothing is decided, the walk stops there (the next sessions + would fail the same way), and the outage is logged once, with the new runbook entry + `broker-state-unavailable`. The first answer after it is logged once too. +- **A log that does not read** (a damaged complete line): nothing is decided for that session, + logged once with `session-log-damaged`; the walk goes on with the next session. A last line + without its newline is one `loopd` is still writing, and is left out, silently. +- **The ledger at start.** `memoryd` reads `/memory/honcho.jsonl` through before anything + else in `serve::start`. A damaged or unreadable ledger stops `memoryd`: exit 1, one line with + the pointer `honcho-ledger-damaged`, no socket bound (as task 06 does for + `honcho-config-invalid`). A cut-off last line is dropped by the ledger (task 07) and logged + once. A ledger that cannot be **written** while `memoryd` runs is logged with the same pointer; + the walk stops, and the turn is decided at a later report. +- **Without `[honcho]`** nothing changes: no ledger, no question, no line. Every existing test + passes unchanged. + +Epochs are out of scope: `loopd` writes only `sessions//0.jsonl` and numbers turns in it, and +so does this task. (A later milestone that adds compaction must settle how turns are numbered +across epochs.) + +## Interfaces + +In `crates/memoryd/src/honcho/state.rs` (the module `memoryd::honcho::state`): + +```rust +pub const WAIT: Duration = Duration::from_secs(5); +pub const RUNBOOK: &str = "see docs/runbook.md#broker-state-unavailable"; +pub fn ask(socket: &Path, session: &SessionId, wait: Duration) -> Result; +``` + +In `crates/memoryd/src/honcho/turns.rs` (the module `memoryd::honcho::turns`): + +```rust +pub const LEDGER: &str = "honcho.jsonl"; +pub const LOG_RUNBOOK: &str = "see docs/runbook.md#session-log-damaged"; +pub type Ask<'a> = &'a dyn Fn(&SessionId) -> Result; +pub fn log_path(sessions: &Path, session: &SessionId) -> PathBuf; +pub fn read_log(path: &Path) -> Result, String>; +pub fn finished(turn: &[LogRecord]) -> bool; +pub fn decidable(records: &[LogRecord], reported: u64) -> (u64, bool); +pub struct TurnSync { /* private */ } +impl TurnSync { + pub fn new(ledger: Ledger, agent: &str, sessions: &Path) -> TurnSync; + pub fn ledger(&self) -> &Ledger; + pub fn waiting(&self) -> Vec<(SessionId, u64)>; + pub fn updated(&mut self, session: &SessionId, turn: u64, ask: Ask<'_>, now: Timestamp, + log: &dyn Fn(&str)); +} +``` + +In `crates/memoryd/src/serve.rs`: `pub enum StartError { Failed(String), Ledger(LedgerError) }` +(Debug, Display, Error, `From`), and `start` returns `Result`. +Task 11 uses `TurnSync`, `decidable` and `waiting` when it rebuilds what a restart forgets. + +## The lines `memoryd` logs + +| When | The line | +|---|---| +| the ledger's last line was cut off (at start) | `memoryd: /memory/honcho.jsonl: a last line cut off by a crash was dropped` | +| `brokerd` gives no answer, the first time | `memoryd: ; turns wait undecided; see docs/runbook.md#broker-state-unavailable` | +| `brokerd` answers again | `memoryd: brokerd answers on its state socket again; turns are decided` | +| a session's log does not read, the first time | `memoryd: :: ; see docs/runbook.md#session-log-damaged` | +| the ledger cannot be written | `memoryd: : ; see docs/runbook.md#honcho-ledger-damaged` | + +`` for `brokerd` is `ask`'s error: `: ` (connect, timeout, frame) or +`brokerd: ` for an error frame. At start, a damaged ledger is `memoryd: :: +; see docs/runbook.md#honcho-ledger-damaged` on stderr, and an unreadable one `memoryd: +: ; see docs/runbook.md#honcho-ledger-damaged`: one line each, then exit 1. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/memoryd/src/honcho/mod.rs` | `pub mod state;` and `pub mod turns;` (edit H1) | +| `crates/memoryd/src/honcho/state.rs` | new: a skeleton you copy, then fill 1 body | +| `crates/memoryd/src/honcho/turns.rs` | new: a skeleton you copy, then fill 6 bodies | +| `crates/memoryd/src/serve.rs` | module comment, imports, `Memoryd.honcho`, `StartError`, `start`, `connection`, `decide_turns` (edits S1 to S7) | +| `crates/memoryd/src/main.rs` | a ledger error printed with its own pointer (edit M1) | +| `docs/runbook.md` | `session-log-damaged` and `honcho-ledger-damaged` changed (edits R1 to R4); `broker-state-unavailable` appended | +| everything else (`decide.rs`, `ledger.rs`, `config.rs`, other crates) | **no change** | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - new tests: `crates/memoryd/tests/honcho_state.rs`, `crates/memoryd/tests/honcho_turns.rs`, + `crates/memoryd/tests/honcho_serve.rs` + - new test support: `crates/memoryd/tests/support/brokerd.rs` (a fake `brokerd` state + socket), `crates/memoryd/tests/support/turns.rs` (logs of turns with speakers) + - the skeletons `crates/memoryd/src/honcho/state.rs` and `crates/memoryd/src/honcho/turns.rs` +- Append, from `docs/plans/M5c1/files/`, to the end of `docs/runbook.md`, with one blank line + before it: `runbook-broker-state-unavailable.md` (step 7 gives the command). +- Modify: `crates/memoryd/src/honcho/mod.rs`, `crates/memoryd/src/serve.rs`, + `crates/memoryd/src/main.rs`, `docs/runbook.md`, `docs/implementer-log.md`. +- Do not touch: any other source file, any other test or support file (`support/logs.rs` and + `support/tmp.rs` in particular, which the new tests use), any file of another crate, any + `Cargo.toml`, and in the skeletons anything but the 7 `todo!()` bodies. + +## Facts you need + +- **The socket.** `brokerd` serves `state.sock` (task 05): one frame in, one final frame out, the + connection then closed. The question is `Envelope { v: PROTOCOL_VERSION, id: 1, r#final: true, + msg: Message::SessionState(SessionState { session }) }`; the answer + `Message::SessionStateReply(SessionStateReply { class, untrusted })`, or `Message::Error` (for + example a damaged state file, `broker-state-damaged` in `brokerd`'s own log). Frames are + `bxcore::write_frame` and `read_frame`, as on every socket. `Config::state_socket()` (task 06) + is the path. +- **Timeouts.** The read and the write timeouts are 5 s (`WAIT`): "no byte for 5 s", as `loopd`'s + `MemoryPort` does it with `set_read_timeout` and `set_write_timeout`. There is no connect + timeout: `std::os::unix::net::UnixStream` has none, and a connect fails at once when nothing + listens. A read that times out is `frame I/O error: Resource temporarily unavailable (os error + 11)` on Linux. +- **The log.** `sessions//0.jsonl` under `[paths] home` (`Config::sessions_dir()`), one + `LogRecord` per line. `loopd` writes a record and its newline in two writes, so `memoryd` can + read a last line without its newline: it is left out. A turn is finished when its last + `Assistant` record asks for no tool (`finished`); it need not have text (task 08 skips an empty + reply as `the turn has no reply`). +- **The walk** is `TurnSync::updated`, given whole with `walk`: it calls your `records`, + `decidable`, `state` and task 08's `decide`. Read both before you fill the bodies. A session + leaves the map when every turn up to its reported one (as far as its log goes) is decided; it + stays while a turn waits (for `brokerd`, a readable log, or a reply). +- **The ledger's time** is the walk's `now` (`Timestamp::now()` in `serve.rs`), so a decision + made late carries the time it was made. +- **What a restart forgets.** The map lives in memory. After a restart a waiting session is + walked again only at its next report. Rebuilding the map at start (from the ledger and the + logs, with `decidable` so that a running turn is never decided) is task 11's. +- **The given tests.** + - `honcho_state.rs` (5 tests): the question as `brokerd` reads it and its answer; an error + frame; no listener (an error at once); silence (an error after the wait, here 300 ms). + - `honcho_turns.rs` (16 tests): `TurnSync` with `brokerd` as a closure: every undecided turn up + to the report and the log; the running last turn; one question per session, none when + nothing waits; the outage and the return logged once; the walk stopped at the first failure; + an earlier report never lowering the turn; a damaged log and a line being written; a ledger + that cannot be written; `read_log`, `finished`, `decidable`. + - `honcho_serve.rs` (6 tests): `serve::start` on a thread with a fake `brokerd`: a turn decided + after `loopd` has its answer (and a report on the wrong socket deciding nothing); a silent + `brokerd` holding up neither a second report nor a memory call (each answered within 1 s), + then logged once after 5 s; nothing asked or written without `[honcho]`; a damaged or + unreadable ledger stopping `start` before any socket is bound; a cut-off last line logged; + and the `memoryd` process exiting with exactly the ledger's line. It takes about 6 s. +- The calls the bodies use: + + | Call | Signature, and what it gives | + |---|---| + | `UnixStream::connect` | `UnixStream::connect(socket) -> io::Result` | + | `set_read_timeout`, `set_write_timeout` | `stream.set_read_timeout(Some(wait)) -> io::Result<()>` | + | `write_frame`, `read_frame` | `write_frame(&mut stream, &env) -> Result<(), FrameError>`; `read_frame(&mut stream) -> Result` | + | `std::fs::read` | `std::fs::read(path) -> io::Result>` | + | `Iterator::rposition` | `text.iter().rposition(\|b\| *b == b'\n') -> Option` | + | `slice::split_inclusive` | `bytes.split_inclusive(\|b\| *b == b'\n')`: each line with its newline, no empty last piece | + | `serde_json::from_slice` | `serde_json::from_slice::(line) -> serde_json::Result` | + | `Path::join` | `sessions.join(session.as_str()).join("0.jsonl")` | + | `BTreeSet::insert`, `remove` | `self.unreadable.insert(s) -> bool` (true when it was not there); `remove(s: &str)` | + | `turns`, `turn_records` | task 08's: `turns(records) -> u64`; `turn_records(records, turn) -> Option<&[LogRecord]>` | + | `Option::is_some_and` | `turn_records(records, limit).is_some_and(finished)` | + | `matches!` with a guard | `matches!(last, Some(LogRecord::Assistant { tool_calls, .. }) if tool_calls.is_empty())` | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/memoryd/src/honcho/mod.rs` + +**H1.** The line + +```rust +pub mod ledger; +``` + +becomes + +```rust +pub mod ledger; +pub mod state; +pub mod turns; +``` + +### `crates/memoryd/src/serve.rs` + +**S1.** The lines + +```rust +//! Serving: three sockets, one request per connection, each answered by one frame (M5a spec, +//! section 7). One lock covers the files and the index, so writes never interleave. +``` + +become + +```rust +//! Serving: three sockets, one request per connection, each answered by one frame (M5a spec, +//! section 7). One lock covers the files and the index, so writes never interleave. With +//! `[honcho]`, a reported turn is decided after `loopd` has its answer, under a lock of its own +//! (M5c spec, section 5). +``` + +**S2.** The lines + +```rust + MemoryCall, MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, Timestamp, + ToolResponse, WireError, read_frame, write_frame, +}; +``` + +become + +```rust + MemoryCall, MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, SessionId, + SessionUpdated, Timestamp, ToolResponse, WireError, read_frame, write_frame, +}; +``` + +**S3.** The line + +```rust +use crate::files::Store; +``` + +becomes + +```rust +use crate::files::Store; +use crate::honcho::ledger::{self, Ledger, LedgerError}; +use crate::honcho::state; +use crate::honcho::turns::{self, TurnSync}; +``` + +**S4.** In `pub struct Memoryd`, the lines + +```rust + state: Mutex, + log: Log, +} +``` + +become + +```rust + state: Mutex, + /// With `[honcho]`: the turns reported and not yet decided, and the ledger. Never taken + /// while `state` is held, and never held by a request that waits for its answer. + honcho: Option>, + log: Log, +} + +/// Why `start` failed. `main` prints a start failure with the pointer `memoryd-start-failed`, and +/// a ledger that does not read through with its own. +#[derive(Debug)] +pub enum StartError { + Failed(String), + Ledger(LedgerError), +} + +impl std::fmt::Display for StartError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + StartError::Failed(why) => write!(f, "{why}"), + // `Damaged` names its pointer itself. + StartError::Ledger(e @ LedgerError::Damaged { .. }) => write!(f, "{e}"), + StartError::Ledger(e) => write!(f, "{e}; {}", ledger::RUNBOOK), + } + } +} + +impl std::error::Error for StartError {} + +impl From for StartError { + fn from(why: String) -> StartError { + StartError::Failed(why) + } +} +``` + +**S5.** The lines + +```rust +/// Create what is missing under `memory/`, open the index and catch it up, then bind the three +/// sockets. The error names what failed; the caller points at `memoryd-start-failed`. +pub fn start(config: Config, log: Log) -> Result { + let store = Store::new(&config.memory_dir()); +``` + +become + +```rust +/// With `[honcho]`, read the ledger through first. Then create what is missing under `memory/`, +/// open the index and catch it up, and bind the three sockets. The error names what failed. +pub fn start(config: Config, log: Log) -> Result { + let honcho = match config.honcho_config()? { + Some(honcho) => { + let path = config.memory_dir().join(turns::LEDGER); + let ledger = Ledger::open(&path).map_err(StartError::Ledger)?; + if ledger.repaired() { + (log)( + format!( + "memoryd: {}: a last line cut off by a crash was dropped", + path.display() + ) + .as_str(), + ); + } + Some(Mutex::new(TurnSync::new( + ledger, + &honcho.agent, + &config.sessions_dir(), + ))) + } + None => None, + }; + let store = Store::new(&config.memory_dir()); +``` + +**S6.** In `start`, where `Memoryd` is built, the lines + +```rust + state: Mutex::new(State { index }), + log, +``` + +become + +```rust + state: Mutex::new(State { index }), + honcho, + log, +``` + +**S7.** In `Memoryd::connection`, the lines + +```rust + let reply = self.handle(request.msg, which); + let env = Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: true, + msg: reply, + }; + let _ = write_frame(&mut stream, &env); + } +``` + +become + +```rust + let update = match (which, &request.msg) { + (Which::Loop, Message::SessionUpdated(update)) => Some(update.clone()), + _ => None, + }; + let reply = self.handle(request.msg, which); + let env = Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: true, + msg: reply, + }; + let _ = write_frame(&mut stream, &env); + // `loopd` has its answer before `brokerd` is asked: it sends the turn's reply only after + // this frame, and `brokerd` may take seconds. + drop(stream); + if let Some(update) = update { + self.decide_turns(&update); + } + } + + /// With `[honcho]`: decide the reported turn, and every other that waits (`TurnSync`). + fn decide_turns(&self, update: &SessionUpdated) { + let Some(honcho) = &self.honcho else { + return; + }; + let socket = self.config.state_socket(); + let ask = |session: &SessionId| state::ask(&socket, session, state::WAIT); + let mut sync = honcho.lock().unwrap_or_else(|p| p.into_inner()); + sync.updated( + &update.session, + update.turn, + &ask, + Timestamp::now(), + &*self.log, + ); + } +``` + +### `crates/memoryd/src/main.rs` + +**M1.** The lines + +```rust + Ok(running) => running, + Err(e) => { + eprintln!("memoryd: {e}\n{START_FAILED}"); +``` + +become + +```rust + Ok(running) => running, + // The ledger's line names its own pointer, `honcho-ledger-damaged`. + Err(e @ serve::StartError::Ledger(_)) => { + eprintln!("memoryd: {e}"); + return ExitCode::from(1); + } + Err(e) => { + eprintln!("memoryd: {e}\n{START_FAILED}"); +``` + +### `docs/runbook.md` + +**R1.** In `session-log-damaged`, the lines + +```text +**What you see.** A turn on an existing session fails with `:: ` naming +`sessions//.jsonl`, and this entry. Other sessions work. +``` + +become + +```text +**What you see.** A turn on an existing session fails with `:: ` naming +`sessions//.jsonl`, and this entry. Other sessions work. With `[honcho]`, `memoryd` +logs the same line once, starting `memoryd: `, and that session's turns wait undecided for Honcho +until the log reads again. +``` + +**R2.** In `honcho-ledger-damaged` (the entry task 07 appended), the lines + +````text +**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 +``` +```` + +become + +````text +**What you see.** `memoryd` exits 1 at start, before it binds its sockets, with one line naming +`memory/honcho.jsonl`, a line number and a reason, then this entry: + +```text +memoryd: /srv/bx/memory/honcho.jsonl:12: turn 3 of session chat-1: decided already; see docs/runbook.md#honcho-ledger-damaged +``` + +Or, while it runs, `memoryd: : ; see docs/runbook.md#honcho-ledger-damaged` when the +ledger cannot be written (a full disk, a wrong owner). The turn waits undecided, and is decided at +a later turn's end once the ledger can be written. +```` + +**R3.** In `honcho-ledger-damaged`, the lines + +```text +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. +``` + +become + +```text +does not guess: it does not start. While it is stopped no memory is served either, and `loopd` +cannot start a session (`memory-unavailable`). 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. +``` + +**R4.** In `honcho-ledger-damaged`, the lines + +```text +**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. +``` + +become + +```text +**Fix.** If the ledger cannot be written, free space or correct the owner of `memory/` +(`df -h`, `ls -l "$f"`); nothing else is needed. For a damaged line, 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. +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, five commands: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_state.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_turns.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/support/brokerd.rs crates/memoryd/tests/support/ + cp docs/plans/M5c1/files/crates/memoryd/tests/support/turns.rs crates/memoryd/tests/support/ + ``` + +- [ ] **2. See them fail.** `cargo build --tests --keep-going -p memoryd`, with a shell timeout + of 600000 ms: three test targets do not compile, one `E0432` each: + `honcho_state` (``unresolved import `memoryd::honcho::state` ``), `honcho_turns` + (``unresolved import `memoryd::honcho::turns` ``) and `honcho_serve` (``unresolved import + `memoryd::serve::StartError` ``). Every other target compiles. +- [ ] **3. The modules.** Make edit H1, then copy the two skeletons: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/state.rs crates/memoryd/src/honcho/ + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/turns.rs crates/memoryd/src/honcho/ + ``` + + `cargo check -p memoryd`: no error, and 19 warnings, all from the unfilled bodies: four + `unused import` lines (`std::os::unix::net::UnixStream`; `Envelope`, `Message`, + `PROTOCOL_VERSION`, `SessionState`, `read_frame` and `write_frame`; `crate::honcho::state`; + `turn_records` and `turns`), fourteen `unused variable` (`session` four times, `log` twice, + and `socket`, `wait`, `path`, `sessions`, `records`, `reported`, `turn`, `ask` once each), and + ``fields `sessions`, `broker_down`, and `unreadable` are never read``. They go as you fill the bodies. +- [ ] **4. Fill the seven bodies, one at a time**, each from its comment, and run + `cargo check -p memoryd` after each: `ask` in `state.rs`, then in `turns.rs` `log_path`, + `read_log`, `finished`, `records`, `state`, `decidable`. 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 two files: `TurnSync`, `new`, `waiting`, + `updated` and `walk` are given. After the seventh, `cargo check -p memoryd` prints no warning, + and this command passes all its tests (5 and 16): + + ```sh + cargo test -p memoryd --test honcho_state --test honcho_turns + ``` +- [ ] **5. Serving.** Make edits S1 to S7 in `serve.rs`, then M1 in `main.rs`. + `cargo check -p memoryd`: no error and no warning. +- [ ] **6. The tests.** This command, with a shell timeout of 600000 ms, and every test passes + (6, 16, 5, and the existing `serve` and `honcho_start` tests unchanged): + + ```sh + cargo test -p memoryd --test honcho_serve --test honcho_turns --test honcho_state \ + --test serve --test honcho_start + ``` +- [ ] **7. The runbook.** Make edits R1 to R4 in `docs/runbook.md`, then from the repository + root: + + ```sh + { echo; cat docs/plans/M5c1/files/runbook-broker-state-unavailable.md; } >> docs/runbook.md + ``` + + Then `grep -c '^## broker-state-unavailable$' docs/runbook.md` prints `1`, and + `grep -c 'does not start its Honcho sync' docs/runbook.md` prints `0`. +- [ ] **8. Check.** Three commands: + - `grep -n 'decide_turns\|let _ = write_frame(&mut stream, &env);' crates/memoryd/src/serve.rs` + prints three lines, in this order: the `write_frame` line, the call + `self.decide_turns(&update);`, and `fn decide_turns(`. `loopd` has its answer first. + - `grep -rn 'runbook.md#broker-state-unavailable' crates/memoryd/src` prints exactly one line, + the `RUNBOOK` constant in `honcho/state.rs`. + - `git diff --stat -- crates/bxcore crates/loopd crates/brokerd crates/gatewayd crates/bxctl + Cargo.toml crates/memoryd/Cargo.toml crates/memoryd/src/honcho/decide.rs + crates/memoryd/src/honcho/ledger.rs crates/memoryd/src/config.rs` prints nothing. +- [ ] **9. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1257 tests: 1230 before this task, plus 27 new ones (5 in + `honcho_state.rs`, 16 in `honcho_turns.rs`, 6 in `honcho_serve.rs`). +- [ ] **10. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/09-memoryd-honcho-turns`. `git status --short` lists `honcho/mod.rs`, `serve.rs`, + `main.rs`, `docs/runbook.md` and your log as modified, and `honcho/state.rs`, `honcho/turns.rs`, + the three test files and the two support files 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 1257 tests. +- `git diff HEAD~1 --stat` lists only the files of step 10. + +## Stop and report if + +- A "before" block of H1, S1 to S7, M1 or R1 to R4 is not in its file exactly as shown, or is + there more than once. +- Step 2 or 3 shows a different result: other test targets, errors of another kind, or other + warnings. +- `cargo check` reports an error after a body you filled or an edit you made, and the comment or + the edit does not tell you how to fix it. Quote the error, and stop. +- A comment in a skeleton does not fit the code under it, or disagrees with this task's text or + "The lines `memoryd` logs". Quote both and stop. +- After step 4 or 6 a test fails although each body follows its comment and each edit is as + shown. Do not change a test, a support file or the skeletons' given code: stop, and give the + failing tests' names and their messages. +- The fix seems to need a change anywhere else: `decide.rs`, `ledger.rs`, `config.rs`, another + crate, a new thread, or a lock held while `brokerd` is asked other than `honcho`'s. It does + not; stop and say why you think so. +- A test outside the new 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 1257. + +## What the given tests pin + +Each of these changes to the finished code makes at least one given test fail (42 in all): the +map keeping the lowest or the latest report instead of the highest; walking only the reported +session; keeping a finished session, or dropping one whose last turn waits; going on after a +silent `brokerd`, or after a ledger write failed; ignoring the log's length or the report; deciding +a running last turn; deciding a turn twice; asking with nothing undecided; logging the outage, the +return or a damaged log more than once, or not at all, or without its pointer; deciding a +session whose log does not read; parsing a line being written; line numbers from 0; a missing log +taken as an error; any `Assistant` record, or the first, ending a turn; another epoch's file; +deciding before `loopd` has its answer, or under the index lock, or not at all, or for a report on +another socket; the repaired ledger not logged; a damaged ledger ignored, or read after a socket +is bound; the ledger's error printed without its pointer, with it twice, or with the start +pointer too; no read timeout; an error frame taken for a clean session; the wrong session asked; +another frame id. + +## After this task + +- A restart forgets the sessions that wait. Task 11 rebuilds them at start (from the logs and the + ledger, with `decidable`, so that a running turn is never decided). +- Whether turns from before `[honcho]` was turned on are decided at all is the owner's open + question; if not, a later task adds a cutoff as a new skip row in `decide`. diff --git a/docs/plans/M5c1/README.md b/docs/plans/M5c1/README.md index 2685fa2..ca6e06d 100644 --- a/docs/plans/M5c1/README.md +++ b/docs/plans/M5c1/README.md @@ -43,11 +43,11 @@ the question on a new socket; `memoryd` gains a `honcho` module (configuration, | 02 | `loopd` logs the speaker on the `User` record and sends the turn number in `SessionUpdated` | loopd | | 03 | `gatewayd` sets the speaker: the Mattermost username when every post of the burst is from one person, else none | gatewayd | | 04 | `bxctl chat --speaker NAME`, opt-in, no default | bxctl | -| 05 | `brokerd` serves `run/memory-broker/state.sock` from its state store; runbook `broker-state-unavailable` | brokerd | +| 05 | `brokerd` serves `run/memory-broker/state.sock` from its state store; `brokerd-listener-lost` and `socket-forbidden` name it | brokerd | | 06 | `memoryd`'s `[honcho]` section and its keys, checked at start; runbook `honcho-config-invalid` | memoryd | | 07 | The ledger, `memory/honcho.jsonl`: its records, append, read-through, and the outbox | memoryd | | 08 | The decision for a turn (a pure function: state, labels, speaker, text) | memoryd | -| 09 | On `SessionUpdated`: ask `brokerd`, read the labels, decide, record; undecided turns retried | memoryd | +| 09 | On `SessionUpdated`, after `loopd` has its answer: ask `brokerd`, read the log, decide, record; undecided turns retried at the next report of any session; the ledger read through at start; runbook `broker-state-unavailable` | memoryd | | 10 | `roncho` clients per workspace, and creating workspaces, peers and sessions (names, the agent unobserved, the fleet with reasoning, summaries and dreaming off, the `-2` sessions); the fake Honcho server | memoryd | | 11 | The sync worker for turns: messages in parts of 25,000, neutralised; backoff; outage logged once; refused sends and keys; runbooks `honcho-unreachable`, `honcho-send-refused`, `honcho-key-refused` | memoryd | | 12 | Memory writes mirrored: each journalled edit decided by spec 4's table (conclusions, fleet messages, retractions, undo) | memoryd | diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/state.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/state.rs new file mode 100644 index 0000000..bc5f35f --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/state.rs @@ -0,0 +1,36 @@ +//! Asking `brokerd` for a session's state on its `state.sock` (M5c spec, section 7): one question +//! per connection, one final answer, as on every socket between the roles. + +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::time::Duration; + +use bxcore::{ + Envelope, Message, PROTOCOL_VERSION, SessionId, SessionState, SessionStateReply, read_frame, + write_frame, +}; + +/// How long `brokerd` may leave the connection without a byte, sending or answering. +pub const WAIT: Duration = Duration::from_secs(5); + +/// The runbook entry for a `brokerd` that does not answer. +pub const RUNBOOK: &str = "see docs/runbook.md#broker-state-unavailable"; + +/// `brokerd`'s answer for `session`, asked on `socket`. `wait` is the read and the write timeout; +/// the connect has none, and fails at once when nothing listens. Any error, an error frame +/// included, is `Err` with one line that names the socket or `brokerd`. +pub fn ask( + socket: &Path, + session: &SessionId, + wait: Duration, +) -> Result { + // `let line = |e: &dyn std::fmt::Display| format!("{}: {e}", socket.display());`. Then, + // each error through `.map_err(|e| line(&e))?`: `UnixStream::connect(socket)` (`mut stream`), + // `set_read_timeout(Some(wait))`, `set_write_timeout(Some(wait))`, and `write_frame` of an + // `Envelope { v: PROTOCOL_VERSION, id: 1, r#final: true, msg }` with `msg` the + // `Message::SessionState(SessionState { session: session.clone() })` (name it `question`). + // Then `match read_frame(&mut stream).map_err(|e| line(&e))?.msg`: `SessionStateReply(reply)` + // -> `Ok(reply)`; `Error(e)` -> `Err(format!("brokerd: {}", e.detail))`; anything else -> + // `Err("brokerd answered with the wrong message".to_string())`. + todo!() +} diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/turns.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/turns.rs new file mode 100644 index 0000000..23d9e7a --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/turns.rs @@ -0,0 +1,201 @@ +//! Deciding finished turns when `loopd` reports one (M5c spec, section 5). At each report every +//! session with undecided turns is walked: its log is read, `brokerd` is asked once for its state, +//! and each undecided turn is decided and appended to the ledger at once. A turn that cannot be +//! decided now (`brokerd` silent, the log unreadable) waits for the next report of any session. + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; + +use bxcore::{LogRecord, SessionId, SessionStateReply, Timestamp}; + +use crate::honcho::decide::{decide, turn_records, turns}; +use crate::honcho::ledger::{self, Item, Ledger, Record}; +use crate::honcho::state; + +/// The ledger's file name, in `memory/`. +pub const LEDGER: &str = "honcho.jsonl"; + +/// The runbook entry for a session log with a damaged line. +pub const LOG_RUNBOOK: &str = "see docs/runbook.md#session-log-damaged"; + +/// `brokerd`'s answer for a session, or why there is none. +pub type Ask<'a> = &'a dyn Fn(&SessionId) -> Result; + +/// How one session's walk ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Walk { + /// Every turn up to the reported one is decided: the session leaves the map. + Done, + /// Some turn waits: the session stays in the map. + Waiting, + /// `brokerd` did not answer, or the ledger could not be written: the walk stops. + Stop, +} + +/// The turns reported and not yet decided, and the ledger their decisions go to. +pub struct TurnSync { + ledger: Ledger, + agent: String, + sessions: PathBuf, + /// Each session with undecided turns, by its id, and the highest turn `loopd` reported for + /// it. `SessionId` has no order, so the map is keyed by its text. + reported: BTreeMap, + /// Whether the last question to `brokerd` went unanswered: an outage is logged once. + broker_down: bool, + /// Sessions whose log last had a damaged line: logged once each. + unreadable: BTreeSet, +} + +/// The path of `session`'s log under `sessions`: epoch 0, the only one `loopd` writes. +pub fn log_path(sessions: &Path, session: &SessionId) -> PathBuf { + // `sessions.join(session.as_str()).join("0.jsonl")`. + todo!() +} + +/// A session log's complete lines as records. A missing file has none. A last line without its +/// newline is one `loopd` is still writing, and is left out. A complete line that does not parse +/// is an error naming the file and the line, with the runbook pointer. +pub fn read_log(path: &Path) -> Result, String> { + // `std::fs::read(path)` into `text` (bytes): NotFound -> `return Ok(Vec::new())`; another + // error `e` -> `Err(format!("{}: {e}; {LOG_RUNBOOK}", path.display()))`. `whole` is one past + // the last `b'\n'` (`text.iter().rposition(..)`: `Some(last)` -> `last + 1`, `None` -> 0). + // Walk `text.get(..whole).unwrap_or_default().split_inclusive(|b| *b == b'\n').enumerate()`: + // each line through `serde_json::from_slice::`, pushed; an error `e` -> `return + // Err(format!("{}:{}: {e}; {LOG_RUNBOOK}", path.display(), index + 1))`. `Ok(records)`. + todo!() +} + +/// Whether a turn has ended with an answer: its last `Assistant` record asks for no tool. +pub fn finished(turn: &[LogRecord]) -> bool { + // `let last = turn.iter().rev().find(|r| matches!(r, LogRecord::Assistant { .. }));` then + // `matches!(last, Some(LogRecord::Assistant { tool_calls, .. }) if tool_calls.is_empty())`. + todo!() +} + +impl TurnSync { + /// `ledger` read through at start; `agent` is `[honcho] agent`; `sessions` the directory of + /// `loopd`'s session logs. + pub fn new(ledger: Ledger, agent: &str, sessions: &Path) -> TurnSync { + TurnSync { + ledger, + agent: agent.to_string(), + sessions: sessions.to_path_buf(), + reported: BTreeMap::new(), + broker_down: false, + unreadable: BTreeSet::new(), + } + } + + pub fn ledger(&self) -> &Ledger { + &self.ledger + } + + /// The sessions that still have undecided turns, with the highest turn reported for each. + pub fn waiting(&self) -> Vec<(SessionId, u64)> { + self.reported.values().cloned().collect() + } + + /// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk every session + /// with undecided turns, in the order of their ids, until one stops the walk. + pub fn updated( + &mut self, + session: &SessionId, + turn: u64, + ask: Ask<'_>, + now: Timestamp, + log: &dyn Fn(&str), + ) { + let entry = self + .reported + .entry(session.as_str().to_string()) + .or_insert((session.clone(), 0)); + entry.1 = entry.1.max(turn); + for (session, turn) in self.waiting() { + match self.walk(&session, turn, ask, now, log) { + Walk::Done => { + self.reported.remove(session.as_str()); + } + Walk::Waiting => {} + Walk::Stop => return, + } + } + } + + /// Decide `session`'s undecided turns from 1 to `reported`, as far as its log goes. The log's + /// last turn waits until it has an answer: it may still be running. + fn walk( + &mut self, + session: &SessionId, + reported: u64, + ask: Ask<'_>, + now: Timestamp, + log: &dyn Fn(&str), + ) -> Walk { + let Some(records) = self.records(session, log) else { + return Walk::Waiting; + }; + let (limit, held) = decidable(&records, reported); + let item = |turn: u64| Item::Turn { + session: session.clone(), + turn, + }; + let undecided: Vec = (1..=limit) + .filter(|turn| !self.ledger.is_decided(&item(*turn))) + .collect(); + if undecided.is_empty() { + return if held { Walk::Waiting } else { Walk::Done }; + } + let Some(state) = self.state(session, ask, log) else { + return Walk::Stop; + }; + for turn in undecided { + let Ok(action) = decide(&state, &records, turn, &self.agent) else { + continue; + }; + let record = Record::Decided { + time: now, + item: item(turn), + action, + }; + if let Err(e) = self.ledger.append(record) { + log(&format!("memoryd: {e}; {}", ledger::RUNBOOK)); + return Walk::Stop; + } + } + if held { Walk::Waiting } else { Walk::Done } + } + + /// `session`'s log, read. A log that does not read is logged once, until it reads again. + fn records(&mut self, session: &SessionId, log: &dyn Fn(&str)) -> Option> { + // `match read_log(&log_path(&self.sessions, session))`: `Ok(records)` -> + // `self.unreadable.remove(session.as_str());` and `Some(records)`. `Err(e)` -> when + // `self.unreadable.insert(session.as_str().to_string())` is true (not logged yet), + // `log(&format!("memoryd: {e}"))`; then `None`. + todo!() + } + + /// `brokerd`'s answer for `session`. The first failure after an answer is logged, and so is + /// the first answer after a failure. + fn state( + &mut self, + session: &SessionId, + ask: Ask<'_>, + log: &dyn Fn(&str), + ) -> Option { + // `match ask(session)`. `Ok(state)`: if `self.broker_down`, set it `false` and log exactly + // `memoryd: brokerd answers on its state socket again; turns are decided`; `Some(state)`. + // `Err(e)`: if `!self.broker_down`, set it `true` and log + // `format!("memoryd: {e}; turns wait undecided; {}", state::RUNBOOK)`; then `None`. + todo!() + } +} + +/// How far a log's turns may be decided for a report of turn `reported`: up to the smaller of +/// `reported` and the log's turns, but for the log's last turn when it has no answer yet. The +/// limit, and whether that last turn was held back. +pub fn decidable(records: &[LogRecord], reported: u64) -> (u64, bool) { + // `let count = turns(records);` and `let limit = reported.min(count);`. When `limit > 0 && + // limit == count && !turn_records(records, limit).is_some_and(finished)`: `(limit - 1, true)`; + // otherwise `(limit, false)`. + todo!() +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs new file mode 100644 index 0000000..2392f75 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs @@ -0,0 +1,329 @@ +//! `memoryd` deciding turns as it serves (M5c spec, section 5): `loopd` has its answer to +//! `session_updated` before `brokerd` is asked, and a silent `brokerd` holds up neither `loopd` +//! nor a memory call; without `[honcho]` nothing is asked or written; a damaged ledger stops the +//! start with its pointer, and a cut-off last line is dropped and logged. Every wait has a +//! deadline. Do not edit. + +#[path = "support/brokerd.rs"] +mod brokerd; +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; + +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use brokerd::{Answer, FakeBrokerd}; +use bxcore::{ + CallId, DataClass, Envelope, MemoryCall, Message, SessionId, SessionUpdated, ToolRequest, + read_frame, write_frame, +}; +use memoryd::config::Config; +use memoryd::honcho::ledger::LedgerError; +use memoryd::serve::{self, StartError}; +use tmp::TempDir; +use turns::{clean, write_turns}; + +const SECTION: &str = "[honcho] +url = \"http://127.0.0.1:9\" +agent = \"boxmaker0\" +owner = \"kyle\" +trusted = \"boxmaker0-trusted\" +untrusted = \"boxmaker0-untrusted\" +fleet = \"fleet\" +"; + +const DECIDED: &str = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"mm-a","turn":1},"action":{"type":"turn","workspace":"trusted","speaker":"kyle"}}"#; + +struct Home { + config: Config, + text: String, + dir: TempDir, +} + +impl Home { + fn new(tag: &str, honcho: bool) -> Home { + let dir = TempDir::new(tag); + let soul = dir.write("etc/soul.md", "You are Boxmaker.\n"); + let text = format!( + "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n{}", + dir.path().display(), + soul.display(), + if honcho { SECTION } else { "" } + ); + Home { + config: Config::parse(&text).unwrap(), + text, + dir, + } + } + + fn ledger(&self) -> PathBuf { + self.dir.path().join("memory/honcho.jsonl") + } + + fn ledger_text(&self) -> String { + std::fs::read_to_string(self.ledger()).unwrap_or_default() + } + + /// `serve::start`, run on a thread; its log lines are collected. + fn serve(&self) -> Arc>> { + let lines = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&lines); + let log: serve::Log = Arc::new(move |l: &str| sink.lock().unwrap().push(l.to_string())); + let running = serve::start(self.config.clone(), log).unwrap(); + std::thread::spawn(move || running.run()); + lines + } +} + +/// One request and its answer, with a 10 s limit; how long the answer took. +fn ask(socket: &Path, msg: Message) -> (Message, Duration) { + let started = Instant::now(); + let mut stream = UnixStream::connect(socket).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let env = Envelope { + v: 1, + id: 7, + r#final: true, + msg, + }; + write_frame(&mut stream, &env).unwrap(); + let reply = read_frame(&mut stream).unwrap(); + (reply.msg, started.elapsed()) +} + +fn updated(home: &Home, session: &str, turn: u64) -> (Message, Duration) { + let msg = Message::SessionUpdated(SessionUpdated { + session: SessionId::new(session).unwrap(), + turn, + }); + ask(&home.config.loop_socket(), msg) +} + +/// Waits up to 10 s for `done`; true if it came. +fn within_10s(done: impl Fn() -> bool) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if done() { + return true; + } + std::thread::sleep(Duration::from_millis(10)); + } + false +} + +#[test] +fn a_reported_turn_is_decided_after_loopd_has_its_answer() { + let home = Home::new("serve-decide", true); + write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]); + write_turns(home.dir.path(), "mm-z", &[turns::finished(Some("kyle"))]); + let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::State(clean())); + let _lines = home.serve(); + // A report on another socket is refused, and decides nothing. + let wrong = Message::SessionUpdated(SessionUpdated { + session: SessionId::new("mm-z").unwrap(), + turn: 1, + }); + let (answer, _) = ask(&home.config.admin_socket(), wrong); + assert!(matches!(answer, Message::Error(_)), "{answer:?}"); + let (answer, _) = updated(&home, "mm-a", 1); + assert!(matches!(answer, Message::Ok(_)), "{answer:?}"); + assert!(within_10s(|| home.ledger_text().contains("\"turn\":1}"))); + // The line as `DECIDED`, but for its time, which is the decision's. + let text = home.ledger_text(); + assert_eq!(text.lines().count(), 1, "{text}"); + let mut got: serde_json::Value = serde_json::from_str(&text).unwrap(); + let mut want: serde_json::Value = serde_json::from_str(DECIDED).unwrap(); + got["time"] = serde_json::Value::Null; + want["time"] = serde_json::Value::Null; + assert_eq!(got, want); + assert_eq!(fake.seen().len(), 1); +} + +#[test] +fn a_silent_brokerd_holds_up_neither_loopd_nor_memory() { + let home = Home::new("serve-silent", true); + write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]); + write_turns(home.dir.path(), "mm-b", &[turns::finished(Some("kyle"))]); + let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::Hang); + let lines = home.serve(); + let (answer, took) = updated(&home, "mm-a", 1); + assert!(matches!(answer, Message::Ok(_)), "{answer:?}"); + assert!(took < Duration::from_secs(1), "loopd waited {took:?}"); + assert!(fake.wait_for(1), "brokerd was never asked"); + // While the question hangs: another report, and a memory call. + let (answer, took) = updated(&home, "mm-b", 1); + assert!(matches!(answer, Message::Ok(_)), "{answer:?}"); + assert!(took < Duration::from_secs(1), "loopd waited {took:?}"); + let call = Message::MemoryCall(MemoryCall { + request: ToolRequest { + session: SessionId::new("mm-b").unwrap(), + call: CallId(1), + tool: "recall".to_string(), + arguments: r#"{"query":"hello"}"#.to_string(), + }, + taint: DataClass::Private, + untrusted: false, + }); + let (answer, took) = ask(&home.config.broker_socket(), call); + assert!(matches!(answer, Message::ToolResponse(_)), "{answer:?}"); + assert!( + took < Duration::from_secs(1), + "the memory call waited {took:?}" + ); + // After 5 s without a byte, the outage is logged once, and nothing is decided. + let pointer = "turns wait undecided; see docs/runbook.md#broker-state-unavailable"; + assert!(within_10s(|| lines + .lock() + .unwrap() + .iter() + .any(|l| l.contains(pointer)))); + let logged = lines.lock().unwrap().clone(); + let outage: Vec<&String> = logged.iter().filter(|l| l.contains(pointer)).collect(); + assert_eq!(outage.len(), 1, "{logged:?}"); + let start = format!("memoryd: {}: ", home.config.state_socket().display()); + assert!(outage[0].starts_with(&start), "{}", outage[0]); + assert_eq!(home.ledger_text(), ""); +} + +#[test] +fn without_honcho_nothing_is_asked_or_written() { + let home = Home::new("serve-off", false); + write_turns(home.dir.path(), "mm-a", &[turns::finished(Some("kyle"))]); + let fake = FakeBrokerd::start(&home.config.state_socket(), Answer::State(clean())); + let lines = home.serve(); + let (answer, _) = updated(&home, "mm-a", 1); + assert!(matches!(answer, Message::Ok(_)), "{answer:?}"); + // A negative: nothing may happen, so there is nothing to wait for but a while. + std::thread::sleep(Duration::from_millis(300)); + assert!(fake.seen().is_empty()); + assert!(!home.ledger().exists()); + assert!( + lines.lock().unwrap().is_empty(), + "{:?}", + lines.lock().unwrap() + ); +} + +#[test] +fn a_damaged_ledger_stops_the_start() { + let home = Home::new("serve-damaged", true); + std::fs::create_dir_all(home.dir.path().join("memory")).unwrap(); + std::fs::write(home.ledger(), "not json\n").unwrap(); + let log: serve::Log = Arc::new(|_: &str| {}); + let err = match serve::start(home.config.clone(), log) { + Err(err) => err, + Ok(_) => panic!("memoryd started"), + }; + assert!( + matches!( + err, + StartError::Ledger(LedgerError::Damaged { line: 1, .. }) + ), + "{err:?}" + ); + let shown = err.to_string(); + assert!(shown.starts_with(&format!("{}:1: ", home.ledger().display()))); + assert!( + shown.ends_with("; see docs/runbook.md#honcho-ledger-damaged"), + "{shown}" + ); + assert_eq!(shown.matches("see docs/runbook.md").count(), 1, "{shown}"); + assert!(!home.config.loop_socket().exists(), "a socket was bound"); + // A ledger that cannot be read is named with the same pointer. + std::fs::remove_file(home.ledger()).unwrap(); + std::fs::create_dir(home.ledger()).unwrap(); + let log: serve::Log = Arc::new(|_: &str| {}); + let err = match serve::start(home.config.clone(), log) { + Err(err) => err, + Ok(_) => panic!("memoryd started"), + }; + assert!( + matches!(err, StartError::Ledger(LedgerError::Io(..))), + "{err:?}" + ); + let shown = err.to_string(); + assert!(shown.starts_with(&format!("{}: ", home.ledger().display()))); + assert!( + shown.ends_with("; see docs/runbook.md#honcho-ledger-damaged"), + "{shown}" + ); +} + +#[test] +fn a_cut_off_last_line_is_dropped_and_logged() { + let home = Home::new("serve-repaired", true); + std::fs::create_dir_all(home.dir.path().join("memory")).unwrap(); + std::fs::write(home.ledger(), format!("{DECIDED}\n{{\"type\":\"se")).unwrap(); + let lines = home.serve(); + assert_eq!( + *lines.lock().unwrap(), + vec![format!( + "memoryd: {}: a last line cut off by a crash was dropped", + home.ledger().display() + )] + ); + assert_eq!(home.ledger_text(), format!("{DECIDED}\n")); +} + +/// A `memoryd` process, killed and waited for when dropped: declared after its `Home`, so it +/// drops first. +struct Daemon(std::process::Child); + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[test] +fn memoryd_exits_with_the_ledgers_line() { + let home = Home::new("serve-exit", true); + let config = home.dir.write("memoryd.toml", &home.text); + std::fs::create_dir_all(home.dir.path().join("memory")).unwrap(); + std::fs::write(home.ledger(), "not json\n").unwrap(); + let mut child = Daemon( + Command::new(env!("CARGO_BIN_EXE_memoryd")) + .args(["serve", "--config"]) + .arg(&config) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let deadline = Instant::now() + Duration::from_secs(10); + let mut code = None; + while Instant::now() < deadline { + if let Ok(Some(status)) = child.0.try_wait() { + code = status.code(); + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let _ = child.0.kill(); + let _ = child.0.wait(); + let mut stderr = String::new(); + if let Some(mut pipe) = child.0.stderr.take() { + let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr); + } + assert_eq!(code, Some(1), "{stderr}"); + let start = format!("memoryd: {}:1: ", home.ledger().display()); + assert!(stderr.starts_with(&start), "{stderr}"); + assert!( + stderr.ends_with("; see docs/runbook.md#honcho-ledger-damaged\n"), + "{stderr}" + ); + assert_eq!(stderr.lines().count(), 1, "{stderr}"); + assert!(!home.config.loop_socket().exists(), "a socket was bound"); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_state.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_state.rs new file mode 100644 index 0000000..607b0a3 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_state.rs @@ -0,0 +1,87 @@ +//! Asking `brokerd` for a session's state on its `state.sock` (M5c spec, section 7): the +//! question as `brokerd` reads it, its answer, an error frame, silence and no listener, each an +//! error with one line. Against a fake `brokerd` in the test's own directory. Do not edit. + +#[path = "support/brokerd.rs"] +mod brokerd; +#[path = "support/tmp.rs"] +mod tmp; + +use std::time::{Duration, Instant}; + +use brokerd::{Answer, FakeBrokerd}; +use bxcore::{DataClass, Message, PROTOCOL_VERSION, SessionId, SessionState, SessionStateReply}; +use memoryd::honcho::state::{RUNBOOK, WAIT, ask}; +use tmp::TempDir; + +fn session() -> SessionId { + SessionId::new("mm-chat-1").unwrap() +} + +#[test] +fn the_constants() { + assert_eq!(WAIT, Duration::from_secs(5)); + assert_eq!(RUNBOOK, "see docs/runbook.md#broker-state-unavailable"); +} + +#[test] +fn the_question_and_its_answer() { + let dir = TempDir::new("state-answer"); + let socket = dir.path().join("run/memory-broker/state.sock"); + let reply = SessionStateReply { + class: DataClass::Secret, + untrusted: true, + }; + let fake = FakeBrokerd::start(&socket, Answer::State(reply.clone())); + assert_eq!(ask(&socket, &session(), WAIT), Ok(reply)); + let seen = fake.seen(); + assert_eq!(seen.len(), 1); + assert_eq!( + (seen[0].v, seen[0].id, seen[0].r#final), + (PROTOCOL_VERSION, 1, true) + ); + assert_eq!( + seen[0].msg, + Message::SessionState(SessionState { session: session() }) + ); +} + +#[test] +fn an_error_frame_is_an_error() { + let dir = TempDir::new("state-error"); + let socket = dir.path().join("state.sock"); + let _fake = FakeBrokerd::start(&socket, Answer::Error("the state file is damaged".into())); + assert_eq!( + ask(&socket, &session(), WAIT), + Err("brokerd: the state file is damaged".to_string()) + ); +} + +#[test] +fn no_listener_is_an_error_at_once() { + let dir = TempDir::new("state-none"); + let socket = dir.path().join("state.sock"); + let started = Instant::now(); + let err = ask(&socket, &session(), WAIT).unwrap_err(); + assert!( + started.elapsed() < Duration::from_secs(1), + "{:?}", + started.elapsed() + ); + assert!(err.starts_with(&format!("{}: ", socket.display())), "{err}"); + assert!(!err.contains('\n'), "{err}"); +} + +#[test] +fn silence_is_an_error_after_the_wait() { + let dir = TempDir::new("state-hang"); + let socket = dir.path().join("state.sock"); + let fake = FakeBrokerd::start(&socket, Answer::Hang); + let started = Instant::now(); + let err = ask(&socket, &session(), Duration::from_millis(300)).unwrap_err(); + let waited = started.elapsed(); + assert!(waited >= Duration::from_millis(300), "{waited:?}"); + assert!(waited < Duration::from_secs(5), "{waited:?}"); + assert!(err.starts_with(&format!("{}: ", socket.display())), "{err}"); + assert_eq!(fake.seen().len(), 1, "the question was not sent"); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_turns.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_turns.rs new file mode 100644 index 0000000..8e6ba01 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_turns.rs @@ -0,0 +1,428 @@ +//! Deciding reported turns (M5c spec, section 5): every undecided turn of every waiting session up +//! to the one reported, as far as the log goes; the log's last turn waits for its reply; one +//! question to `brokerd` per session per walk, and none when nothing waits; a silent `brokerd` +//! stops the walk and is logged once, and so is its return; a damaged log and a ledger that cannot +//! be written. `brokerd` is a closure here; the socket is in `honcho_state.rs`. Do not edit. + +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; + +use std::cell::RefCell; + +use bxcore::{DataClass, LogRecord, SessionId, SessionStateReply, Timestamp}; +use logs::{assistant, write_log}; +use memoryd::honcho::decide::{NO_REPLY, NO_SPEAKER, SECRET_CLASS}; +use memoryd::honcho::ledger::{Action, Item, Ledger, Workspace}; +use memoryd::honcho::turns::{ + LEDGER, LOG_RUNBOOK, TurnSync, decidable, finished, log_path, read_log, +}; +use tmp::TempDir; +use turns::{clean, fetched, running, sid, spoken, write_turns}; + +const BROKER_DOWN: &str = "see docs/runbook.md#broker-state-unavailable"; +const BACK: &str = "memoryd: brokerd answers on its state socket again; turns are decided"; + +/// A home with a ledger, a `TurnSync`, and a `brokerd` whose answer the test sets. +struct World { + sync: TurnSync, + answer: RefCell>, + asked: RefCell>, + lines: RefCell>, + dir: TempDir, +} + +impl World { + fn new(tag: &str) -> World { + let dir = TempDir::new(tag); + std::fs::create_dir_all(dir.path().join("memory")).unwrap(); + let ledger = Ledger::open(&dir.path().join("memory").join(LEDGER)).unwrap(); + World { + sync: TurnSync::new(ledger, "boxmaker0", &dir.path().join("sessions")), + answer: RefCell::new(Ok(clean())), + asked: RefCell::new(Vec::new()), + lines: RefCell::new(Vec::new()), + dir, + } + } + + fn log(&self, session: &str, turns: &[Vec]) { + write_turns(self.dir.path(), session, turns); + } + + fn updated(&mut self, session: &str, turn: u64) { + let ask = |s: &SessionId| { + self.asked.borrow_mut().push(s.as_str().to_string()); + self.answer.borrow().clone() + }; + let log = |line: &str| self.lines.borrow_mut().push(line.to_string()); + let now = Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap(); + self.sync.updated(&sid(session), turn, &ask, now, &log); + } + + fn action(&self, session: &str, turn: u64) -> Option { + let item = Item::Turn { + session: sid(session), + turn, + }; + let path = self.dir.path().join("memory").join(LEDGER); + let text = std::fs::read_to_string(path).unwrap_or_default(); + let key = serde_json::to_string(&item).unwrap(); + let line = text.lines().find(|l| l.contains(&key))?; + let value: serde_json::Value = serde_json::from_str(line).unwrap(); + Some(serde_json::from_value(value["action"].clone()).unwrap()) + } + + fn waiting(&self) -> Vec<(String, u64)> { + let waiting = self.sync.waiting(); + waiting + .into_iter() + .map(|(s, n)| (s.as_str().to_string(), n)) + .collect() + } +} + +fn to(workspace: Workspace, speaker: &str) -> Option { + Some(Action::Turn { + workspace, + speaker: speaker.to_string(), + }) +} + +fn skip(why: &str) -> Option { + Some(Action::Skip { + why: why.to_string(), + }) +} + +#[test] +fn the_reported_turn_is_decided_and_written_at_once() { + let mut w = World::new("turns-one"); + w.log("mm-a", &[turns::finished(Some("kyle"))]); + w.updated("mm-a", 1); + assert_eq!(w.action("mm-a", 1), to(Workspace::Trusted, "kyle")); + assert_eq!(*w.asked.borrow(), vec!["mm-a"]); + assert!(w.waiting().is_empty()); + assert!(w.lines.borrow().is_empty(), "{:?}", w.lines.borrow()); + let reopened = Ledger::open(&w.dir.path().join("memory").join(LEDGER)).unwrap(); + assert_eq!(reopened.counts().decided, 1); +} + +#[test] +fn every_undecided_turn_up_to_the_reported_one_is_decided() { + let mut w = World::new("turns-all"); + w.log( + "mm-a", + &[ + turns::finished(None), + running(Some("kyle")), + fetched(Some("kyle")), + turns::finished(Some("kyle")), + ], + ); + w.updated("mm-a", 4); + assert_eq!(w.action("mm-a", 1), skip(NO_SPEAKER)); + assert_eq!(w.action("mm-a", 2), skip(NO_REPLY)); + assert_eq!(w.action("mm-a", 3), to(Workspace::Untrusted, "kyle")); + assert_eq!(w.action("mm-a", 4), to(Workspace::Untrusted, "kyle")); + assert_eq!(*w.asked.borrow(), vec!["mm-a"], "one question per session"); +} + +#[test] +fn only_as_far_as_the_report_and_the_log_go() { + let mut w = World::new("turns-limit"); + w.log( + "mm-a", + &[turns::finished(Some("kyle")), turns::finished(Some("kyle"))], + ); + w.updated("mm-a", 9); + assert!(w.action("mm-a", 2).is_some()); + assert!(w.waiting().is_empty(), "turns not in the log do not wait"); + w.updated("mm-a", 9); + assert_eq!( + w.asked.borrow().len(), + 1, + "asked about turns not in the log" + ); + w.log( + "mm-b", + &[ + turns::finished(Some("kyle")), + turns::finished(Some("kyle")), + turns::finished(Some("kyle")), + ], + ); + w.updated("mm-b", 1); + assert!(w.action("mm-b", 1).is_some()); + assert_eq!(w.action("mm-b", 2), None, "turn 2 was not reported"); + assert!(w.waiting().is_empty()); +} + +#[test] +fn the_logs_last_turn_waits_for_its_reply() { + let mut w = World::new("turns-running"); + w.log( + "mm-a", + &[turns::finished(Some("kyle")), running(Some("kyle"))], + ); + w.updated("mm-a", 2); + assert!(w.action("mm-a", 1).is_some()); + assert_eq!(w.action("mm-a", 2), None, "a running turn was decided"); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 2)]); + // Another report while it runs: it still waits, and nothing is asked for it. + w.log("mm-c", &[turns::finished(Some("kyle"))]); + w.updated("mm-c", 1); + assert_eq!(w.action("mm-a", 2), None); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 2)]); + assert_eq!(*w.asked.borrow(), vec!["mm-a", "mm-c"]); + // It ends; the next report of any session decides it. + let mut ended = running(Some("kyle")); + ended.push(assistant(3, Some("found it"), &[])); + w.log("mm-a", &[turns::finished(Some("kyle")), ended]); + w.log("mm-b", &[turns::finished(Some("ana"))]); + w.updated("mm-b", 1); + assert_eq!(w.action("mm-a", 2), to(Workspace::Trusted, "kyle")); + assert_eq!(w.action("mm-b", 1), to(Workspace::Trusted, "ana")); + assert!(w.waiting().is_empty()); + assert_eq!(w.sync.ledger().counts().skipped, 0); +} + +#[test] +fn a_report_past_the_log_does_not_decide_a_running_turn() { + let mut w = World::new("turns-inflated"); + w.log( + "mm-a", + &[turns::finished(Some("kyle")), running(Some("kyle"))], + ); + w.updated("mm-a", 5); + assert!(w.action("mm-a", 1).is_some()); + assert_eq!(w.action("mm-a", 2), None, "a running turn was decided"); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 5)]); +} + +#[test] +fn an_earlier_report_does_not_lower_the_turn() { + let mut w = World::new("turns-order"); + *w.answer.borrow_mut() = Err("no".to_string()); + w.log( + "mm-a", + &[ + turns::finished(Some("kyle")), + turns::finished(Some("kyle")), + turns::finished(Some("kyle")), + ], + ); + w.updated("mm-a", 3); + w.updated("mm-a", 1); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 3)]); + *w.answer.borrow_mut() = Ok(clean()); + w.log("mm-b", &[turns::finished(Some("kyle"))]); + w.updated("mm-b", 1); + assert_eq!(w.action("mm-a", 3), to(Workspace::Trusted, "kyle")); +} + +#[test] +fn a_silent_brokerd_is_logged_once_and_its_return_once() { + let mut w = World::new("turns-down"); + let why = "/h/run/memory-broker/state.sock: Connection refused (os error 111)"; + *w.answer.borrow_mut() = Err(why.to_string()); + w.log("mm-a", &[turns::finished(Some("kyle"))]); + w.log("mm-b", &[turns::finished(Some("kyle"))]); + w.updated("mm-a", 1); + w.updated("mm-b", 1); + w.updated("mm-a", 1); + assert_eq!(w.action("mm-a", 1), None); + assert_eq!(w.action("mm-b", 1), None); + assert_eq!( + *w.lines.borrow(), + vec![format!( + "memoryd: {why}; turns wait undecided; {BROKER_DOWN}" + )] + ); + assert_eq!(w.waiting().len(), 2); + *w.answer.borrow_mut() = Ok(clean()); + w.log("mm-c", &[turns::finished(Some("kyle"))]); + w.updated("mm-c", 1); + for session in ["mm-a", "mm-b", "mm-c"] { + assert_eq!( + w.action(session, 1), + to(Workspace::Trusted, "kyle"), + "{session}" + ); + } + assert_eq!(w.lines.borrow().len(), 2); + assert_eq!(w.lines.borrow()[1], BACK); + assert!(w.waiting().is_empty()); +} + +#[test] +fn a_silent_brokerd_stops_the_walk_at_the_first_question() { + let mut w = World::new("turns-stop"); + *w.answer.borrow_mut() = Err("no".to_string()); + for session in ["mm-a", "mm-b", "mm-c"] { + w.log(session, &[turns::finished(Some("kyle"))]); + } + w.updated("mm-a", 1); + w.updated("mm-b", 1); + w.updated("mm-c", 1); + // One question per walk, to the first waiting session in the order of the ids. + assert_eq!(*w.asked.borrow(), vec!["mm-a", "mm-a", "mm-a"]); + *w.answer.borrow_mut() = Ok(clean()); + w.updated("mm-c", 1); + assert_eq!(w.asked.borrow()[3..], ["mm-a", "mm-b", "mm-c"]); +} + +#[test] +fn nothing_undecided_asks_nothing() { + let mut w = World::new("turns-quiet"); + w.log("mm-a", &[turns::finished(Some("kyle"))]); + w.updated("mm-a", 1); + w.updated("mm-a", 1); + assert_eq!(*w.asked.borrow(), vec!["mm-a"]); + let lines = std::fs::read_to_string(w.dir.path().join("memory").join(LEDGER)).unwrap(); + assert_eq!(lines.lines().count(), 1); + // A restart reads the ledger through: nothing is decided twice. + let ledger = Ledger::open(&w.dir.path().join("memory").join(LEDGER)).unwrap(); + w.sync = TurnSync::new(ledger, "boxmaker0", &w.dir.path().join("sessions")); + w.updated("mm-a", 1); + assert_eq!(w.asked.borrow().len(), 1); +} + +#[test] +fn brokerds_state_decides_the_workspace() { + let mut w = World::new("turns-state"); + w.log("mm-a", &[turns::finished(Some("kyle"))]); + *w.answer.borrow_mut() = Ok(SessionStateReply { + class: DataClass::Private, + untrusted: true, + }); + w.updated("mm-a", 1); + assert_eq!(w.action("mm-a", 1), to(Workspace::Untrusted, "kyle")); + w.log("mm-b", &[turns::finished(Some("kyle"))]); + *w.answer.borrow_mut() = Ok(SessionStateReply { + class: DataClass::Secret, + untrusted: false, + }); + w.updated("mm-b", 1); + assert_eq!(w.action("mm-b", 1), skip(SECRET_CLASS)); +} + +#[test] +fn a_damaged_log_waits_and_is_logged_once() { + let mut w = World::new("turns-damaged"); + let path = write_log(w.dir.path(), "mm-a", &turns::finished(Some("kyle"))); + let text = std::fs::read_to_string(&path).unwrap(); + std::fs::write(&path, format!("not json\n{text}")).unwrap(); + w.log("mm-b", &[turns::finished(Some("kyle"))]); + w.updated("mm-a", 1); + w.updated("mm-b", 1); + assert_eq!(w.action("mm-a", 1), None); + assert_eq!(w.action("mm-b", 1), to(Workspace::Trusted, "kyle")); + let lines = w.lines.borrow().clone(); + assert_eq!(lines.len(), 1, "{lines:?}"); + let start = format!("memoryd: {}:1: ", path.display()); + assert!(lines[0].starts_with(&start), "{}", lines[0]); + assert!( + lines[0].ends_with(&format!("; {LOG_RUNBOOK}")), + "{}", + lines[0] + ); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 1)]); + assert_eq!(*w.asked.borrow(), vec!["mm-b"]); + // Mended, it is decided at the next report. + std::fs::write(&path, text).unwrap(); + w.updated("mm-b", 1); + assert_eq!(w.action("mm-a", 1), to(Workspace::Trusted, "kyle")); +} + +#[test] +fn a_line_loopd_is_still_writing_is_left_out() { + let mut w = World::new("turns-partial"); + let path = write_log(w.dir.path(), "mm-a", &turns::finished(Some("kyle"))); + let half = serde_json::to_string(&spoken("next", Some("kyle"))).unwrap(); + let mut text = std::fs::read_to_string(&path).unwrap(); + text.push_str(&half[..half.len() / 2]); + std::fs::write(&path, text).unwrap(); + w.updated("mm-a", 1); + assert_eq!(w.action("mm-a", 1), to(Workspace::Trusted, "kyle")); + assert!(w.lines.borrow().is_empty(), "{:?}", w.lines.borrow()); +} + +#[test] +fn a_ledger_that_cannot_be_written_stops_the_walk() { + let mut w = World::new("turns-ledger"); + w.log("mm-a", &[turns::finished(Some("kyle"))]); + let ledger = w.dir.path().join("memory").join(LEDGER); + std::fs::create_dir(&ledger).unwrap(); + w.updated("mm-a", 1); + let lines = w.lines.borrow().clone(); + assert_eq!(lines.len(), 1, "{lines:?}"); + assert!(lines[0].starts_with(&format!("memoryd: {}: ", ledger.display()))); + assert!(lines[0].ends_with("; see docs/runbook.md#honcho-ledger-damaged")); + assert!(!w.sync.ledger().is_decided(&Item::Turn { + session: sid("mm-a"), + turn: 1 + })); + assert_eq!(w.waiting(), vec![("mm-a".to_string(), 1)]); + std::fs::remove_dir(&ledger).unwrap(); + w.updated("mm-a", 1); + assert_eq!(w.action("mm-a", 1), to(Workspace::Trusted, "kyle")); +} + +#[test] +fn reading_a_log() { + let dir = TempDir::new("turns-read"); + let path = log_path(&dir.path().join("sessions"), &sid("mm-a")); + assert_eq!(path, dir.path().join("sessions/mm-a/0.jsonl")); + assert_eq!(read_log(&path), Ok(Vec::new()), "a missing log"); + let records = turns::finished(Some("kyle")); + write_log(dir.path(), "mm-a", &records); + assert_eq!(read_log(&path), Ok(records.clone())); + let mut text = std::fs::read_to_string(&path).unwrap(); + text.push_str("{\"type\":\"us"); + std::fs::write(&path, &text).unwrap(); + assert_eq!(read_log(&path), Ok(records)); + std::fs::write(&path, "{\"type\":\"user\"}\n").unwrap(); + let err = read_log(&path).unwrap_err(); + assert!(err.starts_with(&format!("{}:1: ", path.display())), "{err}"); + assert!(err.ends_with(&format!("; {LOG_RUNBOOK}")), "{err}"); + std::fs::remove_file(&path).unwrap(); + std::fs::create_dir(&path).unwrap(); + let err = read_log(&path).unwrap_err(); + assert!(err.starts_with(&format!("{}: ", path.display())), "{err}"); + assert_eq!(LOG_RUNBOOK, "see docs/runbook.md#session-log-damaged"); + assert_eq!(LEDGER, "honcho.jsonl"); +} + +#[test] +fn a_turn_is_finished_by_an_answer_without_tool_calls() { + assert!(finished(&turns::finished(Some("kyle")))); + assert!(finished(&fetched(None))); + assert!(!finished(&running(None))); + assert!(!finished(&[spoken("hello", None)])); + let mut empty = vec![spoken("hello", None), assistant(2, None, &[])]; + assert!(finished(&empty), "an empty answer still ends the turn"); + empty.push(assistant(3, Some("and"), &[("fetch", "{}")])); + assert!(!finished(&empty)); +} + +#[test] +fn how_far_a_log_may_be_decided() { + let done = |n: usize| -> Vec { + (0..n).flat_map(|_| turns::finished(Some("kyle"))).collect() + }; + assert_eq!(decidable(&[], 3), (0, false)); + assert_eq!(decidable(&done(3), 0), (0, false)); + assert_eq!(decidable(&done(3), 2), (2, false)); + assert_eq!(decidable(&done(3), 3), (3, false)); + assert_eq!(decidable(&done(3), 7), (3, false)); + let mut open = done(2); + open.extend(running(Some("kyle"))); + assert_eq!(decidable(&open, 2), (2, false)); + assert_eq!(decidable(&open, 3), (2, true)); + assert_eq!(decidable(&open, 9), (2, true)); + assert_eq!(decidable(&running(None), 1), (0, true)); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/support/brokerd.rs b/docs/plans/M5c1/files/crates/memoryd/tests/support/brokerd.rs new file mode 100644 index 0000000..bc76637 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/brokerd.rs @@ -0,0 +1,127 @@ +//! A fake `brokerd` state socket for tests: it records each question and answers with a fixed +//! state, an error frame, or nothing at all (it holds the connection open without a byte). It +//! stops, and releases every held connection, when dropped; every wait has a deadline. Do not +//! edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use bxcore::{ + Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionStateReply, WireError, read_frame, + write_frame, +}; + +/// How the fake answers. +#[derive(Debug, Clone)] +pub enum Answer { + State(SessionStateReply), + Error(String), + /// Read the question, then send nothing until the fake is dropped (at most 30 s). + Hang, +} + +pub struct FakeBrokerd { + path: PathBuf, + stop: Arc, + seen: Arc>>, + accept: Option>, +} + +impl FakeBrokerd { + /// Listen on `path` (its directory is made) and answer every connection with `answer`. + pub fn start(path: &Path, answer: Answer) -> FakeBrokerd { + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir).unwrap(); + } + let _ = std::fs::remove_file(path); + let listener = UnixListener::bind(path).unwrap(); + let stop = Arc::new(AtomicBool::new(false)); + let seen = Arc::new(Mutex::new(Vec::new())); + let (stop_accept, seen_accept) = (Arc::clone(&stop), Arc::clone(&seen)); + let accept = std::thread::spawn(move || { + for stream in listener.incoming() { + if stop_accept.load(Ordering::SeqCst) { + return; + } + let Ok(stream) = stream else { return }; + let (stop, seen, answer) = ( + Arc::clone(&stop_accept), + Arc::clone(&seen_accept), + answer.clone(), + ); + std::thread::spawn(move || serve(stream, &answer, &seen, &stop)); + } + }); + FakeBrokerd { + path: path.to_path_buf(), + stop, + seen, + accept: Some(accept), + } + } + + /// Every question read so far, in order. + pub fn seen(&self) -> Vec { + self.seen.lock().unwrap().clone() + } + + /// Waits up to 10 s until at least `n` questions were read; true if they were. + pub fn wait_for(&self, n: usize) -> bool { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if self.seen().len() >= n { + return true; + } + std::thread::sleep(Duration::from_millis(10)); + } + false + } +} + +fn serve(mut stream: UnixStream, answer: &Answer, seen: &Mutex>, stop: &AtomicBool) { + let _ = stream.set_read_timeout(Some(Duration::from_secs(10))); + let Ok(question) = read_frame(&mut stream) else { + return; + }; + let id = question.id; + seen.lock().unwrap().push(question); + let msg = match answer { + Answer::State(state) => Message::SessionStateReply(state.clone()), + Answer::Error(detail) => Message::Error(WireError { + code: ErrorCode::Internal, + detail: detail.clone(), + }), + Answer::Hang => { + let deadline = Instant::now() + Duration::from_secs(30); + while !stop.load(Ordering::SeqCst) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + return; + } + }; + let env = Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg, + }; + let _ = write_frame(&mut stream, &env); +} + +impl Drop for FakeBrokerd { + fn drop(&mut self) { + self.stop.store(true, Ordering::SeqCst); + // Wakes the accept loop, which then sees `stop`. + let _ = UnixStream::connect(&self.path); + if let Some(accept) = self.accept.take() { + let _ = accept.join(); + } + let _ = std::fs::remove_file(&self.path); + } +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/support/turns.rs b/docs/plans/M5c1/files/crates/memoryd/tests/support/turns.rs new file mode 100644 index 0000000..c1a66bd --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/turns.rs @@ -0,0 +1,57 @@ +//! Session logs of turns with speakers, for the tests of deciding turns. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::path::Path; + +use bxcore::{DataClass, LogRecord, SessionId, SessionStateReply}; + +use crate::logs::{assistant, at, result}; + +pub fn sid(s: &str) -> SessionId { + SessionId::new(s).unwrap() +} + +pub fn clean() -> SessionStateReply { + SessionStateReply { + class: DataClass::Private, + untrusted: false, + } +} + +pub fn spoken(content: &str, speaker: Option<&str>) -> LogRecord { + LogRecord::User { + time: at(1), + content: content.to_string(), + speaker: speaker.map(str::to_string), + } +} + +/// A finished turn: the speaker's message and the reply. +pub fn finished(speaker: Option<&str>) -> Vec { + vec![spoken("hello", speaker), assistant(2, Some("hi"), &[])] +} + +/// A finished turn that read an untrusted page. +pub fn fetched(speaker: Option<&str>) -> Vec { + vec![ + spoken("read this page", speaker), + assistant(2, None, &[("fetch", "{}")]), + result(3, "the page", DataClass::Public, true), + assistant(4, Some("it says hello"), &[]), + ] +} + +/// A turn still running, or one that failed: no reply. +pub fn running(speaker: Option<&str>) -> Vec { + vec![ + spoken("look it up", speaker), + assistant(2, None, &[("fetch", "{}")]), + ] +} + +/// `sessions//0.jsonl` under `home`, the turns one after another. +pub fn write_turns(home: &Path, session: &str, turns: &[Vec]) { + let records: Vec = turns.iter().flatten().cloned().collect(); + crate::logs::write_log(home, session, &records); +} diff --git a/docs/plans/M5c1/files/runbook-broker-state-unavailable.md b/docs/plans/M5c1/files/runbook-broker-state-unavailable.md new file mode 100644 index 0000000..c6cd644 --- /dev/null +++ b/docs/plans/M5c1/files/runbook-broker-state-unavailable.md @@ -0,0 +1,40 @@ +## broker-state-unavailable + +**What you see.** `memoryd` logs one line when it cannot get a session's state from `brokerd`, +then this entry: + +```text +memoryd: /srv/bx/run/memory-broker/state.sock: Connection refused (os error 111); turns wait undecided; see docs/runbook.md#broker-state-unavailable +``` + +The reason may instead be `No such file or directory` (nothing listens), a timeout +(`frame I/O error: Resource temporarily unavailable`, after 5 seconds without a byte), or +`brokerd: ` when `brokerd` answered with an error. When `brokerd` answers again, +`memoryd` logs `memoryd: brokerd answers on its state socket again; turns are decided`. +Conversations are not affected. + +**Why.** With `[honcho]`, `memoryd` decides where each finished turn goes in Honcho: the trusted +workspace, the untrusted one, or nowhere (M5c spec, section 5). It needs `brokerd`'s record of the +session's class and trust for that, and never guesses it: a guess could put a turn from a session +that read a web page in the trusted workspace. So nothing is decided while `brokerd` does not +answer. The turns wait, in memory, and are decided at the next turn's end of any session once +`brokerd` answers, from the session's state then (which can only have moved towards untrusted). +The outage is logged once, not at every turn. A restart of `memoryd` forgets the waiting turns +until each of their sessions has another turn. + +**Confirm.** + +```sh +ls -l "$BOXMAKER_HOME/run/memory-broker/state.sock" +grep -n 'state' /etc/boxmaker/memoryd.toml /etc/boxmaker/brokerd.toml +``` + +`memoryd`'s `[sockets] state` and `brokerd`'s `[sockets] state` must name the same path (both +default to `/run/memory-broker/state.sock`). If `brokerd` answered with an error, its own +log names the problem, usually `broker-state-damaged`. + +**Fix.** Start `brokerd` if it is stopped; make the two `[sockets] state` paths agree; or, for +`brokerd`'s error, follow the entry its log names. + +**Check.** At the next turn's end `memoryd` logs that `brokerd` answers again, and +`memory/honcho.jsonl` gains a `decided` line for each turn that waited.