From d4b7f42b56348d6c6eb0adfb1b2d26c2274459c4 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 06:14:58 -0700 Subject: [PATCH] M5b plan: task 08, the self-test checks the tags against the tokenizer loopd's self-test gains a check, run second (after "server matches config", before any prompt): each tag in bxcore::sanitize::TAGS must be one token from /tokenize, its recorded id, and its neutralised form more than one. A failure names the tag by its id, never its text, and points at a new runbook entry, loopd-tags-unexpected, given as a fragment; SelfTestError::pointer picks the entry per failure. One request per text (66): /tokenize has no batch form. The fake server gains respond and a fake tokenizer; selftest.rs, serve.rs, pointers.rs and support/mod.rs get M5b copies. Checks, on a reference at ref 08 (titan, rustc 1.98.1, own target/): - /tokenize fields from llama.cpp tools/server/README.md at b10964 (b29c606) and its handler; on straylight all 33 tags gave their ids and every neutralised form 7 to 11 tokens, 66 requests in ~0.3 s - given tests before the change: 4 fail as the task lists, selftest_tags does not compile (7 E0599); pointers passes - make gate: gate: ok, 1050 tests (1042 + 8); /tmp and $TMPDIR unchanged around it; the four suites 25 runs under load, no failure - 20 mutations, each caught: id not compared; neutralised check dropped; > 1 as >= 1, as == 1 missing 0, as > 2; first or last tag skipped; tag asked twice; pointer dropped or wrong (all or Tokenize); check skipped in serve; check run last; parse_special false; add_special left out; id missing from the message; the tag's text in the message; tokenize error as Infer; wrong count; wrong other id - the task's edits applied to a fresh ref-07 copy give files byte-identical to the reference; every before block found once - no chat-template tag whole in the task or given files; task lines within 100 characters except tables and verbatim lines; given files under 500 lines Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/plans/M5b/08-loopd-selftest-tags.md | 472 +++++++++++++++++ .../M5b/files/crates/loopd/tests/pointers.rs | 247 +++++++++ .../M5b/files/crates/loopd/tests/selftest.rs | 177 +++++++ .../files/crates/loopd/tests/selftest_tags.rs | 349 +++++++++++++ .../M5b/files/crates/loopd/tests/serve.rs | 128 +++++ .../files/crates/loopd/tests/support/mod.rs | 474 ++++++++++++++++++ .../files/runbook-loopd-tags-unexpected.md | 53 ++ 7 files changed, 1900 insertions(+) create mode 100644 docs/plans/M5b/08-loopd-selftest-tags.md create mode 100644 docs/plans/M5b/files/crates/loopd/tests/pointers.rs create mode 100644 docs/plans/M5b/files/crates/loopd/tests/selftest.rs create mode 100644 docs/plans/M5b/files/crates/loopd/tests/selftest_tags.rs create mode 100644 docs/plans/M5b/files/crates/loopd/tests/serve.rs create mode 100644 docs/plans/M5b/files/crates/loopd/tests/support/mod.rs create mode 100644 docs/plans/M5b/files/runbook-loopd-tags-unexpected.md diff --git a/docs/plans/M5b/08-loopd-selftest-tags.md b/docs/plans/M5b/08-loopd-selftest-tags.md new file mode 100644 index 0000000..750d91f --- /dev/null +++ b/docs/plans/M5b/08-loopd-selftest-tags.md @@ -0,0 +1,472 @@ +# M5b task 08: the self-test checks the model's tags against the tokenizer + +**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `loopd: the self-test checks the model's tags against the tokenizer` + +## Read this first: the markup strings + +This task checks your own model's chat-template tags (the markup strings) against the inference +server. Your model uses these same strings to mark its reasoning and its tool calls. **If you +spell one out, even in your thinking, it ends your reasoning or your turn.** So: + +- **Do not open** the given test file `selftest_tags.rs` or the runbook fragment, and do not + work through their cases in your reasoning. You copy them; you never need to read them. No + failure message shows a tag: a failure names a tag by its token id. +- Nothing you write needs a tag. The code walks `bxcore::sanitize::TAGS` (task 03) and calls + `neutralise`; it never looks at what a tag says. Do not write a tag anywhere, as tasks 03 to 07 + said. + +## Goal + +`loopd` neutralises the model's tags in every text from outside (task 07), using the list +`TAGS` and the token id recorded with each tag (task 03). That list was measured against one +model file. After a model or template change, a tag could stop being the token the list says, +and the list would no longer protect the model. So the self-test, which `loopd selftest` and +`loopd serve` both run before `loopd` serves anyone, gains a check (M5b spec, section 8): for +each tag in `TAGS`, the server's `/tokenize` must give **exactly one token, the tag's recorded +id**, and for the tag's neutralised form, **more than one token**. If not, `loopd` does not +start, and says why with a new runbook entry, `loopd-tags-unexpected`. + +The check runs **second**, right after "server matches config" and before the first prompt: like +that check it sends no prompt and uses no slot, so a server whose tokenizer is not the recorded +one is sent no prompt either. The steps become: + +```text +selftest: server matches config +selftest: tags tokenize as recorded +selftest: tool call round trip +selftest: turn 2 cache hit +selftest: ok +``` + +A failure names the tag by its **token id, never by its text**: the line goes to the operator's +terminal and log, which other programs (and models) may read, and a tag written out whole there +is the very thing this milestone removes. The operator finds the tag with +`grep -n crates/bxcore/src/sanitize/tags.rs`. The four failures, as printed: + +```text +selftest: FAILED: the tag with id 248046 is 6 tokens, not one; see docs/runbook.md#loopd-tags-unexpected +selftest: FAILED: the tag with id 248046 is one token, but the token 248047; see docs/runbook.md#loopd-tags-unexpected +selftest: FAILED: the neutralised form of the tag with id 248046 does not split into several tokens (the server gave 1); see docs/runbook.md#loopd-tags-unexpected +selftest: FAILED: could not tokenize the tag with id 248044: the server responded with 404: …; see docs/runbook.md#loopd-tags-unexpected +``` + +Every other self-test failure keeps its line and its entry, `loopd-selftest-failed`. So the +pointer moves from `main.rs` into a method, `SelfTestError::pointer`, which gives each failure +its own. The gate's runbook check fails on a pointer without its entry, so this task appends the +entry to `docs/runbook.md`. + +| Name | What it is | +|---|---| +| `loopd::llama::Client::token_ids` | `pub fn token_ids(&self, text: &str) -> Result, InferError>`: new (I2) | +| `TokenIds` (private, `info.rs`) | `/tokenize`'s answer: `struct TokenIds { tokens: Vec }` (I1) | +| `loopd::selftest::SelfTestError` | gains four variants, last: `TagNotOneToken`, `TagOtherId`, `NeutralisedTooFew`, `Tokenize` (S2) | +| `SelfTestError::pointer` | `pub fn pointer(&self) -> &'static str`: the runbook pointer for the failure (S2) | +| `tags_tokenize`, `check_tag` (private) | the check (S5) | +| `loopd::selftest::run` | runs the check second (S4) | + +## Files + +- Copy, from `docs/plans/M5b/files/` (the same path under it): + - the new test `crates/loopd/tests/selftest_tags.rs` + - **new versions** of `crates/loopd/tests/selftest.rs`, `crates/loopd/tests/serve.rs`, + `crates/loopd/tests/pointers.rs` and `crates/loopd/tests/support/mod.rs` (they replace the + files in the tree). The fake server in `support/mod.rs` gains `respond`, a path whose reply + is computed from the request, and `tokenizer`, a fake `/tokenize`; the three other files + give their healthy fake server that tokenizer and expect the fourth step. +- Append, from `docs/plans/M5b/files/`, to the end of `docs/runbook.md`, with one blank line + before it: `runbook-loopd-tags-unexpected.md` (step 3 gives the command). +- Modify: `crates/loopd/src/llama/info.rs` (I1, I2), `crates/loopd/src/selftest.rs` (S1 to S6), + `crates/loopd/src/main.rs` (M1), `docs/implementer-log.md`. +- Do not touch: `crates/bxcore`, any `Cargo.toml`, any other file of `loopd`, any other crate. + +## Facts you need + +- **`/tokenize`** (llama.cpp `tools/server/README.md` at tag `b10964`, commit `b29c606`, the + build straylight runs; and its handler in `tools/server/server-context.cpp`). A `POST` with a + JSON body: + + | Field | Meaning | Default | + |---|---|---| + | `content` | the text to tokenize (required) | | + | `add_special` | insert special tokens such as BOS | `false` | + | `parse_special` | tokenize special tokens as such; when `false` they are plain text | `true` | + | `with_pieces` | answer objects `{"id", "piece"}` instead of bare ids | `false` | + + `token_ids` sends `model` (the router picks the model by it, as `tokenize` does today), + `content`, `add_special: false` and `parse_special: true`, the last two written out so the + check does not rest on a default: a chat prompt is tokenized with special parsing on, and that + is the case the check is about. It leaves `with_pieces` out, so the answer is bare ids: + + ```json + {"tokens": [248046]} + ``` + + The answer is a format another program defines, so `TokenIds` ignores unknown fields (no + `deny_unknown_fields`). An answer whose `tokens` is not a list of ids that fit in a `u32` (a + count, objects with pieces, a negative number) is `InferError::Protocol`. The request goes + through `Client::call`, like `props`: one connection, the `liveness_ms` limit, no retry. +- **One request per text.** The API has no batch: a `content` given as a list is tokenized into + one flat list of ids, which cannot be split back per text. So the check makes 66 requests, + each tag and then its neutralised form. On straylight (2026-09-26) all 66 took about 0.3 s, and + every tag gave its recorded id and every neutralised form 7 to 11 tokens. +- **What the check cannot do**: find a tag missing from `TAGS`. The server does not list its + special tokens. +- **The fake tokenizer** (`support::tokenizer`, for the tests): a request whose `content` is one + of `TAGS` gets that tag's id; any other text gets one token per byte. The tests then change one + answer at a time, for every tag in turn. +- `crates/loopd/tests/device.rs` (ignored; `make verify-device`, run by the owner against + straylight) runs the self-test, so the check runs there too. You do not run it. +- The calls the edits make, all given in full below: + + | Call | Signature, and what it gives | + |---|---| + | `bxcore::sanitize::TAGS` | `pub const TAGS: &[Tag]`, each `Tag { text: &'static str, id: u32 }` | + | `bxcore::sanitize::neutralise` | `fn neutralise(text: &str) -> Cow<'_, str>`; `&neutralise(t)` gives a `&str` | + | `Client::call` | `fn call(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result, InferError>` | + | `serde_json::from_slice` | `from_slice::(&bytes) -> Result` | + | slice patterns | `match ids.as_slice() { [got] if *got == id => …, [got] => …, _ => … }` | + +## The edits, exactly + +Every "before" block is in its file exactly once. Nothing else in these files changes. + +### `crates/loopd/src/llama/info.rs` + +**I1.** The lines + +```rust +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CacheOutcome { +``` + +become + +```rust +/// `/tokenize`'s answer without `with_pieces`: `{"tokens": [ids]}`. Other fields are ignored. +#[derive(serde::Deserialize)] +struct TokenIds { + tokens: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CacheOutcome { +``` + +**I2.** The lines (the end of `tokenize`) + +```rust + .ok_or_else(|| InferError::Protocol("tokenize response has no tokens".to_string())) + } +``` + +become + +```rust + .ok_or_else(|| InferError::Protocol("tokenize response has no tokens".to_string())) + } + + /// The token ids the server gives `text`, with special tokens parsed as a chat prompt's are + /// and no BOS added. The self-test checks the model's tags with it (M5b spec, section 8). + pub fn token_ids(&self, text: &str) -> Result, InferError> { + let body = serde_json::to_string(&serde_json::json!({ + "model": self.cfg.infer.model, + "content": text, + "add_special": false, + "parse_special": true, + })) + .map_err(|_| InferError::Protocol("could not encode tokenize request".to_string()))?; + let bytes = self.call("POST", "/tokenize", Some(body.as_bytes()))?; + let ids: TokenIds = serde_json::from_slice(&bytes).map_err(|_| { + InferError::Protocol("tokenize response has no list of token ids".to_string()) + })?; + Ok(ids.tokens) + } +``` + +`tokenize` itself does not change: the baseline's token count uses it. + +### `crates/loopd/src/selftest.rs` + +**S1.** The lines + +```rust +//! The startup self-test: before `loopd` serves anyone, it checks that the server is the one its +//! config describes, that a tool call comes back parsed, and that a second turn reuses the first +//! turn's cache. Any check failing means `loopd` does not start. + +use crate::llama::{ +``` + +become + +```rust +//! The startup self-test: before `loopd` serves anyone, it checks that the server is the one its +//! config describes, that the model's tags tokenize as recorded, that a tool call comes back +//! parsed, and that a second turn reuses the first turn's cache. Any check failing means `loopd` +//! does not start. + +use bxcore::sanitize::{TAGS, Tag, neutralise}; + +use crate::llama::{ +``` + +**S2.** The lines (the end of `SelfTestError`) + +```rust + Infer(InferError), + Hash, +} +``` + +become + +```rust + Infer(InferError), + Hash, + /// The tag with the recorded id `id` is `count` tokens, not one. + TagNotOneToken { + id: u32, + count: usize, + }, + /// The tag with the recorded id `id` is one token, but the token `got`. + TagOtherId { + id: u32, + got: u32, + }, + /// The neutralised form of the tag with the recorded id `id` is `count` tokens, not more + /// than one. + NeutralisedTooFew { + id: u32, + count: usize, + }, + /// `/tokenize` failed for the tag with the recorded id `id`, or for its neutralised form. + Tokenize { + id: u32, + error: InferError, + }, +} + +impl SelfTestError { + /// The runbook pointer that ends the line `loopd` prints for this failure. A tag failure has + /// its own entry; every other failure shares the self-test's. + pub fn pointer(&self) -> &'static str { + match self { + SelfTestError::TagNotOneToken { .. } + | SelfTestError::TagOtherId { .. } + | SelfTestError::NeutralisedTooFew { .. } + | SelfTestError::Tokenize { .. } => "see docs/runbook.md#loopd-tags-unexpected", + SelfTestError::Mismatch { .. } + | SelfTestError::ToolCall(_) + | SelfTestError::CacheMiss { .. } + | SelfTestError::Infer(_) + | SelfTestError::Hash => "see docs/runbook.md#loopd-selftest-failed", + } + } +} +``` + +Each pointer is written out whole, never built with `format!`: `scripts/check-runbook.sh` reads +the anchors of the pointers in the source. + +**S3.** In `impl std::fmt::Display for SelfTestError`, the line + +```rust + SelfTestError::Hash => write!(f, "the chat template could not be hashed"), +``` + +becomes + +```rust + SelfTestError::Hash => write!(f, "the chat template could not be hashed"), + SelfTestError::TagNotOneToken { id, count } => { + write!(f, "the tag with id {id} is {count} tokens, not one") + } + SelfTestError::TagOtherId { id, got } => { + write!(f, "the tag with id {id} is one token, but the token {got}") + } + SelfTestError::NeutralisedTooFew { id, count } => write!( + f, + "the neutralised form of the tag with id {id} does not split into several tokens (the server gave {count})" + ), + SelfTestError::Tokenize { id, error } => { + write!(f, "could not tokenize the tag with id {id}: {error}") + } +``` + +**S4.** The lines + +```rust +/// Runs the three checks in order. `on_step` is told the name of each check as it starts. +pub fn run(client: &Client, on_step: &mut dyn FnMut(&str)) -> Result<(), SelfTestError> { + on_step("server matches config"); + server_matches(client)?; +``` + +become + +```rust +/// Runs the four checks in order. `on_step` is told the name of each check as it starts. +pub fn run(client: &Client, on_step: &mut dyn FnMut(&str)) -> Result<(), SelfTestError> { + on_step("server matches config"); + server_matches(client)?; + + on_step("tags tokenize as recorded"); + tags_tokenize(client)?; +``` + +**S5.** The line + +```rust +/// Check 2: a tool call comes back parsed, with a `read_file` tool and a string `path`. +``` + +becomes (two new functions, then the old line numbered 3) + +```rust +/// Check 2: each of the model's tags (`bxcore::sanitize::TAGS`) is one token, the one recorded +/// with it, and its neutralised form is more than one (M5b spec, section 8). Like check 1 it sends +/// no prompt and uses no slot, so a server whose tokenizer is not the recorded one is sent no +/// prompt either. A tag missing from the list cannot be found: the server does not list them. +fn tags_tokenize(client: &Client) -> Result<(), SelfTestError> { + for tag in TAGS { + check_tag(client, tag)?; + } + Ok(()) +} + +/// Two `/tokenize` requests: `tag.text`, then `neutralise(tag.text)`. An error names the tag by +/// its id, never by its text, which the operator's terminal and log would show whole. +fn check_tag(client: &Client, tag: &Tag) -> Result<(), SelfTestError> { + let id = tag.id; + let tokenize = |text: &str| { + client + .token_ids(text) + .map_err(|error| SelfTestError::Tokenize { id, error }) + }; + let ids = tokenize(tag.text)?; + match ids.as_slice() { + [got] if *got == id => {} + [got] => return Err(SelfTestError::TagOtherId { id, got: *got }), + _ => { + return Err(SelfTestError::TagNotOneToken { + id, + count: ids.len(), + }); + } + } + let count = tokenize(&neutralise(tag.text))?.len(); + if count <= 1 { + return Err(SelfTestError::NeutralisedTooFew { id, count }); + } + Ok(()) +} + +/// Check 3: a tool call comes back parsed, with a `read_file` tool and a string `path`. +``` + +**S6.** The line + +```rust +/// Check 3: a second turn that extends the first reuses the first turn's cache. +``` + +becomes + +```rust +/// Check 4: a second turn that extends the first reuses the first turn's cache. +``` + +### `crates/loopd/src/main.rs` + +**M1.** In `run_selftest_check`, the line + +```rust + eprintln!("selftest: FAILED: {e}; see docs/runbook.md#loopd-selftest-failed"); +``` + +becomes + +```rust + eprintln!("selftest: FAILED: {e}; {}", e.pointer()); +``` + +`run_selftest` and `run_serve` both call `run_selftest_check`, so both run the new check and print +its pointer; neither changes. + +## Steps + +- [ ] **1. Copy the given tests.** From the repository root, five commands: + + ```sh + cp docs/plans/M5b/files/crates/loopd/tests/selftest_tags.rs crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/selftest.rs crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/serve.rs crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/pointers.rs crates/loopd/tests/ + cp docs/plans/M5b/files/crates/loopd/tests/support/mod.rs crates/loopd/tests/support/ + ``` + +- [ ] **2. See them fail.** Two commands: + - `cargo test -p loopd --no-fail-fast --test selftest --test serve --test pointers`: all three + compile. `pointers` passes (7 tests). Four tests fail, each on an `assert_eq`: + + | Test | Before this task | + |---|---| + | `selftest`: `a_healthy_server_passes_all_four_checks_in_order` | the steps: three, without `"tags tokenize as recorded"` | + | `selftest`: `a_completion_without_the_tool_call_fails_the_tool_call_check` | `left: 2`, `right: 3` | + | `selftest`: `a_cold_second_turn_fails_the_cache_check` | `left: 3`, `right: 4` | + | `serve`: `serve_runs_the_self_test_then_binds_the_socket_with_mode_0600` | `the tag check ran: …`, `left: 0`, `right: 66` | + + The other three tests of `selftest` and two of `serve` pass. + - `cargo test -p loopd --test selftest_tags`: does not compile, with 7 errors, all `E0599`: + no variant named `TagNotOneToken`, `TagOtherId`, `NeutralisedTooFew` or `Tokenize` for + `SelfTestError`, and no method named `pointer` (three times). +- [ ] **3. The runbook.** From the repository root, without opening the fragment: + + ```sh + { echo; cat docs/plans/M5b/files/runbook-loopd-tags-unexpected.md; } >> docs/runbook.md + ``` + + Then `sh scripts/check-runbook.sh .` prints nothing and exits 0, and + `grep -c '^## loopd-tags-unexpected$' docs/runbook.md` prints `1`. +- [ ] **4. `info.rs`.** Make edits I1 and I2. `cargo check -p loopd`: no error and no warning + (`token_ids` is `pub`, so nothing warns that it is not used yet). +- [ ] **5. `selftest.rs`.** Make edits S1 to S6. `cargo check -p loopd`: no error. +- [ ] **6. `main.rs`.** Make edit M1. `cargo check -p loopd`: no error. Then `cargo fmt --all`. +- [ ] **7. See them pass.** + `cargo test -p loopd --test selftest_tags --test selftest --test serve --test pointers`: + `selftest_tags` 8 passed, `selftest` 6, `serve` 3, `pointers` 7. Then `cargo test -p loopd`: + every suite passes. +- [ ] **8. Check.** `grep -rn 'runbook.md#loopd-' crates/loopd/src` prints exactly two lines, + both in `crates/loopd/src/selftest.rs` (the two pointers of S2). +- [ ] **9. Run the gate.** `make gate`, with a shell timeout of 600000 ms. Expected last line: + `gate: ok`, with 1050 tests: 1042 before this task, plus 8 in `selftest_tags.rs` (the new + versions of the other four files keep their counts). +- [ ] **10. Log and commit.** `git status --short` lists `crates/loopd/src/llama/info.rs`, + `crates/loopd/src/selftest.rs`, `crates/loopd/src/main.rs`, the four replaced test files, + `docs/runbook.md` and your log as modified, and `crates/loopd/tests/selftest_tags.rs` as new. + Stage with `git add crates/loopd docs/runbook.md docs/implementer-log.md Cargo.lock`, then + commit as `AGENTS.md` says. `Cargo.lock` does not change in this task; the `git add` is + harmless. + +## Done when + +- `make gate` prints `gate: ok` with 1050 tests. +- `git diff HEAD~1 --stat` lists only the three sources and five test files under + `crates/loopd/`, `docs/runbook.md` and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of I1, I2, S1 to S6 or M1 is not in its file exactly as shown, or is there + more than once. +- Step 2 or 3 shows a different result: other failing tests, other errors, or other counts. +- After step 4, 5 or 6, `cargo check -p loopd` reports an error although the edits are made as + shown. Quote it and stop. +- After step 7 a test fails. Do not open the test to find out why, and do not change a test, the + fake server or the runbook fragment: stop, and give the failing tests' names and their + messages. +- The fix seems to need a change anywhere else: in `bxcore`, `TAGS`, another file of `loopd`, or + a `Cargo.toml`. It does not; stop and say why you think so. +- A test outside `loopd` fails in the gate: check once whether it fails without your change, then + stop and report both results. +- The gate passes with a count other than 1050. diff --git a/docs/plans/M5b/files/crates/loopd/tests/pointers.rs b/docs/plans/M5b/files/crates/loopd/tests/pointers.rs new file mode 100644 index 0000000..f4f93ff --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/pointers.rs @@ -0,0 +1,247 @@ +//! Every fail-closed message `loopd` produces ends with the runbook entry that explains it. +//! `scripts/check-runbook.sh` checks that the entries exist; these tests check that the messages +//! name them. Do not edit. + +mod support; + +use std::path::Path; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +use bxcore::SessionId; +use loopd::baseline::Baseline; +use loopd::broker_port::{not_configured_line, unavailable_line}; +use loopd::memory::{MemoryPort, MemorySource}; +use loopd::session::{Session, SessionError}; +use loopd::tools::Registry; +use support::{FakeServer, Home, Reply, tokenizer}; + +// `want` is the whole pointer, written out: scripts/check-runbook.sh reads the anchors of every +// pointer in the source, and cannot read one built with `format!`. +fn ends_with_pointer(message: &str, want: &str) { + assert!( + message.ends_with(want), + "{message:?} must end with {want:?}" + ); + assert_eq!( + message.matches("runbook.md#").count(), + 1, + "one pointer, not two: {message:?}" + ); +} + +#[test] +fn a_damaged_session_log_names_its_entry() { + let home = Home::new(); + let cfg = home.config(Path::new("/tmp/unused.sock")); + let id = SessionId::new("a").unwrap(); + let baseline = Baseline::assemble(&cfg, &Registry::m3a()).unwrap(); + drop(Session::create(&home.dir, id.clone(), baseline, 0).unwrap()); + let log = home.dir.join("sessions/a/0.jsonl"); + let mut text = std::fs::read_to_string(&log).unwrap(); + text.push_str("{\"type\":\"user\",\"time\":"); + std::fs::write(&log, text).unwrap(); + match Session::open(&home.dir, id) { + Err(e @ SessionError::Torn { .. }) => { + let message = e.to_string(); + assert!(message.contains("0.jsonl:2: "), "{message}"); + ends_with_pointer(&message, "see docs/runbook.md#session-log-damaged"); + } + Err(other) => panic!("{other:?}"), + Ok(_) => panic!("a torn log was opened"), + } +} + +#[test] +fn only_the_torn_error_points_at_the_damaged_log_entry() { + let home = Home::new(); + let missing = Session::open(&home.dir, SessionId::new("nobody").unwrap()); + let message = match missing { + Err(e) => e.to_string(), + Ok(_) => panic!("a session nobody created was opened"), + }; + assert!(!message.contains("runbook"), "{message}"); +} + +#[test] +fn memoryd_unreachable_or_refusing_names_the_memory_entry() { + let home = Home::new(); + let port = MemoryPort::new( + home.dir.join("run/nobody/memory.sock"), + Duration::from_secs(1), + ); + let message = port.block().expect_err("nobody is listening"); + ends_with_pointer(&message, "see docs/runbook.md#memory-unavailable"); + // A refusal from memoryd itself (a file over its limit) ends with the same entry. + let socket = home.dir.join("run/loop-memory/memory.sock"); + std::fs::create_dir_all(socket.parent().unwrap()).unwrap(); + let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let env = bxcore::read_frame(&mut stream).unwrap(); + let reply = bxcore::Envelope { + v: 1, + id: env.id, + r#final: true, + msg: bxcore::Message::Error(bxcore::WireError { + code: bxcore::ErrorCode::Internal, + detail: "memory/user.md: 1,376 characters; its limit is 1,375".to_string(), + }), + }; + bxcore::write_frame(&mut stream, &reply).unwrap(); + }); + let message = MemoryPort::new(socket, Duration::from_secs(5)) + .block() + .expect_err("memoryd refused"); + assert!(message.contains("user.md: 1,376 characters"), "{message}"); + ends_with_pointer(&message, "see docs/runbook.md#memory-unavailable"); +} + +#[test] +fn the_broker_lines_name_their_entry() { + let line = unavailable_line("cannot connect to /run/x.sock: No such file"); + assert_eq!( + line, + "loopd: the tool broker is unavailable: cannot connect to /run/x.sock: No such file; \ + see docs/runbook.md#broker-unavailable" + ); + ends_with_pointer(&line, "see docs/runbook.md#broker-unavailable"); + let line = not_configured_line(); + assert!( + line.starts_with("loopd: no tool broker is configured"), + "{line}" + ); + ends_with_pointer(&line, "see docs/runbook.md#broker-unavailable"); +} + +fn config_file( + home: &Home, + server: &FakeServer, + expect_slots: u32, + extra: &str, +) -> std::path::PathBuf { + let text = format!( + r#" +[infer] +socket = "{}" +model = "test-model" +[slots] +main = 0 +background = 1 +[expect] +template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b" +n_ctx = 131072 +slots = {expect_slots} +[limits] +poll_ms = 40 +liveness_ms = 500 +retry_backoff_ms = [10] +[paths] +home = "{}" +{extra} +"#, + server.socket.display(), + home.dir.display() + ); + let path = home.dir.join("config.toml"); + std::fs::write(&path, text).unwrap(); + path +} + +fn healthy_routes(server: &FakeServer) { + server.route("/props", vec![Reply::fixture("props")]); + server.respond("/tokenize", tokenizer); + server.route( + "/v1/chat/completions", + vec![ + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + Reply::fixture("plain"), + ], + ); +} + +/// Waits for `loopd serve` to bind its socket, stops it, and returns what it printed. +fn stderr_once_serving(mut child: Child, socket: &Path) -> String { + let mut up = false; + for _ in 0..200 { + if socket.exists() { + up = true; + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + child.kill().unwrap(); + let output = child.wait_with_output().unwrap(); + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + assert!(up, "the socket never appeared; stderr: {stderr}"); + stderr +} + +fn serve(config: &Path) -> Child { + Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config"]) + .arg(config) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap() +} + +#[test] +fn a_failed_self_test_names_its_entry() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + // The server has two slots; the config expects three. + let config = config_file(&home, &server, 3, ""); + for command in ["selftest", "serve"] { + let output = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args([command, "--config"]) + .arg(&config) + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&output.stderr); + assert_eq!(output.status.code(), Some(1), "{command}: {stderr}"); + let line = stderr + .lines() + .find(|l| l.starts_with("selftest: FAILED: ")) + .unwrap_or_else(|| panic!("{command}: no FAILED line: {stderr}")); + ends_with_pointer(line, "see docs/runbook.md#loopd-selftest-failed"); + } +} + +#[test] +fn serve_without_a_broker_says_so_once_with_the_entry() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server, 2, ""); + let socket = home.dir.join("run/loop/loop.sock"); + let stderr = stderr_once_serving(serve(&config), &socket); + let lines: Vec<&str> = stderr + .lines() + .filter(|l| l.contains("no tool broker is configured")) + .collect(); + assert_eq!(lines.len(), 1, "once, at startup: {stderr}"); + ends_with_pointer(lines[0], "see docs/runbook.md#broker-unavailable"); +} + +#[test] +fn serve_with_a_broker_socket_does_not_say_it() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + // The socket need not exist: loopd connects per call, and a missing broker is not a reason + // to refuse to start. + let extra = format!( + "[broker]\nsocket = \"{}\"\n", + home.dir.join("run/loop-broker/broker.sock").display() + ); + let config = config_file(&home, &server, 2, &extra); + let socket = home.dir.join("run/loop/loop.sock"); + let stderr = stderr_once_serving(serve(&config), &socket); + assert!(!stderr.contains("no tool broker"), "{stderr}"); + assert!(stderr.contains("serving on"), "{stderr}"); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/selftest.rs b/docs/plans/M5b/files/crates/loopd/tests/selftest.rs new file mode 100644 index 0000000..db99333 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/selftest.rs @@ -0,0 +1,177 @@ +//! Tests for the startup self-test, against the fake server. Do not edit. + +mod support; + +use loopd::llama::Client; +use loopd::selftest::{SelfTestError, run}; +use support::{FakeServer, Reply, tokenizer}; + +const CHAT: &str = "/v1/chat/completions"; +const TOKENIZE: &str = "/tokenize"; + +/// A change to the config before the self-test runs. +type Edit = fn(&mut loopd::config::Config); + +/// The recorded props and the fake tokenizer, and no completion yet. +fn props_and_tags() -> FakeServer { + let server = FakeServer::start(); + server.route("/props", vec![Reply::fixture("props")]); + server.respond(TOKENIZE, tokenizer); + server +} + +/// A server that passes: the recorded props, the fake tokenizer, then the three recorded +/// completions in order. +fn healthy() -> FakeServer { + let server = props_and_tags(); + server.route( + CHAT, + vec![ + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + ], + ); + server +} + +fn run_with( + server: &FakeServer, + edit: impl FnOnce(&mut loopd::config::Config), +) -> (Result<(), SelfTestError>, Vec) { + let mut cfg = support::test_config(&server.socket); + edit(&mut cfg); + let mut steps = Vec::new(); + let result = run(&Client::new(cfg), &mut |s| steps.push(s.to_string())); + (result, steps) +} + +#[test] +fn a_healthy_server_passes_all_four_checks_in_order() { + let server = healthy(); + let (result, steps) = run_with(&server, |_| {}); + assert!(result.is_ok(), "{result:?}"); + assert_eq!( + steps, + [ + "server matches config", + "tags tokenize as recorded", + "tool call round trip", + "turn 2 cache hit" + ] + ); + + let chats = server.requests_to(CHAT); + assert_eq!(chats.len(), 3); + // The tool-call check offers exactly one tool; the cache check offers none. + assert_eq!(chats[0].json()["tools"].as_array().map(Vec::len), Some(1)); + assert!(chats[1].json().get("tools").is_none()); + // All three run on the main slot. + assert!(chats.iter().all(|c| c.json()["id_slot"] == 0)); + // Turn 2 is turn 1 plus the assistant's answer plus a new question: a strict extension. + let turn1 = chats[1].json()["messages"].as_array().unwrap().clone(); + let turn2 = chats[2].json()["messages"].as_array().unwrap().clone(); + assert_eq!(turn2.len(), turn1.len() + 2); + assert_eq!(turn2[..turn1.len()], turn1[..]); + assert_eq!( + turn2[turn1.len()], + serde_json::json!({"role": "assistant", "content": "Blue"}) + ); +} + +#[test] +fn the_main_slot_comes_from_the_config() { + let server = healthy(); + let (result, _) = run_with(&server, |c| c.slots.main = 1); + assert!(result.is_ok(), "{result:?}"); + assert!( + server + .requests_to(CHAT) + .iter() + .all(|c| c.json()["id_slot"] == 1) + ); +} + +#[test] +fn each_expected_value_is_checked() { + let zero = bxcore::Hash32::ZERO; + let cases: [(&str, Edit); 3] = [ + ("chat template sha256", |c| { + c.expect.template_sha256 = bxcore::Hash32::ZERO + }), + ("context per slot", |c| c.expect.n_ctx = 4096), + ("slot count", |c| c.expect.slots = 3), + ]; + for (want_what, edit) in cases { + let server = healthy(); + let (result, steps) = run_with(&server, edit); + match result { + Err(SelfTestError::Mismatch { + what, + expected, + got, + }) => { + assert_eq!(what, want_what); + assert_ne!(expected, got); + if what == "chat template sha256" { + assert_eq!(expected, zero.to_hex()); + } + } + other => panic!("{want_what}: {other:?}"), + } + assert_eq!( + steps.len(), + 1, + "a server that is not the expected one is not sent any prompt" + ); + assert!(server.requests_to(CHAT).is_empty()); + assert!(server.requests_to(TOKENIZE).is_empty()); + } +} + +#[test] +fn a_completion_without_the_tool_call_fails_the_tool_call_check() { + let server = props_and_tags(); + server.route(CHAT, vec![Reply::fixture("plain")]); + let (result, steps) = run_with(&server, |_| {}); + assert!( + matches!(result, Err(SelfTestError::ToolCall(_))), + "{result:?}" + ); + assert_eq!(steps.len(), 3); +} + +#[test] +fn a_cold_second_turn_fails_the_cache_check() { + // turn1 twice: the "second turn" reuses 15 tokens where 46 were left. + let server = props_and_tags(); + server.route( + CHAT, + vec![ + Reply::fixture("tool_call"), + Reply::fixture("progress"), + Reply::fixture("turn1"), + ], + ); + let (result, steps) = run_with(&server, |_| {}); + match result { + Err(SelfTestError::CacheMiss { expected, got }) => assert_eq!((expected, got), (7052, 15)), + other => panic!("{other:?}"), + } + assert_eq!(steps.len(), 4); +} + +#[test] +fn a_server_that_is_down_is_an_inference_error_after_retries() { + let server = props_and_tags(); + server.route(CHAT, vec![Reply::fixture("tool_call").cut_after(500)]); + let (result, _) = run_with(&server, |c| c.limits.retry_attempts = 2); + assert!(matches!(result, Err(SelfTestError::Infer(_))), "{result:?}"); + assert_eq!( + server.requests_to(CHAT).len(), + 2, + "the self-test retries like everything else" + ); + let e: Box = Box::new(result.unwrap_err()); + assert!(!e.to_string().is_empty()); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/selftest_tags.rs b/docs/plans/M5b/files/crates/loopd/tests/selftest_tags.rs new file mode 100644 index 0000000..6ef1fe7 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/selftest_tags.rs @@ -0,0 +1,349 @@ +//! The self-test's tag check (M5b spec, section 8): each of the model's tags must be one token, +//! the one recorded with it, and its neutralised form more than one, or `loopd` does not start. +//! Against the fake server's tokenizer, with one answer changed at a time. The tags come from +//! `bxcore::sanitize::TAGS`, and no assertion prints one (tip T33): a failure names the tag by its +//! id, and the tests check that it does. Do not edit. + +mod support; +#[path = "support/tags.rs"] +mod tags; + +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use bxcore::sanitize::{TAGS, neutralise}; +use loopd::llama::Client; +use loopd::selftest::{SelfTestError, run}; +use support::{FakeServer, Home, Recorded, Reply, tokenizer, tokens}; +use tags::{holds_a_tag, same}; + +const CHAT: &str = "/v1/chat/completions"; +const TOKENIZE: &str = "/tokenize"; +const POINTER: &str = "see docs/runbook.md#loopd-tags-unexpected"; + +/// The recorded props and completions, and `tokenize` as the server's `/tokenize`. +fn server_with(tokenize: impl Fn(&Recorded) -> Reply + Send + Sync + 'static) -> FakeServer { + let server = FakeServer::start(); + server.route("/props", vec![Reply::fixture("props")]); + server.route( + CHAT, + vec![ + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + Reply::fixture("plain"), + ], + ); + server.respond(TOKENIZE, tokenize); + server +} + +/// The text a `/tokenize` request asked about. +fn content(request: &Recorded) -> String { + let body: serde_json::Value = serde_json::from_slice(&request.body).unwrap_or_default(); + body["content"].as_str().unwrap_or_default().to_string() +} + +/// The fake tokenizer, except that `text` is answered with `ids`. +fn except(text: &str, ids: Vec) -> impl Fn(&Recorded) -> Reply + Send + Sync + 'static { + let text = text.to_string(); + move |request: &Recorded| { + if content(request) == text { + tokens(&ids) + } else { + tokenizer(request) + } + } +} + +/// Runs the self-test against `server` with the test config; the result and the steps started. +fn run_on(server: &FakeServer) -> (Result<(), SelfTestError>, Vec) { + let cfg = support::test_config(&server.socket); + let mut steps = Vec::new(); + let result = run(&Client::new(cfg), &mut |s| steps.push(s.to_string())); + (result, steps) +} + +/// A failure of the tag check stops the self-test before any prompt, names the tag by its id, +/// holds no tag, and points at the tags' own runbook entry. +#[track_caller] +fn stopped_before_any_prompt(server: &FakeServer, e: &SelfTestError, steps: &[String], id: u32) { + let text = e.to_string(); + assert!(!holds_a_tag(&text), "tag {id}: the message holds a tag"); + assert!(text.contains(&id.to_string()), "tag {id}: {text}"); + assert_eq!(e.pointer(), POINTER, "tag {id}"); + assert_eq!(steps.len(), 2, "tag {id}: {steps:?}"); + assert!(server.requests_to(CHAT).is_empty(), "tag {id}: no prompt"); +} + +#[test] +fn every_tag_and_its_neutralised_form_tokenize_as_recorded() { + let server = server_with(tokenizer); + let (result, steps) = run_on(&server); + assert!(result.is_ok(), "{result:?}"); + assert_eq!( + steps, + [ + "server matches config", + "tags tokenize as recorded", + "tool call round trip", + "turn 2 cache hit" + ] + ); + + let asked = server.requests_to(TOKENIZE); + assert_eq!( + asked.len(), + 2 * TAGS.len(), + "one request per tag and per form" + ); + let mut want: Vec = TAGS + .iter() + .flat_map(|t| [t.text.to_string(), neutralise(t.text).into_owned()]) + .collect(); + let mut got: Vec = asked.iter().map(content).collect(); + want.sort(); + got.sort(); + same(&got, &want, "the texts asked about"); + for (i, request) in asked.iter().enumerate() { + assert_eq!(request.method, "POST", "request {i}"); + let body = request.json(); + let fields = serde_json::json!({ + "model": "test-model", + "content": content(request), + "add_special": false, + "parse_special": true, + }); + same(&body, &fields, &format!("request {i}: the body")); + } + + // Every tag is checked before the first prompt is sent. + let paths: Vec = server + .requests() + .iter() + .map(|r| r.path().to_string()) + .collect(); + let last_tokenize = paths.iter().rposition(|p| p == TOKENIZE); + let first_chat = paths.iter().position(|p| p == CHAT); + assert!(last_tokenize < first_chat, "{paths:?}"); +} + +#[test] +fn a_tag_that_is_not_one_token_fails() { + for (i, tag) in TAGS.iter().enumerate() { + // Six ordinary tokens, as a tag the tokenizer no longer knows would be; none for the + // first tag. + let ids = if i == 0 { + Vec::new() + } else { + vec![27, 91, 316, 6018, 91, 29] + }; + let count = ids.len(); + let server = server_with(except(tag.text, ids)); + let (result, steps) = run_on(&server); + match &result { + Err(e @ SelfTestError::TagNotOneToken { id, count: got }) => { + assert_eq!((*id, *got), (tag.id, count), "tag {}", tag.id); + stopped_before_any_prompt(&server, e, &steps, tag.id); + } + other => panic!("tag {}: {other:?}", tag.id), + } + } +} + +#[test] +fn a_tag_that_is_another_token_fails() { + for tag in TAGS { + let other = tag.id + 1; + let server = server_with(except(tag.text, vec![other])); + let (result, steps) = run_on(&server); + match &result { + Err(e @ SelfTestError::TagOtherId { id, got }) => { + assert_eq!((*id, *got), (tag.id, other), "tag {}", tag.id); + stopped_before_any_prompt(&server, e, &steps, tag.id); + } + other => panic!("tag {}: {other:?}", tag.id), + } + } +} + +#[test] +fn a_neutralised_form_that_is_one_token_or_none_fails() { + for (i, tag) in TAGS.iter().enumerate() { + // The tag's own token, as a tokenizer that saw through the escape would give; none for + // the first tag. + let ids = if i == 0 { Vec::new() } else { vec![tag.id] }; + let count = ids.len(); + let server = server_with(except(&neutralise(tag.text), ids)); + let (result, steps) = run_on(&server); + match &result { + Err(e @ SelfTestError::NeutralisedTooFew { id, count: got }) => { + assert_eq!((*id, *got), (tag.id, count), "tag {}", tag.id); + stopped_before_any_prompt(&server, e, &steps, tag.id); + } + other => panic!("tag {}: {other:?}", tag.id), + } + } +} + +#[test] +fn a_neutralised_form_of_two_tokens_passes() { + // "More than one" is the rule; two is enough. + let server = server_with(except(&neutralise(TAGS[0].text), vec![5, 6])); + let (result, steps) = run_on(&server); + assert!(result.is_ok(), "{result:?}"); + assert_eq!(steps.len(), 4); +} + +#[test] +fn a_tokenizer_that_fails_or_answers_wrong_fails_closed() { + let first = TAGS[0].id; + let cases: [(&str, Option); 5] = [ + ("no /tokenize at all", None), + ( + "an HTTP error", + Some(Reply::json( + 500, + r#"{"error":{"code":500,"message":"boom"}}"#, + )), + ), + ( + "a count, not ids", + Some(Reply::json(200, r#"{"tokens": 1}"#)), + ), + ( + "pieces, not ids", + Some(Reply::json(200, r#"{"tokens":[{"id":1,"piece":"x"}]}"#)), + ), + ( + "a negative id", + Some(Reply::json(200, r#"{"tokens":[-1]}"#)), + ), + ]; + for (what, reply) in cases { + let server = FakeServer::start(); + server.route("/props", vec![Reply::fixture("props")]); + server.route(CHAT, vec![Reply::fixture("tool_call")]); + if let Some(reply) = reply { + server.route(TOKENIZE, vec![reply]); + } + let (result, steps) = run_on(&server); + match &result { + Err(e @ SelfTestError::Tokenize { id, .. }) => { + assert_eq!(*id, first, "{what}"); + stopped_before_any_prompt(&server, e, &steps, first); + } + other => panic!("{what}: {other:?}"), + } + assert_eq!(server.requests_to(TOKENIZE).len(), 1, "{what}: no retry"); + } +} + +#[test] +fn the_other_failures_keep_the_self_tests_entry() { + let e = SelfTestError::Hash; + assert_eq!(e.pointer(), "see docs/runbook.md#loopd-selftest-failed"); + let e = SelfTestError::CacheMiss { + expected: 2, + got: 1, + }; + assert_eq!(e.pointer(), "see docs/runbook.md#loopd-selftest-failed"); +} + +/// A `loopd` process that is killed when the test ends, however it ends. +struct Running(Child); + +impl Drop for Running { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Runs `loopd --config ` for at most ten seconds: its exit code (None if it +/// was still running and was killed) and what it printed on standard error. +fn loopd(command: &str, config: &Path) -> (Option, String) { + let child = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args([command, "--config"]) + .arg(config) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut running = Running(child); + let deadline = Instant::now() + Duration::from_secs(10); + let mut code = None; + while Instant::now() < deadline { + if let Some(status) = running.0.try_wait().unwrap() { + code = status.code(); + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let _ = running.0.kill(); + let _ = running.0.wait(); + let mut stderr = String::new(); + if let Some(mut pipe) = running.0.stderr.take() { + std::io::Read::read_to_string(&mut pipe, &mut stderr).unwrap(); + } + (code, stderr) +} + +fn config_file(home: &Home, server: &FakeServer) -> PathBuf { + let text = format!( + r#" +[infer] +socket = "{}" +model = "test-model" +[slots] +main = 0 +background = 1 +[expect] +template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b" +n_ctx = 131072 +slots = 2 +[limits] +poll_ms = 40 +liveness_ms = 500 +retry_backoff_ms = [10] +[paths] +home = "{}" +"#, + server.socket.display(), + home.dir.display() + ); + let path = home.dir.join("config.toml"); + std::fs::write(&path, text).unwrap(); + path +} + +#[test] +fn selftest_and_serve_stop_on_a_tag_and_name_its_entry() { + let home = Home::new(); + // The end-of-turn tag, as six ordinary tokens. + let tag = &TAGS[2]; + let server = server_with(except(tag.text, vec![27, 91, 316, 6018, 91, 29])); + let config = config_file(&home, &server); + for command in ["selftest", "serve"] { + let (code, stderr) = loopd(command, &config); + assert!( + !holds_a_tag(&stderr), + "{command}: what it printed holds a tag" + ); + assert_eq!(code, Some(1), "{command}: {stderr}"); + let line = stderr + .lines() + .find(|l| l.starts_with("selftest: FAILED: ")) + .unwrap_or_else(|| panic!("{command}: no FAILED line: {stderr}")); + assert!(line.ends_with(POINTER), "{command}: {line}"); + assert_eq!(line.matches("runbook.md#").count(), 1, "{command}: {line}"); + assert!(line.contains(&tag.id.to_string()), "{command}: {line}"); + assert!(!stderr.contains("serving on"), "{command}: {stderr}"); + } + assert!( + !home.dir.join("run/loop/loop.sock").exists(), + "serve bound no socket" + ); + assert!(server.requests_to(CHAT).is_empty(), "no prompt was sent"); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/serve.rs b/docs/plans/M5b/files/crates/loopd/tests/serve.rs new file mode 100644 index 0000000..6b0c873 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/serve.rs @@ -0,0 +1,128 @@ +//! Tests for the `loopd serve` command. Do not edit. + +mod support; + +use std::os::unix::fs::PermissionsExt; +use std::process::{Command, Stdio}; +use std::time::Duration; +use support::{FakeServer, Home, Reply, tokenizer}; + +fn config_file(home: &Home, server: &FakeServer, expect_slots: u32) -> std::path::PathBuf { + let text = format!( + r#" +[infer] +socket = "{}" +model = "test-model" +[slots] +main = 0 +background = 1 +[expect] +template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b" +n_ctx = 131072 +slots = {expect_slots} +[limits] +poll_ms = 40 +liveness_ms = 500 +retry_backoff_ms = [10] +[paths] +home = "{}" +"#, + server.socket.display(), + home.dir.display() + ); + let path = home.dir.join("config.toml"); + std::fs::write(&path, text).unwrap(); + path +} + +fn healthy_routes(server: &FakeServer) { + server.route("/props", vec![Reply::fixture("props")]); + server.respond("/tokenize", tokenizer); + server.route( + "/v1/chat/completions", + vec![ + Reply::fixture("tool_call"), + Reply::fixture("turn1"), + Reply::fixture("turn2"), + Reply::fixture("plain"), + ], + ); +} + +#[test] +fn serve_runs_the_self_test_then_binds_the_socket_with_mode_0600() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server, 2); + let socket = home.dir.join("run").join("loop").join("loop.sock"); + let mut child = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config"]) + .arg(&config) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut up = false; + for _ in 0..200 { + if socket.exists() { + up = true; + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + let mode = std::fs::metadata(&socket).map(|m| m.permissions().mode() & 0o777); + let _ = child.kill(); + let output = child.wait_with_output().unwrap(); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(up, "the socket never appeared; stderr: {stderr}"); + assert_eq!(mode.unwrap(), 0o600, "{stderr}"); + assert!(stderr.contains("selftest: ok"), "{stderr}"); + assert!(stderr.contains("serving on"), "{stderr}"); + assert_eq!( + server.requests_to("/v1/chat/completions").len(), + 3, + "the three self-test completions ran" + ); + assert_eq!( + server.requests_to("/tokenize").len(), + 2 * bxcore::sanitize::TAGS.len(), + "the tag check ran: each tag and its neutralised form" + ); +} + +#[test] +fn serve_refuses_to_start_when_the_self_test_fails() { + let home = Home::new(); + let server = FakeServer::start(); + healthy_routes(&server); + let config = config_file(&home, &server, 3); // the fixture reports 2 slots + let output = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config"]) + .arg(&config) + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&output.stderr); + assert_eq!(output.status.code(), Some(1), "{stderr}"); + assert!(stderr.contains("selftest: FAILED"), "{stderr}"); + assert!(stderr.contains("slot count"), "{stderr}"); + assert!( + !home.dir.join("run").join("loop").join("loop.sock").exists(), + "no socket was left behind" + ); +} + +#[test] +fn usage_and_a_bad_config_are_reported() { + let output = Command::new(env!("CARGO_BIN_EXE_loopd")) + .arg("dance") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&output.stderr).contains("usage")); + let output = Command::new(env!("CARGO_BIN_EXE_loopd")) + .args(["serve", "--config", "/nonexistent/config.toml"]) + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(1)); + assert!(String::from_utf8_lossy(&output.stderr).contains("config.toml")); +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/support/mod.rs b/docs/plans/M5b/files/crates/loopd/tests/support/mod.rs new file mode 100644 index 0000000..2092dab --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/support/mod.rs @@ -0,0 +1,474 @@ +//! A scripted stand-in for `llama-server`, for tests. Do not edit. +//! +//! It listens on a Unix socket in a temporary directory. Each path has a list of replies that are +//! served in order; the last one repeats. A reply is raw bytes, normally a response recorded from +//! the real server (`tests/fixtures/http/*.http`), and can be delayed, sent in small pieces, cut +//! short, or left hanging. A path can instead compute each reply from its request, as the fake +//! tokenizer does. Every request is recorded. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::collections::VecDeque; +use std::io::{Read, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +static NEXT: AtomicU32 = AtomicU32::new(0); + +pub fn fixture_path(kind: &str, name: &str) -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/fixtures") + .join(kind) + .join(name) +} + +pub fn fixture_bytes(kind: &str, name: &str) -> Vec { + let path = fixture_path(kind, name); + std::fs::read(&path).unwrap_or_else(|e| panic!("{}: {e}", path.display())) +} + +/// A config that points at `socket`, with every limit short enough for a test. +pub fn test_config(socket: &Path) -> loopd::config::Config { + let text = format!( + r#" +[infer] +socket = "{}" +model = "test-model" +[slots] +main = 0 +background = 1 +[expect] +template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b" +n_ctx = 131072 +slots = 2 +[limits] +poll_ms = 40 +busy_wait_ms = 400 +load_wait_ms = 300 +idle_grace_ms = 200 +liveness_ms = 150 +retry_backoff_ms = [10, 20, 30] +retry_window_ms = 5000 +"#, + socket.display() + ); + loopd::config::Config::parse(&text).unwrap() +} + +/// A temporary `BOXMAKER_HOME` with a `system.md` beside a `config.toml`, for session tests. +pub struct Home { + pub dir: PathBuf, +} + +impl Home { + pub fn new() -> Home { + let n = NEXT.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!("loopd-home-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("system.md"), "You are Boxmaker, a test agent.\n").unwrap(); + Home { dir } + } + + /// A config for `socket` whose home, system prompt and channel socket are all under this + /// directory, with the fast test limits. + pub fn config(&self, socket: &Path) -> loopd::config::Config { + let mut cfg = test_config(socket); + cfg.paths.home = self.dir.clone(); + cfg.baseline.system = self.dir.join("system.md"); + cfg.channel.socket = self.dir.join("loop.sock"); + cfg + } + + pub fn write(&self, relative: &str, text: &str) { + let path = self.dir.join(relative); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, text).unwrap(); + } + + pub fn read(&self, relative: &str) -> String { + std::fs::read_to_string(self.dir.join(relative)).unwrap() + } + + /// The records of a session's log, epoch 0. + pub fn records(&self, session: &str) -> Vec { + let text = self.read(&format!("sessions/{session}/0.jsonl")); + text.lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect() + } +} + +/// A tool port that answers from a script and records what it was asked. +pub struct ScriptedPort { + replies: Mutex>, + calls: Mutex>, +} + +impl ScriptedPort { + /// Replies are given in order; when they run out, every call gets `fallback`. + pub fn new(replies: Vec) -> ScriptedPort { + ScriptedPort { + replies: Mutex::new(replies.into()), + calls: Mutex::new(Vec::new()), + } + } + + pub fn calls(&self) -> Vec { + self.calls.lock().unwrap().clone() + } +} + +pub fn ok_result(content: &str) -> bxcore::ToolResponse { + bxcore::ToolResponse::Result { + content: content.to_string(), + class: bxcore::DataClass::Private, + untrusted: true, + truncated: false, + } +} + +pub fn pending(approval: u64, expires: &str) -> bxcore::ToolResponse { + bxcore::ToolResponse::PendingApproval { + approval, + expires: bxcore::Timestamp::parse(expires).unwrap(), + } +} + +impl loopd::tools::ToolPort for ScriptedPort { + /// A scripted `PendingApproval` is reported through `on_pending`, as the real port does with + /// the broker's pending frame, and the reply after it is the answer. Use + /// `ReturnsPendingPort` for a port that breaks the rule and returns one. + fn call( + &self, + request: &bxcore::ToolRequest, + on_pending: &mut dyn FnMut(&loopd::tools::Pending), + ) -> bxcore::ToolResponse { + self.calls.lock().unwrap().push(request.clone()); + let mut replies = self.replies.lock().unwrap(); + let mut reply = replies.pop_front(); + if let Some(bxcore::ToolResponse::PendingApproval { approval, expires }) = reply { + on_pending(&loopd::tools::Pending { approval, expires }); + reply = replies.pop_front(); + } + reply.unwrap_or_else(|| ok_result("scripted")) + } +} + +/// A port that breaks the rule: it returns a pending frame as its final answer. +pub struct ReturnsPendingPort; + +impl loopd::tools::ToolPort for ReturnsPendingPort { + fn call( + &self, + _request: &bxcore::ToolRequest, + _on_pending: &mut dyn FnMut(&loopd::tools::Pending), + ) -> bxcore::ToolResponse { + pending(1, "2026-09-18T12:00:00.000Z") + } +} + +/// The fake `/tokenize`, for `FakeServer::respond`: the text of the request's `content` is one +/// token when it is one of the model's tags (`bxcore::sanitize::TAGS`), the tag's id, and +/// otherwise one token per byte, the byte's value. A body without a string `content` gets a 400. +pub fn tokenizer(request: &Recorded) -> Reply { + let body: serde_json::Value = serde_json::from_slice(&request.body).unwrap_or_default(); + let Some(content) = body.get("content").and_then(|c| c.as_str()) else { + return Reply::json(400, r#"{"error":"no content"}"#); + }; + let ids: Vec = match bxcore::sanitize::TAGS.iter().find(|t| t.text == content) { + Some(tag) => vec![tag.id], + None => content.bytes().map(u32::from).collect(), + }; + tokens(&ids) +} + +/// A `/tokenize` answer as llama-server gives one without `with_pieces`: `{"tokens": [ids]}`. +pub fn tokens(ids: &[u32]) -> Reply { + Reply::json(200, &serde_json::json!({ "tokens": ids }).to_string()) +} + +/// What an independent script derived from a recorded stream: `tests/fixtures/expected/`. +pub fn expected(name: &str) -> serde_json::Value { + serde_json::from_slice(&fixture_bytes("expected", &format!("{name}.json"))).unwrap() +} + +#[derive(Clone)] +pub struct Reply { + bytes: Vec, + head_delay_ms: u64, + piece: usize, + piece_delay_ms: u64, + stop_after: Option, + hang_ms: u64, +} + +impl Reply { + /// Exactly these bytes, then close. + pub fn raw(bytes: impl Into>) -> Reply { + Reply { + bytes: bytes.into(), + head_delay_ms: 0, + piece: usize::MAX, + piece_delay_ms: 0, + stop_after: None, + hang_ms: 0, + } + } + + /// A response recorded from the real server: `tests/fixtures/http/.http`. + pub fn fixture(name: &str) -> Reply { + Reply::raw(fixture_bytes("http", &format!("{name}.http"))) + } + + /// A small JSON response with a content length. + pub fn json(status: u16, body: &str) -> Reply { + Reply::raw(format!( + "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + )) + } + + /// Wait this long before sending the first byte, as a queued request does. + pub fn head_delay(mut self, ms: u64) -> Reply { + self.head_delay_ms = ms; + self + } + + /// Send `piece` bytes at a time, waiting `delay_ms` before each piece after the first. + pub fn trickle(mut self, piece: usize, delay_ms: u64) -> Reply { + self.piece = piece.max(1); + self.piece_delay_ms = delay_ms; + self + } + + /// Send only the first `bytes` bytes, then close, as a server that dies does. + pub fn cut_after(mut self, bytes: usize) -> Reply { + self.stop_after = Some(bytes); + self + } + + /// Send only the first `bytes` bytes, then stay silent for `ms` before closing. + pub fn hang_after(mut self, bytes: usize, ms: u64) -> Reply { + self.stop_after = Some(bytes); + self.hang_ms = ms; + self + } + + /// The offset just after the `n`th `data:` line of the body, for use with `cut_after`. + pub fn offset_after_events(&self, n: usize) -> usize { + let mut seen = 0; + let mut at = 0; + while let Some(found) = find(&self.bytes[at..], b"\n\n") { + at += found + 2; + seen += 1; + if seen == n { + return at; + } + } + panic!("the reply has only {seen} events"); + } +} + +#[derive(Debug, Clone)] +pub struct Recorded { + pub method: String, + /// Path and query as sent. + pub target: String, + /// Header lines as sent, without the request line. + pub headers: Vec, + pub body: Vec, +} + +impl Recorded { + pub fn path(&self) -> &str { + self.target.split('?').next().unwrap_or("") + } + + pub fn json(&self) -> serde_json::Value { + serde_json::from_slice(&self.body).expect("request body is JSON") + } +} + +/// A reply computed from its request. +type Responder = Arc Reply + Send + Sync>; + +struct State { + routes: Mutex)>>, + responders: Mutex>, + requests: Mutex>, +} + +pub struct FakeServer { + pub socket: PathBuf, + state: Arc, + dir: PathBuf, +} + +impl FakeServer { + pub fn start() -> FakeServer { + let n = NEXT.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!("loopd-fake-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let socket = dir.join("infer.sock"); + let _ = std::fs::remove_file(&socket); + let listener = UnixListener::bind(&socket).unwrap(); + let state = Arc::new(State { + routes: Mutex::new(Vec::new()), + responders: Mutex::new(Vec::new()), + requests: Mutex::new(Vec::new()), + }); + let accept_state = Arc::clone(&state); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(stream) = stream else { break }; + let state = Arc::clone(&accept_state); + thread::spawn(move || serve(stream, &state)); + } + }); + FakeServer { socket, state, dir } + } + + /// Replies for `path` (the query is ignored), served in order. The last one repeats. They + /// replace a responder for `path`. + pub fn route(&self, path: &str, replies: Vec) { + assert!(!replies.is_empty()); + self.state + .responders + .lock() + .unwrap() + .retain(|(p, _)| p != path); + let mut routes = self.state.routes.lock().unwrap(); + routes.retain(|(p, _)| p != path); + routes.push((path.to_string(), replies.into())); + } + + /// For each request to `path` (the query is ignored), the reply `reply` computes from it. It + /// replaces the replies `route` gave `path`. + pub fn respond(&self, path: &str, reply: impl Fn(&Recorded) -> Reply + Send + Sync + 'static) { + self.state.routes.lock().unwrap().retain(|(p, _)| p != path); + let mut responders = self.state.responders.lock().unwrap(); + responders.retain(|(p, _)| p != path); + responders.push((path.to_string(), Arc::new(reply))); + } + + pub fn requests(&self) -> Vec { + self.state.requests.lock().unwrap().clone() + } + + pub fn requests_to(&self, path: &str) -> Vec { + self.requests() + .into_iter() + .filter(|r| r.path() == path) + .collect() + } +} + +fn find(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +} + +fn read_request(stream: &mut UnixStream) -> Option { + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let head_end = loop { + if let Some(end) = find(&buf, b"\r\n\r\n") { + break end; + } + match stream.read(&mut chunk) { + Ok(0) | Err(_) => return None, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + } + }; + let head = String::from_utf8_lossy(&buf[..head_end]).to_string(); + let mut lines = head.split("\r\n"); + let mut request_line = lines.next()?.split(' '); + let method = request_line.next()?.to_string(); + let target = request_line.next()?.to_string(); + let headers: Vec = lines.map(str::to_string).collect(); + let length = headers + .iter() + .filter_map(|h| h.split_once(':')) + .find(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .and_then(|(_, v)| v.trim().parse::().ok()) + .unwrap_or(0); + let mut body = buf[head_end + 4..].to_vec(); + while body.len() < length { + match stream.read(&mut chunk) { + Ok(0) | Err(_) => return None, + Ok(n) => body.extend_from_slice(&chunk[..n]), + } + } + Some(Recorded { + method, + target, + headers, + body, + }) +} + +fn serve(mut stream: UnixStream, state: &State) { + let Some(request) = read_request(&mut stream) else { + return; + }; + let path = request.path().to_string(); + let responder = state + .responders + .lock() + .unwrap() + .iter() + .find(|(p, _)| *p == path) + .map(|(_, reply)| Arc::clone(reply)); + let computed = responder.map(|reply| reply(&request)); + state.requests.lock().unwrap().push(request); + let reply = computed.or_else(|| { + let mut routes = state.routes.lock().unwrap(); + match routes.iter_mut().find(|(p, _)| *p == path) { + Some((_, replies)) if replies.len() > 1 => replies.pop_front(), + Some((_, replies)) => replies.front().cloned(), + None => None, + } + }); + let reply = + reply.unwrap_or_else(|| Reply::json(404, r#"{"error":"no route in the fake server"}"#)); + thread::sleep(Duration::from_millis(reply.head_delay_ms)); + let end = reply + .stop_after + .unwrap_or(reply.bytes.len()) + .min(reply.bytes.len()); + for (i, piece) in reply.bytes[..end].chunks(reply.piece).enumerate() { + if i > 0 { + thread::sleep(Duration::from_millis(reply.piece_delay_ms)); + } + if stream.write_all(piece).is_err() { + return; // the client went away, which some tests do on purpose + } + let _ = stream.flush(); + } + thread::sleep(Duration::from_millis(reply.hang_ms)); +} + +/// Removes a test's directory, first putting back the permissions a test may have taken away, +/// or the removal fails (tip T44). +fn remove_test_dir(dir: &Path) { + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)); + let _ = std::fs::remove_dir_all(dir); +} + +impl Drop for Home { + fn drop(&mut self) { + remove_test_dir(&self.dir); + } +} + +impl Drop for FakeServer { + fn drop(&mut self) { + remove_test_dir(&self.dir); + } +} diff --git a/docs/plans/M5b/files/runbook-loopd-tags-unexpected.md b/docs/plans/M5b/files/runbook-loopd-tags-unexpected.md new file mode 100644 index 0000000..62be857 --- /dev/null +++ b/docs/plans/M5b/files/runbook-loopd-tags-unexpected.md @@ -0,0 +1,53 @@ +## loopd-tags-unexpected + +**What you see.** `loopd selftest` or `loopd serve` prints `selftest: tags tokenize as recorded`, +then `selftest: FAILED: ; see docs/runbook.md#loopd-tags-unexpected`, and exits 1. No +prompt was sent. The reason names a tag by its token id, never by its text: + +- `the tag with id is tokens, not one` +- `the tag with id is one token, but the token ` +- `the neutralised form of the tag with id does not split into several tokens (the server + gave )` +- `could not tokenize the tag with id : ` + +**Why.** `loopd` neutralises the model's chat-template tags in every text from outside, writing +their angle brackets as `<` and `>` (M5b spec, sections 6.2 and 8). The list of tags, each +with its token id, is compiled into `bxcore` (`crates/bxcore/src/sanitize/tags.rs`), as measured +on straylight. At startup `loopd` asks the server's `/tokenize` about each tag and its neutralised +form. If a tag is no longer the one token recorded with it, the model or its tokenizer has changed +and the list may no longer match it; if a neutralised form is a single token, neutralising no +longer keeps that tag from the model. Either way the model could receive text from outside as its +own control tokens, so `loopd` does not start. The check cannot find a new tag missing from the +list: the server does not list its special tokens. + +**Confirm.** Find the tag: `grep -n '' crates/bxcore/src/sanitize/tags.rs` shows its entry, +built from its name with `bar!`, `open!` or `close!`. Ask the server yourself, building the tag +from its name as the file does. For a `bar!` tag named `im_end`: + +```sh +printf '{"model":"ornith-1.5-35b-a3b","content":"<|%s|>","parse_special":true}' im_end | + curl -s -d @- http://straylight:11434/tokenize +printf '{"model":"ornith-1.5-35b-a3b","content":"<|%s|>","parse_special":true}' im_end | + curl -s -d @- http://straylight:11434/tokenize +``` + +For an `open!` tag the content is `<%s>`, for a `close!` tag ``. The first answer should be +`{"tokens":[]}`, the second a list of several ids. Check which model file the server loaded: +`curl -s 'http://straylight:11434/props?model=ornith-1.5-35b-a3b' | jq .model_path`. + +**Fix, by reason.** + +- `could not tokenize …`: the server did not answer `/tokenize`. A router in front of + llama-server may not forward it (crossbar does not, `docs/decisions.md`, P22): point + `inferproxy` at llama-server itself. Otherwise check that the server is up: + `curl -s http://straylight:11434/health`, and run `loopd selftest --config ` again. +- A tag with another id, or several tokens: the model file changed. Do not run `loopd` against + it. Measure the new model's tags as M5b task 03 did (every vocabulary entry of type control or + user-defined, each checked with `/tokenize`), and change `TAGS` and the tests' list in + `crates/bxcore/tests/support/tags.rs` together; the change reaches the agent when it is rebuilt. + A new model also invalidates the measurements in `docs/inference-contract.md`. +- A neutralised form that is one token: the tokenizer has an entry for the escaped spelling, and + the replacement of the M5b spec, section 6.2, no longer works for that tag. That is a design + change: record it in `docs/decisions.md` and do not run `loopd` until it is settled. + +**Check.** `loopd selftest --config ` prints `selftest: ok`.