gatewayd: deliver, turns on loop.sock and answers posted in their thread

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
2026-09-23 21:34:34 -07:00
parent 24cac6a4eb
commit d1d8531495
5 changed files with 607 additions and 0 deletions
@@ -0,0 +1,200 @@
+//! One turn on `loop.sock` and its answer in the thread (M4a spec, section 8). Typing is shown by
+//! `serve`, which owns the WebSocket; this module only sends the turn and posts what comes back.
+
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+
+use proto::{
+    Envelope, ErrorCode, Message, PROTOCOL_VERSION, Turn, TurnEvent, WireError, read_frame,
+    write_frame,
+};
+
+use crate::mm::MmError;
+use crate::mm::rest::Client;
+use crate::sessions::{Batch, Thread};
+
+/// The longest reply we post, in characters (a post holds at most 16,383).
+pub const MAX_POST: usize = 16_000;
+pub const LOOP_DOWN: &str = "Boxmaker's loop is not running (see docs/runbook.md#loop-unavailable)";
+pub const EMPTY_ANSWER: &str = "(the answer was empty)";
+
+/// Somewhere to post: Mattermost, or a test's record.
+pub trait Poster: Send + Sync {
+    fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError>;
+}
+
+impl Poster for Client {
+    fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError> {
+        self.create_post(channel, root, text).map(|_| ())
+    }
+}
+
+/// How a turn ended.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Outcome {
+    Answer(String),
+    Refused(WireError),
+    /// `loop.sock` could not be reached or closed early; why, for the log.
+    LoopDown(String),
+}
+
+pub fn approval_text(approval: u64) -> String {
+    format!(
+        "waiting for approval {approval}: approve or deny it with `bxctl` (Mattermost approvals arrive in M4b)"
+    )
+}
+
+/// "Error: <code>: <detail>"; the detail carries `loopd`'s runbook pointer when there is one.
+pub fn error_text(error: &WireError) -> String {
+    // "Error: <code>: <detail>", where <code> is the snake_case name serde gives the ErrorCode
+    // (`serde_json::to_value(code)` is a JSON string, e.g. "no_such_session").
+    let code = match serde_json::to_value(error.code) {
+        Ok(serde_json::Value::String(code)) => code,
+        _ => return format!("Error: <unknown>: {}", error.detail),
+    };
+    format!("Error: {}: {}", code, error.detail)
+}
+
+/// An answer in posts of at most `MAX_POST` characters: each cut at the last newline before the
+/// limit (the newline is dropped), or at the limit when there is none.
+pub fn split_answer(text: &str) -> Vec<String> {
+    // Blank (only whitespace) -> [EMPTY_ANSWER]. Otherwise, while the rest is longer than MAX_POST
+    // *characters*: take the first MAX_POST characters; if they hold a newline past position 0, cut
+    // at the last one and drop that newline; else cut at MAX_POST characters. The last part is the
+    // rest.
+    if text.trim().is_empty() {
+        return vec![EMPTY_ANSWER.to_string()];
+    }
+
+    let mut parts = Vec::new();
+    let mut rest = text;
+    while rest.chars().count() > MAX_POST {
+        let cut = rest
+            .char_indices()
+            .take(MAX_POST)
+            .filter(|&(i, c)| i != 0 && c == '\n')
+            .last()
+            .map(|(i, _)| i);
+        let (end, skip) = match cut {
+            // A newline past position 0: drop it by cutting before and skipping past it.
+            Some(bi) => (bi, bi + 1),
+            // No newline: cut at the byte index of the MAX_POST-th character, never inside one.
+            None => {
+                let byte = rest
+                    .char_indices()
+                    .nth(MAX_POST)
+                    .map(|(i, _)| i)
+                    .unwrap_or(rest.len());
+                (byte, byte)
+            }
+        };
+        parts.push(rest[..end].to_string());
+        rest = &rest[skip..];
+    }
+    parts.push(rest.to_string());
+    parts
+}
+
+/// Send one turn and read it to its end; `on_event` sees every event.
+fn one_turn(
+    socket: &Path,
+    batch: &Batch,
+    resume: bool,
+    on_event: &mut dyn FnMut(&TurnEvent),
+) -> Outcome {
+    // 1. Connect to the socket (else LoopDown("cannot connect to <path>: <e>")).
+    let mut stream = match UnixStream::connect(socket) {
+        Ok(stream) => stream,
+        Err(e) => {
+            return Outcome::LoopDown(format!("cannot connect to {}: {}", socket.display(), e));
+        }
+    };
+
+    // 2. `write_frame` one Envelope: v PROTOCOL_VERSION, id 1, final true, msg Turn (else
+    //    LoopDown).
+    let turn = Turn {
+        session: batch.session.clone(),
+        content: batch.text.clone(),
+        resume,
+    };
+    if let Err(e) = write_frame(
+        &mut stream,
+        &Envelope {
+            v: PROTOCOL_VERSION,
+            id: 1,
+            r#final: true,
+            msg: Message::Turn(turn),
+        },
+    ) {
+        return Outcome::LoopDown(format!("cannot send the turn: {}", e));
+    }
+
+    // 3. `read_frame` until the end: (id 1, not final, TurnEvent) -> on_event; (1, final, TurnDone)
+    //    -> Answer with its content; (1, final, Error) -> Refused; a read error -> LoopDown("the
+    //    turn ended early: <e>"); anything else -> LoopDown("an unexpected frame").
+    loop {
+        let env = match read_frame(&mut stream) {
+            Ok(env) => env,
+            Err(e) => return Outcome::LoopDown(format!("the turn ended early: {}", e)),
+        };
+        if env.id != 1 {
+            return Outcome::LoopDown("an unexpected frame".to_string());
+        }
+        match (env.r#final, &env.msg) {
+            (false, Message::TurnEvent(e)) => on_event(e),
+            (true, Message::TurnDone(d)) => return Outcome::Answer(d.content.clone()),
+            (true, Message::Error(w)) => return Outcome::Refused(w.clone()),
+            _ => return Outcome::LoopDown("an unexpected frame".to_string()),
+        }
+    }
+}
+
+/// A turn for a batch. A reply in a thread `loopd` does not know creates the session, as
+/// `bxctl chat --session` does.
+pub fn run_turn(socket: &Path, batch: &Batch, on_event: &mut dyn FnMut(&TurnEvent)) -> Outcome {
+    // `one_turn` with batch.resume. When that is Refused with NoSuchSession and batch.resume was
+    // true, one more `one_turn` with resume false. Otherwise the first outcome.
+    match one_turn(socket, batch, batch.resume, on_event) {
+        Outcome::Refused(err) if err.code == ErrorCode::NoSuchSession && batch.resume => {
+            one_turn(socket, batch, false, on_event)
+        }
+        outcome => outcome,
+    }
+}
+
+/// Post one thing in a thread, logging (via `log`) when Mattermost refuses it, and carry on.
+fn post(poster: &dyn Poster, thread: &Thread, text: String, log: &dyn Fn(&str)) {
+    if let Err(e) = poster.post(&thread.channel, &thread.root, &text) {
+        log(&format!(
+            "gatewayd: cannot post in {} (thread {}): {}",
+            thread.channel, thread.root, e
+        ));
+    }
+}
+
+/// Run a batch's turn and post what comes of it in its thread. A post that fails is logged.
+pub fn deliver(poster: &dyn Poster, socket: &Path, batch: &Batch, log: &dyn Fn(&str)) {
+    // Post in the batch thread. An ApprovalPending event posts `approval_text(approval)` at once.
+    // Then: Answer -> every part of `split_answer`, in order; Refused -> `error_text`;
+    // LoopDown(why) -> log "gatewayd: <session>: <why>" and post LOOP_DOWN. A post that fails is
+    // logged, exactly "gatewayd: cannot post in <channel> (thread <root>): <error>", and the rest
+    // goes on.
+    let outcome = run_turn(socket, batch, &mut |event| {
+        if let TurnEvent::ApprovalPending { approval, .. } = event {
+            post(poster, &batch.thread, approval_text(*approval), log);
+        }
+    });
+
+    match outcome {
+        Outcome::Answer(content) => {
+            for part in split_answer(&content) {
+                post(poster, &batch.thread, part, log);
+            }
+        }
+        Outcome::Refused(err) => post(poster, &batch.thread, error_text(&err), log),
+        Outcome::LoopDown(why) => {
+            log(&format!("gatewayd: {}: {}", batch.session.as_str(), why));
+            post(poster, &batch.thread, LOOP_DOWN.to_string(), log);
+        }
+    }
+}
@@ -2,6 +2,7 @@
 //! the answers back, over outbound connections only. M4a spec: `docs/specs/2026-09-23-m4a-gateway.md`.

 pub mod config;
+pub mod deliver;
 pub mod http;
 pub mod mm;
 pub mod net;
@@ -0,0 +1,304 @@
+//! A turn on `loop.sock` and its answer in the thread, against a fake `loopd`: the answer, long
+//! answers, approvals, errors, an unknown session, and a loop that is not there (M4a spec, section
+//! 8). Do not edit.
+
+#[path = "support/fake_loop.rs"]
+mod fake_loop;
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::sync::Mutex;
+use std::time::Duration;
+
+use fake_loop::{Reply, done, error, event, serve_loop};
+use gatewayd::deliver::{EMPTY_ANSWER, LOOP_DOWN, MAX_POST, Poster, deliver, split_answer};
+use gatewayd::mm::MmError;
+use gatewayd::sessions::{Batch, Thread};
+use proto::{DataClass, ErrorCode, SessionId, Timestamp, TurnEvent};
+use tmp::TempDir;
+
+const DM: &str = "d0000000000000000000000000";
+const ROOT: &str = "r0000000000000000000000000";
+
+#[derive(Default)]
+struct Record {
+    posts: Mutex<Vec<(String, String, String)>>,
+    fail: bool,
+}
+
+impl Poster for Record {
+    fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError> {
+        if self.fail {
+            return Err(MmError::Status(500, "down".to_string()));
+        }
+        self.posts
+            .lock()
+            .unwrap()
+            .push((channel.to_string(), root.to_string(), text.to_string()));
+        Ok(())
+    }
+}
+
+impl Record {
+    fn texts(&self) -> Vec<String> {
+        let posts = self.posts.lock().unwrap();
+        assert!(
+            posts.iter().all(|(c, r, _)| c == DM && r == ROOT),
+            "every post in the thread"
+        );
+        posts.iter().map(|(_, _, t)| t.clone()).collect()
+    }
+}
+
+fn batch(resume: bool, text: &str) -> Batch {
+    Batch {
+        session: SessionId::new(&format!("mm-{ROOT}")).unwrap(),
+        thread: Thread {
+            channel: DM.to_string(),
+            root: ROOT.to_string(),
+        },
+        resume,
+        text: text.to_string(),
+    }
+}
+
+fn run(dir: &TempDir, poster: &Record, batch: &Batch) -> Vec<String> {
+    let log = Mutex::new(Vec::new());
+    deliver(poster, &dir.path().join("loop.sock"), batch, &|line| {
+        log.lock().unwrap().push(line.to_string())
+    });
+    log.into_inner().unwrap()
+}
+
+#[test]
+fn the_answer_is_posted_and_nothing_else() {
+    let dir = TempDir::new("deliver-answer");
+    let turns = serve_loop(&dir.path().join("loop.sock"), |_, _| {
+        vec![
+            event(TurnEvent::Progress {
+                total: 10,
+                cache: 0,
+                processed: 10,
+            }),
+            event(TurnEvent::Reasoning {
+                text: "private thoughts".to_string(),
+            }),
+            event(TurnEvent::ToolCallStarted {
+                name: "read_file".to_string(),
+            }),
+            event(TurnEvent::ToolResult {
+                name: "read_file".to_string(),
+                class: DataClass::Private,
+                truncated: false,
+            }),
+            event(TurnEvent::Content {
+                text: "The ans".to_string(),
+            }),
+            done("The answer."),
+        ]
+    });
+    let poster = Record::default();
+    let log = run(&dir, &poster, &batch(true, "one\n\ntwo"));
+    assert_eq!(poster.texts(), ["The answer."]);
+    assert!(log.is_empty(), "{log:?}");
+    let turn = turns.recv_timeout(Duration::from_secs(5)).unwrap();
+    assert_eq!(
+        (turn.session.as_str(), turn.content.as_str(), turn.resume),
+        (format!("mm-{ROOT}").as_str(), "one\n\ntwo", true)
+    );
+}
+
+#[test]
+fn an_approval_is_announced_once_before_the_answer() {
+    let dir = TempDir::new("deliver-approval");
+    let expires = Timestamp::from_unix_millis(1_758_650_000_000).unwrap();
+    let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| {
+        vec![
+            event(TurnEvent::ApprovalPending {
+                approval: 42,
+                tool: "shell".to_string(),
+                expires,
+            }),
+            done("done"),
+        ]
+    });
+    let poster = Record::default();
+    run(&dir, &poster, &batch(false, "go"));
+    assert_eq!(
+        poster.texts(),
+        [
+            "waiting for approval 42: approve or deny it with `bxctl` (Mattermost approvals arrive in M4b)",
+            "done"
+        ]
+    );
+}
+
+#[test]
+fn errors_are_posted_with_their_code() {
+    let dir = TempDir::new("deliver-error");
+    let _turns = serve_loop(&dir.path().join("loop.sock"), |_, _| {
+        vec![error(
+            ErrorCode::Inference,
+            "the model server failed\nsee docs/runbook.md#loopd-selftest-failed",
+        )]
+    });
+    let poster = Record::default();
+    run(&dir, &poster, &batch(false, "go"));
+    assert_eq!(
+        poster.texts(),
+        ["Error: inference: the model server failed\nsee docs/runbook.md#loopd-selftest-failed"]
+    );
+}
+
+#[test]
+fn a_reply_in_a_thread_loopd_does_not_know_creates_the_session() {
+    let dir = TempDir::new("deliver-unknown");
+    let turns = serve_loop(&dir.path().join("loop.sock"), |n, _| {
+        if n == 0 {
+            vec![error(ErrorCode::NoSuchSession, "no such session")]
+        } else {
+            vec![done("hello")]
+        }
+    });
+    let poster = Record::default();
+    run(&dir, &poster, &batch(true, "hi"));
+    assert_eq!(poster.texts(), ["hello"]);
+    let first = turns.recv_timeout(Duration::from_secs(5)).unwrap();
+    let second = turns.recv_timeout(Duration::from_secs(5)).unwrap();
+    assert_eq!(
+        (first.resume, second.resume, second.content.as_str()),
+        (true, false, "hi")
+    );
+}
+
+#[test]
+fn a_new_session_is_not_retried() {
+    let dir = TempDir::new("deliver-noretry");
+    let turns = serve_loop(&dir.path().join("loop.sock"), |_, _| {
+        vec![error(ErrorCode::NoSuchSession, "odd")]
+    });
+    let poster = Record::default();
+    run(&dir, &poster, &batch(false, "hi"));
+    assert_eq!(poster.texts(), ["Error: no_such_session: odd"]);
+    assert!(turns.recv_timeout(Duration::from_secs(5)).is_ok());
+    assert!(
+        turns.recv_timeout(Duration::from_millis(200)).is_err(),
+        "one turn only"
+    );
+}
+
+#[test]
+fn a_loop_that_is_not_there_or_goes_away() {
+    let dir = TempDir::new("deliver-down");
+    let poster = Record::default();
+    let log = run(&dir, &poster, &batch(false, "hi"));
+    assert_eq!(poster.texts(), [LOOP_DOWN]);
+    assert_eq!(log.len(), 1, "{log:?}");
+    assert!(
+        log[0].starts_with(&format!("gatewayd: mm-{ROOT}: cannot connect to ")),
+        "{log:?}"
+    );
+
+    for (n, replies) in [
+        vec![
+            event(TurnEvent::Content {
+                text: "x".to_string(),
+            }),
+            Reply::Close,
+        ],
+        vec![Reply::Bytes(b"\x00\x00\x00\x05{bad}".to_vec())],
+        vec![
+            Reply::Frame(proto::Envelope {
+                v: 1,
+                id: 2,
+                r#final: true,
+                msg: proto::Message::Ok(proto::Empty {}),
+            }),
+            done("late"),
+        ],
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let dir = TempDir::new(&format!("deliver-early-{n}"));
+        let replies = Mutex::new(Some(replies));
+        let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| {
+            replies.lock().unwrap().take().unwrap_or_default()
+        });
+        let poster = Record::default();
+        let log = run(&dir, &poster, &batch(false, "hi"));
+        assert_eq!(poster.texts(), [LOOP_DOWN], "case {n}");
+        assert_eq!(log.len(), 1, "case {n}: {log:?}");
+    }
+}
+
+#[test]
+fn a_post_that_fails_is_logged() {
+    let dir = TempDir::new("deliver-postfail");
+    let _turns = serve_loop(&dir.path().join("loop.sock"), |_, _| vec![done("lost")]);
+    let poster = Record {
+        fail: true,
+        ..Record::default()
+    };
+    let log = run(&dir, &poster, &batch(false, "hi"));
+    assert_eq!(
+        log,
+        [format!(
+            "gatewayd: cannot post in {DM} (thread {ROOT}): status 500: \"down\""
+        )]
+    );
+}
+
+#[test]
+fn long_answers_are_split_at_newlines() {
+    let short = "a".repeat(MAX_POST);
+    assert_eq!(split_answer(&short), [short.as_str()]);
+    let over = "a".repeat(MAX_POST + 1);
+    assert_eq!(split_answer(&over), ["a".repeat(MAX_POST), "a".to_string()]);
+    let lines = format!(
+        "{}\n{}\n{}",
+        "a".repeat(10_000),
+        "b".repeat(5_000),
+        "c".repeat(2_000)
+    );
+    assert_eq!(
+        split_answer(&lines),
+        [
+            format!("{}\n{}", "a".repeat(10_000), "b".repeat(5_000)),
+            "c".repeat(2_000)
+        ]
+    );
+    let wide = "é".repeat(MAX_POST + 5);
+    let parts = split_answer(&wide);
+    assert_eq!(
+        parts.iter().map(|p| p.chars().count()).collect::<Vec<_>>(),
+        [MAX_POST, 5],
+        "characters, not bytes"
+    );
+    let leading = format!("\n{}", "x".repeat(MAX_POST + 1));
+    let parts = split_answer(&leading);
+    assert!(
+        parts
+            .iter()
+            .all(|p| !p.is_empty() && p.chars().count() <= MAX_POST),
+        "{:?}",
+        parts.iter().map(|p| p.len()).collect::<Vec<_>>()
+    );
+    assert_eq!(parts.concat(), leading, "a hard cut drops nothing");
+    assert_eq!(split_answer(""), [EMPTY_ANSWER]);
+    assert_eq!(split_answer(" \n "), [EMPTY_ANSWER]);
+}
+
+#[test]
+fn a_long_answer_is_posted_in_order() {
+    let dir = TempDir::new("deliver-long");
+    let answer = format!("{}\n{}", "a".repeat(MAX_POST - 1), "b".repeat(MAX_POST));
+    let sent = answer.clone();
+    let _turns = serve_loop(&dir.path().join("loop.sock"), move |_, _| vec![done(&sent)]);
+    let poster = Record::default();
+    run(&dir, &poster, &batch(false, "hi"));
+    assert_eq!(
+        poster.texts(),
+        ["a".repeat(MAX_POST - 1), "b".repeat(MAX_POST)]
+    );
+}
@@ -0,0 +1,101 @@
+//! A fake `loopd` on a Unix socket: it records each turn and answers with the frames the test's
+//! function gives for it. Do not edit.
+
+#![allow(dead_code)] // each test file uses a different part of this module
+
+use std::io::Write;
+use std::os::unix::net::UnixListener;
+use std::path::Path;
+use std::sync::mpsc;
+
+use proto::{
+    Envelope, ErrorCode, Message, PROTOCOL_VERSION, Turn, TurnDone, TurnEvent, Usage, WireError,
+    read_frame, write_frame,
+};
+
+const USAGE: Usage = Usage {
+    cache_n: 0,
+    prompt_n: 1,
+    predicted_n: 1,
+    reasoning_tokens: 0,
+    thinking_capped: false,
+};
+
+/// What the fake sends back for a turn.
+pub enum Reply {
+    Frame(Envelope),
+    /// Raw bytes, for broken frames.
+    Bytes(Vec<u8>),
+    /// Stop answering this connection (the caller sees it close).
+    Close,
+}
+
+pub fn event(e: TurnEvent) -> Reply {
+    Reply::Frame(Envelope {
+        v: PROTOCOL_VERSION,
+        id: 1,
+        r#final: false,
+        msg: Message::TurnEvent(e),
+    })
+}
+
+pub fn done(content: &str) -> Reply {
+    let msg = Message::TurnDone(TurnDone {
+        content: content.to_string(),
+        usage: USAGE,
+    });
+    Reply::Frame(Envelope {
+        v: PROTOCOL_VERSION,
+        id: 1,
+        r#final: true,
+        msg,
+    })
+}
+
+pub fn error(code: ErrorCode, detail: &str) -> Reply {
+    let msg = Message::Error(WireError {
+        code,
+        detail: detail.to_string(),
+    });
+    Reply::Frame(Envelope {
+        v: PROTOCOL_VERSION,
+        id: 1,
+        r#final: true,
+        msg,
+    })
+}
+
+/// Listen on `socket`; `script(n, &turn)` gives the replies to the n-th turn (from 0).
+pub fn serve_loop<F>(socket: &Path, script: F) -> mpsc::Receiver<Turn>
+where
+    F: Fn(usize, &Turn) -> Vec<Reply> + Send + 'static,
+{
+    let listener = UnixListener::bind(socket).unwrap();
+    let (tx, rx) = mpsc::channel();
+    std::thread::spawn(move || {
+        for (n, stream) in listener.incoming().enumerate() {
+            let Ok(mut stream) = stream else { continue };
+            let Ok(Envelope {
+                msg: Message::Turn(turn),
+                ..
+            }) = read_frame(&mut stream)
+            else {
+                continue;
+            };
+            // Reported before the script runs: a script may wait for the test.
+            let _ = tx.send(turn.clone());
+            let replies = script(n, &turn);
+            for reply in replies {
+                let ok = match reply {
+                    Reply::Frame(env) => write_frame(&mut stream, &env).is_ok(),
+                    Reply::Bytes(b) => stream.write_all(&b).is_ok(),
+                    Reply::Close => false,
+                };
+                if !ok {
+                    break;
+                }
+            }
+        }
+    });
+    rx
+}