From d41c41eb4a12b247eae3382687b4394f2c04a1ea Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 28 Sep 2026 19:30:58 -0700 Subject: [PATCH] M5c-2 plan: task 08, recall's Honcho part; the wiring split out as task 09 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5c2/01-wire-ask-memory.md | 4 +- docs/plans/M5c2/02-brokerd-ask-memory.md | 2 +- docs/plans/M5c2/05-memoryd-honcho-readers.md | 6 +- docs/plans/M5c2/06-memoryd-honcho-recalled.md | 2 +- docs/plans/M5c2/08-memoryd-honcho-search.md | 212 +++++++++ docs/plans/M5c2/README.md | 10 +- .../files/crates/memoryd/src/honcho/search.rs | 165 +++++++ .../crates/memoryd/tests/honcho_search.rs | 419 ++++++++++++++++++ .../crates/memoryd/tests/support/honcho.rs | 321 ++++++++++++++ 9 files changed, 1130 insertions(+), 11 deletions(-) create mode 100644 docs/plans/M5c2/08-memoryd-honcho-search.md create mode 100644 docs/plans/M5c2/files/crates/memoryd/src/honcho/search.rs create mode 100644 docs/plans/M5c2/files/crates/memoryd/tests/honcho_search.rs create mode 100644 docs/plans/M5c2/files/crates/memoryd/tests/support/honcho.rs diff --git a/docs/plans/M5c2/01-wire-ask-memory.md b/docs/plans/M5c2/01-wire-ask-memory.md index d5300c4..13cfacc 100644 --- a/docs/plans/M5c2/01-wire-ask-memory.md +++ b/docs/plans/M5c2/01-wire-ask-memory.md @@ -12,7 +12,7 @@ agents, a question about one or two of its workspaces (M5c spec, section 9). Thi the tool's arguments to `bxcore`, with their shape checks, the way `recall`'s are there (`RecallArgs`, its name constant `RECALL`, its limit `MAX_QUERY`, and `RecallArgs::check`). Later tasks use them: task 02 (`brokerd` parses the arguments with them), task 04 (`loopd` declares the -tool), tasks 09 and 10 (`memoryd` answers it). +tool), tasks 10 and 11 (`memoryd` answers it). ```json {"question":"Who wrote the launch note?","workspaces":["fleet","boxmaker1"]} @@ -24,7 +24,7 @@ tool), tasks 09 and 10 (`memoryd` answers it). - **`workspaces`**: optional; at most two names, none empty, none repeated, each 1 to 512 characters of ASCII letters, digits, `_` and `-`. Missing, or `[]`, means the one workspace `own`, the agent's own trusted one. Whether a name is one this agent's `[honcho]` configuration - has is **not** checked here: `memoryd` checks that in task 09. + has is **not** checked here: `memoryd` checks that in task 10. - There is no other field, and an unknown field is refused, as in every argument type we define. (An earlier draft of the spec had a `reasoning` field; it was removed, and a given test sends it and expects a refusal.) diff --git a/docs/plans/M5c2/02-brokerd-ask-memory.md b/docs/plans/M5c2/02-brokerd-ask-memory.md index 0990931..ca8a134 100644 --- a/docs/plans/M5c2/02-brokerd-ask-memory.md +++ b/docs/plans/M5c2/02-brokerd-ask-memory.md @@ -25,7 +25,7 @@ section 9): `brokerd`'s wait for `memoryd` stays 30 s for every memory tool in this task; task 03 gives `ask_memory` its own. Nothing yet asks `brokerd` for `ask_memory`: `loopd` declares it in task 04 -and `memoryd` answers it from task 10. Until then, a call reaching `memoryd` is answered +and `memoryd` answers it from task 11. Until then, a call reaching `memoryd` is answered `memoryd runs only recall and memory`, a failed result. ## Every place that changes diff --git a/docs/plans/M5c2/05-memoryd-honcho-readers.md b/docs/plans/M5c2/05-memoryd-honcho-readers.md index 8d092a6..9f5fe08 100644 --- a/docs/plans/M5c2/05-memoryd-honcho-readers.md +++ b/docs/plans/M5c2/05-memoryd-honcho-readers.md @@ -12,7 +12,7 @@ and untrusted ones and the fleet's), which move into the sync worker's thread. M (M5c spec, sections 3, 8 and 9): `recall` searches other agents' workspaces and the fleet's, and `ask_memory` asks one or two workspaces by name. Reading needs clients of its own. This task builds them at start and gives later tasks the types they use; nothing searches or asks yet -(tasks 08 and 09). +(tasks 08 to 10). - **`Readers`** (new module `crates/memoryd/src/honcho/read.rs`, given whole): a `Reader` for - each `[[honcho.read]]` workspace, in the order written, with that entry's `trust` and @@ -37,7 +37,7 @@ builds them at start and gives later tasks the types they use; nothing searches - **At start**, `start::open` builds the readers beside the writers' `Clients` and keeps them in `Opened::readers`. A read key that cannot be read stops `memoryd`, as a written workspace's does. The three written workspaces' keys are read by both; each key file's warning is logged - once. `serve.rs` does not change: nothing uses the readers until task 08, which takes + once. `serve.rs` does not change: nothing uses the readers until task 09, which takes `opened.readers` before the worker's thread takes the rest of `Opened`. ## Every place that changes @@ -77,7 +77,7 @@ This is the complete list. From `config.rs` it needs `FLEET_KEY`, `TRUSTED_KEY`, `UNTRUSTED_KEY`, `read_key`, `Trust` and `HonchoConfig`, all public already, and from `bxcore` `DataClass` and `bxcore::memory::OWN_WORKSPACE` (`"own"`, task 01). -- **What `Readers` offers** (for tasks 08 and 09; read `read.rs` for the whole): +- **What `Readers` offers** (for tasks 08 to 10; read `read.rs` for the whole): ```rust pub const SEARCH_READ: Duration; // 5 s diff --git a/docs/plans/M5c2/06-memoryd-honcho-recalled.md b/docs/plans/M5c2/06-memoryd-honcho-recalled.md index 9edb3a0..de948fb 100644 --- a/docs/plans/M5c2/06-memoryd-honcho-recalled.md +++ b/docs/plans/M5c2/06-memoryd-honcho-recalled.md @@ -11,7 +11,7 @@ From task 08 on, `recall` searches other agents' Honcho workspaces and the fleet local index (M5c spec, section 8). This task writes the pure part: what `recall` makes of Honcho's results. A new module, `crates/memoryd/src/honcho/recalled.rs`, turns one source's results into items, interleaves the sources' items after the local hits, and labels the answer. -It does no I/O; nothing calls it until task 08. +It does no I/O; nothing calls it until task 08, and `recall` uses it from task 09. - **The inputs, per source**: its `Label` (task 05); the messages Honcho's workspace search returned (`roncho::Message`); and for another agent's workspace, the conclusions of two diff --git a/docs/plans/M5c2/08-memoryd-honcho-search.md b/docs/plans/M5c2/08-memoryd-honcho-search.md new file mode 100644 index 0000000..0ad375f --- /dev/null +++ b/docs/plans/M5c2/08-memoryd-honcho-search.md @@ -0,0 +1,212 @@ +# M5c-2 task 08: `recall`'s Honcho part, every source at once with its own limit + +**Branch:** `m5c2`. Check it: `git branch --show-current` prints `m5c2` and `git status --short` +is empty; otherwise stop. Never switch branch. +**Commit subject:** `memoryd: recall's Honcho part, every source at once with its own limit` +**Log row:** the Task cell is `M5c2/08-memoryd-honcho-search`. + +## Goal + +With `[honcho]`, `recall` searches other agents' Honcho workspaces and the fleet's beside the +local index (M5c spec, section 8). Tasks 05 to 07 gave the readers, the items and the local +side. This task writes the Honcho part itself, as a new module +`crates/memoryd/src/honcho/search.rs`, and the answer made from both parts. Task 09 wires it into +`memoryd`'s `recall`, outside the state lock; until then nothing calls it. + +- **`search(readers, owner, args, wait, log) -> Searched`**: + - **A query holding a secret asks nothing.** If `decide::has_secret(&args.query)` (M5c-1's + known-prefix secret test, with its masked and deleted forms), no source is asked: + `Searched::Secret`. + - **Every source at once.** Each source of `readers.recall_sources()` (the `[[honcho.read]]` + workspaces in the order written, then the fleet's; never the agent's own) is asked on a thread + of its own. + - **A source's calls**, one after another on its thread: `search_workspace` with `limit = k`; + then, for another agent's workspace only, `query_conclusions` with `top_k = k` for that + agent's peer, then for the owner's peer (observer and observed the same peer, as `memoryd` + writes conclusions). The fleet's is the one search. A source has answered only when all its + calls have; its kept items are task 06's `items`. + - **Each source has `wait`** (5 s in `memoryd`, `SOURCE_WAIT`; the tests give less), counted + from when the Honcho part starts, not added up. A source that failed, or has not answered by + then, is unavailable. A late thread is left to finish on its own, holding no lock; what it + finds is dropped. + - **Logging**: each unavailable source is one line, `memoryd: recall: {name} unavailable: + {error}`, where the error is `HonchoError`'s kind and status (`Honcho is unreachable (server, + HTTP 503)`) or `no answer in {wait} ms`. Never the query, never any text. A Honcho that is + down gives one line per source per `recall`, and the model calls `recall` a few times a turn + at most. + - The answer, `Searched::Asked { found, unavailable }`: each source's kept items in + `recall_sources()` order (empty for one that did not answer), and the unavailable sources' + names, in the same order. +- **`answer(local, query, k, searched) -> Recalled`**: the local items (task 07's + `recall::Local`), then Honcho's, interleaved up to `k` in all and numbered after the local + ones (task 06's `interleave` and `render`); `(no matches for "…")` when there is neither; then + the `unavailable` lines, or, for `Secret`, the line `honcho: not asked; the query looks like it + holds a secret`. The label is the stricter of the local part's and the shown Honcho items' + (task 06's `label`): the higher class, and untrusted when either is. +- **The fake Honcho** of the tests (`crates/memoryd/tests/support/honcho.rs`, a given file) gets a + new version: it answers a workspace search from fixture 30 and a conclusion query from fixture + 32 (about `kyle`) or 33 (about anyone else), can answer as a workspace that does not exist + (`Mode::Missing`: fixtures 34, 35 and 36), and can take a mode for one workspace + (`set_workspace`). Everything it did before is unchanged. + +## Every place that changes + +This is the complete list. + +| Place | What happens there | +|---|---| +| `crates/memoryd/src/honcho/search.rs` | new: a skeleton you copy, then fill 2 bodies | +| `crates/memoryd/src/honcho/mod.rs` | `pub mod search;` (edit H1) | +| `crates/memoryd/tests/honcho_search.rs` | new, given | +| `crates/memoryd/tests/support/honcho.rs` | new version, given | +| everything else | **no change**: not `serve.rs` (task 09), not `recall.rs`, no `Cargo.toml` | + +## Files + +- Copy, from `docs/plans/M5c2/files/` (the same path under it): the new test + `crates/memoryd/tests/honcho_search.rs`; a **new version** of + `crates/memoryd/tests/support/honcho.rs` (it replaces the file in the tree); and the skeleton + `crates/memoryd/src/honcho/search.rs`. +- Modify: `crates/memoryd/src/honcho/mod.rs`, `docs/implementer-log.md`. +- Do not touch: any other file, and in the skeleton anything but its two `todo!()` bodies. + +## Facts you need + +- **The skeleton** is written except two bodies: `source` (one source's calls) and `answer`. + `conclusions` (one conclusion query), `search` (the threads, the limit, the log) and the types + are written; read them before you fill the bodies. `search` calls `source` on each thread. +- **`roncho`'s calls and types** (this table is complete; `roncho`'s source, in the Cargo + registry, is outside the repository and off limits): + + | Item | What it does | + |---|---| + | `Client::search_workspace(&self, &MessageSearch) -> Result, roncho::blocking::Error>` | `POST /v3/workspaces/{w}/search` | + | `MessageSearch { query: String, filters, limit: Option, scope }`, `MessageSearch::new(query)` | all fields public: `MessageSearch { limit: Some(k), ..MessageSearch::new(query) }` sets the limit | + | `Client::query_conclusions`, `ConclusionQuery` | used in `conclusions`, written | + | `roncho::Message`, `roncho::Conclusion` | what `items` (task 06) takes | + +- **From earlier tasks**: `classify(&roncho::blocking::Error) -> HonchoError` (M5c-1, in + `client.rs`); `Reader::search() -> &Client`, `Reader::source`, `Reader::label` (task 05); + `items`, `interleave`, `render`, `label`, `unavailable` (task 06, `recalled.rs`); `Local { + items, class, untrusted }`, `Recalled { text, class, untrusted }` (task 07, `recall.rs`). +- **A name clash** in `answer`: `Searched::Asked`'s field `unavailable` has the name of task 06's + function `unavailable`. Bind the field under another name (`unavailable: missing`), as the + comment says, or the function is hidden inside that arm. +- **The given test** `honcho_search.rs` (12 tests), against the fake Honcho: + - the limit and the secret line; + - every source searched with its calls, their paths and bodies (limit, `top_k`, the peers); + - the agent's own workspaces never searched; + - each source's hits in the answer with their tags, numbered after a local item; + - a fleet hit making the answer untrusted, and a trusted read workspace's hit not (T58); + - a hanging source: one `unavailable` line after the limit (1.5 s), while the others answer; + - every source hanging, answered after one limit, not one per source; + - a failing source (503, 422): `unavailable`, logged by kind and status only; + - a missing workspace: no hits, not unavailable; + - a query holding a secret: no request at all, and the secret line; + - nothing found: `(no matches for "zebra")`, then any `unavailable` line; + - the local hits filling `k`, and the two labels joined. + + A key in it is built from pieces; the waits have limits (the whole file takes about 3 s). +- The calls the bodies use: `MessageSearch::new`, `Client::search_workspace`, + `Result::map_err`, `classify`, `conclusions` (written), `items`; `usize::try_from`, + `interleave`, `unavailable`, `label`, `render`, `DataClass::max`, `slice::join`. + +## The edits, exactly + +### `crates/memoryd/src/honcho/mod.rs` + +**H1.** Add one line. The line + +```rust +pub mod report; +``` + +becomes + +```rust +pub mod report; +pub mod search; +``` + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, two commands: + + ```sh + cp docs/plans/M5c2/files/crates/memoryd/tests/honcho_search.rs crates/memoryd/tests/ + cp docs/plans/M5c2/files/crates/memoryd/tests/support/honcho.rs crates/memoryd/tests/support/ + ``` + + Then `git status --short` lists `crates/memoryd/tests/support/honcho.rs` as modified and + `honcho_search.rs` as new. +- [ ] **2. See them fail.** `cargo build --tests --keep-going -p memoryd`, with a shell timeout of + 600000 ms: one test target, `honcho_search`, does not compile: `E0432`, ``unresolved import + `memoryd::honcho::search` ``. Every other target compiles. +- [ ] **3. The skeleton and the edit.** Copy the skeleton, then make edit H1: + + ```sh + cp docs/plans/M5c2/files/crates/memoryd/src/honcho/search.rs crates/memoryd/src/honcho/ + ``` + + `cargo check -p memoryd`: no error, and 12 warnings, all from the unfilled bodies: unused + imports (`MessageSearch`, `Source`, and `interleave`, `items`, `label`, `render`, + `unavailable`), unused variables (`reader`, `owner`, `query`, `k`, `local`, `searched`), and + the function `conclusions`, never used. +- [ ] **4. Fill the bodies**, one at a time, from their comments, with `cargo check -p memoryd` + after each: `source`, then `answer`. Delete each instruction comment and keep the doc comment + above the function. After the last, `cargo check -p memoryd` prints no warning. +- [ ] **5. The tests.** With a shell timeout of 600000 ms, every test passes: 13 in + `honcho_client`, 3 in `honcho_keys` (and 1 ignored, as before), 8 in `honcho_read`, 12 in + `honcho_search`, 12 in `honcho_worker`: + + ```sh + cargo test -p memoryd --test honcho_search --test honcho_read --test honcho_client \ + --test honcho_keys --test honcho_worker + ``` +- [ ] **6. Check.** `git diff --stat` lists exactly `crates/memoryd/src/honcho/mod.rs` and + `crates/memoryd/tests/support/honcho.rs`, and `git status --short` shows `search.rs` and + `honcho_search.rs` as new. +- [ ] **7. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`. The test count is the passed total of `cargo test --workspace` + plus one, for the ignored `end_to_end` test that the gate runs on its own: **1430** after this + task, 1418 before it; the 12 new ones are the tests of `honcho_search.rs`. To count, add up + the `N passed` of every `test result:` line the gate's `cargo test --workspace` printed, and + add 1. Never run the gate again only to count. +- [ ] **8. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is + `M5c2/08-memoryd-honcho-search`. 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 1430 tests. +- `git diff HEAD~1 --stat` lists only `mod.rs`, `search.rs`, `honcho_search.rs`, + `support/honcho.rs` and `docs/implementer-log.md`. + +## Stop and report if + +- The branch check at the top fails. +- The "before" line of H1 is not in the file exactly once. +- Step 2 or 3 shows a different result: other targets failing, other errors, or warnings about + anything but the unfilled bodies. +- `cargo check` reports an error after a body, and its comment does not tell you how to fix it. + Quote the error, and stop. +- After step 4 a test fails although each body follows its comment. Do not change a test, the + fake or the written parts of the skeleton: stop, and give the failing tests' names and their + messages. +- The fix seems to need a change anywhere else: `serve.rs`, `recall.rs`, `read.rs`, + `recalled.rs`, a new dependency. It does not; stop and say why you think so. +- A test outside the given files fails in the gate: check once whether it fails without your + change (`git stash -u`, run it, `git stash pop`), then stop and report both results. +- The gate passes with a count other than 1430. +- A comment in the skeleton does not fit its signature. Quote it and stop. + +## What the given tests pin + +Each of these changes to the finished code makes at least one given test fail (19 tried, all +caught): the secret check skipped; the search without its limit, or a conclusion query without +`top_k` or with another observer; the conclusions about the agent's peer or the owner's left +out, or asked of the fleet; a failed or a late source not reported unavailable; no limit on the +wait; the sources asked one after another; the local items left out, or Honcho's numbered from +1 or interleaved as if there were none; no `(no matches …)` line; the `unavailable` lines left +out; either part's label ignored; the secret line's words. diff --git a/docs/plans/M5c2/README.md b/docs/plans/M5c2/README.md index b584018..ff0d9b9 100644 --- a/docs/plans/M5c2/README.md +++ b/docs/plans/M5c2/README.md @@ -80,11 +80,12 @@ Honcho sources and chats outside its state lock, and answers `ask_memory`. | 05 | `memoryd`: read clients, one per `[[honcho.read]]` workspace, the fleet and the agent's own two, with a 5 s limit for searches and 150 s for chat; each source's label; `recall_sources` and `resolve`; `own`, `untrusted` and `fleet` reserved as read agents' names | memoryd | | 06 | `memoryd`, pure: Honcho messages and conclusions as `recall` items (tag, source name, class, neutralised, capped), kept only when they share a term with the query, interleaved with the local hits to `k`, the local index first | memoryd | | 07 | `memoryd`: local `recall` items in task 06's form, one indented line each, through one shared `snippet` and `item`; `recall::local` gives the items, their count and their label apart (T37) | memoryd | -| 08 | `memoryd`: `recall`'s Honcho sources, each on its own thread with its own limit, after the local search and outside the lock; `: unavailable`; a query holding a secret sends nothing | memoryd | -| 09 | `memoryd`: `ask_memory`: workspace names checked against `[honcho]`, both chats at once, each answer tagged and capped, labels, `: unavailable`; a question holding a secret sends nothing | memoryd | -| 10 | `memoryd` serves `ask_memory` beside `recall` and `memory`, outside the lock; `bxctl`'s usage and the runbook where a new message needs a pointer | memoryd, bxctl | +| 08 | `memoryd`: `recall`'s Honcho part (`honcho/search.rs`): every source of `recall_sources` at once, each on its own thread with a limit of 5 s from the start; a read agent's workspace is a search and two conclusion queries; `: unavailable`; a query holding a secret asks nothing; the answer from the local items and Honcho's, with the stricter label | memoryd | +| 09 | `memoryd`: `recall` with `[honcho]` runs the local part under the state lock and the Honcho part outside it; the readers taken at start; without `[honcho]`, `recall` as before | memoryd | +| 10 | `memoryd`: `ask_memory`: workspace names checked against `[honcho]`, both chats at once, each answer tagged and capped, labels, `: unavailable`; a question holding a secret sends nothing | memoryd | +| 11 | `memoryd` serves `ask_memory` beside `recall` and `memory`, outside the lock; `bxctl`'s usage and the runbook where a new message needs a pointer | memoryd, bxctl | -After task 10, the design model runs the M5c-2 device check (spec 13) on straylight, rift and the +After task 11, the design model runs the M5c-2 device check (spec 13) on straylight, rift and the owner's Mattermost: a second `memoryd` under its own name writes a fleet note that `recall` finds, tagged `memory-tainted`, turning the session untrusted; `ask_memory` answers from the agent's own trusted workspace within its limit; with rift down, `recall` answers from the local index with @@ -105,6 +106,7 @@ The gate count is the passed total of `cargo test --workspace` plus one, for the | 05 | 1403 | 24, 24 | | 06 | 1414 | 22, 22 | | 07 | 1418 | 15, 15 | +| 08 | 1430 | 19, 19 | **A flaky test, fixed before the run:** `bxctl`'s `chat_approvals` test `json_prints_only_json_and_never_asks_brokerd` failed once in 5 loaded runs of the whole workspace diff --git a/docs/plans/M5c2/files/crates/memoryd/src/honcho/search.rs b/docs/plans/M5c2/files/crates/memoryd/src/honcho/search.rs new file mode 100644 index 0000000..fec0a14 --- /dev/null +++ b/docs/plans/M5c2/files/crates/memoryd/src/honcho/search.rs @@ -0,0 +1,165 @@ +//! `recall`'s Honcho part (M5c spec, section 8): every source `Readers::recall_sources` names, +//! searched at once, each on its own thread with its own limit, then its answer made from the +//! local index's items followed by Honcho's. A query that holds a secret asks nothing. Runs +//! outside `memoryd`'s state lock; a source still running at its limit is left to finish, and +//! what it finds is dropped. + +use std::sync::{Arc, mpsc}; +use std::time::{Duration, Instant}; + +use bxcore::RecallArgs; +use roncho::{Conclusion, ConclusionQuery, MessageSearch}; +use serde_json::{Map, json}; + +use crate::honcho::client::{HonchoError, classify}; +use crate::honcho::decide::has_secret; +use crate::honcho::read::{Reader, Readers, Source}; +use crate::honcho::recalled::{Item, interleave, items, label, render, unavailable}; +use crate::recall::{Local, Recalled}; +use crate::serve::Log; + +/// How long each source has, counted from when the Honcho part starts (M5c spec, section 8). +pub const SOURCE_WAIT: Duration = Duration::from_secs(5); + +/// The line in place of Honcho's items when the query holds a known-prefix secret. +pub const SECRET: &str = "honcho: not asked; the query looks like it holds a secret"; + +/// What the Honcho part found. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Searched { + /// The query holds a secret: no source was asked. + Secret, + /// Each source's kept items, in `recall_sources` order (none for a source that did not + /// answer), and the names of those that did not, in the same order. + Asked { + found: Vec>, + unavailable: Vec, + }, +} + +/// The conclusions about `peer` that `peer` drew, as `memoryd` writes them: the observer and the +/// observed are the same peer (M5c spec, section 8). +fn conclusions( + reader: &Reader, + peer: &str, + query: &str, + k: u32, +) -> Result, HonchoError> { + let mut filters = Map::new(); + filters.insert("observer_id".to_string(), json!(peer)); + filters.insert("observed_id".to_string(), json!(peer)); + let asked = ConclusionQuery::new(query).top_k(k).filters(filters); + reader + .search() + .query_conclusions(&asked) + .map_err(|e| classify(&e)) +} + +/// One source, its calls one after another: the workspace's messages, then for another agent's +/// workspace the conclusions about that agent's peer and about the owner's. Its kept items, or +/// the first call's failure. +pub fn source(reader: &Reader, owner: &str, query: &str, k: u32) -> Result, HonchoError> { + // 1. messages: `reader.search().search_workspace(&MessageSearch { limit: Some(k), + // ..MessageSearch::new(query) })`, its error mapped with `classify(&e)`, then `?`. + // 2. For `Source::Agent(peer)`: about_agent = `conclusions(reader, peer, query, k)?`, then + // about_owner = `conclusions(reader, owner, query, k)?`, in that order. Any other source: + // both `Vec::new()`, and no conclusion is asked for. + // 3. `Ok(items(&reader.label, query, &messages, &about_agent, &about_owner))`. + todo!() +} + +/// Ask every source of `readers.recall_sources()` at once for `args`, each on a thread of its +/// own, and wait for each at most `wait` from now. A source that failed, or that has not answered +/// by then, is unavailable; each is logged by its name and the error's kind and status, never the +/// query. `Secret`, with nothing sent, when the query holds a known-prefix secret. +pub fn search( + readers: &Arc, + owner: &str, + args: &RecallArgs, + wait: Duration, + log: &Log, +) -> Searched { + if has_secret(&args.query) { + return Searched::Secret; + } + let deadline = Instant::now() + wait; + let k = u32::try_from(args.k).unwrap_or(u32::MAX); + let names: Vec = readers + .recall_sources() + .iter() + .map(|r| r.label.name.clone()) + .collect(); + let (tx, rx) = mpsc::channel::<(usize, Result, HonchoError>)>(); + let mut answers: Vec, HonchoError>>> = vec![None; names.len()]; + for (i, slot) in answers.iter_mut().enumerate() { + let (readers, sender) = (Arc::clone(readers), tx.clone()); + let (owner, query) = (owner.to_string(), args.query.clone()); + let spawned = std::thread::Builder::new() + .name("honcho-recall".to_string()) + .spawn(move || { + if let Some(reader) = readers.recall_sources().get(i) { + let _ = sender.send((i, source(reader, &owner, &query, k))); + } + }); + if spawned.is_err() { + *slot = Some(Err(HonchoError::Unreachable { + kind: "thread", + status: None, + })); + } + } + drop(tx); + while answers.iter().any(Option::is_none) { + let left = deadline.saturating_duration_since(Instant::now()); + match rx.recv_timeout(left) { + Ok((i, answer)) => { + if let Some(slot) = answers.get_mut(i) { + *slot = Some(answer); + } + } + Err(_) => break, + } + } + let mut found = Vec::with_capacity(names.len()); + let mut missing = Vec::new(); + for (name, answer) in names.into_iter().zip(answers) { + match answer { + Some(Ok(kept)) => found.push(kept), + Some(Err(e)) => { + (log)(&format!("memoryd: recall: {name} unavailable: {e}")); + found.push(Vec::new()); + missing.push(name); + } + None => { + (log)(&format!( + "memoryd: recall: {name} unavailable: no answer in {} ms", + wait.as_millis() + )); + found.push(Vec::new()); + missing.push(name); + } + } + } + Searched::Asked { + found, + unavailable: missing, + } +} + +/// `recall`'s answer from the local index's items and what Honcho found: the local items, then +/// Honcho's up to `k` in all, numbered after them; `(no matches for "…")` when there is neither; +/// then the `unavailable` lines, or the secret line. The label is the stricter of the two parts'. +pub fn answer(local: Local, query: &str, k: u64, searched: Searched) -> Recalled { + // 1. k as a usize (`usize::try_from(k).unwrap_or(usize::MAX)`). + // 2. (shown, tail): `Secret` -> (no items, `SECRET.to_string()`); `Asked { found, + // unavailable: missing }` (bound as `missing`: the field has the function's name) -> + // (`interleave(local.items.len(), k, found)`, `unavailable(&names)` where names is + // `missing` as a `Vec<&str>`). + // 3. (class, untrusted) = `label(&shown)`; `first` = local.items.len() + 1, taken before the + // items move. parts: `local.items`; then, if `shown` is not empty, `render(first, + // &shown)`; then, if parts is still empty, format!("(no matches for \"{query}\")"); then + // `tail` if it is not empty. + // 4. Recalled { text: parts joined by "\n", class: `local.class.max(class)`, untrusted: + // `local.untrusted || untrusted` }. + todo!() +} diff --git a/docs/plans/M5c2/files/crates/memoryd/tests/honcho_search.rs b/docs/plans/M5c2/files/crates/memoryd/tests/honcho_search.rs new file mode 100644 index 0000000..82123c6 --- /dev/null +++ b/docs/plans/M5c2/files/crates/memoryd/tests/honcho_search.rs @@ -0,0 +1,419 @@ +//! `recall`'s Honcho part (M5c spec, section 8), against a fake Honcho that answers from the +//! responses recorded from rift: every read workspace and the fleet's searched at once, never the +//! agent's own; a source that hangs or fails is one `unavailable` line after its limit while the +//! others answer; a missing workspace is no hits; a query holding a secret asks nothing; and the +//! answer made from the local items and Honcho's, with the stricter label. Do not edit. + +#[path = "support/honcho.rs"] +mod honcho; + +use std::collections::BTreeMap; +use std::ffi::OsString; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bxcore::{DataClass, RecallArgs}; +use honcho::{FakeHoncho, Mode}; +use memoryd::honcho::config::{HonchoConfig, ReadWorkspace, Trust}; +use memoryd::honcho::read::{ReadTimeouts, Readers}; +use memoryd::honcho::search::{SECRET, SOURCE_WAIT, Searched, answer, search}; +use memoryd::recall::Local; +use memoryd::serve::Log; + +const FAST: ReadTimeouts = ReadTimeouts { + connect: Duration::from_secs(5), + search: Duration::from_secs(5), + chat: Duration::from_secs(5), +}; +/// The limit the tests give each source: long, so that a loaded machine never makes a source +/// that answers late; a search where every source answers returns at once. +const WAIT: Duration = Duration::from_secs(10); +/// The limit when a source hangs: the others still answer well inside it. +const HANG: Duration = Duration::from_millis(1_500); + +fn read(agent: &str, trust: Trust) -> ReadWorkspace { + ReadWorkspace { + agent: agent.to_string(), + workspace: format!("{agent}-trusted"), + trust, + class: DataClass::Private, + } +} + +/// `boxmaker0`, owned by `kyle`, reading `boxmaker1` (labelled trusted) and `boxmaker2`. +fn readers(fake: &FakeHoncho, read: Vec) -> Arc { + let config = HonchoConfig { + url: fake.url.clone(), + agent: "boxmaker0".to_string(), + owner: "kyle".to_string(), + trusted: "boxmaker0-trusted".to_string(), + untrusted: "boxmaker0-untrusted".to_string(), + fleet: "fleet".to_string(), + read, + keys: BTreeMap::new(), + }; + let env = |_: &str| -> Option { None }; + Arc::new(Readers::new(&config, &env, FAST).unwrap().0) +} + +fn two(fake: &FakeHoncho) -> Arc { + readers( + fake, + vec![ + read("boxmaker1", Trust::Trusted), + read("boxmaker2", Trust::Untrusted), + ], + ) +} + +/// A log that keeps its lines. +fn log() -> (Log, Arc>>) { + let lines = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&lines); + let log: Log = Arc::new(move |line: &str| sink.lock().unwrap().push(line.to_string())); + (log, lines) +} + +fn args(query: &str, k: u64) -> RecallArgs { + RecallArgs { + query: query.to_string(), + k, + } +} + +fn nothing_local() -> Local { + Local { + items: Vec::new(), + class: DataClass::Private, + untrusted: false, + } +} + +fn asked(searched: &Searched) -> (Vec>, Vec) { + match searched { + Searched::Asked { found, unavailable } => ( + found + .iter() + .map(|items| items.iter().map(|i| i.text.clone()).collect()) + .collect(), + unavailable.clone(), + ), + Searched::Secret => panic!("not asked"), + } +} + +#[test] +fn the_limit() { + assert_eq!(SOURCE_WAIT, Duration::from_secs(5)); + assert_eq!( + SECRET, + "honcho: not asked; the query looks like it holds a secret" + ); +} + +/// A read agent's workspace: its messages, the conclusions about its peer and about the owner's; +/// the fleet's: its messages. Each kept only with a word of the query, once. +#[test] +fn every_source_is_searched_with_its_calls() { + let fake = FakeHoncho::start(); + let (log, lines) = log(); + let searched = search(&two(&fake), "kyle", &args("tea", 7), WAIT, &log); + let (found, unavailable) = asked(&searched); + let tea = [ + "I prefer tea to coffee in the afternoon.", + "Noted: tea in the afternoon.", + ]; + let with_owner = [tea[0], tea[1], "Kyle prefers tea in the afternoon."]; + assert_eq!(found, [&with_owner[..], &with_owner[..], &tea[..]]); + assert_eq!(unavailable, Vec::::new()); + assert!(lines.lock().unwrap().is_empty()); + let mut seen: Vec<(String, serde_json::Value)> = fake + .requests() + .into_iter() + .map(|r| (r.path, r.body.unwrap_or_default())) + .collect(); + seen.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.to_string().cmp(&b.1.to_string()))); + let query = |peer: &str| { + serde_json::json!({ "query": "tea", "top_k": 7, + "filters": { "observer_id": peer, "observed_id": peer } }) + }; + let search = serde_json::json!({ "query": "tea", "limit": 7 }); + let w = "/v3/workspaces"; + assert_eq!( + seen, + [ + ( + format!("{w}/boxmaker1-trusted/conclusions/query"), + query("boxmaker1") + ), + ( + format!("{w}/boxmaker1-trusted/conclusions/query"), + query("kyle") + ), + (format!("{w}/boxmaker1-trusted/search"), search.clone()), + ( + format!("{w}/boxmaker2-trusted/conclusions/query"), + query("boxmaker2") + ), + ( + format!("{w}/boxmaker2-trusted/conclusions/query"), + query("kyle") + ), + (format!("{w}/boxmaker2-trusted/search"), search.clone()), + (format!("{w}/fleet/search"), search), + ] + ); +} + +/// The agent's own workspaces hold what the local index holds: `recall` never searches them. +#[test] +fn the_agents_own_workspaces_are_never_searched() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + search(&two(&fake), "kyle", &args("tea", 5), WAIT, &log); + let requests = fake.requests(); + assert_eq!(requests.len(), 7); + for request in requests { + assert!(!request.path.contains("boxmaker0"), "{}", request.path); + } +} + +/// The answer shows each source's hits with its tag, numbered after the local items, one from +/// each source in turn. +#[test] +fn the_answer_shows_each_sources_hits_with_its_tag() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + let searched = search(&two(&fake), "kyle", &args("tea", 4), WAIT, &log); + let local = Local { + items: vec![ + "[1] memory · session mm-a, 2026-09-20 14:02 UTC, the owner said · private\n tea time" + .to_string(), + ], + class: DataClass::Private, + untrusted: false, + }; + let got = answer(local, "tea", 4, searched); + let headers: Vec<&str> = got + .text + .lines() + .filter(|l| l.starts_with('[')) + .map(|l| l.split(" · ").next().unwrap()) + .collect(); + assert_eq!( + headers, + [ + "[1] memory", + "[2] memory", + "[3] memory-tainted", + "[4] memory-tainted" + ] + ); + assert!( + got.text.contains("\n[2] memory · boxmaker1: kyle said, "), + "{}", + got.text + ); + assert!( + got.text + .contains("\n[4] memory-tainted · fleet: kyle said, "), + "{}", + got.text + ); + assert_eq!((got.class, got.untrusted), (DataClass::Private, true)); +} + +/// T58: a fleet hit makes the answer untrusted; a hit from a workspace the owner labelled trusted +/// does not. +#[test] +fn a_fleet_hit_taints_the_answer_and_a_trusted_workspaces_does_not() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + let one = readers(&fake, vec![read("boxmaker1", Trust::Trusted)]); + fake.set_workspace("fleet", Mode::Missing); + let got = answer( + nothing_local(), + "tea", + 5, + search(&one, "kyle", &args("tea", 5), WAIT, &log), + ); + assert!( + got.text.contains("[1] memory · boxmaker1: "), + "{}", + got.text + ); + assert_eq!((got.class, got.untrusted), (DataClass::Private, false)); + fake.set_workspace("fleet", Mode::Fixtures); + fake.set_workspace("boxmaker1-trusted", Mode::Missing); + let got = answer( + nothing_local(), + "tea", + 5, + search(&one, "kyle", &args("tea", 5), WAIT, &log), + ); + assert!( + got.text.starts_with("[1] memory-tainted · fleet: "), + "{}", + got.text + ); + assert_eq!((got.class, got.untrusted), (DataClass::Private, true)); +} + +/// A source that does not answer is one `unavailable` line after its limit; the others answer. +#[test] +fn a_hanging_source_is_unavailable_and_the_others_still_answer() { + let fake = FakeHoncho::start(); + fake.set_workspace("boxmaker2-trusted", Mode::Hang); + let (log, lines) = log(); + let started = Instant::now(); + let searched = search(&two(&fake), "kyle", &args("tea", 10), HANG, &log); + let took = started.elapsed(); + assert!(took >= HANG, "{took:?}"); + assert!(took < Duration::from_secs(6), "{took:?}"); + let (found, unavailable) = asked(&searched); + assert_eq!(unavailable, ["boxmaker2"]); + assert_eq!(found.iter().map(Vec::len).collect::>(), [3, 0, 2]); + assert_eq!( + *lines.lock().unwrap(), + ["memoryd: recall: boxmaker2 unavailable: no answer in 1500 ms"] + ); + let got = answer(nothing_local(), "tea", 10, searched); + assert!( + got.text.ends_with("\nboxmaker2: unavailable"), + "{}", + got.text + ); +} + +/// The sources are asked at once: when every one hangs, the answer comes after one limit, not +/// one per source. +#[test] +fn every_source_is_asked_at_once() { + let fake = FakeHoncho::start(); + fake.set(Mode::Hang); + let (log, _) = log(); + let started = Instant::now(); + let (found, unavailable) = asked(&search(&two(&fake), "kyle", &args("tea", 5), HANG, &log)); + let took = started.elapsed(); + assert!(took < Duration::from_millis(4_500), "{took:?}"); + assert_eq!(unavailable, ["boxmaker1", "boxmaker2", "fleet"]); + assert!(found.iter().all(Vec::is_empty)); +} + +/// A source whose call fails is unavailable, logged by the error's kind and status only. +#[test] +fn a_failing_source_is_unavailable_and_logged_without_the_query() { + let fake = FakeHoncho::start(); + fake.set_workspace("fleet", Mode::Status(503)); + fake.set_workspace("boxmaker1-trusted", Mode::Status(422)); + let (log, lines) = log(); + let searched = search(&two(&fake), "kyle", &args("tea", 5), WAIT, &log); + let (found, unavailable) = asked(&searched); + assert_eq!(unavailable, ["boxmaker1", "fleet"]); + assert_eq!(found[1].len(), 3); + assert_eq!( + *lines.lock().unwrap(), + [ + "memoryd: recall: boxmaker1 unavailable: Honcho refused the request (status, HTTP 422)", + "memoryd: recall: fleet unavailable: Honcho is unreachable (server, HTTP 503)", + ] + ); + let got = answer(nothing_local(), "tea", 5, searched); + assert!( + got.text + .ends_with("\nboxmaker1: unavailable\nfleet: unavailable"), + "{}", + got.text + ); +} + +/// Honcho answers a search or a conclusion query in a workspace that does not exist with 200 and +/// no hits: the source answered, with nothing. +#[test] +fn a_missing_workspace_is_no_hits() { + let fake = FakeHoncho::start(); + fake.set_workspace("boxmaker1-trusted", Mode::Missing); + let (log, lines) = log(); + let (found, unavailable) = asked(&search(&two(&fake), "kyle", &args("tea", 5), WAIT, &log)); + assert_eq!(found[0], Vec::::new()); + assert_eq!(unavailable, Vec::::new()); + assert!(lines.lock().unwrap().is_empty()); +} + +/// A query that holds a known-prefix secret is sent to no source (M5c spec, section 8). +#[test] +fn a_query_holding_a_secret_asks_nothing() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + let token = format!("{}{}", ["gh", "p_"].concat(), "A1b2".repeat(9)); + let searched = search( + &two(&fake), + "kyle", + &args(&format!("tea {token}"), 5), + WAIT, + &log, + ); + assert_eq!(searched, Searched::Secret); + assert_eq!(fake.requests().len(), 0); + let local = Local { + items: vec!["[1] memory · x · private\n tea".to_string()], + class: DataClass::Private, + untrusted: false, + }; + let got = answer(local, "tea", 5, Searched::Secret); + assert_eq!( + got.text, + format!("[1] memory · x · private\n tea\n{SECRET}") + ); + assert_eq!((got.class, got.untrusted), (DataClass::Private, false)); +} + +/// Neither the local index nor Honcho has anything: the sentence local `recall` gives, then any +/// `unavailable` lines. +#[test] +fn when_nothing_is_found_the_answer_says_so() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + let searched = search(&two(&fake), "kyle", &args("zebra", 5), WAIT, &log); + let got = answer(nothing_local(), "zebra", 5, searched); + assert_eq!(got.text, "(no matches for \"zebra\")"); + assert_eq!((got.class, got.untrusted), (DataClass::Private, false)); + let got = answer( + nothing_local(), + "zebra", + 5, + Searched::Asked { + found: vec![Vec::new()], + unavailable: vec!["fleet".to_string()], + }, + ); + assert_eq!(got.text, "(no matches for \"zebra\")\nfleet: unavailable"); +} + +/// The local index first: when its hits fill `k`, no Honcho hit is shown and none labels the +/// answer. The label is the stricter of the two parts'. +#[test] +fn local_hits_come_first_and_the_labels_join() { + let fake = FakeHoncho::start(); + let (log, _) = log(); + let local = Local { + items: vec![ + "[1] memory · a · secret\n tea".to_string(), + "[2] memory · b · private\n tea".to_string(), + ], + class: DataClass::Secret, + untrusted: false, + }; + let searched = search(&two(&fake), "kyle", &args("tea", 2), WAIT, &log); + let got = answer(local.clone(), "tea", 2, searched.clone()); + assert_eq!(got.text, local.items.join("\n")); + assert_eq!((got.class, got.untrusted), (DataClass::Secret, false)); + let got = answer(local, "tea", 3, searched); + assert_eq!(got.text.lines().filter(|l| l.starts_with('[')).count(), 3); + assert!( + got.text.contains("\n[3] memory · boxmaker1: "), + "{}", + got.text + ); + assert_eq!((got.class, got.untrusted), (DataClass::Secret, false)); +} diff --git a/docs/plans/M5c2/files/crates/memoryd/tests/support/honcho.rs b/docs/plans/M5c2/files/crates/memoryd/tests/support/honcho.rs new file mode 100644 index 0000000..1376157 --- /dev/null +++ b/docs/plans/M5c2/files/crates/memoryd/tests/support/honcho.rs @@ -0,0 +1,321 @@ +//! 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 (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, to answer nothing, or to answer as a workspace that does not exist, for every +//! workspace or for one. 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::{HashMap, 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, + /// As a workspace that does not exist: a search or a conclusion query answers 200 with no + /// hits (fixtures 34, 35), chat 404 (36). + Missing, +} + +struct Shared { + mode: Mutex, + /// A mode for one workspace's requests, by its name, instead of `mode`. + workspaces: 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), + workspaces: Mutex::new(HashMap::new()), + 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; + } + + /// Answer `workspace`'s requests in `mode`, whatever `set` says for the others. + pub fn set_workspace(&self, workspace: &str, mode: Mode) { + self.shared + .workspaces + .lock() + .unwrap() + .insert(workspace.to_string(), 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"]) => { + first(fixture("08-messages-add-batch.json"), request, "messages") + } + ("POST", ["v3", "workspaces", _, "conclusions"]) => first( + fixture("11-conclusions-create-batch.json"), + request, + "conclusions", + ), + // A search answers the three recorded messages about tea (fixture 30), whatever it asked. + ("POST", ["v3", "workspaces", _, "search"]) => fixture("30-workspace-search-limit.json"), + // A conclusion query answers the recorded one about the owner (32) when it asks about + // `kyle`, else the one about an agent (33). + ("POST", ["v3", "workspaces", _, "conclusions", "query"]) => { + let about = request + .body + .as_ref() + .and_then(|b| b["filters"]["observed_id"].as_str().map(str::to_string)); + if about.as_deref() == Some("kyle") { + fixture("32-conclusions-query-owner.json") + } else { + fixture("33-conclusions-query-agent.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()), + } +} + +/// 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 { + return; + }; + shared.requests.lock().unwrap().push(request.clone()); + let workspace = request.path.split('/').nth(3).unwrap_or("").to_string(); + let mode = match shared.workspaces.lock().unwrap().get(&workspace) { + Some(mode) => *mode, + None => *shared.mode.lock().unwrap(), + }; + let (status, body) = match mode { + Mode::Fixtures => answer(&request, &shared.made), + Mode::Missing if request.path.ends_with("/search") => { + fixture("34-workspace-search-missing-workspace.json") + } + Mode::Missing if request.path.ends_with("/conclusions/query") => { + fixture("35-conclusions-query-missing-workspace.json") + } + Mode::Missing => fixture("36-workspace-chat-missing-workspace.json"), + 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()); +}