Plan M5a: local memory in 15 tasks

memoryd (files, the check, edits, journal, index, recall, serving),
the memory tools and built-in grants in brokerd, the memory block in
loopd, bxctl reindex and memory, and the device checks with memoryd.

Each task was built as a reference and gated. Each was replayed on its
own at its end state (854 to 936 tests) and mutation-checked. Each
skeleton compiles and fails its tests. Every todo comment was read
beside its signature.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-24 15:32:54 -07:00
co-authored by Claude Opus 5.5
parent 19fd74db6e
commit de1032157b
76 changed files with 8382 additions and 0 deletions
@@ -0,0 +1,122 @@
//! Memory, as `loopd` sees it (M5a spec, section 9): the block `memoryd` renders for a new
//! session's baseline, the schemas of the two core memory tools, and the notice after a turn.
//! `loopd` keeps no memory itself; `brokerd` decides the tool calls and `memoryd` runs them.
use std::os::unix::net::UnixStream;
use std::path::PathBuf;
use std::time::Duration;
use proto::{
Empty, Envelope, Message, PROTOCOL_VERSION, SessionId, SessionUpdated, read_frame, write_frame,
};
use crate::llama::ToolSchema;
/// Where a new session's baseline points when there is no memory block.
pub const UNAVAILABLE: &str = "see docs/runbook.md#memory-unavailable";
/// How long `loopd` waits for `memoryd`.
pub const MEMORY_WAIT: Duration = Duration::from_secs(10);
/// Where the memory block comes from, and who hears that a session has new records.
pub trait MemorySource: Send + Sync {
/// The block for a new session. `Err` ends in `UNAVAILABLE`: no session starts without it.
fn block(&self) -> Result<String, String>;
/// The session has new records; best effort, a failure is only logged by the caller.
fn updated(&self, session: &SessionId) -> Result<(), String>;
}
/// No memory: for tests and the self-test, never for `loopd serve`.
pub struct NoMemory;
impl MemorySource for NoMemory {
fn block(&self) -> Result<String, String> {
Ok(String::new())
}
fn updated(&self, _session: &SessionId) -> Result<(), String> {
Ok(())
}
}
/// A client of `memoryd`'s `loopd` socket. One connection per request.
pub struct MemoryPort {
socket: PathBuf,
wait: Duration,
}
impl MemoryPort {
pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort {
MemoryPort { socket, wait }
}
fn ask(&self, msg: Message) -> Result<Message, String> {
// Every error is format!("memoryd at {}: {e}; {UNAVAILABLE}", self.socket.display()).
// `UnixStream::connect(&self.socket)`, read and write timeouts of `self.wait`,
// `write_frame` an Envelope { v: PROTOCOL_VERSION, id: 1, final: true, msg }, then
// `read_frame`; Ok(its msg).
todo!()
}
}
impl MemorySource for MemoryPort {
fn block(&self) -> Result<String, String> {
// `self.ask(Message::MemoryBlock(Empty {}))?`: MemoryBlockReply(reply) -> Ok(reply.text);
// Error(e) -> Err(format!("memoryd: {}; {UNAVAILABLE}", e.detail)); anything else ->
// Err(format!("memoryd answered with the wrong message; {UNAVAILABLE}")).
todo!()
}
fn updated(&self, session: &SessionId) -> Result<(), String> {
// `self.ask(Message::SessionUpdated(SessionUpdated { session: session.clone() }))?`: Ok(_)
// -> Ok(()); Error(e) -> Err(format!("memoryd: {}", e.detail)); anything else ->
// Err("memoryd answered with the wrong message").
todo!()
}
}
pub const RECALL: &str = "recall";
pub const MEMORY: &str = "memory";
pub fn recall_schema() -> ToolSchema {
ToolSchema {
name: RECALL.to_string(),
description: "Search your memory: past conversations and notes. Each hit says where it \
came from: identity, memory, or memory-tainted (from a session that read \
untrusted content)."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"query": { "type": "string", "description": "Words to search for." },
"k": { "type": "integer", "description": "How many hits, 1 to 10. Default 5." }
},
"required": ["query"]
}),
}
}
pub fn memory_schema() -> ToolSchema {
ToolSchema {
name: MEMORY.to_string(),
description: "Write memory. note: record something seen or told. add, replace, remove: \
edit memory.md (your work) or user.md (your owner); entries are named by \
their exact text. consolidate: rewrite a whole file, moving or dropping \
entries to notes. Takes effect in the next session."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"action": { "type": "string", "enum": ["note", "add", "replace", "remove", "consolidate"] },
"file": { "type": "string", "enum": ["memory", "user"] },
"text": { "type": "string", "description": "The new entry, or the note." },
"old": { "type": "string", "description": "The exact text of the entry to replace or remove." },
"entries": {
"type": "array",
"description": "consolidate: every current entry once, as {old, keep: true}, {old, text}, {old, move: true} or {old, drop: true}; new entries as {text}.",
"items": { "type": "object" }
}
},
"required": ["action"]
}),
}
}
@@ -0,0 +1,235 @@
//! Tests for the baseline and the replay function. Do not edit.
mod support;
use loopd::baseline::{Baseline, messages};
use loopd::llama::ChatMessage;
use loopd::tools::Registry;
use proto::{CallId, DataClass, Epoch, Hash32, LogRecord, SessionId, Timestamp, ToolCall};
use std::path::Path;
use support::Home;
fn ts() -> Timestamp {
Timestamp::parse("2026-09-18T08:00:00.000Z").unwrap()
}
#[test]
fn assembles_system_prompt_and_core_schemas() {
let home = Home::new();
let cfg = home.config(Path::new("/tmp/unused.sock"));
let b = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
assert_eq!(
b.system, "You are Boxmaker, a test agent.",
"trailing newline trimmed"
);
let names: Vec<&str> = b.tools.iter().map(|t| t.name.as_str()).collect();
assert_eq!(
names,
["clock", "find_tool", "call_tool"],
"core tools, then the two meta-tools; not echo"
);
}
#[test]
fn the_memory_block_is_appended_when_present_and_only_then() {
let home = Home::new();
let cfg = home.config(Path::new("/tmp/unused.sock"));
let block = "# soul.md\nYou are kind.\n\n# memory.md (0 of 2,200 characters)\n(empty)\n";
let b = Baseline::assemble_with(&cfg, &Registry::m2b(), block).unwrap();
assert_eq!(
b.system,
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind.\n\n# memory.md (0 of 2,200 characters)\n(empty)"
);
home.write("memory/core.md", "An old core file is ignored.\n");
let b = Baseline::assemble_with(&cfg, &Registry::m2b(), " \n").unwrap();
assert_eq!(
b.system, "You are Boxmaker, a test agent.",
"an empty block adds nothing, and core.md is no longer read"
);
assert_eq!(b, Baseline::assemble(&cfg, &Registry::m2b()).unwrap());
}
#[test]
fn a_missing_system_prompt_is_an_error_naming_the_file() {
let home = Home::new();
let mut cfg = home.config(Path::new("/tmp/unused.sock"));
cfg.baseline.system = home.dir.join("nope.md");
let e = Baseline::assemble(&cfg, &Registry::m2b()).unwrap_err();
assert!(e.to_string().contains("nope.md"), "{e}");
}
#[test]
fn json_round_trip_and_hash() {
let home = Home::new();
let cfg = home.config(Path::new("/tmp/unused.sock"));
let b = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
let text = b.to_json().unwrap();
assert_eq!(Baseline::from_json(&text).unwrap(), b);
assert_eq!(
b.hash().unwrap(),
proto::sha256(text.as_bytes()).unwrap(),
"the hash is of the JSON as written"
);
assert_ne!(b.hash().unwrap(), Hash32::ZERO);
let mut other = b.clone();
other.system.push('!');
assert_ne!(other.hash().unwrap(), b.hash().unwrap());
assert!(
Baseline::from_json(&text.replacen("\"system\"", "\"zz\":1,\"system\"", 1)).is_err(),
"unknown keys are rejected"
);
}
#[test]
fn replay_keeps_only_messages_in_order_and_unchanged() {
let b = Baseline {
system: "sys".to_string(),
tools: vec![],
};
let call = ToolCall {
id: "c1".to_string(),
name: "clock".to_string(),
arguments: "{}".to_string(),
};
let records = vec![
LogRecord::SessionStart {
time: ts(),
session: SessionId::new("s").unwrap(),
epoch: Epoch(0),
slot: 0,
baseline: Hash32::ZERO,
},
LogRecord::User {
time: ts(),
content: "hi".to_string(),
},
LogRecord::Assistant {
time: ts(),
content: None,
reasoning_content: Some("think".to_string()),
tool_calls: vec![call.clone()],
},
LogRecord::Usage {
time: ts(),
cache_n: 1,
prompt_n: 2,
predicted_n: 3,
reasoning_tokens: 1,
thinking_capped: false,
},
LogRecord::ToolResult {
time: ts(),
call: CallId(1),
tool_call_id: "c1".to_string(),
content: "noon".to_string(),
class: DataClass::Public,
untrusted: false,
truncated: false,
},
LogRecord::CacheLoss {
time: ts(),
expected: 10,
got: 0,
},
LogRecord::Assistant {
time: ts(),
content: Some("It is noon.".to_string()),
reasoning_content: None,
tool_calls: vec![],
},
LogRecord::EpochEnd {
time: ts(),
next: Epoch(1),
summary: "x".to_string(),
},
];
let want = vec![
ChatMessage::System {
content: "sys".to_string(),
},
ChatMessage::User {
content: "hi".to_string(),
},
ChatMessage::Assistant {
content: None,
reasoning_content: Some("think".to_string()),
tool_calls: vec![call],
},
ChatMessage::Tool {
tool_call_id: "c1".to_string(),
content: "noon".to_string(),
},
ChatMessage::Assistant {
content: Some("It is noon.".to_string()),
reasoning_content: None,
tool_calls: vec![],
},
];
assert_eq!(messages(&b, &records), want);
assert_eq!(
messages(&b, &[]),
vec![ChatMessage::System {
content: "sys".to_string()
}]
);
}
/// Replaying a prefix of the log gives a prefix of the messages: the function never reorders
/// or rewrites. Checked for every prefix of a generated log.
#[test]
fn replay_of_a_prefix_is_a_prefix() {
let b = Baseline {
system: "sys".to_string(),
tools: vec![],
};
let mut records = Vec::new();
let mut seed: u64 = 0x9e3779b97f4a7c15;
let mut next = || {
seed ^= seed << 13;
seed ^= seed >> 7;
seed ^= seed << 17;
seed
};
for i in 0..200u64 {
let r = match next() % 5 {
0 => LogRecord::User {
time: ts(),
content: format!("u{i}"),
},
1 => LogRecord::Assistant {
time: ts(),
content: Some(format!("a{i}")),
reasoning_content: None,
tool_calls: vec![],
},
2 => LogRecord::ToolResult {
time: ts(),
call: CallId(i),
tool_call_id: format!("t{i}"),
content: "x".repeat((next() % 50) as usize),
class: DataClass::Private,
untrusted: true,
truncated: false,
},
3 => LogRecord::Usage {
time: ts(),
cache_n: i,
prompt_n: 1,
predicted_n: 1,
reasoning_tokens: 0,
thinking_capped: false,
},
_ => LogRecord::CacheLoss {
time: ts(),
expected: i,
got: 0,
},
};
records.push(r);
}
let whole = messages(&b, &records);
for n in 0..=records.len() {
let part = messages(&b, &records[..n]);
assert_eq!(whole[..part.len()], part[..], "prefix of {n} records");
}
}
@@ -0,0 +1,382 @@
//! Tests for the channel protocol on `loop.sock`. Do not edit.
//!
//! A real `channel::serve` runs on a socket in a temporary home, with the fake inference server
//! behind it. The tests speak the frame protocol to it directly.
mod support;
use loopd::channel::{Context, serve};
use loopd::llama::Client;
use loopd::memory::NoMemory;
use loopd::tools::Registry;
use proto::{
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnEvent, read_frame,
write_frame,
};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use support::{FakeServer, Home, Reply, ScriptedPort};
const CHAT: &str = "/v1/chat/completions";
struct Loopd {
home: Home,
server: FakeServer,
socket: PathBuf,
}
fn start(replies: Vec<proto::ToolResponse>) -> Loopd {
let home = Home::new();
let server = FakeServer::start();
let cfg = home.config(&server.socket);
let socket = cfg.channel_socket();
let listener = UnixListener::bind(&socket).unwrap();
let ctx = Arc::new(Context::new(
cfg.clone(),
Client::new(cfg),
Box::new(ScriptedPort::new(replies)),
Registry::m2b(),
Box::new(NoMemory),
));
thread::spawn(move || serve(listener, ctx));
Loopd {
home,
server,
socket,
}
}
fn id(s: &str) -> SessionId {
SessionId::new(s).unwrap()
}
/// Sends one turn and collects every frame that comes back.
fn turn(socket: &PathBuf, session: &str, content: &str, resume: bool) -> Vec<Envelope> {
let mut stream = UnixStream::connect(socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.unwrap();
let msg = Message::Turn(Turn {
session: id(session),
content: content.to_string(),
resume,
});
write_frame(
&mut stream,
&Envelope {
v: PROTOCOL_VERSION,
id: 42,
r#final: true,
msg,
},
)
.unwrap();
let mut frames = Vec::new();
loop {
let frame = read_frame(&mut stream).unwrap();
let last = frame.r#final;
frames.push(frame);
if last {
break;
}
}
frames
}
fn error_code(frames: &[Envelope]) -> Option<ErrorCode> {
match &frames.last()?.msg {
Message::Error(e) => Some(e.code),
_ => None,
}
}
#[test]
fn a_turn_streams_events_and_ends_with_turn_done() {
let l = start(vec![]);
l.server.route(CHAT, vec![Reply::fixture("thinking")]);
let frames = turn(&l.socket, "a", "what is 17 * 23?", false);
assert!(frames.len() > 3, "{frames:?}");
assert!(
frames.iter().all(|f| f.id == 42),
"every frame carries the request id"
);
assert!(frames.iter().all(|f| f.v == PROTOCOL_VERSION));
let (last, events) = frames.split_last().unwrap();
assert!(
events
.iter()
.all(|f| !f.r#final && matches!(f.msg, Message::TurnEvent(_)))
);
assert!(
events
.iter()
.any(|f| matches!(f.msg, Message::TurnEvent(TurnEvent::Reasoning { .. })))
);
assert!(
events
.iter()
.any(|f| matches!(f.msg, Message::TurnEvent(TurnEvent::Content { .. })))
);
match &last.msg {
Message::TurnDone(done) => {
assert_eq!(
done.content,
support::expected("thinking")["content"].as_str().unwrap()
);
assert_eq!(done.usage.reasoning_tokens, 49);
assert_eq!(done.usage.prompt_n, 40);
}
other => panic!("{other:?}"),
}
// The content events, concatenated, are the answer.
let streamed: String = events
.iter()
.filter_map(|f| match &f.msg {
Message::TurnEvent(TurnEvent::Content { text }) => Some(text.as_str()),
_ => None,
})
.collect();
let Message::TurnDone(done) = &last.msg else {
unreachable!()
};
assert_eq!(streamed, done.content);
// And the session is on disk.
assert_eq!(l.home.records("a").len(), 4);
}
#[test]
fn tool_calls_are_reported_by_name_only() {
let l = start(vec![support::ok_result("straylight\n")]);
l.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let frames = turn(&l.socket, "a", "hostname?", false);
let tool_events: Vec<&TurnEvent> = frames
.iter()
.filter_map(|f| match &f.msg {
Message::TurnEvent(
e @ (TurnEvent::ToolCallStarted { .. } | TurnEvent::ToolResult { .. }),
) => Some(e),
_ => None,
})
.collect();
assert_eq!(tool_events.len(), 2);
assert_eq!(
tool_events[0],
&TurnEvent::ToolCallStarted {
name: "read_file".to_string()
}
);
assert_eq!(
tool_events[1],
&TurnEvent::ToolResult {
name: "read_file".to_string(),
class: proto::DataClass::Private,
truncated: false
}
);
let text = serde_json::to_string(&frames).unwrap();
assert!(
!text.contains("straylight"),
"the result body is not on the channel: {text}"
);
}
#[test]
fn resume_and_create_are_checked() {
let l = start(vec![]);
l.server
.route(CHAT, vec![Reply::fixture("turn1"), Reply::fixture("turn2")]);
assert_eq!(
error_code(&turn(&l.socket, "a", "x", true)),
Some(ErrorCode::NoSuchSession),
"resume needs an existing session"
);
assert!(matches!(
turn(&l.socket, "a", "one", false).last().unwrap().msg,
Message::TurnDone(_)
));
assert_eq!(
error_code(&turn(&l.socket, "a", "x", false)),
Some(ErrorCode::SessionExists),
"create needs a new one"
);
let frames = turn(&l.socket, "a", "two", true);
assert!(
matches!(frames.last().unwrap().msg, Message::TurnDone(_)),
"{frames:?}"
);
let sent = l.server.requests_to(CHAT);
assert_eq!(sent.len(), 2);
assert_eq!(
sent[1].json()["messages"].as_array().unwrap().len(),
4,
"the second turn carried the first"
);
}
#[test]
fn a_busy_session_is_refused_at_once_and_another_session_is_not() {
let l = start(vec![]);
let size = support::fixture_bytes("http", "plain.http").len();
l.server
.route(CHAT, vec![Reply::fixture("plain").trickle(size / 4, 120)]);
let socket = l.socket.clone();
let first = thread::spawn(move || turn(&socket, "a", "slow", false));
thread::sleep(Duration::from_millis(100));
let started = std::time::Instant::now();
assert_eq!(
error_code(&turn(&l.socket, "a", "again", true)),
Some(ErrorCode::SessionBusy)
);
assert!(
started.elapsed() < Duration::from_millis(100),
"refused at once"
);
assert!(
matches!(
turn(&l.socket, "b", "other", false).last().unwrap().msg,
Message::TurnDone(_)
),
"another session runs"
);
assert!(matches!(
first.join().unwrap().last().unwrap().msg,
Message::TurnDone(_)
));
}
#[test]
fn limits_and_server_errors_come_back_as_error_codes() {
let l = start(vec![]);
l.server.route(CHAT, vec![Reply::fixture("context_full")]);
assert_eq!(
error_code(&turn(&l.socket, "a", "x", false)),
Some(ErrorCode::SessionFull)
);
l.server.route(CHAT, vec![Reply::fixture("tool_call")]);
let frames = turn(&l.socket, "b", "x", false);
assert_eq!(
error_code(&frames),
Some(ErrorCode::TurnLimit),
"a repeated call twice"
);
assert!(
frames
.iter()
.any(|f| matches!(f.msg, Message::TurnEvent(TurnEvent::ToolCallStarted { .. }))),
"events before the error were delivered"
);
l.server.route(CHAT, vec![Reply::fixture("bad_request")]);
let frames = turn(&l.socket, "c", "x", false);
assert_eq!(error_code(&frames), Some(ErrorCode::Inference));
let Message::Error(e) = &frames.last().unwrap().msg else {
unreachable!()
};
assert!(e.detail.contains("400"), "{}", e.detail);
}
#[test]
fn bad_frames_get_an_error_and_a_close() {
let l = start(vec![]);
// A frame that is not a turn.
let mut stream = UnixStream::connect(&l.socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
let msg = Message::Error(proto::WireError {
code: ErrorCode::Internal,
detail: String::new(),
});
write_frame(
&mut stream,
&Envelope {
v: PROTOCOL_VERSION,
id: 1,
r#final: true,
msg,
},
)
.unwrap();
let reply = read_frame(&mut stream).unwrap();
assert!(reply.r#final);
assert!(
matches!(
reply.msg,
Message::Error(proto::WireError {
code: ErrorCode::BadMessage,
..
})
),
"{reply:?}"
);
assert!(
matches!(read_frame(&mut stream), Err(proto::FrameError::Closed)),
"then the connection is closed"
);
// Bytes that are not a frame at all.
let mut stream = UnixStream::connect(&l.socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
std::io::Write::write_all(&mut stream, &[0, 0, 0, 3, b'{', b'{', b'{']).unwrap();
let reply = read_frame(&mut stream).unwrap();
assert!(
matches!(
reply.msg,
Message::Error(proto::WireError {
code: ErrorCode::BadMessage,
..
})
),
"{reply:?}"
);
// A wrong protocol version.
let mut stream = UnixStream::connect(&l.socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
let msg = Message::Turn(Turn {
session: id("a"),
content: "x".to_string(),
resume: false,
});
write_frame(
&mut stream,
&Envelope {
v: 2,
id: 1,
r#final: true,
msg,
},
)
.unwrap();
let reply = read_frame(&mut stream).unwrap();
assert!(
matches!(
reply.msg,
Message::Error(proto::WireError {
code: ErrorCode::BadVersion,
..
})
),
"{reply:?}"
);
// A client that connects and leaves.
drop(UnixStream::connect(&l.socket).unwrap());
thread::sleep(Duration::from_millis(50));
l.server.route(CHAT, vec![Reply::fixture("plain")]);
assert!(matches!(
turn(&l.socket, "z", "still up", false).last().unwrap().msg,
Message::TurnDone(_)
));
}
@@ -0,0 +1,480 @@
//! Checks against the real server. Ignored unless run by `make verify-device`. Do not edit.
//!
//! They need two environment variables:
//! BOXMAKER_INFERPROXY path to the built `inferproxy` binary
//! BOXMAKER_UPSTREAM host:port of the llama-server router, for example straylight:11434
//! and, optionally, BOXMAKER_MODEL (default ornith-1.5-35b-a3b). They use slot 0.
//!
//! They never touch the real `llama-server` process. The "server dies" test kills and restarts
//! its own `inferproxy`.
#[path = "support/memoryd.rs"]
mod memoryd_run;
use loopd::config::Config;
use loopd::llama::info::{CacheOutcome, cache_outcome};
use loopd::llama::{ChatEvent, ChatMessage, ChatRequest, Client};
use std::path::{Path, PathBuf};
use std::process::{Child, Command};
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
struct Proxy {
child: Child,
socket: PathBuf,
}
impl Proxy {
fn start(socket: &Path) -> Proxy {
let binary = std::env::var("BOXMAKER_INFERPROXY").expect("BOXMAKER_INFERPROXY is not set");
let upstream = std::env::var("BOXMAKER_UPSTREAM").expect("BOXMAKER_UPSTREAM is not set");
let child = Command::new(binary)
.arg("--listen")
.arg(socket)
.arg("--upstream")
.arg(upstream)
.spawn()
.expect("cannot start inferproxy");
let mut proxy = Proxy {
child,
socket: socket.to_path_buf(),
};
for _ in 0..100 {
if socket.exists() {
return proxy;
}
thread::sleep(Duration::from_millis(20));
}
proxy.kill();
panic!("inferproxy did not create {}", socket.display());
}
fn kill(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
impl Drop for Proxy {
fn drop(&mut self) {
self.kill();
}
}
/// The server as deployed on straylight, recorded in one place. When the deployment changes on
/// purpose, change this and `tools/check-m3a-device.sh` together, and note it in
/// `docs/inference-contract.md`. Since 2026-09-22: four slots over one 262,144-token pool.
const EXPECT: &str = r#"template_sha256 = "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b"
n_ctx = 262144
slots = 4"#;
fn socket_path(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("loopd-device-{}-{name}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
dir.join("infer.sock")
}
fn config(socket: &Path) -> Config {
let model =
std::env::var("BOXMAKER_MODEL").unwrap_or_else(|_| "ornith-1.5-35b-a3b".to_string());
let text = format!(
r#"
[infer]
socket = "{}"
model = "{model}"
[slots]
main = 0
background = 1
[expect]
{EXPECT}
"#,
socket.display()
);
Config::parse(&text).unwrap()
}
fn user(text: &str) -> ChatMessage {
// A different prompt each run, so that an earlier run's cache cannot make a check pass.
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
ChatMessage::User {
content: format!("{text} (run {nonce})"),
}
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn the_startup_self_test_passes() {
let socket = socket_path("selftest");
let _proxy = Proxy::start(&socket);
let client = Client::new(config(&socket));
loopd::selftest::run(&client, &mut |step| eprintln!("selftest: {step}")).unwrap();
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn a_capped_thinking_block_ends_and_the_answer_arrives() {
let socket = socket_path("cap");
let _proxy = Proxy::start(&socket);
let mut cfg = config(&socket);
cfg.limits.thinking_cap = 60;
let client = Client::new(cfg);
let ask = "Think at great length about how many ways 8 rooks fit on a chessboard without attacking \
each other and with none on the main diagonal. Then answer in one sentence.";
let req = ChatRequest {
slot: 0,
messages: vec![user(ask)],
tools: vec![],
thinking: true,
};
let mut capped = Vec::new();
let done = client
.chat_with_retry(&req, &mut |e| {
if let ChatEvent::ThinkingCapped { tokens } = e {
capped.push(*tokens);
}
})
.unwrap();
assert!(done.thinking_capped);
assert_eq!(capped.len(), 1);
assert!(capped[0] >= 60 && capped[0] < 70, "capped at {capped:?}");
assert!(
done.reasoning_tokens < 60 + 256,
"thinking went on to {}",
done.reasoning_tokens
);
assert!(
done.content.is_some_and(|c| !c.trim().is_empty()),
"an answer followed the forced end"
);
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn a_request_survives_its_proxy_being_killed_and_restarted() {
let socket = socket_path("restart");
let mut proxy = Proxy::start(&socket);
let mut cfg = config(&socket);
cfg.limits.retry_backoff_ms = vec![1_500];
let client = Client::new(cfg);
let ask = "Write about 300 words on the history of cork.";
let req = ChatRequest {
slot: 0,
messages: vec![user(ask)],
tools: vec![],
thinking: false,
};
let (tx, rx) = mpsc::channel();
let worker = thread::spawn(move || {
let mut retries = 0;
let mut told = false;
let result = client.chat_with_retry(&req, &mut |e| match e {
ChatEvent::Content(_) if !told => {
told = true;
let _ = tx.send(());
}
ChatEvent::Retrying { .. } => retries += 1,
_ => {}
});
(result, retries)
});
rx.recv_timeout(Duration::from_secs(120))
.expect("no content arrived");
proxy.kill(); // the stream dies in the middle of the answer
thread::sleep(Duration::from_millis(300));
let _proxy = Proxy::start(&proxy.socket);
let (result, retries) = worker.join().unwrap();
let done = result.expect("the retry should have succeeded");
assert!(retries >= 1, "the request was retried");
assert!(
done.content
.is_some_and(|c| c.split_whitespace().count() > 100)
);
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn a_second_turn_reuses_the_first_turns_cache() {
let socket = socket_path("cache");
let _proxy = Proxy::start(&socket);
let client = Client::new(config(&socket));
let mut messages = vec![user(
"What is 17 * 23? Think briefly, then answer in one short sentence.",
)];
let req = ChatRequest {
slot: 0,
messages: messages.clone(),
tools: vec![],
thinking: true,
};
let turn1 = client.chat_with_retry(&req, &mut |_| {}).unwrap();
assert!(
turn1.reasoning_content.is_some(),
"this check is about replaying a thinking block"
);
messages.push(ChatMessage::Assistant {
content: turn1.content.clone(),
reasoning_content: turn1.reasoning_content.clone(),
tool_calls: turn1.tool_calls.clone(),
});
messages.push(user("And 17 * 24?"));
let req = ChatRequest {
slot: 0,
messages,
tools: vec![],
thinking: true,
};
let turn2 = client.chat_with_retry(&req, &mut |_| {}).unwrap();
assert_eq!(
cache_outcome(&turn1.timings, &turn2.timings),
CacheOutcome::Hit,
"{:?} then {:?}",
turn1.timings,
turn2.timings
);
}
// ---- M2b: the agent loop on the real server ----
/// A `loopd serve` on a private home, killed on drop.
struct Served {
child: Child,
home: PathBuf,
socket: PathBuf,
memoryd: Child,
}
fn config_text(infer: &Path, home: &Path) -> String {
let model =
std::env::var("BOXMAKER_MODEL").unwrap_or_else(|_| "ornith-1.5-35b-a3b".to_string());
format!(
r#"
[infer]
socket = "{}"
model = "{model}"
[slots]
main = 0
background = 1
[expect]
{EXPECT}
[paths]
home = "{}"
"#,
infer.display(),
home.display()
)
}
impl Served {
/// Writes the config and the repository's `system.md` into `home`, and starts `memoryd serve`
/// (with the repository's starter `soul.md`), then `loopd serve`: `loopd` starts no session
/// without the memory block.
fn start(infer: &Path, home: &Path) -> Served {
std::fs::create_dir_all(home).unwrap();
let memoryd = memoryd_run::start(home);
let config = home.join("config.toml");
std::fs::write(&config, config_text(infer, home)).unwrap();
let prompt = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../config/system.md");
std::fs::copy(&prompt, home.join("system.md"))
.expect("config/system.md exists in the repository");
let socket = home.join("run").join("loop").join("loop.sock");
let _ = std::fs::remove_file(&socket);
let child = Command::new(env!("CARGO_BIN_EXE_loopd"))
.arg("serve")
.arg("--config")
.arg(&config)
.spawn()
.expect("cannot start loopd");
let mut served = Served {
child,
home: home.to_path_buf(),
socket,
memoryd,
};
for _ in 0..600 {
if served.socket.exists() {
return served;
}
thread::sleep(Duration::from_millis(100));
}
served.kill();
panic!("loopd did not come up within 60 s");
}
fn kill(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
let _ = self.memoryd.kill();
let _ = self.memoryd.wait();
}
/// One `bxctl chat --say` turn. Returns the answer.
fn say(&self, session: &str, text: &str) -> String {
let bxctl = std::env::var("BOXMAKER_BXCTL").expect("BOXMAKER_BXCTL is not set");
let output = Command::new(bxctl)
.arg("chat")
.arg("--socket")
.arg(&self.socket)
.args(["--session", session, "--no-thinking", "--say", text])
.output()
.expect("cannot run bxctl");
assert!(
output.status.success(),
"bxctl failed: {}",
String::from_utf8_lossy(&output.stderr)
);
String::from_utf8_lossy(&output.stdout)
.trim_end()
.to_string()
}
fn records(&self, session: &str) -> Vec<proto::LogRecord> {
let text =
std::fs::read_to_string(self.home.join("sessions").join(session).join("0.jsonl"))
.unwrap();
text.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect()
}
}
impl Drop for Served {
fn drop(&mut self) {
self.kill();
}
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn the_baseline_fits_the_token_budget() {
let socket = socket_path("budget");
let _proxy = Proxy::start(&socket);
let home = socket.parent().unwrap().join("home");
std::fs::create_dir_all(&home).unwrap();
let mut cfg = config(&socket);
cfg.paths.home = home.clone();
cfg.baseline.system = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../config/system.md");
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let soul = std::fs::read_to_string(root.join("config/soul.md")).unwrap();
let prose = std::fs::read_to_string(root.join("docs/design.md")).unwrap();
let prose: String = prose.split_whitespace().collect::<Vec<_>>().join(" ");
let client = Client::new(cfg.clone());
// The memory block as memoryd renders it (M5a spec, section 4): the starter soul with empty
// files, as a new install sends it; then every file at its limit, the most it can be.
let empty = format!(
"# soul.md\n{}\n\n# memory.md (0 of 2,200 characters)\n(empty)\n\n# user.md (0 of 1,375 characters)\n(empty)",
soul.trim()
);
let full = format!(
"# soul.md\n{}\n\n# memory.md (2,200 of 2,200 characters)\n{}\n\n# user.md (1,375 of 1,375 characters)\n{}",
prose.chars().take(4000).collect::<String>(),
prose.chars().skip(4000).take(2200).collect::<String>(),
prose.chars().skip(6200).take(1375).collect::<String>()
);
for (what, block) in [("starter", empty), ("full", full)] {
let baseline =
loopd::baseline::Baseline::assemble_with(&cfg, &loopd::tools::Registry::m5a(), &block)
.unwrap();
// The system text plus every tool schema as the request carries it.
let mut text = baseline.system.clone();
for tool in &baseline.tools {
text.push('\n');
text.push_str(&serde_json::to_string(&tool).unwrap());
}
let tokens = client.tokenize(&text).unwrap();
eprintln!("baseline, {what} memory: {tokens} tokens");
assert!(
tokens <= 3000,
"the baseline with {what} memory is {tokens} tokens; the brief allows 3000"
);
}
}
#[test]
#[ignore = "needs the real server; run with make verify-device"]
fn a_conversation_uses_tools_survives_a_restart_and_keeps_its_cache() {
let socket = socket_path("loop");
let _proxy = Proxy::start(&socket);
let home = socket.parent().unwrap().join("home");
let session = format!("device-{}", std::process::id());
let mut served = Served::start(&socket, &home);
let a1 = served.say(&session, "Reply with exactly: box made.");
assert!(a1.to_lowercase().contains("box made"), "{a1}");
let a2 = served.say(
&session,
"What is the current time? Use your clock tool, then tell me the year.",
);
assert!(a2.contains("2026") || a2.contains("202"), "{a2}");
// No broker is configured here, so the call fails in plain words and the turn goes on. What
// is checked is the path: find_tool, then call_tool, then an answer.
let a3 = served.say(
&session,
"Find a tool that reads files, use it to read /etc/hostname, and tell me in one \
sentence what happened.",
);
assert!(!a3.trim().is_empty(), "the turn must end in an answer");
served.kill();
served = Served::start(&socket, &home);
let a4 = served.say(
&session,
"What were the exact words I first asked you to reply with?",
);
assert!(
a4.to_lowercase().contains("box made"),
"after a restart the session must still know: {a4}"
);
let records = served.records(&session);
let tool_names: Vec<String> = records
.iter()
.filter_map(|r| match r {
proto::LogRecord::Assistant { tool_calls, .. } => Some(
tool_calls
.iter()
.map(|c| c.name.clone())
.collect::<Vec<_>>(),
),
_ => None,
})
.flatten()
.collect();
assert!(tool_names.contains(&"clock".to_string()), "{tool_names:?}");
assert!(
tool_names.contains(&"find_tool".to_string())
&& tool_names.contains(&"call_tool".to_string()),
"{tool_names:?}"
);
let usages = records
.iter()
.filter(|r| matches!(r, proto::LogRecord::Usage { .. }))
.count();
let assistants = records
.iter()
.filter(|r| matches!(r, proto::LogRecord::Assistant { .. }))
.count();
assert_eq!(usages, assistants, "one usage record per completion");
let losses: Vec<&proto::LogRecord> = records
.iter()
.filter(|r| matches!(r, proto::LogRecord::CacheLoss { .. }))
.collect();
assert!(
losses.is_empty(),
"every request hit the cache, including the one after the restart: {losses:?}"
);
let results = records
.iter()
.filter(|r| matches!(r, proto::LogRecord::ToolResult { .. }))
.count();
assert!(
results >= 3,
"clock, find_tool and call_tool each left a result: {results}"
);
}
@@ -0,0 +1,184 @@
//! `loopd` and memory: the memory block in a new session's baseline and not in a running one's,
//! no session without it, `memoryd` told after a finished turn, and the two core memory tools
//! (M5a spec, section 9). Do not edit.
mod support;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use loopd::channel::{Context, serve};
use loopd::llama::Client;
use loopd::memory::{MemorySource, UNAVAILABLE};
use loopd::tools::Registry;
use proto::{
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, read_frame, write_frame,
};
use support::{FakeServer, Home, Reply, ScriptedPort};
const CHAT: &str = "/v1/chat/completions";
/// A `memoryd` stand-in: the block it serves, and the sessions it was told about.
struct Recorder {
block: Mutex<Result<String, String>>,
updated: Mutex<Vec<String>>,
}
/// Lets the test keep its `Arc<Recorder>` while the context owns a `Box<dyn MemorySource>`.
struct Shared(Arc<Recorder>);
impl MemorySource for Shared {
fn block(&self) -> Result<String, String> {
self.0.block.lock().unwrap().clone()
}
fn updated(&self, session: &SessionId) -> Result<(), String> {
self.0
.updated
.lock()
.unwrap()
.push(session.as_str().to_string());
Ok(())
}
}
struct Loopd {
_home: Home,
server: FakeServer,
socket: PathBuf,
memory: Arc<Recorder>,
}
fn start(block: Result<&str, &str>) -> Loopd {
let home = Home::new();
let server = FakeServer::start();
let cfg = home.config(&server.socket);
let socket = cfg.channel_socket();
let listener = UnixListener::bind(&socket).unwrap();
let memory = Arc::new(Recorder {
block: Mutex::new(block.map(str::to_string).map_err(str::to_string)),
updated: Mutex::new(Vec::new()),
});
let ctx = Arc::new(Context::new(
cfg.clone(),
Client::new(cfg),
Box::new(ScriptedPort::new(vec![])),
Registry::m2b(),
Box::new(Shared(Arc::clone(&memory))),
));
std::thread::spawn(move || serve(listener, ctx));
Loopd {
_home: home,
server,
socket,
memory,
}
}
/// One turn; the last frame.
fn turn(l: &Loopd, session: &str, resume: bool) -> Message {
let mut stream = UnixStream::connect(&l.socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.unwrap();
let msg = Message::Turn(Turn {
session: SessionId::new(session).unwrap(),
content: "hello".to_string(),
resume,
});
let env = Envelope {
v: PROTOCOL_VERSION,
id: 7,
r#final: true,
msg,
};
write_frame(&mut stream, &env).unwrap();
loop {
let frame = read_frame(&mut stream).unwrap();
if frame.r#final {
return frame.msg;
}
}
}
fn system_prompt(l: &Loopd, n: usize) -> String {
let requests = l.server.requests_to(CHAT);
requests[n].json()["messages"][0]["content"]
.as_str()
.unwrap()
.to_string()
}
#[test]
fn a_new_session_starts_with_the_block_and_keeps_it() {
let l = start(Ok("# soul.md\nYou are kind."));
l.server
.route(CHAT, vec![Reply::fixture("plain"), Reply::fixture("plain")]);
assert!(matches!(turn(&l, "a", false), Message::TurnDone(_)));
assert_eq!(
system_prompt(&l, 0),
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind."
);
*l.memory.block.lock().unwrap() = Ok("# soul.md\nYou changed.".to_string());
assert!(matches!(turn(&l, "a", true), Message::TurnDone(_)));
assert_eq!(
system_prompt(&l, 1),
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind.",
"a running session keeps the bytes it started with"
);
}
#[test]
fn no_session_starts_without_the_block() {
let l = start(Err(
"memoryd at /x: refused; see docs/runbook.md#memory-unavailable",
));
match turn(&l, "b", false) {
Message::Error(e) => {
assert_eq!(e.code, ErrorCode::Internal);
assert!(e.detail.ends_with(UNAVAILABLE), "{}", e.detail);
}
other => panic!("started without memory: {other:?}"),
}
assert!(l.server.requests_to(CHAT).is_empty(), "no request was made");
}
#[test]
fn memoryd_is_told_after_a_finished_turn_only() {
let l = start(Ok(""));
l.server.route(
CHAT,
vec![Reply::fixture("plain"), Reply::fixture("bad_request")],
);
turn(&l, "c", false);
assert!(
matches!(turn(&l, "d", false), Message::Error(_)),
"a turn that ran and failed"
);
assert!(matches!(
turn(&l, "nobody-made-this", true),
Message::Error(_)
));
assert_eq!(*l.memory.updated.lock().unwrap(), ["c"]);
}
#[test]
fn recall_and_memory_are_core_tools_of_the_m5a_registry() {
let names: Vec<String> = Registry::m5a()
.core_schemas()
.into_iter()
.map(|s| s.name)
.collect();
assert_eq!(
names,
["clock", "recall", "memory", "find_tool", "call_tool"]
);
let m3a: Vec<String> = Registry::m3a()
.core_schemas()
.into_iter()
.map(|s| s.name)
.collect();
assert_eq!(m3a, ["clock", "find_tool", "call_tool"], "m3a is unchanged");
assert!(Registry::m5a().get("read_file").is_some());
}
@@ -0,0 +1,246 @@
//! 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 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 proto::SessionId;
use support::{FakeServer, Home, Reply};
// `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 = proto::read_frame(&mut stream).unwrap();
let reply = proto::Envelope {
v: 1,
id: env.id,
r#final: true,
msg: proto::Message::Error(proto::WireError {
code: proto::ErrorCode::Internal,
detail: "memory/user.md: 1,376 characters; its limit is 1,375".to_string(),
}),
};
proto::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.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,40 @@
//! Starting `memoryd` for the device tests, with the repository's starter `soul.md`: `loopd`
//! starts no session without the memory block. Do not edit.
use std::path::Path;
use std::process::{Child, Command};
use std::thread;
use std::time::Duration;
/// Write `memoryd.toml` into `home`, start `memoryd serve`, and wait for its `loopd` socket.
pub fn start(home: &Path) -> Child {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let config = home.join("memoryd.toml");
std::fs::write(
&config,
format!(
"[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n",
home.display(),
root.join("config/soul.md").display()
),
)
.unwrap();
let socket = home.join("run").join("loop-memory").join("memory.sock");
let _ = std::fs::remove_file(&socket);
let bin = std::env::var("BOXMAKER_MEMORYD").expect("BOXMAKER_MEMORYD is not set");
let mut child = Command::new(bin)
.arg("serve")
.arg("--config")
.arg(&config)
.spawn()
.expect("cannot start memoryd");
for _ in 0..600 {
if socket.exists() {
return child;
}
thread::sleep(Duration::from_millis(100));
}
let _ = child.kill();
let _ = child.wait();
panic!("memoryd did not come up within 60 s");
}