loopd: the self-test checks the model's tags against the tokenizer

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/loopd/tests/pointers.rs

@@ -14,7 +14,7 @@ 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};
+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!`.
@@ -150,6 +150,7 @@ home = "{}"
 
 fn healthy_routes(server: &FakeServer) {
     server.route("/props", vec![Reply::fixture("props")]);
+    server.respond("/tokenize", tokenizer);
     server.route(
         "/v1/chat/completions",
         vec![

crates/loopd/tests/selftest.rs

@@ -4,17 +4,26 @@ mod support;
 
 use loopd::llama::Client;
 use loopd::selftest::{SelfTestError, run};
-use support::{FakeServer, Reply};
+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);
 
-/// A server that passes: the recorded props, then the three recorded completions in order.
-fn healthy() -> FakeServer {
+/// 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![
@@ -38,7 +47,7 @@ fn run_with(
 }
 
 #[test]
-fn a_healthy_server_passes_all_three_checks_in_order() {
+fn a_healthy_server_passes_all_four_checks_in_order() {
     let server = healthy();
     let (result, steps) = run_with(&server, |_| {});
     assert!(result.is_ok(), "{result:?}");
@@ -46,6 +55,7 @@ fn a_healthy_server_passes_all_three_checks_in_order() {
         steps,
         [
             "server matches config",
+            "tags tokenize as recorded",
             "tool call round trip",
             "turn 2 cache hit"
         ]
@@ -115,27 +125,26 @@ fn each_expected_value_is_checked() {
             "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_second_check() {
-    let server = FakeServer::start();
-    server.route("/props", vec![Reply::fixture("props")]);
+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(), 2);
+    assert_eq!(steps.len(), 3);
 }
 
 #[test]
-fn a_cold_second_turn_fails_the_third_check() {
+fn a_cold_second_turn_fails_the_cache_check() {
     // turn1 twice: the "second turn" reuses 15 tokens where 46 were left.
-    let server = FakeServer::start();
-    server.route("/props", vec![Reply::fixture("props")]);
+    let server = props_and_tags();
     server.route(
         CHAT,
         vec![
@@ -149,13 +158,12 @@ fn a_cold_second_turn_fails_the_third_check() {
         Err(SelfTestError::CacheMiss { expected, got }) => assert_eq!((expected, got), (7052, 15)),
         other => panic!("{other:?}"),
     }
-    assert_eq!(steps.len(), 3);
+    assert_eq!(steps.len(), 4);
 }
 
 #[test]
 fn a_server_that_is_down_is_an_inference_error_after_retries() {
-    let server = FakeServer::start();
-    server.route("/props", vec![Reply::fixture("props")]);
+    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:?}");

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<u32>) -> 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<String>) {
+    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<String> = TAGS
+        .iter()
+        .flat_map(|t| [t.text.to_string(), neutralise(t.text).into_owned()])
+        .collect();
+    let mut got: Vec<String> = 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<String> = 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<Reply>); 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 <command> --config <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<i32>, 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");
+}

crates/loopd/tests/serve.rs

@@ -5,7 +5,7 @@ mod support;
 use std::os::unix::fs::PermissionsExt;
 use std::process::{Command, Stdio};
 use std::time::Duration;
-use support::{FakeServer, Home, Reply};
+use support::{FakeServer, Home, Reply, tokenizer};
 
 fn config_file(home: &Home, server: &FakeServer, expect_slots: u32) -> std::path::PathBuf {
     let text = format!(
@@ -37,6 +37,7 @@ home = "{}"
 
 fn healthy_routes(server: &FakeServer) {
     server.route("/props", vec![Reply::fixture("props")]);
+    server.respond("/tokenize", tokenizer);
     server.route(
         "/v1/chat/completions",
         vec![
@@ -82,6 +83,11 @@ fn serve_runs_the_self_test_then_binds_the_socket_with_mode_0600() {
         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]

crates/loopd/tests/support/mod.rs

@@ -3,7 +3,8 @@
 //! 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. Every request is recorded.
+//! 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
 
@@ -171,6 +172,26 @@ impl loopd::tools::ToolPort for ReturnsPendingPort {
     }
 }
 
+/// 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<u32> = 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()
@@ -273,8 +294,12 @@ impl Recorded {
     }
 }
 
+/// A reply computed from its request.
+type Responder = Arc<dyn Fn(&Recorded) -> Reply + Send + Sync>;
+
 struct State {
     routes: Mutex<Vec<(String, VecDeque<Reply>)>>,
+    responders: Mutex<Vec<(String, Responder)>>,
     requests: Mutex<Vec<Recorded>>,
 }
 
@@ -295,6 +320,7 @@ impl FakeServer {
         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);
@@ -308,14 +334,29 @@ impl FakeServer {
         FakeServer { socket, state, dir }
     }
 
-    /// Replies for `path` (the query is ignored), served in order. The last one repeats.
+    /// 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<Reply>) {
         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<Recorded> {
         self.state.requests.lock().unwrap().clone()
     }
@@ -376,15 +417,23 @@ fn serve(mut stream: UnixStream, state: &State) {
         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 = {
+    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));