diff --git a/crates/loopd/src/channel.rs b/crates/loopd/src/channel.rs new file mode 100644 index 0000000..0237e0b --- /dev/null +++ b/crates/loopd/src/channel.rs @@ -0,0 +1,240 @@ +//! The channel server: one turn per connection over the M1 frame protocol. A session runs one +//! turn at a time; the busy set keeps a second turn on the same session from starting until the +//! first finishes, and is released before the final frame so a client can send the next turn the +//! moment it reads the last one. + +use std::collections::HashSet; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::sync::{Arc, Mutex}; + +use crate::baseline::Baseline; +use crate::config::Config; +use crate::llama::Client; +use crate::session::{Session, SessionError}; +use crate::tools::{Registry, ToolPort}; +use crate::turn::{Runtime, TurnError, run_turn}; +use proto::{ + Envelope, ErrorCode, FrameError, Message, PROTOCOL_VERSION, SessionId, TurnDone, WireError, + read_frame, write_frame, +}; + +/// Everything a turn needs, shared by every connection. +pub struct Context { + pub cfg: Config, + pub client: Client, + pub port: Box, + pub registry: Registry, + busy: Mutex>, +} + +impl Context { + pub fn new( + cfg: Config, + client: Client, + port: Box, + registry: Registry, + ) -> Context { + Context { + cfg, + client, + port, + registry, + busy: Mutex::new(HashSet::new()), + } + } +} + +/// While a turn runs, the session is in the busy set. The guard holds only an immutable borrow of +/// the context and a clone of the id, never the lock, so another connection can still check +/// membership while this turn is slow. The id is removed when the guard drops. +struct Held<'a> { + ctx: &'a Context, + id: SessionId, +} + +impl Drop for Held<'_> { + fn drop(&mut self) { + if let Ok(mut busy) = self.ctx.busy.lock() { + busy.remove(&self.id); + } + } +} + +/// Accepts connections forever, one thread each. Returns only if `accept` fails. +pub fn serve(listener: UnixListener, ctx: Arc) -> std::io::Result<()> { + for stream in listener.incoming() { + let stream = stream?; + let ctx = Arc::clone(&ctx); + std::thread::spawn(move || handle(stream, ctx)); + } + Ok(()) +} + +/// One connection: read a turn, run it, stream its events, send the final frame. +pub fn handle(stream: UnixStream, ctx: Arc) { + let mut stream = stream; + + // 1. Read the turn. A closed stream before anything is not an error; any other read failure + // is reported and closes the connection. + let request = match read_frame(&mut stream) { + Ok(env) => env, + Err(FrameError::Closed) => return, + Err(e) => { + let code = match &e { + FrameError::BadVersion(_) => ErrorCode::BadVersion, + FrameError::Json(_) => ErrorCode::BadMessage, + _ => ErrorCode::BadFrame, + }; + let _ = write_frame(&mut stream, &error_frame(code, 0, e.to_string())); + return; + } + }; + + // 2. Only a turn is served; every reply after this carries the request's id and version. + let Message::Turn(turn) = request.msg else { + let _ = write_frame( + &mut stream, + &error_frame(ErrorCode::BadMessage, request.id, String::new()), + ); + return; + }; + + // 3. Mark the session busy. If it already is, refuse at once. The guard releases the lock here + // and holds only the id plus an immutable borrow, so the turn below does not hold it. + let held = { + let mut busy = match ctx.busy.lock() { + Ok(busy) => busy, + Err(poisoned) => poisoned.into_inner(), + }; + if busy.contains(&turn.session) { + let _ = write_frame( + &mut stream, + &error_frame(ErrorCode::SessionBusy, request.id, String::new()), + ); + return; + } + busy.insert(turn.session.clone()); + Held { + ctx: &ctx, + id: turn.session.clone(), + } + }; + + // 4. Open or create the session, then run the turn, streaming each event as it happens. + let runtime = Runtime { + cfg: &ctx.cfg, + client: &ctx.client, + port: &*ctx.port, + registry: &ctx.registry, + }; + + let mut session = if turn.resume { + match Session::open(&ctx.cfg.paths.home, turn.session.clone()) { + Ok(session) => session, + Err(e) => { + let _ = write_frame( + &mut stream, + &error_frame(session_code(&e), request.id, e.to_string()), + ); + return; + } + } + } else { + let baseline = match Baseline::assemble(&ctx.cfg, &ctx.registry) { + Ok(baseline) => baseline, + Err(e) => { + let _ = write_frame( + &mut stream, + &error_frame(ErrorCode::Internal, request.id, e.to_string()), + ); + return; + } + }; + match Session::create( + &ctx.cfg.paths.home, + turn.session.clone(), + baseline, + ctx.cfg.slots.main, + ) { + Ok(session) => session, + Err(e) => { + let _ = write_frame( + &mut stream, + &error_frame(session_code(&e), request.id, e.to_string()), + ); + return; + } + } + }; + + let outcome = run_turn(&mut session, &runtime, &turn.content, &mut |event| { + let _ = write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: false, + msg: Message::TurnEvent(event.clone()), + }, + ); + }); + + // 5. Release the session before the final frame, so the next turn does not find it busy. + drop(held); + + // 6. and 7. A turn done is success; a turn error maps to a code. A write failure means the + // channel is gone, so we stop and finish quietly. + match outcome { + Ok(outcome) => { + let _ = write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: true, + msg: Message::TurnDone(TurnDone { + content: outcome.content, + usage: outcome.usage, + }), + }, + ); + } + Err(e) => { + let _ = write_frame( + &mut stream, + &error_frame(turn_code(&e), request.id, e.to_string()), + ); + } + } +} + +/// One error envelope: final, the given id and version. +fn error_frame(code: ErrorCode, id: u64, detail: String) -> Envelope { + Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg: Message::Error(WireError { code, detail }), + } +} + +/// The channel code for a session that could not be opened or created. +fn session_code(e: &SessionError) -> ErrorCode { + match e { + SessionError::Exists(_) => ErrorCode::SessionExists, + SessionError::NotFound(_) => ErrorCode::NoSuchSession, + _ => ErrorCode::Internal, + } +} + +/// The channel code for a turn that ran but failed. +fn turn_code(e: &TurnError) -> ErrorCode { + match e { + TurnError::SessionFull => ErrorCode::SessionFull, + TurnError::TurnLimit => ErrorCode::TurnLimit, + TurnError::Infer(_) => ErrorCode::Inference, + TurnError::Session(SessionError::Exists(_)) => ErrorCode::SessionExists, + TurnError::Session(SessionError::NotFound(_)) => ErrorCode::NoSuchSession, + TurnError::Session(_) => ErrorCode::Internal, + } +} diff --git a/crates/loopd/src/lib.rs b/crates/loopd/src/lib.rs index 5c74b73..5d4d50c 100644 --- a/crates/loopd/src/lib.rs +++ b/crates/loopd/src/lib.rs @@ -1,6 +1,7 @@ //! The agent loop: sessions, prompt assembly and memory. It holds no authority. pub mod baseline; +pub mod channel; pub mod config; pub mod http; pub mod llama; diff --git a/crates/loopd/tests/channel.rs b/crates/loopd/tests/channel.rs new file mode 100644 index 0000000..5d3fa4d --- /dev/null +++ b/crates/loopd/tests/channel.rs @@ -0,0 +1,380 @@ +//! 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::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) -> 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(), + )); + 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 { + 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 { + 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(_) + )); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index e142b81..00c0d72 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -36,6 +36,7 @@ reviewer adds findings under "Reviews" once per milestone. | M2b/04-loopd-baseline | 2026-09-18 | done | 2 | fail | none | Wrote crates/loopd/src/baseline.rs: `Baseline` (system prompt + core tool schemas, `deny_unknown_fields`), `BaselineError` (Read/Parse name the file, plus Hash) with Display/std::error::Error, `assemble` (system prompt trimmed of trailing whitespace, core memory appended with a blank line when its trimmed content is non-empty), `to_json`/`from_json`, `load`, and `hash` (sha256 of the canonical JSON). `messages` prepends the system message and replays every `LogRecord` variant explicitly named, so a new one is a compile error. The `\n\n` separator between system prompt and core memory had to be two newlines (a blank line), not one. All 6 baseline tests pass and all loopd tests pass with the new support module; first gate run failed on a rustfmt import-order diff, fixed with `cargo fmt`. | ? | | 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), `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) 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>`; `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. | ? | ## Reviews