Add the channel server on loop.sock

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

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<dyn ToolPort>,
+    pub registry: Registry,
+    busy: Mutex<HashSet<SessionId>>,
+}
+
+impl Context {
+    pub fn new(
+        cfg: Config,
+        client: Client,
+        port: Box<dyn ToolPort>,
+        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<Context>) -> 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<Context>) {
+    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,
+    }
+}

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;

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<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(),
+    ));
+    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(_)
+    ));
+}

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<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. | ? |
 
 
 ## Reviews