diff --git a/docs/plans/M5c1/10-memoryd-honcho-client.md b/docs/plans/M5c1/10-memoryd-honcho-client.md index 418163b..a15afb5 100644 --- a/docs/plans/M5c1/10-memoryd-honcho-client.md +++ b/docs/plans/M5c1/10-memoryd-honcho-client.md @@ -143,7 +143,8 @@ This is the complete list. - Copy, from `docs/plans/M5c1/files/` (the same path under it): - new tests: `crates/memoryd/tests/honcho_client.rs`, `crates/memoryd/tests/honcho_keys.rs`, `crates/memoryd/tests/honcho_naming.rs` - - new test support: `crates/memoryd/tests/support/honcho.rs` (the fake Honcho) + - new test support: `crates/memoryd/tests/support/honcho.rs` (the fake Honcho), from + `honcho.rs-task10` (this task's version; task 13 gives a later one) - the directory `crates/memoryd/tests/fixtures/honcho/` (29 JSON files and a README), whole - the skeletons `crates/memoryd/src/honcho/client.rs` and `crates/memoryd/src/honcho/naming.rs` - Modify: `crates/memoryd/src/honcho/mod.rs`, `crates/memoryd/src/honcho/ledger.rs`, @@ -260,7 +261,7 @@ become cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_client.rs crates/memoryd/tests/ cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_keys.rs crates/memoryd/tests/ cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_naming.rs crates/memoryd/tests/ - cp docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs crates/memoryd/tests/support/ + cp docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs-task10 crates/memoryd/tests/support/honcho.rs ``` `ls crates/memoryd/tests/fixtures/honcho | wc -l` prints `30`. diff --git a/docs/plans/M5c1/11-memoryd-honcho-worker.md b/docs/plans/M5c1/11-memoryd-honcho-worker.md index 1f3a76b..63776ce 100644 --- a/docs/plans/M5c1/11-memoryd-honcho-worker.md +++ b/docs/plans/M5c1/11-memoryd-honcho-worker.md @@ -146,7 +146,8 @@ This is the complete list. ## 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 tests: `crates/memoryd/tests/honcho_worker.rs`, and `crates/memoryd/tests/honcho_worker_more.rs` + from `honcho_worker_more.rs-task11` (this task's version; task 13 gives a later one) - 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) @@ -515,7 +516,7 @@ become ```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_worker_more.rs-task11 crates/memoryd/tests/honcho_worker_more.rs 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/ ``` diff --git a/docs/plans/M5c1/13-memoryd-honcho-edits.md b/docs/plans/M5c1/13-memoryd-honcho-edits.md new file mode 100644 index 0000000..a808a48 --- /dev/null +++ b/docs/plans/M5c1/13-memoryd-honcho-edits.md @@ -0,0 +1,740 @@ +# M5c-1 task 13: the Honcho worker sends memory writes + +**Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `memoryd: the Honcho worker sends conclusions, their deletions and fleet notes` +**Log row:** the Task cell is `M5c1/13-memoryd-honcho-edits`. + +## Goal + +Task 12 decides each memory write into the ledger; task 11's worker sends only turns. This task +makes the worker send the rest (M5c spec, sections 4 and 5): one outbox, oldest first, across +every kind of item. + +- **`Conclude`**: the conclusion is made in the item's workspace (trusted), about its `peer` + (observer and observed are both that peer), stamped with the Boxmaker session's **memory + session**, `_memory`. That Honcho session holds the agent's peer and the owner's and is + made on first use, whether or not any turn of the session is synced (spec section 4, as of + 2026-09-28). Neither peer observes in it: it holds conclusions, not messages, and the spec gives + no flags for it, so both get the agent's (`observe_me: false`, `observe_others: false`). The text + is neutralised as it leaves. `Sent` records the one id Honcho gives. +- **`Unconclude`**: the id to delete is `ledger.conclusion(workspace, text)` (task 07): the latest + sent `Conclude` of that text not yet undone. It is deleted (`delete_conclusion`, task 10: 204 or + 404 is done), and `Sent` records the deleted id, which pops it from `conclusion`'s stack. With + none (never sent, its `Conclude` failed, or already undone) nothing is asked: `Sent` with no ids, + and no log line. A `Conclude` still waiting cannot be behind its `Unconclude`: it was decided + first, and the outbox keeps that order. +- **`Fleet`**: the untrusted note is added to the fleet workspace (made with task 10's settings) as + messages from the agent's peer, neutralised, in parts of 25,000 characters, in one batch, to the + Honcho session named after the Boxmaker session, ``, holding only the agent's peer (spec + section 3). `Sent` records the ids. +- **`Retract`**: nothing decides one any more (a note cannot be undone), so it waits in the outbox, + untouched. +- **Errors**, as for turns (task 11): unreachable waits and tries the same item again; refused + marks it failed, logged once; a refused key stops that workspace. Log lines name an edit's item + `part of edit `. + +To stay under 500 lines, `worker.rs`'s start code (`Opened`, `open`, `spawn`) moves, unchanged +but for its imports, to a new file `honcho/start.rs`; `worker.rs` re-exports them, so `serve.rs` +does not change. + +## Interfaces + +In `crates/memoryd/src/honcho/edits.rs` (the module `memoryd::honcho::edits`): + +```rust +pub const MEMORY_SUFFIX: &str = "_memory"; +pub fn memory_session(session: &SessionId) -> String; +pub fn conclude(clients: &mut Clients, workspace: Workspace, peer: &str, session: &SessionId, + text: &str) -> Result, HonchoError>; +pub fn unconclude(clients: &Clients, workspace: Workspace, id: Option<&str>) + -> Result, HonchoError>; +pub fn fleet_batch(text: &str, agent: &str, session: &SessionId, edit: &str) -> Vec; +pub fn fleet(clients: &mut Clients, session: &SessionId, messages: &[MessageCreate]) + -> Result, HonchoError>; +``` + +In `client.rs`: `Clients` keeps `[honcho] owner`, and gains +`ensure_memory_session(&mut self, workspace: Workspace, name: &str) -> Result<(), HonchoError>` +(edits C1 to C3). In `worker.rs`, the private `Next` holds a `Job` (`Turn`, `Conclude`, +`Unconclude`, `Fleet`), `next` builds it for any kind of item, and `send` dispatches on it +(edits W1 to W7). + +## The requests + +Fixture 11 is the conclusion's shape: + +| Item | Requests | +|---|---| +| `Conclude` | make sure of the workspace, the agent's peer, the owner's peer and `_memory` (`{"id":"_memory","peers":{"":{"observe_me":false,"observe_others":false},"":{"observe_me":false,"observe_others":false}}}`), then `POST /v3/workspaces//conclusions` with `{"conclusions":[{"content":"","observer_id":"","observed_id":"","session_id":"_memory"}]}` | +| `Unconclude` | `DELETE /v3/workspaces//conclusions/`, or none | +| `Fleet` | make sure of the fleet workspace, the agent's peer and `` (`{"id":"","peers":{"":{"observe_me":false,"observe_others":false}}}`), then `POST /v3/workspaces//sessions//messages` with each part `{"content":"","peer_id":"","metadata":{"agent":"","boxmaker_session":"","edit":""}}` | + +What was made sure of is remembered for the run (task 10), so a second conclusion in the same +session asks only for the conclusion. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/memoryd/src/honcho/mod.rs` | `pub mod edits;` and `pub mod start;` (edits H1, H2) | +| `crates/memoryd/src/honcho/client.rs` | `owner`, `ensure_memory_session` (edits C1 to C3) | +| `crates/memoryd/src/honcho/edits.rs` | new: a skeleton you copy, then fill 5 bodies | +| `crates/memoryd/src/honcho/start.rs` | new, given whole: the start code from `worker.rs` | +| `crates/memoryd/src/honcho/worker.rs` | imports, `Job` and `Next`, `label`, the two log lines, `next`/`send`/`send_turn`, the start code removed (edits W1 to W7) | +| `crates/memoryd/tests/support/honcho.rs` | a new version, given: the fake answers a conclusion batch (fixture 11), and a batch with as many items as it asked for | +| `crates/memoryd/tests/honcho_worker_more.rs` | a new version, given: its "later actions wait" test is now about a retraction | +| everything else | **no change** | + +## Files + +- Copy, from `docs/plans/M5c1/files/` (the same path under it): + - the new test `crates/memoryd/tests/honcho_edits.rs` + - **new versions** (they replace the files in the tree) of + `crates/memoryd/tests/support/honcho.rs` and `crates/memoryd/tests/honcho_worker_more.rs`; + do not copy `honcho.rs-task10` or `honcho_worker_more.rs-task11`, the older ones + - the skeleton `crates/memoryd/src/honcho/edits.rs` and the file + `crates/memoryd/src/honcho/start.rs` (given whole: nothing to fill) +- Modify: the files of edits H1 to W7, `docs/implementer-log.md`. +- Do not touch: any other source file (`serve.rs` in particular), any other test or support file, + the fixtures, any `Cargo.toml`, and in the skeleton anything but the 5 `todo!()` bodies. + +## Facts you need + +- **`roncho`'s conclusion call** (`roncho::blocking::Client`): `create_conclusions(&ConclusionBatchCreate) + -> Result, Error>` (`POST .../conclusions`; a batch of 1 to 100, checked + before sending). `ConclusionCreate::new(content, observer_id, observed_id)`, + `.with_session_id(name)`; `ConclusionBatchCreate::new(vec![..])`. `Conclusion.id` is the id. + Honcho's `src/schemas/api.py` caps a conclusion batch at 100, as it does messages. +- **Task 07's `conclusion`** replays the `Sent` records in order: a sent `Conclude` of a workspace + and text pushes `(item, ids[0])`; a sent `Unconclude` of the same pops. So an `Unconclude`'s + `Sent` pops whether it carries the deleted id or none; the id is kept for the record. +- **Session names.** `_` is allowed in Honcho names and in no Boxmaker session id, so + `_memory` is never a turn session (``, `-2`, …) or another session's id. +- **The given code** of the worker, after W6: `next` matches `(item, action)` pairs; an + `Unconclude`'s id is looked up there, under the ledger's lock, when the item is picked; + `send` calls `send_turn` (task 11's turn code, unchanged but for its arguments) or `edits`. +- **The given tests.** + - `honcho_edits.rs` (10 tests): the memory session's name; a conclusion's five requests, exactly, + its id, and `conclusion`; a replace and its undo (the requests in order, the deleted ids, each + thing made sure of once); an `Unconclude` with nothing to delete (never sent, or its + `Conclude` failed); a 404 on delete; a fleet note's four requests, exactly, and a long note in + parts; a note too long for one batch; neutralising; one outbox across a turn, an edit and a + turn, and a restart that sends nothing; errors (an outage, a 422, a refused fleet key that + leaves the trusted workspace going), with no text in a log line. + - `honcho_worker_more.rs` (10 tests): task 11's, with its "later actions" test about a + retraction, which waits. +- The calls the bodies use: + + | Call | Signature, and what it gives | + |---|---| + | `Clients::ensure_memory_session` | edit C3: `(&mut self, workspace, name: &str) -> Result<(), HonchoError>` | + | `Clients::ensure_session`, `delete_conclusion`, `client` | task 10 | + | `ConclusionCreate`, `ConclusionBatchCreate` | above | + | `Client::create_conclusions`, `add_messages` | above; task 11 | + | `neutralise` | `bxcore::sanitize::neutralise(text) -> Cow<'_, str>`; `ConclusionCreate::new` takes it as `impl Into` | + | `worker::parts` | task 11: parts of at most 25,000 characters | + | `classify` | task 10: `roncho`'s error as a `HonchoError` | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. +Edit W6 replaces `next` and `send` as they are after task 12 (`next`'s body is yours from task +11, with task 12's W5); edit W7 removes the start code, which moves to `start.rs`. If either +"before" block differs from your file, stop and report. + +### `crates/memoryd/src/honcho/worker.rs` + +**W1.** The lines + +```rust +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::mirror::Mirror; +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}; +``` + +become + +```rust +use crate::honcho::client::{Clients, HonchoError, classify}; +use crate::honcho::decide::{final_reply, turn_records, user_text}; +use crate::honcho::edits; +use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace}; +use crate::honcho::naming::{HonchoSession, honcho_session}; +use crate::honcho::turns::{TurnSync, log_path, read_log}; +use crate::serve::Log; + +pub use crate::honcho::start::{Opened, open, spawn}; +``` + +**W2.** The lines + +```rust +/// 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, +} +``` + +become + +```rust +/// What to send for one waiting item (M5c spec, section 5). +enum Job { + /// A turn: its session and number, its speaker, and its Honcho session. + Turn { + session: SessionId, + turn: u64, + speaker: String, + honcho: Option, + }, + /// A conclusion about `peer`, stamped with `session`'s memory session. + Conclude { + peer: String, + session: SessionId, + text: String, + }, + /// Delete the conclusion with this id; `None` when no sent conclusion is live. + Unconclude { id: Option }, + /// An untrusted note: a fleet message from the agent. + Fleet { + session: SessionId, + edit: String, + text: String, + }, +} + +/// The oldest waiting item: what it is, where it goes, and what to send. +struct Next { + item: Item, + workspace: Workspace, + job: Job, +} +``` + +**W3.** The line + +```rust +/// The ids of the messages Honcho made, in order. +``` + +becomes + +```rust +/// An item as the log lines name it: `turn 3 of session mm-a`, `part 0 of edit 2026-09-28-0001`. +fn label(item: &Item) -> String { + match item { + Item::Turn { session, turn } => format!("turn {turn} of session {}", session.as_str()), + Item::Edit { id, part } => format!("part {part} of edit {id}"), + } +} + +/// The ids of the messages Honcho made, in order. +``` + +**W4.** In `step`, in the arm `Err(Unsent::Honcho(e))`, the lines + +```rust + "memoryd: turn {} of session {} was not sent: {e}; it is marked failed; {SEND_REFUSED}", + next.turn, + next.session.as_str() +``` + +become + +```rust + "memoryd: {} was not sent: {e}; it is marked failed; {SEND_REFUSED}", + label(&next.item) +``` + +**W5.** In `step`, in the arm `Err(Unsent::Local(kind))`, the lines + +```rust + "memoryd: turn {} of session {} cannot be sent ({kind}); it is marked failed; {pointer}", + next.turn, + next.session.as_str() +``` + +become + +```rust + "memoryd: {} cannot be sent ({kind}); it is marked failed; {pointer}", + label(&next.item) +``` + +**W6.** The lines (the whole of `next` and `send`, from `next`'s doc comment to the line before `unreachable`'s) + +```rust + /// The oldest waiting turn whose workspace's key was not refused. Other actions wait. + fn next(&self) -> Option { + let ledger = self.ledger.lock().unwrap_or_else(|p| p.into_inner()); + ledger.outbox().into_iter().find_map(|(item, action)| { + let Item::Turn { session, turn } = &item else { + return None; + }; + let Action::Turn { workspace, speaker } = action else { + return None; + }; + if self.refused.contains(&workspace) { + return None; + } + Some(Next { + honcho: honcho_session(ledger.decisions(), session, *turn), + session: session.clone(), + turn: *turn, + item: item.clone(), + workspace, + speaker, + }) + }) + } + + /// 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 (text, reply) = read_turn(&log_path(&self.sessions, &next.session), next.turn) + .ok_or(Unsent::Local("log"))?; + let honcho = next.honcho.as_ref().ok_or(Unsent::Local("log"))?; + let messages = batch( + &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))), + } + } + +``` + +become + +```rust + /// The oldest waiting item whose workspace's key was not refused. A retraction, which + /// nothing decides any more, waits. + fn next(&self) -> Option { + let ledger = self.ledger.lock().unwrap_or_else(|p| p.into_inner()); + ledger.outbox().into_iter().find_map(|(item, action)| { + let (workspace, job) = match (&item, action) { + (Item::Turn { session, turn }, Action::Turn { workspace, speaker }) => ( + workspace, + Job::Turn { + honcho: honcho_session(ledger.decisions(), session, *turn), + session: session.clone(), + turn: *turn, + speaker, + }, + ), + ( + Item::Edit { .. }, + Action::Conclude { + workspace, + peer, + session, + text, + }, + ) => ( + workspace, + Job::Conclude { + peer, + session, + text, + }, + ), + (Item::Edit { .. }, Action::Unconclude { workspace, text }) => ( + workspace, + Job::Unconclude { + id: ledger.conclusion(workspace, &text).map(|(_, id)| id), + }, + ), + (Item::Edit { id, .. }, Action::Fleet { session, text }) => ( + Workspace::Fleet, + Job::Fleet { + session, + edit: id.clone(), + text, + }, + ), + _ => return None, + }; + if self.refused.contains(&workspace) { + return None; + } + Some(Next { + item: item.clone(), + workspace, + job, + }) + }) + } + + /// Send one item, by what it is. + fn send(&mut self, next: &Next) -> Result, Unsent> { + let honcho = |e| Unsent::Honcho(e); + match &next.job { + Job::Turn { + session, + turn, + speaker, + honcho: name, + } => self.send_turn(next.workspace, session, *turn, speaker, name.as_ref()), + Job::Conclude { + peer, + session, + text, + } => edits::conclude(&mut self.clients, next.workspace, peer, session, text) + .map_err(honcho), + Job::Unconclude { id } => { + edits::unconclude(&self.clients, next.workspace, id.as_deref()).map_err(honcho) + } + Job::Fleet { + session, + edit, + text, + } => { + let messages = edits::fleet_batch(text, &self.agent, session, edit); + if messages.len() > MAX_BATCH { + return Err(Unsent::Local("too-long")); + } + edits::fleet(&mut self.clients, session, &messages).map_err(honcho) + } + } + } + + /// Read the turn again, make sure of its Honcho session, and add its messages as one batch. + fn send_turn( + &mut self, + workspace: Workspace, + session: &SessionId, + turn: u64, + speaker: &str, + honcho: Option<&HonchoSession>, + ) -> Result, Unsent> { + let (text, reply) = + read_turn(&log_path(&self.sessions, session), turn).ok_or(Unsent::Local("log"))?; + let honcho = honcho.ok_or(Unsent::Local("log"))?; + let messages = batch(&text, &reply, speaker, &self.agent, session, turn); + if messages.len() > MAX_BATCH { + return Err(Unsent::Local("too-long")); + } + self.clients + .ensure_session(workspace, honcho) + .map_err(Unsent::Honcho)?; + let client = self.clients.client(workspace); + match client.add_messages(&honcho.name, &messages) { + Ok(made) => Ok(ids(&made)), + Err(e) => Err(Unsent::Honcho(classify(&e))), + } + } + +``` + +**W7.** At the end of the file, the lines (the end of `impl Worker`, then the start code) + +```rust +} + +/// 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, + owner: String, +} + +impl Opened { + /// The memory writes' side of the ledger (task 12). + pub fn mirror(&self) -> Mirror { + let ledger = self + .sync + .lock() + .unwrap_or_else(|p| p.into_inner()) + .shared_ledger(); + Mirror::new(ledger, &self.agent, &self.owner) + } +} + +/// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the +/// 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(), + owner: honcho.owner.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}")) +} +``` + +become + +```rust +} +``` + +### `crates/memoryd/src/honcho/client.rs` + +**C1.** In `pub struct Clients`, the lines + +```rust + fleet: Client, + agent: String, +``` + +become + +```rust + fleet: Client, + agent: String, + owner: String, +``` + +**C2.** In `Clients::new`, the lines + +```rust + agent: config.agent.clone(), + ensured: HashSet::new(), +``` + +become + +```rust + agent: config.agent.clone(), + owner: config.owner.clone(), + ensured: HashSet::new(), +``` + +**C3.** The line + +```rust + /// Delete conclusion `id` in `workspace`. Gone already (404) counts as done. +``` + +becomes + +```rust + /// Make sure memory session `name` (a Boxmaker session's `_memory`) exists in + /// `workspace`, after the workspace, the agent's peer and the owner's. It holds conclusions, + /// no messages, so neither peer observes (M5c spec, section 4). + pub fn ensure_memory_session( + &mut self, + workspace: Workspace, + name: &str, + ) -> Result<(), HonchoError> { + self.ensure_workspace(workspace)?; + let (agent, owner) = (self.agent.clone(), self.owner.clone()); + self.ensure_peer(workspace, &agent)?; + self.ensure_peer(workspace, &owner)?; + let key = (workspace, Ensured::Session(name.to_string())); + if self.ensured.contains(&key) { + return Ok(()); + } + let unobserved = || { + SessionPeerConfig::new() + .observe_me(false) + .observe_others(false) + }; + let mut peers = HashMap::new(); + peers.insert(agent, unobserved()); + peers.insert(owner, unobserved()); + let create = SessionCreate::new(name).with_peers(peers); + created(self.client(workspace).session(&create))?; + self.ensured.insert(key); + Ok(()) + } + + /// Delete conclusion `id` in `workspace`. Gone already (404) counts as done. +``` + +### `crates/memoryd/src/honcho/mod.rs` + +**H1.** The line + +```rust +pub mod decide; +``` + +becomes + +```rust +pub mod decide; +pub mod edits; +``` + +**H2.** The line + +```rust +pub mod naming; +``` + +becomes + +```rust +pub mod naming; +pub mod start; +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, three commands: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_edits.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/support/honcho.rs crates/memoryd/tests/support/ + ``` + +- [ ] **2. See them fail.** With a shell timeout of 600000 ms: + - `cargo build --tests --keep-going -p memoryd`: one test target does not compile, + `honcho_edits`, with one error, `E0432` (``unresolved import `memoryd::honcho::edits` ``). + Every other target compiles. + - `cargo test -p memoryd --test honcho_worker_more --test honcho_client --test honcho_worker` + passes (10, 13, 12): the new fake and the retraction test agree with the code as it is. +- [ ] **3. The new files.** Make edits H1, H2 and C1 to C3, then copy: + + ```sh + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/edits.rs crates/memoryd/src/honcho/ + cp docs/plans/M5c1/files/crates/memoryd/src/honcho/start.rs crates/memoryd/src/honcho/ + ``` + + `cargo check -p memoryd`: no error, and 22 warnings, all from the unfilled bodies: six + `unused import` lines (`bxcore::sanitize::neutralise`; `classify`; + `crate::honcho::naming::HonchoSession`; `crate::honcho::worker::parts`; + `ConclusionBatchCreate` and `ConclusionCreate`; `Map` and `json`), and sixteen `unused + variable` (`session` four times, `clients` three times, `text` and `workspace` twice each, and + `agent`, `edit`, `id`, `messages`, `peer` once each). +- [ ] **4. Fill the five bodies of `edits.rs`, one at a time, in the order of the file**, each + from its comment, and run `cargo check -p memoryd` after each: `memory_session`, `conclude`, + `unconclude`, `fleet_batch`, `fleet`. When a body is filled, delete its instruction comment + (the `//` lines inside the body) and keep the function's doc comment (`///`) as it is. After + the fifth, `cargo check -p memoryd` prints no warning. +- [ ] **5. The worker.** Make edits W1 to W7. `cargo check -p memoryd`: no error and no warning. + `wc -l crates/memoryd/src/honcho/worker.rs` prints a number under 500. +- [ ] **6. The tests.** With a shell timeout of 600000 ms, and every test passes (10, 10, 12, and + the earlier ones unchanged): + + ```sh + cargo test -p memoryd --test honcho_edits --test honcho_worker_more --test honcho_worker \ + --test honcho_client --test honcho_keys --test honcho_mirror_serve --test honcho_serve + ``` +- [ ] **7. Check.** Three commands: + - `grep -n 'fn open\|fn spawn' crates/memoryd/src/honcho/worker.rs + crates/memoryd/src/honcho/start.rs` prints two lines, both in `start.rs`. + - `grep -n 'Action::Retract' crates/memoryd/src/honcho/worker.rs` prints nothing: a retraction + waits. + - `git diff --stat -- crates/bxcore crates/loopd crates/brokerd crates/gatewayd crates/bxctl + Cargo.toml Cargo.lock crates/memoryd/Cargo.toml crates/memoryd/src/serve.rs + crates/memoryd/src/honcho/ledger.rs crates/memoryd/src/honcho/mirror.rs` prints nothing. +- [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1324 tests: 1314 before this task, plus 10 new ones in + `honcho_edits.rs`. +- [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c1/13-memoryd-honcho-edits`. `git status --short` lists `honcho/mod.rs`, `client.rs`, + `worker.rs`, `tests/support/honcho.rs`, `tests/honcho_worker_more.rs` and your log as modified, + and `edits.rs`, `start.rs` and `tests/honcho_edits.rs` as new. Stage with + `git add crates/memoryd 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 1324 tests. +- `git diff HEAD~1 --stat` lists only the files of step 9. + +## Stop and report if + +- A "before" block of H1, H2, C1 to C3 or W1 to W7 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, other + warnings, or other tests failing. +- `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 or 6 a test fails although each body follows its comment and each edit is as + shown. Do not change a test, the fake or the given code: stop, and give the failing tests' + names and their messages. +- A test seems to need a real Honcho or a new fixture. It does not; stop and say why. +- The fix seems to need a change anywhere else: `serve.rs`, `ledger.rs`, `mirror.rs`, another + crate. 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. (`gatewayd`'s `gateway_sock` test + `brokerd_closing_first_ends_it_as_no_longer_pending` fails now and then on a loaded machine, + without this task's change; if it is the one, run the gate once more.) +- The gate passes with a count other than 1324. + +## What the given tests pin + +Each of these changes to the finished code makes at least one given test fail (23 tried): the +suffix; a conclusion not neutralised, observed by the agent, not stamped, or without its memory +session made first; its id dropped; an `Unconclude` recording no id, or deleting something when +there is nothing; a fleet note not neutralised, from the speaker, without its edit in the +metadata, its session with a speaker or not made first, or over the batch limit; the memory +session's owner observed, left out, or not remembered; edits left waiting; an `Unconclude` +looked up in another workspace; a fleet note sent to the trusted workspace; a retraction sent; +an edit's log label; a refused workspace ignored for edits. + +## After this task + +- Task 14: `bxctl memory sync` shows the ledger's counts and its failed items. +- Task 15: the cutoff, turns from before Honcho was turned on skipped. diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/edits.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/edits.rs new file mode 100644 index 0000000..386ef15 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/edits.rs @@ -0,0 +1,78 @@ +//! Memory writes sent to Honcho (M5c spec, sections 4 and 5), for the worker: a conclusion made +//! in the Boxmaker session's memory session, `_memory`; a conclusion deleted by the id its +//! making was given; an untrusted note added to the fleet workspace as a message from the agent. +//! The text is neutralised here, as it leaves. + +use bxcore::SessionId; +use bxcore::sanitize::neutralise; +use roncho::{ConclusionBatchCreate, ConclusionCreate, MessageCreate}; +use serde_json::{Map, json}; + +use crate::honcho::client::{Clients, HonchoError, classify}; +use crate::honcho::ledger::Workspace; +use crate::honcho::naming::HonchoSession; +use crate::honcho::worker::parts; + +/// What names a Boxmaker session's memory session: `_` is in no Boxmaker id, so the name is no +/// other session's. +pub const MEMORY_SUFFIX: &str = "_memory"; + +/// The memory session of `session`: `_memory`. +pub fn memory_session(session: &SessionId) -> String { + // `format!("{}{MEMORY_SUFFIX}", session.as_str())`. + todo!() +} + +/// Make conclusion `text` about `peer` (its observer and the observed) in `workspace`, stamped +/// with `session`'s memory session, which is made sure of first. Its one id. +pub fn conclude( + clients: &mut Clients, + workspace: Workspace, + peer: &str, + session: &SessionId, + text: &str, +) -> Result, HonchoError> { + // `let name = memory_session(session);`, then + // `clients.ensure_memory_session(workspace, &name)?;`. The conclusion: + // `ConclusionCreate::new(neutralise(text), peer, peer).with_session_id(name)` (observer and + // observed both `peer`), in `ConclusionBatchCreate::new(vec![conclusion])`. Then `match + // clients.client(workspace).create_conclusions(&batch)`: `Ok(made)` -> + // `Ok(made.into_iter().map(|c| c.id).collect())`; `Err(e)` -> `Err(classify(&e))`. + todo!() +} + +/// Delete conclusion `id` in `workspace`; with none to delete, nothing is asked. The ids of what +/// was deleted. +pub fn unconclude( + clients: &Clients, + workspace: Workspace, + id: Option<&str>, +) -> Result, HonchoError> { + // `match id`: `Some(id)` -> `clients.delete_conclusion(workspace, id)?;` then + // `Ok(vec![id.to_string()])`; `None` -> `Ok(Vec::new())`: nothing is asked. + todo!() +} + +/// An untrusted note's messages for the fleet workspace: from the agent's peer, neutralised, in +/// parts of at most `MAX_PART` characters, each with the note's metadata. +pub fn fleet_batch(text: &str, agent: &str, session: &SessionId, edit: &str) -> Vec { + // A `Map` `metadata` with `"agent"` -> `json!(agent)`, `"boxmaker_session"` -> + // `json!(session.as_str())`, `"edit"` -> `json!(edit)` (in this order). Then + // `parts(&neutralise(text)).into_iter().map(|part| MessageCreate::new(part, agent) + // .with_metadata(metadata.clone())).collect()`. + todo!() +} + +/// Add `messages` to `session`'s Honcho session in the fleet workspace, which holds only the +/// agent's peer, made sure of first. The ids of the messages. +pub fn fleet( + clients: &mut Clients, + session: &SessionId, + messages: &[MessageCreate], +) -> Result, HonchoError> { + // `let honcho = HonchoSession { name: session.as_str().to_string(), speakers: Vec::new() };` + // then `clients.ensure_session(Workspace::Fleet, &honcho)?;`. Then `match + // clients.client(Workspace::Fleet).add_messages(&honcho.name, messages)`: `Ok(made)` -> + // `Ok(made.into_iter().map(|m| m.id).collect())`; `Err(e)` -> `Err(classify(&e))`. + todo!() +} diff --git a/docs/plans/M5c1/files/crates/memoryd/src/honcho/start.rs b/docs/plans/M5c1/files/crates/memoryd/src/honcho/start.rs new file mode 100644 index 0000000..40af6a8 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/src/honcho/start.rs @@ -0,0 +1,85 @@ +//! `memoryd`'s Honcho side at start (M5c spec, sections 5 and 10): the ledger read through, the +//! sessions that wait again, the clients with their keys, and the worker's thread. + +use std::sync::{Arc, Mutex}; + +use bxcore::SessionId; + +use crate::config::Config; +use crate::honcho::client::{Clients, Timeouts}; +use crate::honcho::config::HonchoConfig; +use crate::honcho::ledger::Ledger; +use crate::honcho::mirror::Mirror; +use crate::honcho::state; +use crate::honcho::turns::{LEDGER, TurnSync}; +use crate::honcho::worker::{AskBox, Tuning, Wake, Worker}; +use crate::serve::{Log, StartError}; + +/// 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, + owner: String, +} + +impl Opened { + /// The memory writes' side of the ledger (task 12). + pub fn mirror(&self) -> Mirror { + let ledger = self + .sync + .lock() + .unwrap_or_else(|p| p.into_inner()) + .shared_ledger(); + Mirror::new(ledger, &self.agent, &self.owner) + } +} + +/// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the +/// 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(), + owner: honcho.owner.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_edits.rs b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_edits.rs new file mode 100644 index 0000000..7013297 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_edits.rs @@ -0,0 +1,411 @@ +//! The worker sending memory writes (M5c spec, sections 4 and 5), stepped by hand against the fake +//! Honcho: a conclusion made in the session's `_memory` session with the agent's and the +//! owner's peers; one deleted by the id its making got, or nothing to delete; a 404 on delete as +//! done; an untrusted note as a fleet message from the agent; the text neutralised; one outbox +//! across turns and edits; errors as for turns; a restart sending nothing again. 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 bxcore::sanitize::TAGS; +use bxcore::{SessionId, Timestamp}; +use honcho::{FakeHoncho, Mode, Request}; +use logs::assistant; +use memoryd::honcho::edits::{MEMORY_SUFFIX, memory_session}; +use memoryd::honcho::ledger::{Action, Item, Record, Workspace}; +use memoryd::honcho::worker::{MAX_PART, Step}; +use serde_json::{Value, json}; +use turns::spoken; +use worker::{FAST, Home}; + +/// The id fixture 11 gives the first conclusion it makes. +const CONCLUSION: &str = "YUBMqIqHhuJprV4o-inrg"; + +fn sid(s: &str) -> SessionId { + SessionId::new(s).unwrap() +} + +fn edit(id: &str, part: u32) -> Item { + Item::Edit { + id: id.to_string(), + part, + } +} + +fn conclude(peer: &str, text: &str) -> Action { + Action::Conclude { + workspace: Workspace::Trusted, + peer: peer.to_string(), + session: sid("mm-a"), + text: text.to_string(), + } +} + +fn unconclude(text: &str) -> Action { + Action::Unconclude { + workspace: Workspace::Trusted, + text: text.to_string(), + } +} + +fn fleet(text: &str) -> Action { + Action::Fleet { + session: sid("mm-c"), + text: text.to_string(), + } +} + +/// Decide `action` for `item`, as task 12 does when an edit is journalled. +fn decide(home: &Home, item: Item, action: Action) { + let sync = home.sync.lock().unwrap(); + sync.ledger() + .append(Record::Decided { + time: Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap(), + item, + action, + }) + .unwrap(); +} + +fn sent_ids(home: &Home, item: &Item) -> Option> { + home.records().into_iter().find_map(|r| match r { + Record::Sent { item: i, ids, .. } if i == *item => Some(ids), + _ => None, + }) +} + +fn posts(fake: &FakeHoncho, suffix: &str) -> Vec { + fake.requests() + .into_iter() + .filter(|r| r.path.ends_with(suffix)) + .collect() +} + +#[test] +fn the_memory_session_is_named_for_the_boxmaker_session() { + assert_eq!(MEMORY_SUFFIX, "_memory"); + assert_eq!(memory_session(&sid("mm-a")), "mm-a_memory"); +} + +#[test] +fn a_conclusion_goes_into_the_sessions_memory_session() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-conclude"); + decide(&home, edit("e1", 0), conclude("kyle", "Prefers UTC.")); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let w_path = "/v3/workspaces/boxmaker0-trusted"; + let seen: Vec<(String, Value)> = fake + .requests() + .into_iter() + .map(|r| (r.path, r.body.unwrap_or(Value::Null))) + .collect(); + assert_eq!( + seen, + vec![ + ( + "/v3/workspaces".to_string(), + json!({ "id": "boxmaker0-trusted" }) + ), + ( + format!("{w_path}/peers"), + json!({ "id": "boxmaker0", "configuration": { "observe_me": false } }) + ), + (format!("{w_path}/peers"), json!({ "id": "kyle" })), + ( + format!("{w_path}/sessions"), + json!({ "id": "mm-a_memory", "peers": { + "boxmaker0": { "observe_me": false, "observe_others": false }, + "kyle": { "observe_me": false, "observe_others": false }, + } }) + ), + ( + format!("{w_path}/conclusions"), + json!({ "conclusions": [{ + "content": "Prefers UTC.", + "observer_id": "kyle", + "observed_id": "kyle", + "session_id": "mm-a_memory", + }] }) + ), + ] + ); + assert_eq!( + sent_ids(&home, &edit("e1", 0)), + Some(vec![CONCLUSION.to_string()]) + ); + let sync = home.sync.lock().unwrap(); + assert_eq!( + sync.ledger().conclusion(Workspace::Trusted, "Prefers UTC."), + Some((edit("e1", 0), CONCLUSION.to_string())) + ); +} + +#[test] +fn a_replace_deletes_then_makes_and_an_undo_reverses_it() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-replace"); + decide(&home, edit("e1", 0), conclude("kyle", "old")); + decide(&home, edit("e2", 0), unconclude("old")); + decide(&home, edit("e2", 1), conclude("kyle", "new")); + // The owner's undo of e2: the new deleted, the old made again. + decide(&home, edit("e3", 0), unconclude("new")); + decide(&home, edit("e3", 1), conclude("kyle", "old")); + let mut w = home.worker(&fake, FAST); + for _ in 0..5 { + assert_eq!(w.step(), Step::Done); + } + assert_eq!(w.step(), Step::Idle); + let order: Vec<(String, String)> = fake + .requests() + .into_iter() + .filter(|r| r.path.contains("/conclusions")) + .map(|r| { + let text = r.body.map_or(String::new(), |b| { + b["conclusions"][0]["content"].as_str().unwrap().to_string() + }); + (r.method, text) + }) + .collect(); + let post = |t: &str| ("POST".to_string(), t.to_string()); + let delete = ("DELETE".to_string(), String::new()); + assert_eq!( + order, + vec![ + post("old"), + delete.clone(), + post("new"), + delete, + post("old") + ] + ); + // Each deletion names the id its conclusion got, and records it. + let deletes = fake + .requests() + .into_iter() + .filter(|r| r.method == "DELETE") + .map(|r| r.path) + .collect::>(); + let path = format!("/v3/workspaces/boxmaker0-trusted/conclusions/{CONCLUSION}"); + assert_eq!(deletes, vec![path.clone(), path]); + // The memory session, its workspace and peers are made sure of once in this run. + assert_eq!(posts(&fake, "/sessions").len(), 1); + assert_eq!(posts(&fake, "/peers").len(), 2); + assert_eq!( + sent_ids(&home, &edit("e2", 0)), + Some(vec![CONCLUSION.to_string()]) + ); + let sync = home.sync.lock().unwrap(); + assert_eq!(sync.ledger().conclusion(Workspace::Trusted, "new"), None); + assert_eq!( + sync.ledger().conclusion(Workspace::Trusted, "old"), + Some((edit("e3", 1), CONCLUSION.to_string())) + ); +} + +#[test] +fn a_deletion_with_nothing_to_delete_is_sent_with_no_id() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-nothing"); + decide(&home, edit("e1", 0), unconclude("never made")); + // A conclusion that failed is not live either. + decide(&home, edit("e2", 0), conclude("kyle", "refused")); + decide(&home, edit("e3", 0), unconclude("refused")); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert!(fake.requests().is_empty()); + assert_eq!(sent_ids(&home, &edit("e1", 0)), Some(Vec::new())); + fake.set(Mode::Status(422)); + assert_eq!(w.step(), Step::Done); + fake.set(Mode::Fixtures); + fake.clear(); + assert_eq!(w.step(), Step::Done); + assert!(fake.requests().is_empty(), "{:?}", fake.requests()); + assert_eq!(sent_ids(&home, &edit("e3", 0)), Some(Vec::new())); + assert_eq!( + home.lines().len(), + 1, + "only the refusal: {:?}", + home.lines() + ); +} + +#[test] +fn a_404_on_delete_is_done() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-404"); + decide(&home, edit("e1", 0), conclude("kyle", "x")); + decide(&home, edit("e2", 0), unconclude("x")); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + fake.set(Mode::Status(404)); + assert_eq!(w.step(), Step::Done); + assert_eq!( + sent_ids(&home, &edit("e2", 0)), + Some(vec![CONCLUSION.to_string()]) + ); + assert!(home.lines().is_empty(), "{:?}", home.lines()); +} + +#[test] +fn an_untrusted_note_is_a_fleet_message_from_the_agent() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-fleet"); + decide(&home, edit("e1", 0), fleet("The page says 3.2.")); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + let seen: Vec<(String, Value)> = fake + .requests() + .into_iter() + .map(|r| (r.path, r.body.unwrap_or(Value::Null))) + .collect(); + assert_eq!(seen.len(), 4); + assert_eq!(seen[0].1["configuration"]["reasoning"]["enabled"], false); + assert_eq!(seen[2].0, "/v3/workspaces/fleet/sessions"); + assert_eq!( + seen[2].1, + json!({ "id": "mm-c", "peers": { + "boxmaker0": { "observe_me": false, "observe_others": false }, + } }) + ); + assert_eq!( + seen[3], + ( + "/v3/workspaces/fleet/sessions/mm-c/messages".to_string(), + json!({ "messages": [{ + "content": "The page says 3.2.", + "peer_id": "boxmaker0", + "metadata": { "agent": "boxmaker0", "boxmaker_session": "mm-c", "edit": "e1" }, + }] }) + ) + ); + assert_eq!(sent_ids(&home, &edit("e1", 0)).map(|i| i.len()), Some(1)); + // A long note in parts, in one batch. + decide(&home, edit("e2", 0), fleet(&"n".repeat(MAX_PART + 1))); + assert_eq!(w.step(), Step::Done); + let last = posts(&fake, "/messages").pop().unwrap().body.unwrap(); + assert_eq!(last["messages"].as_array().unwrap().len(), 2); +} + +#[test] +fn a_note_too_long_for_one_batch_is_marked_failed() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-long"); + decide(&home, edit("e1", 0), fleet(&"n".repeat(MAX_PART * 100 + 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()); +} + +#[test] +fn the_text_is_neutralised() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-neutral"); + decide( + &home, + edit("e1", 0), + conclude("boxmaker0", &format!("a{}b", TAGS[0].text)), + ); + decide(&home, edit("e2", 0), fleet(&format!("c{}d", TAGS[1].text))); + let mut w = home.worker(&fake, FAST); + assert_eq!(w.step(), Step::Done); + assert_eq!(w.step(), Step::Done); + let concluded = posts(&fake, "/conclusions")[0].body.clone().unwrap(); + let content = concluded["conclusions"][0]["content"].as_str().unwrap(); + assert!( + !content.contains(TAGS[0].text) && content.contains("<"), + "{content}" + ); + let message = posts(&fake, "/messages")[0].body.clone().unwrap(); + let content = message["messages"][0]["content"].as_str().unwrap(); + assert!( + !content.contains(TAGS[1].text) && content.contains("<"), + "{content}" + ); +} + +#[test] +fn one_outbox_in_order_across_turns_and_edits() { + let fake = FakeHoncho::start(); + let mut home = Home::new("edits-order"); + home.log( + "mm-a", + &[ + vec![spoken("one", Some("kyle")), assistant(2, Some("a"), &[])], + vec![spoken("two", Some("kyle")), assistant(2, Some("b"), &[])], + ], + ); + home.report("mm-a", 1); + decide(&home, edit("e1", 0), conclude("kyle", "Prefers UTC.")); + home.report("mm-a", 2); + let mut w = home.worker(&fake, FAST); + for _ in 0..3 { + assert_eq!(w.step(), Step::Done); + } + let order: Vec = fake + .requests() + .into_iter() + .filter(|r| r.path.ends_with("/messages") || r.path.ends_with("/conclusions")) + .map(|r| r.path) + .collect(); + assert_eq!( + order, + vec![ + "/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages".to_string(), + "/v3/workspaces/boxmaker0-trusted/conclusions".to_string(), + "/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages".to_string(), + ] + ); + // A restart sends nothing again. + home.restart(); + let fake2 = FakeHoncho::start(); + let mut w = home.worker(&fake2, FAST); + assert_eq!(w.step(), Step::Idle); + assert!(fake2.requests().is_empty()); +} + +#[test] +fn errors_as_for_turns() { + let fake = FakeHoncho::start(); + let home = Home::new("edits-errors"); + decide(&home, edit("e1", 0), conclude("kyle", "private-fact")); + decide(&home, edit("e2", 0), fleet("private-note")); + decide(&home, edit("e3", 0), conclude("kyle", "second")); + let mut w = home.worker(&fake, FAST); + fake.set(Mode::Status(503)); + assert!(matches!(w.step(), Step::Backoff(_))); + fake.set(Mode::Status(422)); + assert_eq!(w.step(), Step::Done); + assert!(matches!( + home.records().last(), + Some(Record::Failed { status: Some(422), kind, .. }) if kind == "status" + )); + fake.set(Mode::Status(401)); + assert_eq!(w.step(), Step::Done, "the fleet's key refused"); + fake.set(Mode::Fixtures); + assert_eq!(w.step(), Step::Done, "the trusted workspace goes on"); + assert!(sent_ids(&home, &edit("e3", 0)).is_some()); + assert_eq!(w.step(), Step::Idle, "the fleet note waits"); + let lines = home.lines(); + assert_eq!(lines.len(), 4, "{lines:?}"); + assert!( + lines[2].starts_with("memoryd: part 0 of edit e1 was not sent: "), + "{lines:?}" + ); + assert!(lines[3].contains("workspace fleet"), "{lines:?}"); + for line in lines { + assert!(!line.contains("private"), "{line}"); + } +} 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 index 95a0414..f79ddd0 100644 --- a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs @@ -1,9 +1,9 @@ //! 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. +//! a retraction waits (nothing decides one); 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; @@ -150,15 +150,17 @@ fn a_restart_never_decides_a_running_turn() { } #[test] -fn actions_of_later_tasks_wait() { +fn a_retraction_waits() { + // Nothing decides a retraction any more (M5c spec, section 4, 2026-09-28): one in the + // ledger is neither sent nor failed. 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(); + let retract = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"retract","session":"mm-a","text":"a note"}}"#; + std::fs::write(home.ledger_path(), format!("{retract}\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_eq!(home.records().len(), 1, "the retraction was failed or sent"); assert!(fake.requests().is_empty()); } diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs-task11 b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs-task11 new file mode 100644 index 0000000..95a0414 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs-task11 @@ -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/honcho.rs b/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs index 2262f25..91daddf 100644 --- a/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs @@ -1,6 +1,7 @@ //! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the -//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path, and records every -//! request, body included. It can be set to answer every request with a status instead, or to +//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path (a batch of +//! messages or conclusions answered with as many as it asked for, up to the two recorded), and +//! records every request, body included. It can be set to answer every request with a status instead, or to //! answer nothing. Stopped, and its held connections released, when dropped; every wait has a //! deadline. Do not edit. @@ -214,8 +215,13 @@ fn answer(request: &Request, made: &Mutex>) -> (u16, String) { } ("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"), ("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => { - fixture("08-messages-add-batch.json") + first(fixture("08-messages-add-batch.json"), request, "messages") } + ("POST", ["v3", "workspaces", _, "conclusions"]) => first( + fixture("11-conclusions-create-batch.json"), + request, + "conclusions", + ), ("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => { fixture("16-conclusion-delete-again.json") } @@ -226,6 +232,19 @@ fn answer(request: &Request, made: &Mutex>) -> (u16, String) { } } +/// A recorded batch answer cut to as many items as the request's `field` holds (the recorded ones +/// hold two), so that one message or conclusion gets one id. +fn first(recorded: (u16, String), request: &Request, field: &str) -> (u16, String) { + let asked = request + .body + .as_ref() + .and_then(|b| b[field].as_array()) + .map_or(0, Vec::len); + let mut items: Vec = serde_json::from_str(&recorded.1).unwrap(); + items.truncate(asked); + (recorded.0, Value::Array(items).to_string()) +} + fn serve(mut stream: TcpStream, shared: &Shared) { let _ = stream.set_read_timeout(Some(Duration::from_secs(10))); let Some(request) = read_request(&mut stream) else { diff --git a/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs-task10 b/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs-task10 new file mode 100644 index 0000000..2262f25 --- /dev/null +++ b/docs/plans/M5c1/files/crates/memoryd/tests/support/honcho.rs-task10 @@ -0,0 +1,260 @@ +//! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the +//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path, and records every +//! request, body included. It can be set to answer every request with a status instead, or to +//! answer nothing. Stopped, and its held connections released, 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::collections::HashSet; +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use serde_json::Value; + +/// One request as the fake read it. +#[derive(Debug, Clone, PartialEq)] +pub struct Request { + pub method: String, + pub path: String, + /// Header names in lower case. + pub headers: Vec<(String, String)>, + pub body: Option, +} + +impl Request { + pub fn header(&self, name: &str) -> Option<&str> { + self.headers + .iter() + .find(|(k, _)| k == name) + .map(|(_, v)| v.as_str()) + } +} + +/// How the fake answers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Mode { + /// From the fixtures. + Fixtures, + /// Every request with this status. A 422 echoes the request's body, as Honcho does. + Status(u16), + /// Read the request, then answer nothing until the fake is dropped (at most 30 s). + Hang, +} + +struct Shared { + mode: Mutex, + requests: Mutex>, + /// What was made, by method and path and id: a second create answers 200, a second delete + /// 404. + made: Mutex>, + stop: AtomicBool, +} + +pub struct FakeHoncho { + pub url: String, + port: u16, + shared: Arc, + accept: Option>, +} + +/// A recorded fixture: its response's status and body. +fn fixture(name: &str) -> (u16, String) { + let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("tests/fixtures/honcho") + .join(name); + let text = std::fs::read_to_string(&path).unwrap(); + let value: Value = serde_json::from_str(&text).unwrap(); + let status = u16::try_from(value["response"]["status"].as_u64().unwrap()).unwrap(); + let body = match &value["response"]["body"] { + Value::String(s) if s.is_empty() => String::new(), + other => other.to_string(), + }; + (status, body) +} + +impl FakeHoncho { + pub fn start() -> FakeHoncho { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let shared = Arc::new(Shared { + mode: Mutex::new(Mode::Fixtures), + requests: Mutex::new(Vec::new()), + made: Mutex::new(HashSet::new()), + stop: AtomicBool::new(false), + }); + let for_accept = Arc::clone(&shared); + let accept = std::thread::spawn(move || { + for stream in listener.incoming() { + if for_accept.stop.load(Ordering::SeqCst) { + return; + } + let Ok(stream) = stream else { return }; + let shared = Arc::clone(&for_accept); + std::thread::spawn(move || serve(stream, &shared)); + } + }); + FakeHoncho { + url: format!("http://127.0.0.1:{port}"), + port, + shared, + accept: Some(accept), + } + } + + pub fn set(&self, mode: Mode) { + *self.shared.mode.lock().unwrap() = mode; + } + + /// Every request read so far, in order. + pub fn requests(&self) -> Vec { + self.shared.requests.lock().unwrap().clone() + } + + /// Forget the requests read so far (not what was made). + pub fn clear(&self) { + self.shared.requests.lock().unwrap().clear(); + } + + /// Waits up to 10 s until at least `n` requests 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.requests().len() >= n { + return true; + } + std::thread::sleep(Duration::from_millis(10)); + } + false + } +} + +impl Drop for FakeHoncho { + fn drop(&mut self) { + self.shared.stop.store(true, Ordering::SeqCst); + let _ = TcpStream::connect(("127.0.0.1", self.port)); + if let Some(accept) = self.accept.take() { + let _ = accept.join(); + } + } +} + +/// Reads one request: the head up to its blank line, then `Content-Length` bytes of body. +fn read_request(stream: &mut TcpStream) -> Option { + let mut buf = Vec::new(); + let mut byte = [0u8; 1]; + while !buf.ends_with(b"\r\n\r\n") { + if stream.read(&mut byte).ok()? == 0 { + return None; + } + buf.push(byte[0]); + } + let head = String::from_utf8(buf).ok()?; + let mut lines = head.split("\r\n"); + let mut first = lines.next()?.split(' '); + let (method, path) = (first.next()?.to_string(), first.next()?.to_string()); + let headers: Vec<(String, String)> = lines + .filter_map(|l| l.split_once(": ")) + .map(|(k, v)| (k.to_ascii_lowercase(), v.to_string())) + .collect(); + let length: usize = headers + .iter() + .find(|(k, _)| k == "content-length") + .map_or(Some(0), |(_, v)| v.parse().ok())?; + let mut body = vec![0u8; length]; + stream.read_exact(&mut body).ok()?; + let body = if length == 0 { + None + } else { + serde_json::from_slice(&body).ok() + }; + Some(Request { + method, + path, + headers, + body, + }) +} + +/// The fixture that answers `request`, by its method and the shape of its path; the second of a +/// pair when the same thing was made (or deleted) before. +fn answer(request: &Request, made: &Mutex>) -> (u16, String) { + let parts: Vec<&str> = request.path.trim_start_matches('/').split('/').collect(); + let id = request + .body + .as_ref() + .and_then(|b| b["id"].as_str()) + .unwrap_or_default() + .to_string(); + let again = !made + .lock() + .unwrap() + .insert(format!("{} {} {id}", request.method, request.path)); + let configured = request + .body + .as_ref() + .is_some_and(|b| b.get("configuration").is_some()); + match (request.method.as_str(), parts.as_slice()) { + ("POST", ["v3", "workspaces"]) if again => fixture("02-workspace-create-again.json"), + ("POST", ["v3", "workspaces"]) => fixture("01-workspace-create.json"), + ("POST", ["v3", "workspaces", _, "peers"]) if again => { + fixture("05-peer-create-boxmaker0-again.json") + } + ("POST", ["v3", "workspaces", _, "peers"]) if configured => { + fixture("04-peer-create-boxmaker0.json") + } + ("POST", ["v3", "workspaces", _, "peers"]) => fixture("03-peer-create-kyle.json"), + ("POST", ["v3", "workspaces", _, "sessions"]) if again => { + fixture("07-session-create-again.json") + } + ("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"), + ("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => { + fixture("08-messages-add-batch.json") + } + ("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => { + fixture("16-conclusion-delete-again.json") + } + ("DELETE", ["v3", "workspaces", _, "conclusions", _]) => { + fixture("15-conclusion-delete.json") + } + _ => (404, r#"{"detail":"Not Found"}"#.to_string()), + } +} + +fn serve(mut stream: TcpStream, shared: &Shared) { + let _ = stream.set_read_timeout(Some(Duration::from_secs(10))); + let Some(request) = read_request(&mut stream) else { + return; + }; + shared.requests.lock().unwrap().push(request.clone()); + let mode = *shared.mode.lock().unwrap(); + let (status, body) = match mode { + Mode::Fixtures => answer(&request, &shared.made), + Mode::Status(422) => { + let input = request.body.clone().unwrap_or(Value::Null); + ( + 422, + serde_json::json!({ "detail": [{ "input": input }] }).to_string(), + ) + } + Mode::Status(status) => (status, r#"{"detail":"refused"}"#.to_string()), + Mode::Hang => { + let deadline = Instant::now() + Duration::from_secs(30); + while !shared.stop.load(Ordering::SeqCst) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + return; + } + }; + let head = format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(body.as_bytes()); +}