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) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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}");
|
||||
}
|
||||
@@ -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<String>) {
|
||||
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<dyn std::error::Error> = Box::new(result.unwrap_err());
|
||||
assert!(!e.to_string().is_empty());
|
||||
}
|
||||
@@ -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");
|
||||
}
|
||||
@@ -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"));
|
||||
}
|
||||
@@ -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<u8> {
|
||||
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<bxcore::LogRecord> {
|
||||
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<VecDeque<bxcore::ToolResponse>>,
|
||||
calls: Mutex<Vec<bxcore::ToolRequest>>,
|
||||
}
|
||||
|
||||
impl ScriptedPort {
|
||||
/// Replies are given in order; when they run out, every call gets `fallback`.
|
||||
pub fn new(replies: Vec<bxcore::ToolResponse>) -> ScriptedPort {
|
||||
ScriptedPort {
|
||||
replies: Mutex::new(replies.into()),
|
||||
calls: Mutex::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn calls(&self) -> Vec<bxcore::ToolRequest> {
|
||||
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<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()
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Reply {
|
||||
bytes: Vec<u8>,
|
||||
head_delay_ms: u64,
|
||||
piece: usize,
|
||||
piece_delay_ms: u64,
|
||||
stop_after: Option<usize>,
|
||||
hang_ms: u64,
|
||||
}
|
||||
|
||||
impl Reply {
|
||||
/// Exactly these bytes, then close.
|
||||
pub fn raw(bytes: impl Into<Vec<u8>>) -> 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/<name>.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<String>,
|
||||
pub body: Vec<u8>,
|
||||
}
|
||||
|
||||
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<dyn Fn(&Recorded) -> Reply + Send + Sync>;
|
||||
|
||||
struct State {
|
||||
routes: Mutex<Vec<(String, VecDeque<Reply>)>>,
|
||||
responders: Mutex<Vec<(String, Responder)>>,
|
||||
requests: Mutex<Vec<Recorded>>,
|
||||
}
|
||||
|
||||
pub struct FakeServer {
|
||||
pub socket: PathBuf,
|
||||
state: Arc<State>,
|
||||
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<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()
|
||||
}
|
||||
|
||||
pub fn requests_to(&self, path: &str) -> Vec<Recorded> {
|
||||
self.requests()
|
||||
.into_iter()
|
||||
.filter(|r| r.path() == path)
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
|
||||
haystack.windows(needle.len()).position(|w| w == needle)
|
||||
}
|
||||
|
||||
fn read_request(stream: &mut UnixStream) -> Option<Recorded> {
|
||||
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<String> = 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::<usize>().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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user