diff --git a/docs/plans/M5c1/09-memoryd-honcho-turns.md b/docs/plans/M5c1/09-memoryd-honcho-turns.md index 4c52ed3..6d1f0ad 100644 --- a/docs/plans/M5c1/09-memoryd-honcho-turns.md +++ b/docs/plans/M5c1/09-memoryd-honcho-turns.md @@ -114,7 +114,8 @@ This is the complete list. - 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` + and `crates/memoryd/tests/honcho_serve.rs` from `honcho_serve.rs-task09` (this task's + version; task 11 gives a later one) - 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` @@ -529,7 +530,7 @@ message, so a turn sent twice stays twice. ```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/honcho_serve.rs-task09 crates/memoryd/tests/honcho_serve.rs 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/ ``` diff --git a/docs/plans/M5c1/11-memoryd-honcho-worker.md b/docs/plans/M5c1/11-memoryd-honcho-worker.md new file mode 100644 index 0000000..1dba521 --- /dev/null +++ b/docs/plans/M5c1/11-memoryd-honcho-worker.md @@ -0,0 +1,624 @@ +# M5c-1 task 11: `memoryd`'s Honcho worker sends decided turns + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `memoryd: the Honcho worker sends decided turns, waits out outages, retries undecided turns` +**Log row:** the Task cell is `M5c1/11-memoryd-honcho-worker`. + +## Goal + +Task 09 decides each finished turn into the ledger; nothing sends it yet. This task adds the +worker that does (M5c spec, section 5): one thread in `memoryd`, started with `[honcho]` only. + +- **Start** (`serve::start`, `[honcho]` only), before anything under the home is made and before + any socket is bound: the ledger is read through (as in task 09), the sessions it knows wait + again (`TurnSync::resume`, below), and the three clients are built (task 10) with their keys. + A key that cannot be read stops `memoryd`: exit 1, `bxcore::secrets`' own lines with the + pointer `secret-unavailable`, no socket bound. The key files' plaintext warnings are logged. + After the sockets are bound the worker's thread starts. +- **Sending**, oldest first, `Action::Turn` items only: the other actions (conclusions, fleet + messages, retractions) wait in the outbox untouched, for task 13. For each turn the worker + reads its records from `loopd`'s log again, takes the speaker's text and the final reply (task + 08's `user_text` and `final_reply`, made public here), neutralises both + (`bxcore::sanitize::neutralise`), cuts each into parts of at most 25,000 characters, makes sure + of its Honcho session (`honcho_session` and `ensure_session`, task 10) and adds every part of + both as one batch, the speaker's first, then the agent's. The ids Honcho gives go in a `Sent` + record. A turn the log no longer holds (a damaged log, a missing turn or reply) is marked + `Failed` with kind `log` and no status; one that needs more than 100 messages, with kind + `too-long`. +- **Honcho unreachable** (a connection error, a timeout, a 5xx): wait, then try the same turn + again: 5 s, 10 s, 20 s and so on up to 10 minutes. The outage is logged once when it starts + and once when it ends (the first answer of any kind), not at every try. +- **Refused** (another 4xx, or a body that cannot be read): the turn is marked `Failed` with the + error's kind and status, logged once, and the next turn goes on. +- **Key refused** (401, 403): nothing more is sent to that workspace until `memoryd` restarts, + logged once; its turns stay waiting, and the other workspaces go on. +- **Nothing to send:** the worker walks task 09's waiting sessions (`TurnSync::retry`: the + "worker's retry" of spec 5, step 3), and so it does right after an outage ends. Then it sleeps + 30 s, or until the next `SessionUpdated` wakes it. +- **A ledger that cannot be written** stops the worker until `memoryd` restarts (logged once): a + send whose `Sent` line could not be written would otherwise be sent again at every step. + +**What a restart knows.** The ledger says what was sent: nothing is sent twice (unless a crash +falls between a send and its `Sent` line). The sessions task 09 was waiting on are rebuilt from +the ledger: each session with a decided turn waits again at the turns its log holds, and the +next walk decides what is left (never a turn still running: task 09's `decidable`). A session +with no decided turn is not scanned: it is picked up at its next `SessionUpdated`. So a session +first reported while `brokerd` was down, and never again before a restart, waits until its next +turn. (Scanning every session would decide old history; task 15 adds the cutoff for turns from +before Honcho was turned on.) + +`memoryd` has no shutdown: it runs until its process ends, and the worker with it. Its waits are +plain sleeps; only the idle sleep is woken early, by a `SessionUpdated`. + +## Interfaces + +In `crates/memoryd/src/honcho/worker.rs` (the module `memoryd::honcho::worker`): + +```rust +pub const MAX_PART: usize = 25_000; +pub const MAX_BATCH: usize = 100; +pub const UNREACHABLE: &str = "see docs/runbook.md#honcho-unreachable"; +pub const SEND_REFUSED: &str = "see docs/runbook.md#honcho-send-refused"; +pub const KEY_REFUSED: &str = "see docs/runbook.md#honcho-key-refused"; +pub const LOG_DAMAGED: &str = "see docs/runbook.md#session-log-damaged"; +pub struct Tuning { pub backoff: Duration, pub backoff_max: Duration, pub idle: Duration } +// Tuning::DEFAULT = 5 s, 600 s, 30 s +pub struct Wake { /* private */ } // Default; notify(&self); wait(&self, timeout) +pub enum Step { Done, Backoff(Duration), Idle, Stopped } +pub type AskBox = Box Result + Send>; +pub struct Worker { /* private */ } +impl Worker { + pub fn new(sync: Arc>, clients: Clients, wake: Arc, ask: AskBox, + log: Log, tuning: Tuning, agent: &str) -> Worker; + pub fn run(self); + pub fn step(&mut self) -> Step; +} +pub fn parts(text: &str) -> Vec; +pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)>; +pub fn batch(text: &str, reply: &str, speaker: &str, agent: &str, session: &SessionId, + turn: u64) -> Vec; +pub struct Opened { pub sync: Arc>, /* private */ } +pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result; +pub fn spawn(opened: Opened, config: &Config, wake: Arc, log: Log) -> Result<(), String>; +``` + +`run` is `step` in a loop: `Backoff(d)` sleeps `d`, `Idle` sleeps on the `Wake` for +`tuning.idle`, `Stopped` returns. The tests call `step` directly, with short waits, and see each +`Backoff` it returns. + +In `crates/memoryd/src/honcho/turns.rs`, `TurnSync` gains `sessions(&self) -> &Path`, +`resume(&mut self)`, `append(&mut self, Record) -> Result<(), LedgerError>` and +`retry(&mut self, ask, now, log)` (the walk `updated` did, now also callable alone). In +`serve.rs`, `StartError` gains `Secret(String)`, and `Memoryd` holds the `TurnSync` in an `Arc` +shared with the worker, and a `Wake` it notifies after each `SessionUpdated`'s decisions. + +## The batch + +Fixture `08-messages-add-batch.json` is the shape: `POST +/v3/workspaces//sessions//messages` with + +```json +{"messages": [ + {"content": "", "peer_id": "", "metadata": {"agent": "", "boxmaker_session": "", "turn": 1}}, + {"content": "", "peer_id": "", "metadata": {"agent": "", "boxmaker_session": "", "turn": 1}} +]} +``` + +answered 201 with the messages made, whose `id`s go in the `Sent` record in order. Fixtures 09 +and 10: a message of 25,000 characters is taken, one of 25,001 refused (422). Characters, not +bytes: `parts` cuts by `char`. The number of messages in one batch Honcho takes is not in the +recorded responses; `MAX_BATCH` is 100, `roncho`'s limit for a batch of conclusions, and a turn +that needs more (over 2,500,000 characters) is marked failed without a request. + +## The lines the worker logs + +| When | The line | +|---|---| +| an outage starts | `memoryd: ; sends wait and are retried; see docs/runbook.md#honcho-unreachable` | +| it ends | `memoryd: Honcho answers again; waiting sends go on` | +| a send is refused | `memoryd: turn of session was not sent: ; it is marked failed; see docs/runbook.md#honcho-send-refused` | +| a turn cannot be read | `memoryd: turn of session cannot be sent (log); it is marked failed; see docs/runbook.md#session-log-damaged` | +| a turn is too long | the same with `(too-long)` and `see docs/runbook.md#honcho-send-refused` | +| a key is refused | `memoryd: Honcho refused the key of workspace (HTTP ); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused` | +| the ledger cannot be written | `memoryd: ; the Honcho worker stops until memoryd restarts; see docs/runbook.md#honcho-ledger-damaged` | + +`` is a `HonchoError`'s `Display` (task 10): its kind and status, never a body. No line +holds a turn's text or a key. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/memoryd/src/honcho/decide.rs` | `user_text` and `final_reply` made public (edits D1, D2) | +| `crates/memoryd/src/honcho/mod.rs` | `pub mod worker;` (edit H1) | +| `crates/memoryd/src/honcho/turns.rs` | `sessions`, `resume`, `append`, `retry` (edits T1 to T3) | +| `crates/memoryd/src/honcho/worker.rs` | new: a skeleton you copy, then fill 11 bodies | +| `crates/memoryd/src/serve.rs` | imports, `Memoryd`, `StartError::Secret`, `start`, `decide_turns` (edits S1 to S8) | +| `crates/memoryd/src/main.rs` | a key's error printed with its own pointer (edit M1) | +| `crates/memoryd/tests/honcho_serve.rs` | a new version, given (task 09's test with a skipped turn in its cut-off ledger, since a worker now runs beside it) | +| `docs/runbook.md` | three entries appended | +| everything else | **no change** | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - new tests: `crates/memoryd/tests/honcho_worker.rs`, `crates/memoryd/tests/honcho_worker_more.rs` + - new test support: `crates/memoryd/tests/support/worker.rs` + - a **new version** of `crates/memoryd/tests/honcho_serve.rs` (it replaces the file in the + tree; `honcho_serve.rs-task09` beside it is the older one: do not copy that) + - the skeleton `crates/memoryd/src/honcho/worker.rs` +- Append, from `docs/plans/M5c1/files/`, to the end of `docs/runbook.md`, each with one blank + line before it, in this order: `runbook-honcho-unreachable.md`, `runbook-honcho-send-refused.md`, + `runbook-honcho-key-refused.md` (step 7 gives the commands). +- Modify: the files of edits D1 to M1, `docs/implementer-log.md`. +- Do not touch: any other source file, any other test or support file, the fixtures, any + `Cargo.toml`, and in the skeleton anything but the 11 `todo!()` bodies. + +## Facts you need + +- **What the skeleton gives whole:** the types, `Worker::new`, `run`, `step` (the table of what + each outcome does), `send` (read, batch, check the size, make sure of the session, add), + `open` and `spawn`. Read `step` and `send` before you fill the helpers they call. +- **The shared state.** `TurnSync` (task 09) holds the ledger and the waiting sessions. The + socket threads take its lock to decide; the worker takes it to find the next turn, to append + its records and to walk, and never holds it while it talks to Honcho or reads a log. +- **`Condvar`.** `Wake` is a `Mutex` and a `Condvar`: `notify` sets the flag and + `notify_one`s; `wait` waits with `wait_timeout_while(guard, timeout, |woken| !*woken)`, which + returns at once if the flag is already set (a `SessionUpdated` that came while the worker was + busy is not lost) and then clears it. +- **`Duration::saturating_mul(2)`** doubles a wait without overflow; `.min(max)` caps it. +- **The given tests.** + - `honcho_worker.rs` (12 tests): the limits and `parts`; a turn sent once as one batch, + exactly, then idle; parts over 25,000 characters; both texts neutralised (a tag from `TAGS`); + a new speaker's turn to `mm-a-2`; an outage (backoffs 1, 2, 4, 8, 8, 8 ms with the tests' + `FAST` waits), its two lines, the sends in order after it, a new outage from the first wait + again; a 422 marked failed and the next sent; a 401 stopping one workspace only; a 403 + ending an outage; a turn gone from the log, or without its reply, marked failed; a turn too + long for a batch. + - `honcho_worker_more.rs` (10 tests): the default waits; undecided turns decided when idle + and after an outage; a restart sending nothing again and resuming only the sessions the + ledger knows; a running turn never decided; a conclusion left waiting; a ledger that cannot + be written stopping the worker; no log line with a turn's text; `memoryd` sending a reported + turn at once (woken, not after 30 s); a key that cannot be read stopping the start and the + process. + - `support/worker.rs`: a home, a `TurnSync`, `brokerd` as a closure, collected log lines, and + a worker against the fake Honcho with short waits. +- The calls the bodies use: + + | Call | Signature, and what it gives | + |---|---| + | `Mutex::lock` | `m.lock().unwrap_or_else(\|p\| p.into_inner())`: the guard, even after a panic elsewhere | + | `Condvar::notify_one`, `wait_timeout_while` | `cv.wait_timeout_while(guard, timeout, \|v\| cond) -> LockResult<(MutexGuard, WaitTimeoutResult)>` | + | `str::chars`, `slice::chunks` | `text.chars().collect::>()`; `chars.chunks(n)`: slices of at most `n` | + | `neutralise` | `bxcore::sanitize::neutralise(text: &str) -> Cow<'_, str>` | + | `MessageCreate` | `roncho::MessageCreate::new(content, peer_id).with_metadata(Map)` | + | `Ledger::outbox`, `decisions` | tasks 07 and 10: `Vec<(Item, Action)>` in decision order; an iterator of `(&Item, &Action)` | + | `honcho_session` | task 10: `honcho_session(ledger.decisions(), session, turn) -> Option` | + | `read_log`, `turn_records`, `user_text`, `final_reply` | tasks 09 and 08 | + | `TurnSync::retry`, `append` | this task's edits T2 and T3 | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/memoryd/src/honcho/decide.rs` + +**D1.** In `decide.rs`, the line + +```rust +fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) { +``` + +becomes + +```rust +pub fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) { +``` + +**D2.** The line + +```rust +fn final_reply(turn: &[LogRecord]) -> Option<&str> { +``` + +becomes + +```rust +pub fn final_reply(turn: &[LogRecord]) -> Option<&str> { +``` + +### `crates/memoryd/src/honcho/mod.rs` + +**H1.** The line + +```rust +pub mod turns; +``` + +becomes + +```rust +pub mod turns; +pub mod worker; +``` + +### `crates/memoryd/src/honcho/turns.rs` + +**T1.** In `impl TurnSync`, the line + +```rust + /// The sessions that still have undecided turns, with the highest turn reported for each. +``` + +becomes + +```rust + /// The directory of `loopd`'s session logs. + pub fn sessions(&self) -> &Path { + &self.sessions + } + + /// The sessions that still have undecided turns, with the highest turn reported for each. +``` + +**T2.** The lines + +```rust + /// `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. +``` + +become + +```rust + /// After a restart: each session with a decided turn in the ledger waits again, at the turns + /// its log holds, so that a turn left undecided before the restart is decided at the next + /// walk (`decidable` holds back one still running). A session with no decided turn is left + /// to its next report: sessions from before Honcho was turned on are never walked. + pub fn resume(&mut self) { + let mut sessions: Vec = Vec::new(); + for (item, _) in self.ledger.decisions() { + if let Item::Turn { session, .. } = item + && !sessions.contains(session) + { + sessions.push(session.clone()); + } + } + for session in sessions { + if let Ok(records) = read_log(&log_path(&self.sessions, &session)) { + self.reported.insert( + session.as_str().to_string(), + (session.clone(), turns(&records)), + ); + } + } + } + + /// Append `record` to the ledger (the worker's `Sent` and `Failed`). + pub fn append(&mut self, record: Record) -> Result<(), ledger::LedgerError> { + self.ledger.append(record) + } + + /// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk. +``` + +**T3.** In `updated`, the lines + +```rust + entry.1 = entry.1.max(turn); + for (session, turn) in self.waiting() { +``` + +become + +```rust + entry.1 = entry.1.max(turn); + self.retry(ask, now, log); + } + + /// Walk every session with undecided turns, in the order of their ids, until one stops the + /// walk: at a report, and from the worker when it has nothing to send (M5c spec, section 5). + pub fn retry(&mut self, ask: Ask<'_>, now: Timestamp, log: &dyn Fn(&str)) { + for (session, turn) in self.waiting() { +``` + +### `crates/memoryd/src/serve.rs` + +**S1.** The lines + +```rust +use crate::honcho::ledger::{self, Ledger, LedgerError}; +use crate::honcho::state; +use crate::honcho::turns::{self, TurnSync}; +``` + +become + +```rust +use crate::honcho::ledger::{self, LedgerError}; +use crate::honcho::state; +use crate::honcho::turns::TurnSync; +use crate::honcho::worker::{self, Wake}; +``` + +**S2.** In `pub struct Memoryd`, the lines + +```rust + honcho: Option>, + log: Log, +} +``` + +become + +```rust + honcho: Option>>, + /// Wakes the Honcho worker when a turn has been decided. + wake: Arc, + log: Log, +} +``` + +**S3.** In `pub enum StartError`, the lines + +```rust + Failed(String), + Ledger(LedgerError), +} +``` + +become + +```rust + Failed(String), + Ledger(LedgerError), + /// A Honcho key that cannot be read: `bxcore::secrets`' line, with its own pointer. + Secret(String), +} +``` + +**S4.** In `impl std::fmt::Display for StartError`, the line + +```rust + StartError::Failed(why) => write!(f, "{why}"), +``` + +becomes + +```rust + StartError::Failed(why) | StartError::Secret(why) => write!(f, "{why}"), +``` + +**S5.** The lines + +```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, + }; +``` + +become + +```rust +/// With `[honcho]`, read the ledger through and the keys first. Then create what is missing +/// under `memory/`, open the index and catch it up, bind the three sockets, and start the Honcho +/// worker. The error names what failed. +pub fn start(config: Config, log: Log) -> Result { + let opened = match config.honcho_config()? { + Some(honcho) => Some(worker::open(&config, &honcho, &log)?), + None => None, + }; +``` + +**S6.** In `start`, the lines + +```rust + (admin_listener, Which::Admin), + ]; +``` + +become + +```rust + (admin_listener, Which::Admin), + ]; + let wake = Arc::new(Wake::default()); + let honcho = match opened { + Some(opened) => { + let sync = Arc::clone(&opened.sync); + worker::spawn(opened, &config, Arc::clone(&wake), Arc::clone(&log))?; + Some(sync) + } + None => None, + }; +``` + +**S7.** In `start`, where `Memoryd` is built, the lines + +```rust + honcho, + log, +``` + +become + +```rust + honcho, + wake, + log, +``` + +**S8.** In `decide_turns`, the lines + +```rust + Timestamp::now(), + &*self.log, + ); + } +``` + +become + +```rust + Timestamp::now(), + &*self.log, + ); + drop(sync); + self.wake.notify(); + } +``` + +### `crates/memoryd/src/main.rs` + +**M1.** The lines + +```rust + // The ledger's line names its own pointer, `honcho-ledger-damaged`. + Err(e @ serve::StartError::Ledger(_)) => { +``` + +become + +```rust + // The ledger's and a key's lines name their own pointers. + Err(e @ (serve::StartError::Ledger(_) | serve::StartError::Secret(_))) => { +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, four commands: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker.rs crates/memoryd/tests/ + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.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/worker.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_worker`, 3 errors: `E0432` + (``unresolved import `memoryd::honcho::worker` ``) in the test file and in + `support/worker.rs`, and `E0599` (``no method named `resume` found for struct `TurnSync` ``). + `honcho_worker_more`, 4 errors: the same three, and `E0599` (``no variant, associated function, + or constant named `Secret` found for enum `StartError` ``). Every other target compiles, and + `cargo test -p memoryd --test honcho_serve` passes its 6 tests already. +- [ ] **3. The module.** Make edits D1, D2, H1, T1 to T3, S1 to S8 and M1, then copy the + skeleton: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/worker.rs crates/memoryd/src/honcho/ + ``` + + `cargo check -p memoryd`: no error, and 21 warnings, all from the unfilled bodies: six + `unused import` lines (`bxcore::sanitize::neutralise`; `Map` and `json`; `final_reply`, + `turn_records` and `user_text`; `Action` and `self`; `honcho_session`; `read_log`), thirteen + `unused variable` (`text` and `turn` twice each, and `agent`, `error`, `messages`, `path`, + `record`, `reply`, `session`, `speaker`, `timeout` once each), ``fields `woken` and `ready` are + never read`` and ``fields `ask` and `outage` are never read``. +- [ ] **4. Fill the eleven bodies of `worker.rs`, one at a time, in the order of the file**, each + from its comment, and run `cargo check -p memoryd` after each: `notify`, `wait`, `parts`, + `read_turn`, `batch`, `ids`, `next`, `unreachable`, `recovered`, `retry`, `record`. 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. After the last, + `cargo check -p memoryd` prints no warning. +- [ ] **5. The tests.** With a shell timeout of 600000 ms, and every test passes (12, 10, 6, and + the existing ones unchanged): + + ```sh + cargo test -p memoryd --test honcho_worker --test honcho_worker_more --test honcho_serve \ + --test honcho_turns --test honcho_client --test honcho_start --test serve + ``` +- [ ] **6. Check.** Three commands: + - `grep -n 'thread::spawn\|Builder::new()' crates/memoryd/src/honcho/worker.rs` prints one + line, in `spawn`: the worker is the one new thread. + - `grep -rn 'runbook.md#honcho-unreachable\|runbook.md#honcho-send-refused\|runbook.md#honcho-key-refused' crates/memoryd/src` + prints three lines, the constants in `worker.rs`. + - `git diff --stat -- crates/bxcore crates/loopd crates/brokerd crates/gatewayd crates/bxctl + Cargo.toml Cargo.lock crates/memoryd/Cargo.toml` prints nothing. +- [ ] **7. The runbook.** From the repository root, three commands: + + ```sh + { echo; cat docs/plans/M5c1/files/runbook-honcho-unreachable.md; } >> docs/runbook.md + { echo; cat docs/plans/M5c1/files/runbook-honcho-send-refused.md; } >> docs/runbook.md + { echo; cat docs/plans/M5c1/files/runbook-honcho-key-refused.md; } >> docs/runbook.md + ``` + + Then `grep -c '^## honcho-' docs/runbook.md` prints `5`. +- [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1300 tests: 1278 before this task, plus 22 new ones (12 in + `honcho_worker.rs`, 10 in `honcho_worker_more.rs`). +- [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/11-memoryd-honcho-worker`. `git status --short` lists `decide.rs`, `honcho/mod.rs`, + `turns.rs`, `serve.rs`, `main.rs`, `tests/honcho_serve.rs`, `docs/runbook.md` and your log as + modified, and `worker.rs`, the two new test files and `tests/support/worker.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 1300 tests. +- `git diff HEAD~1 --stat` lists only the files of step 9. + +## Stop and report if + +- A "before" block of D1 to M1 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 the skeleton does not fit the code under it, or disagrees with this task's tables. + Quote both and stop. +- After step 4 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 skeleton's given code: stop, and give the failing + tests' names and their messages. +- A test seems to need a real Honcho, a new fixture or a longer wait. It does not; stop and say + why you think so. +- The fix seems to need a change anywhere else: `client.rs`, `ledger.rs`, another crate, a + second thread. 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 1300. + +## What the given tests pin + +Each of these changes to the finished code makes at least one given test fail (36 tried; one +cannot change behaviour): a part of 25,001, or parts cut by bytes; no neutralising; the agent +first, or the reply from the speaker; the wrong turn in the metadata; no batch limit; no session +made first; the messages sent to the Boxmaker id instead of the Honcho session; the ids dropped; +the wait not doubled or not capped; the outage logged at every try, its end not logged, no +retry after it, or a refused key not ending it; a refused key not remembered, its workspace not +skipped, or its turns failed; a refused send not recorded; a local failure not recorded; the +newest turn first; no retry when idle, or idle after a retry that decided something; a ledger +that cannot be written ignored, or `stopped` ignored; the wake doing nothing; a turn without its +reply sent anyway; every session in `sessions/` resumed, or none; the retry treated as a report; +no worker started, or not woken; a key's error taken for a start failure, or printed with the +start pointer. + +## After this task + +- Task 13 sends the other actions (conclusions, their deletion, fleet messages and retractions) + from the same worker. +- Task 15 adds the cutoff: turns from before Honcho was turned on are skipped. diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/worker.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/worker.rs new file mode 100644 index 0000000..1901e46 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/worker.rs @@ -0,0 +1,403 @@ +//! The worker: one thread that sends what the ledger says is waiting, oldest first (M5c spec, +//! section 5). A turn is read again from `loopd`'s log, neutralised, cut into parts Honcho takes, +//! and added to its Honcho session as one batch; the ids Honcho gives go in the ledger. An outage +//! is waited out with a doubling wait and logged when it starts and when it ends; a refused send is +//! marked failed; a refused key stops that workspace until `memoryd` restarts. When nothing waits, +//! the worker decides what `brokerd` could not answer for before, then sleeps until a turn ends. + +use std::collections::HashSet; +use std::path::Path; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +use bxcore::sanitize::neutralise; +use bxcore::{SessionId, SessionStateReply, Timestamp}; +use roncho::MessageCreate; +use serde_json::{Map, json}; + +use crate::config::Config; +use crate::honcho::client::{Clients, HonchoError, Timeouts, classify}; +use crate::honcho::config::HonchoConfig; +use crate::honcho::decide::{final_reply, turn_records, user_text}; +use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace}; +use crate::honcho::naming::{HonchoSession, honcho_session}; +use crate::honcho::state; +use crate::honcho::turns::{LEDGER, TurnSync, log_path, read_log}; +use crate::serve::{Log, StartError}; + +/// The most characters Honcho takes in one message (recorded: 25,000 is taken, 25,001 refused). +pub const MAX_PART: usize = 25_000; +/// The most messages the worker puts in one batch. A turn that needs more is marked failed. +pub const MAX_BATCH: usize = 100; + +pub const UNREACHABLE: &str = "see docs/runbook.md#honcho-unreachable"; +pub const SEND_REFUSED: &str = "see docs/runbook.md#honcho-send-refused"; +pub const KEY_REFUSED: &str = "see docs/runbook.md#honcho-key-refused"; +pub const LOG_DAMAGED: &str = "see docs/runbook.md#session-log-damaged"; + +/// The worker's waits. Tests shorten them. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Tuning { + /// The first wait in an outage; each next one doubles, up to `backoff_max`. + pub backoff: Duration, + pub backoff_max: Duration, + /// How long the worker sleeps with nothing to send, unless a turn ends first. + pub idle: Duration, +} + +impl Tuning { + pub const DEFAULT: Tuning = Tuning { + backoff: Duration::from_secs(5), + backoff_max: Duration::from_secs(600), + idle: Duration::from_secs(30), + }; +} + +/// Wakes the worker from its idle sleep when a turn ends. +#[derive(Debug, Default)] +pub struct Wake { + woken: Mutex, + ready: Condvar, +} + +impl Wake { + /// Wake the worker if it sleeps, or make its next sleep return at once. + pub fn notify(&self) { + // `let mut woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());`, then + // `*woken = true;` and `self.ready.notify_one();`. + todo!() + } + + /// Sleep until `notify`, or at most `timeout`. + pub fn wait(&self, timeout: Duration) { + // `let woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());`. Then + // `let (mut woken, _) = self.ready.wait_timeout_while(woken, timeout, |woken| !*woken)` + // `.unwrap_or_else(|p| p.into_inner());`, and `*woken = false;` (the wake is used up). + todo!() + } +} + +/// What one step of the worker did, and so what the loop does next. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Step { + /// A turn was sent, or marked failed: go on at once. + Done, + /// Honcho is unreachable: wait this long, then try the same turn again. + Backoff(Duration), + /// Nothing to send, and the retry decided nothing new: sleep. + Idle, + /// The ledger cannot be written: the worker stops. + Stopped, +} + +/// `brokerd`'s answer for a session, for the retry. +pub type AskBox = Box Result + Send>; + +/// Why a turn could not be sent. +enum Unsent { + Honcho(HonchoError), + /// The turn cannot be read from the log any more, or is too long for one batch. + Local(&'static str), +} + +/// A turn to send: its item, workspace, speaker and Honcho session. +struct Next { + item: Item, + session: SessionId, + turn: u64, + workspace: Workspace, + speaker: String, + honcho: Option, +} + +pub struct Worker { + sync: Arc>, + clients: Clients, + wake: Arc, + ask: AskBox, + log: Log, + tuning: Tuning, + agent: String, + /// In an outage: the next wait. + outage: Option, + /// Workspaces whose key was refused: nothing more is sent to them until restart. + refused: HashSet, + stopped: bool, +} + +/// `text` in parts of at most `MAX_PART` characters, in order. Empty text has no part. +pub fn parts(text: &str) -> Vec { + // `let chars: Vec = text.chars().collect();` then + // `chars.chunks(MAX_PART).map(|chunk| chunk.iter().collect()).collect()`. + todo!() +} + +/// Turn `turn`'s user text and final reply, read again from the log at `path`. `None` when the log +/// does not read, or has no such turn, or the turn no reply. +pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)> { + // Each step with `?`: `read_log(path).ok()?` into `records`; then + // `let records = turn_records(&records, turn)?;`; `let (text, _) = user_text(records);`; + // `let reply = final_reply(records)?;`. Then `Some((text.to_string(), reply.to_string()))`. + todo!() +} + +/// A turn's messages, in one batch: the speaker's text, then the agent's reply, each neutralised +/// and in parts of at most `MAX_PART` characters, each part with the turn's metadata. +pub fn batch( + text: &str, + reply: &str, + speaker: &str, + agent: &str, + session: &SessionId, + turn: u64, +) -> Vec { + // A `Map` `metadata` with `"agent"` -> `json!(agent)`, `"boxmaker_session"` -> + // `json!(session.as_str())`, `"turn"` -> `json!(turn)` (in this order). Then for each + // `(content, peer)` of `[(text, speaker), (reply, agent)]`, for each `part` of + // `parts(&neutralise(content))`, push + // `MessageCreate::new(part, peer).with_metadata(metadata.clone())`. Return the messages. + todo!() +} + +/// The ids of the messages Honcho made, in order. +fn ids(messages: &[roncho::Message]) -> Vec { + // `messages.iter().map(|m| m.id.clone()).collect()`. + todo!() +} + +impl Worker { + /// `agent` is `[honcho] agent`. `sync` and `wake` are shared with the socket threads. + pub fn new( + sync: Arc>, + clients: Clients, + wake: Arc, + ask: AskBox, + log: Log, + tuning: Tuning, + agent: &str, + ) -> Worker { + Worker { + sync, + clients, + wake, + ask, + log, + tuning, + agent: agent.to_string(), + outage: None, + refused: HashSet::new(), + stopped: false, + } + } + + /// Step until the ledger cannot be written; wait as each step says. + pub fn run(mut self) { + loop { + match self.step() { + Step::Done => {} + Step::Backoff(wait) => std::thread::sleep(wait), + Step::Idle => self.wake.wait(self.tuning.idle), + Step::Stopped => return, + } + } + } + + /// Send the oldest waiting turn, or, with none, retry the undecided ones. + pub fn step(&mut self) -> Step { + if self.stopped { + return Step::Stopped; + } + let Some(next) = self.next() else { + self.retry(); + return match self.next() { + Some(_) => Step::Done, + None => Step::Idle, + }; + }; + match self.send(&next) { + Ok(sent) => { + self.recovered(); + self.record(Record::Sent { + time: Timestamp::now(), + item: next.item, + ids: sent, + }) + } + Err(Unsent::Honcho(e @ HonchoError::Unreachable { .. })) => self.unreachable(&e), + Err(Unsent::Honcho(HonchoError::KeyRefused { status })) => { + self.recovered(); + self.refused.insert(next.workspace); + (self.log)(&format!( + "memoryd: Honcho refused the key of workspace {} (HTTP {status}); nothing more is sent to it until memoryd restarts; {KEY_REFUSED}", + self.clients.client(next.workspace).workspace_id() + )); + Step::Done + } + Err(Unsent::Honcho(e)) => { + self.recovered(); + (self.log)(&format!( + "memoryd: turn {} of session {} was not sent: {e}; it is marked failed; {SEND_REFUSED}", + next.turn, + next.session.as_str() + )); + self.record(Record::Failed { + time: Timestamp::now(), + item: next.item, + status: e.status(), + kind: e.kind().to_string(), + }) + } + Err(Unsent::Local(kind)) => { + let pointer = if kind == "log" { + LOG_DAMAGED + } else { + SEND_REFUSED + }; + (self.log)(&format!( + "memoryd: turn {} of session {} cannot be sent ({kind}); it is marked failed; {pointer}", + next.turn, + next.session.as_str() + )); + self.record(Record::Failed { + time: Timestamp::now(), + item: next.item, + status: None, + kind: kind.to_string(), + }) + } + } + } + + /// The oldest waiting turn whose workspace's key was not refused. Other actions wait. + fn next(&self) -> Option { + // Lock `self.sync` (`unwrap_or_else(|p| p.into_inner())`), `let ledger = + // sync.ledger();`, then `ledger.outbox().into_iter().find_map(|(item, action)| { .. })`: + // `None` unless `item` is `Item::Turn { session, turn }` (by `&item`) and `action` is + // `Action::Turn { workspace, speaker }` (each by `let .. else`), and `None` if + // `self.refused.contains(&workspace)`; else `Some(Next { honcho: + // honcho_session(ledger.decisions(), session, *turn), session: session.clone(), turn: + // *turn, item: item.clone(), workspace, speaker })`. + todo!() + } + + /// Read the turn again, make sure of its Honcho session, and add its messages as one batch. + fn send(&mut self, next: &Next) -> Result, Unsent> { + let sessions = self + .sync + .lock() + .unwrap_or_else(|p| p.into_inner()) + .sessions() + .to_path_buf(); + let (text, reply) = read_turn(&log_path(&sessions, &next.session), next.turn) + .ok_or(Unsent::Local("log"))?; + let honcho = next.honcho.as_ref().ok_or(Unsent::Local("log"))?; + let messages = batch( + &text, + &reply, + &next.speaker, + &self.agent, + &next.session, + next.turn, + ); + if messages.len() > MAX_BATCH { + return Err(Unsent::Local("too-long")); + } + self.clients + .ensure_session(next.workspace, honcho) + .map_err(Unsent::Honcho)?; + let client = self.clients.client(next.workspace); + match client.add_messages(&honcho.name, &messages) { + Ok(made) => Ok(ids(&made)), + Err(e) => Err(Unsent::Honcho(classify(&e))), + } + } + + /// Honcho did not answer: log the outage's start once, and wait longer each time. + fn unreachable(&mut self, error: &HonchoError) -> Step { + // `let wait = match self.outage`: `Some(wait)` -> `wait`; `None` -> log + // `format!("memoryd: {error}; sends wait and are retried; {UNREACHABLE}")`, then + // `self.tuning.backoff`. Then + // `self.outage = Some(wait.saturating_mul(2).min(self.tuning.backoff_max));` and + // `Step::Backoff(wait)`. + todo!() + } + + /// Honcho answered: if that ends an outage, log it once, and decide what waited meanwhile. + fn recovered(&mut self) { + // If `self.outage.take().is_some()`: log exactly + // `memoryd: Honcho answers again; waiting sends go on`, then `self.retry();`. + todo!() + } + + /// Task 09's walk, for the turns `brokerd` could not answer for. + fn retry(&mut self) { + // Lock `self.sync` as in `next` (as `mut sync`), then + // `sync.retry(&*self.ask, Timestamp::now(), &*self.log);`. + todo!() + } + + /// Append the worker's record. A ledger that cannot be written stops the worker: the send + /// would be repeated at every step. + fn record(&mut self, record: Record) -> Step { + // `let result = + // self.sync.lock().unwrap_or_else(|p| p.into_inner()).append(record);`. `Ok(())` -> + // `Step::Done`. `Err(e)` -> log `format!("memoryd: {e}; the Honcho worker stops until + // memoryd restarts; {}", ledger::RUNBOOK)`, then `self.stopped = true;`, `Step::Stopped`. + todo!() + } +} + +/// With `[honcho]`, at start: the ledger read through, the sessions that wait again, and the +/// clients with their keys read. Nothing is bound or started yet. +pub struct Opened { + pub sync: Arc>, + clients: Clients, + agent: String, +} + +/// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the +/// clients (their key files' warnings are logged). A key that cannot be read is `Secret`. +pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result { + let path = config.memory_dir().join(LEDGER); + let ledger = Ledger::open(&path).map_err(StartError::Ledger)?; + if ledger.repaired() { + (log)(&format!( + "memoryd: {}: a last line cut off by a crash was dropped", + path.display() + )); + } + let mut sync = TurnSync::new(ledger, &honcho.agent, &config.sessions_dir()); + sync.resume(); + let env = |name: &str| std::env::var_os(name); + let (clients, warnings) = + Clients::new(honcho, &env, Timeouts::WRITES).map_err(StartError::Secret)?; + for warning in warnings { + (log)(&warning); + } + Ok(Opened { + sync: Arc::new(Mutex::new(sync)), + clients, + agent: honcho.agent.clone(), + }) +} + +/// Start the worker's thread, asking `brokerd` on `config`'s state socket. +pub fn spawn(opened: Opened, config: &Config, wake: Arc, log: Log) -> Result<(), String> { + let socket = config.state_socket(); + let ask: AskBox = + Box::new(move |session: &SessionId| state::ask(&socket, session, state::WAIT)); + let worker = Worker::new( + opened.sync, + opened.clients, + wake, + ask, + log, + Tuning::DEFAULT, + &opened.agent, + ); + std::thread::Builder::new() + .name("honcho".to_string()) + .spawn(move || worker.run()) + .map(|_| ()) + .map_err(|e| format!("cannot start the Honcho worker: {e}")) +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs index 2392f75..34e0b99 100644 --- a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs @@ -2,7 +2,8 @@ //! `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. +//! deadline. (From M5c-1 task 11 a Honcho worker runs beside it; nothing here waits for it, and the +//! cut-off line is a skipped turn, so that it has nothing to send.) Do not edit. #[path = "support/brokerd.rs"] mod brokerd; @@ -41,6 +42,9 @@ 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"}}"#; +/// A turn decided to go nowhere: the worker has nothing to send for it. +const SKIPPED: &str = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"turn","session":"mm-a","turn":1},"action":{"type":"skip","why":"the turn has no single speaker"}}"#; + struct Home { config: Config, text: String, @@ -263,7 +267,7 @@ fn a_damaged_ledger_stops_the_start() { 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(); + std::fs::write(home.ledger(), format!("{SKIPPED}\n{{\"type\":\"se")).unwrap(); let lines = home.serve(); assert_eq!( *lines.lock().unwrap(), @@ -272,7 +276,7 @@ fn a_cut_off_last_line_is_dropped_and_logged() { home.ledger().display() )] ); - assert_eq!(home.ledger_text(), format!("{DECIDED}\n")); + assert_eq!(home.ledger_text(), format!("{SKIPPED}\n")); } /// A `memoryd` process, killed and waited for when dropped: declared after its `Home`, so it diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs-task09 b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs-task09 new file mode 100644 index 0000000..2392f75 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs-task09 @@ -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_worker.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker.rs new file mode 100644 index 0000000..c2e5c44 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker.rs @@ -0,0 +1,380 @@ +//! The Honcho worker, stepped by hand against the fake Honcho (M5c spec, section 5): a turn sent +//! once as one batch, the speaker's parts then the agent's, neutralised, in parts of 25,000 +//! characters, to the `-2` session for a new speaker; an outage waited out with a doubling wait +//! and logged twice in all; a refused send marked failed; a refused key stopping its workspace +//! only; a turn the log no longer holds marked failed; other actions left waiting. The waits and +//! a restart are in `honcho_worker_more.rs`. Do not edit. + +#[path = "support/honcho.rs"] +mod honcho; +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; +#[path = "support/worker.rs"] +mod worker; + +use std::time::Duration; + +use bxcore::sanitize::TAGS; +use bxcore::{DataClass, SessionStateReply}; +use honcho::{FakeHoncho, Mode, Request}; +use logs::assistant; +use memoryd::honcho::ledger::{Item, Record}; +use memoryd::honcho::worker::{MAX_BATCH, MAX_PART, Step, parts}; +use serde_json::{Value, json}; +use turns::{sid, spoken}; +use worker::{FAST, Home}; + +/// The ids fixture 08 answers with. +const IDS: [&str; 2] = ["7HqDWzXcKAdw9tnElCNFg", "mjsyhwuGsXTh1_zG_tWRx"]; + +fn message_posts(fake: &FakeHoncho) -> Vec { + fake.requests() + .into_iter() + .filter(|r| r.path.ends_with("/messages")) + .collect() +} + +fn turn(text: &str, speaker: &str, reply: &str) -> Vec { + vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])] +} + +fn meta(session: &str, turn: u64) -> Value { + json!({ "agent": "boxmaker0", "boxmaker_session": session, "turn": turn }) +} + +fn sent(session: &str, turn: u64) -> impl Fn(&Record) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + move |r| matches!(r, Record::Sent { item: i, .. } if *i == item) +} + +#[test] +fn the_limits() { + assert_eq!(MAX_PART, 25_000); + assert_eq!(MAX_BATCH, 100); + assert!(parts("").is_empty()); + assert_eq!(parts("abc"), vec!["abc".to_string()]); + let long = "é".repeat(MAX_PART); + assert_eq!(parts(&long), vec![long.clone()], "characters, not bytes"); + let over = format!("{long}xy"); + assert_eq!(parts(&over), vec![long, "xy".to_string()]); +} + +#[test] +fn a_turn_is_sent_once_as_one_batch() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-once"); + home.log("mm-a", &[turn("hello", "kyle", "hi")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let posts = message_posts(&fake); + assert_eq!(posts.len(), 1); + assert_eq!( + posts[0].path, + "/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages" + ); + assert_eq!( + posts[0].body, + Some(json!({ "messages": [ + { "content": "hello", "peer_id": "kyle", "metadata": meta("mm-a", 1) }, + { "content": "hi", "peer_id": "boxmaker0", "metadata": meta("mm-a", 1) }, + ] })) + ); + // The session was made first, with both peers. + assert_eq!(fake.requests().len(), 5); + assert!(fake.requests()[3].path.ends_with("/sessions")); + let records = home.records(); + assert_eq!(records.len(), 2); + assert!(matches!(&records[1], Record::Sent { ids, .. } if *ids == IDS)); + fake.clear(); + assert_eq!(w.step(), Step::Idle); + assert!(fake.requests().is_empty()); + assert!(home.lines().is_empty(), "{:?}", home.lines()); +} + +#[test] +fn a_text_over_the_limit_is_sent_in_parts() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-parts"); + let text = "a".repeat(MAX_PART + 1); + let reply = "é".repeat(MAX_PART); + home.log("mm-a", &[turn(&text, "kyle", &reply)]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let body = message_posts(&fake)[0].body.clone().unwrap(); + let got: Vec<(usize, String)> = body["messages"] + .as_array() + .unwrap() + .iter() + .map(|m| { + ( + m["content"].as_str().unwrap().chars().count(), + m["peer_id"].as_str().unwrap().to_string(), + ) + }) + .collect(); + assert_eq!( + got, + vec![ + (MAX_PART, "kyle".to_string()), + (1, "kyle".to_string()), + (MAX_PART, "boxmaker0".to_string()), + ] + ); +} + +#[test] +fn both_texts_are_neutralised() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-neutral"); + let text = format!("a{}b", TAGS[0].text); + let reply = format!("c{}d", TAGS[1].text); + home.log("mm-a", &[turn(&text, "kyle", &reply)]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let body = message_posts(&fake)[0].body.clone().unwrap(); + let contents: Vec = body["messages"] + .as_array() + .unwrap() + .iter() + .map(|m| m["content"].as_str().unwrap().to_string()) + .collect(); + assert_eq!(contents.len(), 2); + for (content, tag) in contents.iter().zip([TAGS[0].text, TAGS[1].text]) { + assert!(!content.contains(tag)); + assert!(content.contains("<"), "not neutralised"); + } +} + +#[test] +fn a_new_speaker_goes_to_the_next_session() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-second"); + home.log( + "mm-a", + &[turn("one", "kyle", "a"), turn("two", "ana_b", "b")], + ); + home.report("mm-a", 2); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert_eq!(w.step(), Step::Done); + let posts = message_posts(&fake); + assert_eq!(posts.len(), 2); + assert!(posts[0].path.ends_with("/sessions/mm-a/messages")); + assert!(posts[1].path.ends_with("/sessions/mm-a-2/messages")); + let second = fake + .requests() + .into_iter() + .find(|r| r.path.ends_with("/sessions") && r.body.as_ref().unwrap()["id"] == "mm-a-2") + .unwrap(); + let peers = second.body.unwrap()["peers"].clone(); + assert_eq!(peers["kyle"]["observe_me"], true); + assert_eq!(peers["ana_b"]["observe_me"], true); + assert_eq!(peers["boxmaker0"]["observe_me"], false); +} + +#[test] +fn an_outage_is_waited_out_and_logged_twice_in_all() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-outage"); + home.log("mm-a", &[turn("private-alpha", "kyle", "a")]); + home.log("mm-b", &[turn("private-beta", "kyle", "b")]); + home.report("mm-a", 1); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(503)); + let waits: Vec = (0..6).map(|_| w.step()).collect(); + let ms = |n| Step::Backoff(Duration::from_millis(n)); + assert_eq!(waits, vec![ms(1), ms(2), ms(4), ms(8), ms(8), ms(8)]); + assert_eq!(home.lines().len(), 1, "{:?}", home.lines()); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + assert_eq!(w.step(), Step::Done); + let lines = home.lines(); + assert_eq!(lines.len(), 2, "{lines:?}"); + assert_eq!( + lines[0], + "memoryd: Honcho is unreachable (server, HTTP 503); sends wait and are retried; see docs/runbook.md#honcho-unreachable" + ); + assert_eq!( + lines[1], + "memoryd: Honcho answers again; waiting sends go on" + ); + // In order: mm-a, then mm-b. + let records = home.records(); + let a = records.iter().position(sent("mm-a", 1)).unwrap(); + let b = records.iter().position(sent("mm-b", 1)).unwrap(); + assert!(a < b); + // A new outage starts from the first wait again. + home.log("mm-c", &[turn("c", "kyle", "c")]); + home.report("mm-c", 1); + fake.set(Mode::Hang); + assert_eq!(w.step(), ms(1)); + assert!(home.lines()[2].contains("(timeout)"), "{:?}", home.lines()); +} + +#[test] +fn a_refused_send_is_marked_failed_and_the_next_goes_on() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-refused"); + home.log("mm-a", &[turn("private-alpha", "kyle", "a")]); + home.log("mm-b", &[turn("private-beta", "kyle", "b")]); + home.report("mm-a", 1); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(422)); + assert_eq!(w.step(), Step::Done); + let failed = home.records().into_iter().find_map(|r| match r { + Record::Failed { + item, status, kind, .. + } => Some((item, status, kind)), + _ => None, + }); + assert_eq!( + failed, + Some(( + Item::Turn { + session: sid("mm-a"), + turn: 1 + }, + Some(422), + "status".to_string() + )) + ); + let lines = home.lines(); + assert_eq!( + lines, + vec![ + "memoryd: turn 1 of session mm-a was not sent: Honcho refused the request (status, HTTP 422); it is marked failed; see docs/runbook.md#honcho-send-refused" + .to_string() + ] + ); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + assert!(home.records().iter().any(sent("mm-b", 1))); + assert_eq!(w.step(), Step::Idle, "the failed turn is not tried again"); +} + +#[test] +fn a_refused_key_stops_its_workspace_only() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-key"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.log("mm-b", &[turn("b", "kyle", "b")]); + home.report("mm-a", 1); + *home.answer.lock().unwrap() = Ok(SessionStateReply { + class: DataClass::Private, + untrusted: true, + }); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(401)); + assert_eq!(w.step(), Step::Done); + assert_eq!( + home.lines(), + vec![ + "memoryd: Honcho refused the key of workspace boxmaker0-trusted (HTTP 401); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused" + .to_string() + ] + ); + fake.set(Mode::Fixtures); + fake.clear(); + assert_eq!(w.step(), Step::Done); + assert!(home.records().iter().any(sent("mm-b", 1))); + assert!( + fake.requests() + .iter() + .all(|r| !r.path.contains("boxmaker0-trusted")) + ); + assert_eq!(w.step(), Step::Idle); + // The trusted turn waits: neither sent nor failed. + let sync = home.sync.lock().unwrap(); + assert_eq!(sync.ledger().outbox().len(), 1); + assert_eq!(sync.ledger().counts().failed, 0); +} + +#[test] +fn a_turn_the_log_no_longer_holds_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-gone"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let path = home.dir.path().join("sessions/mm-a/0.jsonl"); + std::fs::write(&path, "not json\n").unwrap(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "log" + )); + assert_eq!( + home.lines(), + vec![ + "memoryd: turn 1 of session mm-a cannot be sent (log); it is marked failed; see docs/runbook.md#session-log-damaged" + .to_string() + ] + ); + assert!(message_posts(&fake).is_empty()); +} + +#[test] +fn a_turn_without_its_reply_any_more_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-noreply"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + // The log now ends in a tool round: the reply is gone. + home.log("mm-a", &[turns::running(Some("kyle"))]); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "log" + )); + assert!(message_posts(&fake).is_empty()); +} + +#[test] +fn a_refused_key_ends_an_outage() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-key-outage"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(503)); + assert!(matches!(w.step(), Step::Backoff(_))); + fake.set(Mode::Status(403)); + assert_eq!(w.step(), Step::Done); + let lines = home.lines(); + assert_eq!(lines.len(), 3, "{lines:?}"); + assert_eq!( + lines[1], + "memoryd: Honcho answers again; waiting sends go on" + ); + assert!(lines[2].ends_with("see docs/runbook.md#honcho-key-refused")); +} + +#[test] +fn a_turn_too_long_for_one_batch_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-long"); + let text = "a".repeat(MAX_PART * MAX_BATCH); + home.log("mm-a", &[turn(&text, "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: None, kind, .. }) if kind == "too-long" + )); + assert!(fake.requests().is_empty()); +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs new file mode 100644 index 0000000..95a0414 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs @@ -0,0 +1,327 @@ +//! The Honcho worker's other duties (M5c spec, section 5): with nothing to send it decides what +//! `brokerd` could not answer for; after a restart nothing is sent again, and a session with a +//! decided turn waits again (a session without one does not); a running turn is never decided; +//! actions of later tasks wait; a ledger that cannot be written stops it; no log line holds a +//! turn's text. Then `memoryd` itself: a reported turn reaches Honcho, woken at once, and a key +//! that cannot be read stops the start. Do not edit. + +#[path = "support/brokerd.rs"] +mod brokerd; +#[path = "support/honcho.rs"] +mod honcho; +#[path = "support/logs.rs"] +mod logs; +#[path = "support/tmp.rs"] +mod tmp; +#[path = "support/turns.rs"] +mod turns; +#[path = "support/worker.rs"] +mod worker; + +use std::process::{Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use brokerd::{Answer, FakeBrokerd}; +use bxcore::{Envelope, Message, SessionId, SessionUpdated, read_frame, write_frame}; +use honcho::{FakeHoncho, Mode}; +use logs::assistant; +use memoryd::config::Config; +use memoryd::honcho::ledger::{Item, Record}; +use memoryd::honcho::worker::{Step, Tuning}; +use memoryd::serve::{self, StartError}; +use tmp::TempDir; +use turns::{clean, running, sid, spoken}; +use worker::{FAST, Home}; + +fn turn(text: &str, speaker: &str, reply: &str) -> Vec { + vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])] +} + +fn messages(fake: &FakeHoncho) -> usize { + fake.requests() + .iter() + .filter(|r| r.path.ends_with("/messages")) + .count() +} + +fn is_sent(records: &[Record], session: &str, turn: u64) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + records + .iter() + .any(|r| matches!(r, Record::Sent { item: i, .. } if *i == item)) +} + +fn decided(records: &[Record], session: &str, turn: u64) -> bool { + let item = Item::Turn { + session: sid(session), + turn, + }; + records + .iter() + .any(|r| matches!(r, Record::Decided { item: i, .. } if *i == item)) +} + +#[test] +fn the_default_waits() { + assert_eq!( + Tuning::DEFAULT, + Tuning { + backoff: Duration::from_secs(5), + backoff_max: Duration::from_secs(600), + idle: Duration::from_secs(30), + } + ); +} + +#[test] +fn with_nothing_to_send_undecided_turns_are_decided() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-retry"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + *home.answer.lock().unwrap() = Err("brokerd is down".to_string()); + home.report("mm-a", 1); + assert!(!decided(&home.records(), "mm-a", 1)); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle, "brokerd still down"); + *home.answer.lock().unwrap() = Ok(clean()); + assert_eq!(w.step(), Step::Done, "decided now"); + assert!(decided(&home.records(), "mm-a", 1)); + assert_eq!(w.step(), Step::Done, "sent"); + assert!(is_sent(&home.records(), "mm-a", 1)); +} + +#[test] +fn an_outage_that_ends_is_followed_by_the_retry() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-after-outage"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.log("mm-b", &[turn("b", "kyle", "b")]); + home.report("mm-a", 1); + *home.answer.lock().unwrap() = Err("brokerd is down".to_string()); + home.report("mm-b", 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(502)); + assert!(matches!(w.step(), Step::Backoff(_))); + *home.answer.lock().unwrap() = Ok(clean()); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done); + // mm-a is sent, and the outage's end decided mm-b before any idle step. + assert!(decided(&home.records(), "mm-b", 1)); +} + +#[test] +fn a_restart_sends_nothing_again_and_resumes_the_sessions_it_knows() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-restart"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + home.worker(&fake, FAST).step(); + assert_eq!(messages(&fake), 1); + // While memoryd is down: mm-a has a second turn, never reported; mm-z, never seen, one. + home.log("mm-a", &[turn("a", "kyle", "a"), turn("b", "kyle", "b")]); + home.log("mm-z", &[turn("z", "kyle", "z")]); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done, "mm-a's second turn decided"); + assert_eq!(w.step(), Step::Done, "and sent"); + assert_eq!(w.step(), Step::Idle); + let records = home.records(); + assert!(is_sent(&records, "mm-a", 2)); + assert!(!decided(&records, "mm-z", 1), "a session never reported"); + assert_eq!(messages(&fake), 2, "turn 1 was sent again"); +} + +#[test] +fn a_restart_never_decides_a_running_turn() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-running"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + home.worker(&fake, FAST).step(); + home.log("mm-a", &[turn("a", "kyle", "a"), running(Some("kyle"))]); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle); + assert!(!decided(&home.records(), "mm-a", 2)); +} + +#[test] +fn actions_of_later_tasks_wait() { + let fake = FakeHoncho::start(); + let mut home = Home::new("worker-later"); + let conclude = 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":"mm-a","text":"a fact"}}"#; + std::fs::write(home.ledger_path(), format!("{conclude}\n")).unwrap(); + home.restart(); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Idle); + assert_eq!(home.records().len(), 1, "the conclusion was failed or sent"); + assert!(fake.requests().is_empty()); +} + +#[test] +fn a_ledger_that_cannot_be_written_stops_the_worker() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-ledger"); + home.log("mm-a", &[turn("a", "kyle", "a")]); + home.report("mm-a", 1); + let mut w = home.worker(&fake, FAST); + let ledger = home.ledger_path(); + let kept = std::fs::read_to_string(&ledger).unwrap(); + std::fs::remove_file(&ledger).unwrap(); + std::fs::create_dir(&ledger).unwrap(); + assert_eq!(w.step(), Step::Stopped); + assert_eq!(w.step(), Step::Stopped); + assert_eq!( + messages(&fake), + 1, + "sent once; its sent line could not be written" + ); + let lines = home.lines(); + assert_eq!(lines.len(), 1, "{lines:?}"); + assert!(lines[0].ends_with( + "; the Honcho worker stops until memoryd restarts; see docs/runbook.md#honcho-ledger-damaged" + )); + std::fs::remove_dir(&ledger).unwrap(); + std::fs::write(&ledger, kept).unwrap(); +} + +#[test] +fn no_log_line_holds_a_turns_text() { + let fake = FakeHoncho::start(); + let home = Home::new("worker-quiet"); + for (session, status) in [("mm-a", 503), ("mm-b", 422), ("mm-c", 401)] { + home.log( + session, + &[turn("private-question", "kyle", "private-answer")], + ); + home.report(session, 1); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(status)); + w.step(); + } + let lines = home.lines(); + assert_eq!(lines.len(), 3, "{lines:?}"); + for line in lines { + assert!(!line.contains("private"), "{line}"); + } +} + +/// `memoryd serve`'s config for `home`, with `[honcho]` at `url`. +fn serve_config(home: &TempDir, url: &str, extra: &str) -> String { + let soul = home.write("etc/soul.md", "You are Boxmaker.\n"); + format!( + "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n[honcho]\nurl = \"{url}\"\nagent = \"boxmaker0\"\nowner = \"kyle\"\ntrusted = \"boxmaker0-trusted\"\nuntrusted = \"boxmaker0-untrusted\"\nfleet = \"fleet\"\n{extra}", + home.path().display(), + soul.display() + ) +} + +#[test] +fn memoryd_sends_a_reported_turn_at_once() { + let fake = FakeHoncho::start(); + let dir = TempDir::new("worker-serve"); + turns::write_turns(dir.path(), "mm-a", &[turn("a", "kyle", "a")]); + let config = Config::parse(&serve_config(&dir, &fake.url, "")).unwrap(); + let _brokerd = FakeBrokerd::start(&config.state_socket(), Answer::State(clean())); + 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(config.clone(), log).unwrap(); + std::thread::spawn(move || running.run()); + // The worker has slept since start (30 s when idle): the report wakes it. + std::thread::sleep(Duration::from_millis(100)); + let mut stream = std::os::unix::net::UnixStream::connect(config.loop_socket()).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let report = Envelope { + v: 1, + id: 1, + r#final: true, + msg: Message::SessionUpdated(SessionUpdated { + session: SessionId::new("mm-a").unwrap(), + turn: 1, + }), + }; + write_frame(&mut stream, &report).unwrap(); + assert!(matches!( + read_frame(&mut stream).unwrap().msg, + Message::Ok(_) + )); + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline && messages(&fake) == 0 { + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(messages(&fake), 1, "{:?}", lines.lock().unwrap()); +} + +#[test] +fn a_key_that_cannot_be_read_stops_the_start() { + let fake = FakeHoncho::start(); + let dir = TempDir::new("worker-key-start"); + let text = serve_config( + &dir, + &fake.url, + "[secrets.honcho_trusted]\nenv = \"MEMORYD_TEST_UNSET_KEY\"\n", + ); + let config = Config::parse(&text).unwrap(); + let log: serve::Log = Arc::new(|_: &str| {}); + let err = match serve::start(config.clone(), log) { + Err(err) => err, + Ok(_) => panic!("memoryd started"), + }; + assert!(matches!(err, StartError::Secret(_)), "{err:?}"); + assert!(!config.loop_socket().exists(), "a socket was bound"); + // The process: exit 1 with the secret's own lines and pointer, not the start pointer. + let path = dir.write("memoryd.toml", &text); + let mut child = Daemon( + Command::new(env!("CARGO_BIN_EXE_memoryd")) + .args(["serve", "--config"]) + .arg(&path) + .env_remove("MEMORYD_TEST_UNSET_KEY") + .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}"); + assert!( + stderr.starts_with("memoryd: secret honcho_trusted: "), + "{stderr}" + ); + assert!( + stderr.ends_with("\nsee docs/runbook.md#secret-unavailable\n"), + "{stderr}" + ); + assert!(!stderr.contains("memoryd-start-failed"), "{stderr}"); +} + +/// A `memoryd` process, killed and waited for when dropped. +struct Daemon(std::process::Child); + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/support/worker.rs b/docs/plans/M5c1/files/crates/memoryd/tests/support/worker.rs new file mode 100644 index 0000000..d39869c --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/worker.rs @@ -0,0 +1,133 @@ +//! A Honcho worker for tests, stepped by hand: a home with session logs and a ledger, the fake +//! Honcho, `brokerd` as a closure whose answer the test sets, and the log lines collected. Do not +//! edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::collections::BTreeMap; +use std::ffi::OsString; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bxcore::{LogRecord, SessionStateReply, Timestamp}; +use memoryd::honcho::client::{Clients, Timeouts}; +use memoryd::honcho::config::HonchoConfig; +use memoryd::honcho::ledger::{Ledger, Record}; +use memoryd::honcho::turns::{LEDGER, TurnSync}; +use memoryd::honcho::worker::{Tuning, Wake, Worker}; +use memoryd::serve::Log; + +use crate::honcho::FakeHoncho; +use crate::tmp::TempDir; +use crate::turns::{clean, sid, write_turns}; + +pub const FAST: Tuning = Tuning { + backoff: Duration::from_millis(1), + backoff_max: Duration::from_millis(8), + idle: Duration::from_millis(1), +}; + +pub fn config(url: &str) -> HonchoConfig { + HonchoConfig { + url: url.to_string(), + agent: "boxmaker0".to_string(), + owner: "kyle".to_string(), + trusted: "boxmaker0-trusted".to_string(), + untrusted: "boxmaker0-untrusted".to_string(), + fleet: "fleet".to_string(), + read: Vec::new(), + keys: BTreeMap::new(), + } +} + +pub struct Home { + pub sync: Arc>, + /// What `brokerd` answers. + pub answer: Arc>>, + pub lines: Arc>>, + pub dir: TempDir, +} + +impl Home { + pub fn new(tag: &str) -> Home { + let dir = TempDir::new(tag); + std::fs::create_dir_all(dir.path().join("memory")).unwrap(); + Home { + sync: Arc::new(Mutex::new(TurnSync::new( + Ledger::open(&dir.path().join("memory").join(LEDGER)).unwrap(), + "boxmaker0", + &dir.path().join("sessions"), + ))), + answer: Arc::new(Mutex::new(Ok(clean()))), + lines: Arc::new(Mutex::new(Vec::new())), + dir, + } + } + + pub fn ledger_path(&self) -> PathBuf { + self.dir.path().join("memory").join(LEDGER) + } + + /// A new `TurnSync` over the ledger as it is on disk, resumed: what a restart makes. + pub fn restart(&mut self) { + let mut sync = TurnSync::new( + Ledger::open(&self.ledger_path()).unwrap(), + "boxmaker0", + &self.dir.path().join("sessions"), + ); + sync.resume(); + self.sync = Arc::new(Mutex::new(sync)); + } + + pub fn log(&self, session: &str, turns: &[Vec]) { + write_turns(self.dir.path(), session, turns); + } + + /// `loopd` reports turn `turn` of `session`: decided now, with `answer`. + pub fn report(&self, session: &str, turn: u64) { + let answer = Arc::clone(&self.answer); + let ask = move |_: &bxcore::SessionId| answer.lock().unwrap().clone(); + let lines = Arc::clone(&self.lines); + let log = move |line: &str| lines.lock().unwrap().push(line.to_string()); + let now = Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap(); + self.sync + .lock() + .unwrap() + .updated(&sid(session), turn, &ask, now, &log); + } + + pub fn worker(&self, fake: &FakeHoncho, tuning: Tuning) -> Worker { + let no_env = |_: &str| -> Option { None }; + let timeouts = Timeouts { + connect: Duration::from_secs(5), + read: Duration::from_millis(300), + }; + let (clients, _) = Clients::new(&config(&fake.url), &no_env, timeouts).unwrap(); + let answer = Arc::clone(&self.answer); + let lines = Arc::clone(&self.lines); + let log: Log = Arc::new(move |line: &str| lines.lock().unwrap().push(line.to_string())); + Worker::new( + Arc::clone(&self.sync), + clients, + Arc::new(Wake::default()), + Box::new(move |_: &bxcore::SessionId| answer.lock().unwrap().clone()), + log, + tuning, + "boxmaker0", + ) + } + + /// Every record of the ledger on disk, in order. + pub fn records(&self) -> Vec { + std::fs::read_to_string(self.ledger_path()) + .unwrap_or_default() + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect() + } + + pub fn lines(&self) -> Vec { + self.lines.lock().unwrap().clone() + } +} diff --git a/docs/plans/M5c1/files/runbook-honcho-key-refused.md b/docs/plans/M5c1/files/runbook-honcho-key-refused.md new file mode 100644 index 0000000..8f5cd77 --- /dev/null +++ b/docs/plans/M5c1/files/runbook-honcho-key-refused.md @@ -0,0 +1,24 @@ +## honcho-key-refused + +**What you see.** `memoryd` logs one line when Honcho refuses a workspace's key, then this entry: + +```text +memoryd: Honcho refused the key of workspace boxmaker0-trusted (HTTP 401); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused +``` + +**Why.** Each workspace is sent to with its own key, `[secrets.honcho_trusted]`, +`honcho_untrusted` or `honcho_fleet` (M5c spec, section 10), or none. A 401 or 403 means the key +is wrong, expired, or missing where Honcho now wants one. Trying again would be refused again, so +`memoryd` stops sending to that workspace until it restarts; its turns stay waiting in +`memory/honcho.jsonl`, and the other workspaces go on. The key is never printed. + +**Confirm.** Which workspace is named, and where its key comes from: + +```sh +grep -n -A2 '^\[secrets\.honcho_' /etc/boxmaker/memoryd.toml +``` + +**Fix.** Put the right key where `[secrets.]` points (see `secret-unavailable` for the +sources), or remove the table if the workspace needs no key, then restart `memoryd`. + +**Check.** After the restart the waiting turns are sent, and no line names this entry. diff --git a/docs/plans/M5c1/files/runbook-honcho-send-refused.md b/docs/plans/M5c1/files/runbook-honcho-send-refused.md new file mode 100644 index 0000000..447d9b9 --- /dev/null +++ b/docs/plans/M5c1/files/runbook-honcho-send-refused.md @@ -0,0 +1,31 @@ +## honcho-send-refused + +**What you see.** `memoryd` logs one line for a turn that Honcho refused, or that could not be +sent at all, then this entry: + +```text +memoryd: turn 3 of session mm-chat-1 was not sent: Honcho refused the request (status, HTTP 422); it is marked failed; see docs/runbook.md#honcho-send-refused +``` + +or `memoryd: turn 3 of session mm-chat-1 cannot be sent (too-long); it is marked failed; …` for a +turn whose text needs more than 100 messages of 25,000 characters. + +**Why.** Honcho answered with a 4xx other than 401 or 403, a redirect, or a body `roncho` could +not read; sending the same again would get the same answer, or could store it twice. So the turn +is marked `failed` in `memory/honcho.jsonl`, with the kind and the status, and the next turn goes +on. The line never holds the turn's text or Honcho's answer (a 422 echoes the whole request). + +**Confirm.** + +```sh +grep -n '"type":"failed"' "$BOXMAKER_HOME/memory/honcho.jsonl" +``` + +A 422 usually means a name Honcho does not accept, or text over its limit; a 404, a workspace or +session removed on the Honcho side. + +**Fix.** A failed turn is not sent again by itself. To send it again after fixing the cause, +stop `memoryd`, keep a copy of the ledger, remove the turn's `failed` line (see +`honcho-ledger-damaged` for what removing a line means), and start `memoryd`. + +**Check.** The next turns are sent (`sent` lines follow). diff --git a/docs/plans/M5c1/files/runbook-honcho-unreachable.md b/docs/plans/M5c1/files/runbook-honcho-unreachable.md new file mode 100644 index 0000000..c56a0a2 --- /dev/null +++ b/docs/plans/M5c1/files/runbook-honcho-unreachable.md @@ -0,0 +1,30 @@ +## honcho-unreachable + +**What you see.** `memoryd` logs one line when it cannot reach Honcho, then this entry: + +```text +memoryd: Honcho is unreachable (connect); sends wait and are retried; see docs/runbook.md#honcho-unreachable +``` + +The word in parentheses is the kind: `connect` (nothing answers at `[honcho] url`), `timeout` (5 +seconds to connect, or 30 without a byte), `closed`, `io`, or `server, HTTP 5xx` (Honcho answered +with a server error). When a send goes through again, `memoryd` logs `memoryd: Honcho answers +again; waiting sends go on`. Conversations are not affected. + +**Why.** Every finished turn that is decided for Honcho waits in the ledger, +`memory/honcho.jsonl`, until it is sent (M5c spec, section 5). While Honcho does not answer, +`memoryd` tries the oldest waiting turn again, waiting 5 seconds, then 10, 20 and so on up to 10 +minutes between tries, and logs the outage once, not at every try. Nothing is lost: the turns go, +in order, once Honcho answers. A restart of `memoryd` starts the waits again from 5 seconds. + +**Confirm.** + +```sh +grep -n '^url' /etc/boxmaker/memoryd.toml +curl -sS -m 5 "/health" +``` + +**Fix.** Start Honcho, or make the network path to it work (on straylight, rift over the +tailnet: `tailscale status`). If `[honcho] url` is wrong, correct it and restart `memoryd`. + +**Check.** `memoryd` logs `Honcho answers again`, and `memory/honcho.jsonl` gains `sent` lines.