Add the loopd serve command

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
2026-09-18 19:50:32 -07:00
parent 8bc835f623
commit eb8fc5f920
3 changed files with 211 additions and 17 deletions
+88 -17
View File
@@ -1,25 +1,49 @@
//! `loopd`: the agent loop daemon. Its one command today is `selftest`, which runs the startup //! `loopd`: the agent loop daemon. It has two commands: `selftest`, which runs the startup checks
//! checks before `loopd` serves anyone. //! before serving anyone, and `serve`, which runs those checks and then runs the channel server.
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::UnixListener;
use std::path::Path; use std::path::Path;
use std::process::ExitCode; use std::process::ExitCode;
use std::sync::Arc;
use loopd::channel::{self, Context};
use loopd::config::Config; use loopd::config::Config;
use loopd::llama::Client; use loopd::llama::Client;
use loopd::selftest::run; use loopd::selftest::{SelfTestError, run};
use loopd::tools::{FakeTools, Registry};
fn main() -> ExitCode { fn main() -> ExitCode {
let args: Vec<String> = std::env::args().skip(1).collect(); let args: Vec<String> = std::env::args().skip(1).collect();
let args: Vec<&str> = args.iter().map(String::as_str).collect(); let args: Vec<&str> = args.iter().map(String::as_str).collect();
match args.as_slice() { match args.as_slice() {
["selftest", "--config", path] => run_selftest(path), ["selftest", "--config", path] => run_selftest(path),
["serve", "--config", path] => run_serve(path),
_ => { _ => {
eprintln!("usage: loopd selftest --config <path>"); eprintln!("usage: loopd selftest --config <path>");
eprintln!("usage: loopd serve --config <path>");
ExitCode::from(2) ExitCode::from(2)
} }
} }
} }
/// The startup checks, with the lines `loopd` prints. Shared by `selftest` and `serve`.
fn run_selftest_check(client: &Client) -> Result<(), SelfTestError> {
let mut on_step = |step: &str| {
eprintln!("selftest: {step}");
};
match run(client, &mut on_step) {
Ok(()) => {
eprintln!("selftest: ok");
Ok(())
}
Err(e) => {
eprintln!("selftest: FAILED: {e}");
Err(e)
}
}
}
fn run_selftest(path: &str) -> ExitCode { fn run_selftest(path: &str) -> ExitCode {
let cfg = match Config::load(Path::new(path)) { let cfg = match Config::load(Path::new(path)) {
Ok(cfg) => cfg, Ok(cfg) => cfg,
@@ -29,19 +53,66 @@ fn run_selftest(path: &str) -> ExitCode {
} }
}; };
let mut on_step = |step: &str| { let client = Client::new(cfg);
eprintln!("selftest: {step}"); match run_selftest_check(&client) {
}; Ok(()) => ExitCode::SUCCESS,
let result = run(&Client::new(cfg), &mut on_step); Err(_) => ExitCode::from(1),
match result {
Ok(()) => {
eprintln!("selftest: ok");
ExitCode::SUCCESS
}
Err(e) => {
eprintln!("selftest: FAILED: {e}");
ExitCode::from(1)
}
} }
} }
fn run_serve(path: &str) -> ExitCode {
let cfg = match Config::load(Path::new(path)) {
Ok(cfg) => cfg,
Err(e) => {
eprintln!("loopd: {e}");
return ExitCode::from(1);
}
};
let socket = cfg.channel_socket();
if socket.exists()
&& let Err(e) = std::fs::remove_file(&socket)
{
eprintln!(
"loopd: cannot remove the old socket at {}: {e}",
socket.display()
);
return ExitCode::from(1);
}
let client = Client::new(cfg.clone());
if run_selftest_check(&client).is_err() {
return ExitCode::from(1);
}
if let Some(parent) = socket.parent()
&& let Err(e) = std::fs::create_dir_all(parent)
{
eprintln!("loopd: cannot create the socket directory: {e}");
return ExitCode::from(1);
}
let listener = match UnixListener::bind(&socket) {
Ok(listener) => listener,
Err(e) => {
eprintln!("loopd: cannot bind the socket at {}: {e}", socket.display());
return ExitCode::from(1);
}
};
if let Err(e) = std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o600)) {
eprintln!("loopd: cannot set the mode of {}: {e}", socket.display());
return ExitCode::from(1);
}
eprintln!("loopd: serving on {}", socket.display());
let ctx = Arc::new(Context::new(
cfg,
client,
Box::new(FakeTools::new()),
Registry::m2b(),
));
if let Err(e) = channel::serve(listener, ctx) {
eprintln!("loopd: the channel server stopped: {e}");
return ExitCode::from(1);
}
ExitCode::SUCCESS
}
+122
View File
@@ -0,0 +1,122 @@
//! 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};
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.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"
);
}
#[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"));
}
+1
View File
@@ -37,6 +37,7 @@ reviewer adds findings under "Reviews" once per milestone.
| M2b/05-loopd-session | 2026-09-18 | done | 2 | fail | none | Copied the given test byte-identical and wrote crates/loopd/src/session.rs: `Session` (id, dir, baseline, records, appended log file, next_call) and `SessionError` (Exists/NotFound/Io/Torn/Baseline/Encode) with derived Debug, Display and std::error::Error::source. `create` refuses an existing dir, writes `0.baseline.json`, opens `0.jsonl` with `create_new`+`append`, and appends a `SessionStart` (`Timestamp::now()`, epoch 0, the slot, `baseline.hash()`). `open` reads the baseline from the file (not `system.md`), requires every log line to end in `\n` and parse as a `LogRecord` else `Torn` with the 1-based line and reason, and sets `next_call` to one past the highest `ToolResult` call. `append` encodes, writes, `sync_data()`, then pushes to memory. All 7 session tests pass. First gate failed on clippy: split the `source()` arm that bound three different error types into three arms, removed the redundant `.write(true)` (implied by `append`), and used `path.display()` for the `Torn` path. | ? | | M2b/05-loopd-session | 2026-09-18 | done | 2 | fail | none | Copied the given test byte-identical and wrote crates/loopd/src/session.rs: `Session` (id, dir, baseline, records, appended log file, next_call) and `SessionError` (Exists/NotFound/Io/Torn/Baseline/Encode) with derived Debug, Display and std::error::Error::source. `create` refuses an existing dir, writes `0.baseline.json`, opens `0.jsonl` with `create_new`+`append`, and appends a `SessionStart` (`Timestamp::now()`, epoch 0, the slot, `baseline.hash()`). `open` reads the baseline from the file (not `system.md`), requires every log line to end in `\n` and parse as a `LogRecord` else `Torn` with the 1-based line and reason, and sets `next_call` to one past the highest `ToolResult` call. `append` encodes, writes, `sync_data()`, then pushes to memory. All 7 session tests pass. First gate failed on clippy: split the `source()` arm that bound three different error types into three arms, removed the redundant `.write(true)` (implied by `append`), and used `path.display()` for the `Torn` path. | ? |
| M2b/06-loopd-turn | 2026-09-18 | done | 1 | pass | none | Wrote crates/loopd/src/turn.rs (299 lines) and registered `pub mod turn;` in lib.rs; copied the two given tests and three fixtures byte-identical. `TurnError` (SessionFull/TurnLimit/Infer/Session, Display + std::error::Error + From<SessionError>), `TurnOutcome`, `Runtime`, `is_context_full` (the one 400 whose JSON `error.type` is `exceed_context_size_error`), and `run_turn`: append User, build the ChatRequest (slot, messages, tools, thinking), capture `last_usage`, `chat_with_retry` mapping ChatEvent->TurnEvent (dropping ToolCallDelta), append Assistant then Usage, report cache loss between the two conversations, and on no tool calls return `TurnOutcome { content: completion.content.unwrap_or_default(), usage }`; otherwise iterate tool calls under the iteration cap with a repeated-call detector (first repeat returns "already called", a second repeat is TurnLimit), `cap_result`, and `dispatch` (find_tool/call_tool local, every other tool — including read_file — to the port). `run_call` maps Dispatch::Local and every ToolResponse variant to (text, Public, untrusted). Two compile fixes before the gate: `u64::try_from(*ahead).unwrap_or(u64::MAX)` (usize has no From<u64>) and `let Ok(value) = from_str(body) else { return false }` (a temporary borrow); `session.baseline()` returns a reference so it is bound inside the loop. All 6 turn and 9 limits tests pass; `make gate` prints `gate: ok`. | ? | | M2b/06-loopd-turn | 2026-09-18 | done | 1 | pass | none | Wrote crates/loopd/src/turn.rs (299 lines) and registered `pub mod turn;` in lib.rs; copied the two given tests and three fixtures byte-identical. `TurnError` (SessionFull/TurnLimit/Infer/Session, Display + std::error::Error + From<SessionError>), `TurnOutcome`, `Runtime`, `is_context_full` (the one 400 whose JSON `error.type` is `exceed_context_size_error`), and `run_turn`: append User, build the ChatRequest (slot, messages, tools, thinking), capture `last_usage`, `chat_with_retry` mapping ChatEvent->TurnEvent (dropping ToolCallDelta), append Assistant then Usage, report cache loss between the two conversations, and on no tool calls return `TurnOutcome { content: completion.content.unwrap_or_default(), usage }`; otherwise iterate tool calls under the iteration cap with a repeated-call detector (first repeat returns "already called", a second repeat is TurnLimit), `cap_result`, and `dispatch` (find_tool/call_tool local, every other tool — including read_file — to the port). `run_call` maps Dispatch::Local and every ToolResponse variant to (text, Public, untrusted). Two compile fixes before the gate: `u64::try_from(*ahead).unwrap_or(u64::MAX)` (usize has no From<u64>) and `let Ok(value) = from_str(body) else { return false }` (a temporary borrow); `session.baseline()` returns a reference so it is bound inside the loop. All 6 turn and 9 limits tests pass; `make gate` prints `gate: ok`. | ? |
| M2b/07-loopd-channel | 2026-09-18 | done | 3 | fail | none | Wrote crates/loopd/src/channel.rs and registered `pub mod channel;` in lib.rs. `Context` holds a private `Mutex<HashSet<SessionId>>`; `serve` accepts forever with one thread per connection and returns on an accept error; `handle` does the seven steps (read_frame with Closed-before-anything, turn-only, mark busy, open-or-create plus run_turn streaming events, release before the final frame, error-code mapping, quiet write failure). The busy guard is a `Held` struct that borrows the context immutably and holds a clone of the id but never the lock, and it is dropped before sending turn_done or error so a client can send the next turn the moment it reads the last one — that is what keeps `a_busy_session_is_refused_at_once` and the concurrent-session test correct. Channel test reported `6 passed` ten runs in a row, all clean. Two fixes before a clean gate: `cargo fmt` import order and a clippy `question_mark` on the accept loop, re-run after each. | ? | | M2b/07-loopd-channel | 2026-09-18 | done | 3 | fail | none | Wrote crates/loopd/src/channel.rs and registered `pub mod channel;` in lib.rs. `Context` holds a private `Mutex<HashSet<SessionId>>`; `serve` accepts forever with one thread per connection and returns on an accept error; `handle` does the seven steps (read_frame with Closed-before-anything, turn-only, mark busy, open-or-create plus run_turn streaming events, release before the final frame, error-code mapping, quiet write failure). The busy guard is a `Held` struct that borrows the context immutably and holds a clone of the id but never the lock, and it is dropped before sending turn_done or error so a client can send the next turn the moment it reads the last one — that is what keeps `a_busy_session_is_refused_at_once` and the concurrent-session test correct. Channel test reported `6 passed` ten runs in a row, all clean. Two fixes before a clean gate: `cargo fmt` import order and a clippy `question_mark` on the accept loop, re-run after each. | ? |
| M2b/08-loopd-serve | 2026-09-18 | done | 2 | fail | none | Rewrote crates/loopd/src/main.rs into two commands, `selftest` and `serve`, both sharing `run_selftest_check` so the self-test lines are identical. `serve` loads config (exit 1 on failure), removes an existing socket via `channel_socket()` before the self-test, runs the self-test and exits 1 without binding on failure, then creates the socket's parent dir, binds, sets mode 0600 with `std::fs::set_permissions`, prints `serving on`, and calls `channel::serve` with a `Context` from the config, client, `Box::new(FakeTools::new())` and `Registry::m2b()`. Anything else prints both usages and exits 2. The `serve_refuses...` test's "no socket left behind" holds because the socket is removed before the self-test and binding happens only after it passes. First gate run failed on two clippy `collapsible_if` lints; collapsed the two nested `if let` into edition-2024 let-chains and re-ran, which passed. `cargo test -p loopd --test serve` reports 3 passed. | ? |
## Reviews ## Reviews