From d1d8531495e61a845c1c979068f0cc07c6723037 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Wed, 23 Sep 2026 21:34:34 -0700 Subject: [PATCH] gatewayd: deliver, turns on loop.sock and answers posted in their thread Implemented-By: OpenCode session (model recorded in docs/implementer-log.md) --- crates/gatewayd/src/deliver.rs | 200 ++++++++++++++ crates/gatewayd/src/lib.rs | 1 + crates/gatewayd/tests/deliver.rs | 304 +++++++++++++++++++++ crates/gatewayd/tests/support/fake_loop.rs | 101 +++++++ docs/implementer-log.md | 1 + 5 files changed, 607 insertions(+) create mode 100644 crates/gatewayd/src/deliver.rs create mode 100644 crates/gatewayd/tests/deliver.rs create mode 100644 crates/gatewayd/tests/support/fake_loop.rs diff --git a/crates/gatewayd/src/deliver.rs b/crates/gatewayd/src/deliver.rs new file mode 100644 index 0000000..91ab7f3 --- /dev/null +++ b/crates/gatewayd/src/deliver.rs @@ -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: : "; the detail carries `loopd`'s runbook pointer when there is one. +pub fn error_text(error: &WireError) -> String { + // "Error: : ", where 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: : {}", 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 { + // 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 : ")). + 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: "); 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: : " and post LOOP_DOWN. A post that fails is + // logged, exactly "gatewayd: cannot post in (thread ): ", 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); + } + } +} diff --git a/crates/gatewayd/src/lib.rs b/crates/gatewayd/src/lib.rs index b257ff2..90717bb 100644 --- a/crates/gatewayd/src/lib.rs +++ b/crates/gatewayd/src/lib.rs @@ -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; diff --git a/crates/gatewayd/tests/deliver.rs b/crates/gatewayd/tests/deliver.rs new file mode 100644 index 0000000..642e981 --- /dev/null +++ b/crates/gatewayd/tests/deliver.rs @@ -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>, + 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 { + 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 { + 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::>(), + [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::>() + ); + 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)] + ); +} diff --git a/crates/gatewayd/tests/support/fake_loop.rs b/crates/gatewayd/tests/support/fake_loop.rs new file mode 100644 index 0000000..aacb4d2 --- /dev/null +++ b/crates/gatewayd/tests/support/fake_loop.rs @@ -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), + /// 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(socket: &Path, script: F) -> mpsc::Receiver +where + F: Fn(usize, &Turn) -> Vec + 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 +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index cf8af16..4aa0a8e 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M4a/13-gatewayd-deliver | 2026-09-23 | done | 1 | pass | none | Copied `tests/deliver.rs`, `tests/support/fake_loop.rs` and the `src/deliver.rs` skeleton, added `pub mod deliver;` to `lib.rs` (before `http`, alphabetical). Filled `error_text` (the snake_case name serde gives the ErrorCode via `serde_json::to_value`, falling back to `` without ever formatting `ErrorCode` with `{}` since it has no Display); `split_answer` (blank/whitespace-only → [EMPTY_ANSWER]; else while the rest is longer than `MAX_POST` *characters*, cut at the last newline within the first `MAX_POST` chars past position 0 — dropping that newline — else at the byte index of the `MAX_POST`-th char via `char_indices().nth(MAX_POST)`, never inside one char); `one_turn` (connect fails → LoopDown("cannot connect to : "), write one id-1 final Turn envelope (write error → LoopDown), then read: (1, not final, TurnEvent)→on_event, (1, final, TurnDone)→Answer(content), (1, final, Error)→Refused, a read error → LoopDown("the turn ended early: "), anything else → LoopDown("an unexpected frame")); `run_turn` (one_turn with `batch.resume`, and when Refused(NoSuchSession) with `batch.resume` true, one more with resume false to create the session, as `bxctl chat --session`); `deliver` (post in the thread — ApprovalPending posts `approval_text` at once, then Answer→every split part in order, Refused→error_text, LoopDown→log "gatewayd: : " and post LOOP_DOWN; a failing post logs "gatewayd: cannot post in (thread ): " through a small `post` helper). All 9 tests pass; `make gate` prints `gate: ok` first run. | ? | | M4a/12-gatewayd-state | 2026-09-23 | done | 1 | pass | none | Copied `tests/state.rs` and the `src/state.rs` skeleton, added `pub mod state;` to `lib.rs`. Filled `problem` (every `channels` key, `recent` and `threads` entry must be `valid_id` → "not a Mattermost id: "; each `in_flight` entry, session `mm-`+valid id with channel and root valid → "a turn in flight is not valid: "); `load` (NotFound → empty StateFile, any other read error, serde parse error or `problem()` → Read(path, why)); `save`/`persist` (the six atomic steps of brokerd's persist, io error mapped to Write, old file left on failure); `handled` (records unseen ids keeping the newest RECENT_KEPT, moves the channel mark to the max), `mark` (sets only a channel without a mark), `join_thread` (keeps the newest THREADS_KEPT), `start_turn`/`end_turn`/`take_in_flight` (removing by session, save only when in_flight was non-empty). Every mutating method saves before returning. All 4 tests pass in ~0.04 s; `make gate` prints `gate: ok` first run. | ? | | M4a/11-gatewayd-sessions | 2026-09-23 | done | 2 | fail | Removed the `resume` field from the copied `Pending` struct (written by the skeleton but never read) | Filled `named` (byte scan for `@`, the longest run of ASCII alnum/`.` `-` `_` after it, trailing `.` trimmed, lower-cased, empty runs skipped, then continue past the name) and the `Router` (`route` is a straight line of early returns in spec order: own -> Ignore(Own); kind not empty -> System; user not in allow.users -> NotAllowed; then the channel where "D" is always ours and "O"/"P"/"G" needs an allowed channel id plus `for_us`; the thread root is `root_id` or the post id; `!!...` keeps one `!` and queues, a lone `!` or `!approve`/`!deny` is a command where approve/deny answers M4B_COMMAND and anything else UNKNOWN_COMMAND, a command never reaches loopd; then Queue with session `mm-`, `resume = root_id not empty`, `joins_thread = channel_type != "D"`). `for_us` is true when it names this bot (case-insensitive), otherwise a reply in a known thread that names only channel/here/all. `Queues`: `push` starts a turn when idle (Start with that message alone), waits while running and returns Full(thread) at the limit; `finish` joins the waiting texts with "\n\n" as a resume:true Batch, clears the queue and stays running, removing the session when nothing waits. `Pending.resume` was dead code (the next turn is always a continuation so `finish` hardcodes resume:true) so I removed it rather than allow a lint. First gate failed on clippy `manual_strip`; switched `starts_with("!!")`/`starts_with('!')` plus `&message[2..]`/`&message[1..]` slicing to `strip_prefix`. All 8 sessions tests pass; `make gate` prints `gate: ok` on the second run. | ? | | M4a/10-gatewayd-mm | 2026-09-23 | done | 1 | pass | none | Copied `tests/mm_json.rs`, `tests/mm_rest.rs` and `tests/support/http_server.rs`, added the `src/mm/mod.rs` and `src/mm/rest.rs` skeletons to `crates/gatewayd/src/mm/` and `pub mod mm;` to `lib.rs` (before `secrets`, alphabetical). Filled `mod.rs`: `Post::check` requires `id`/`user_id`/`channel_id` `valid_id` and `root_id` empty-or-`valid_id`, else `Json` quoting the offending id with `{:?}`; `json` is `serde_json::from_slice` mapped to `Json(e.to_string())`; `parse_event` matches `hello`/`posted`/other — `posted` takes `data.post` as a JSON *string* (an object or missing is `Json`), parses it, `check`s it, and reads `data.channel_type` (else ""), any other name (or an empty-name reply) is `Other(name)`; `typing` is `serde_json::json!` compacted; `since_list` walks `order` only (skipping ids not in `posts`, keeping `create_at > since && delete_at == 0` after `check`, deduping, then sorting by `(create_at, id)`), `full` when `order.len() >= SINCE_LIMIT`. Filled `rest.rs`: `Client::new` stores the three fields; `once` connects within `timeout`, sets the read timeout, sends `Authorization: Bearer ` (the only `expose`), `Accept`/`Content-Type` headers and `host_header`, mapping every error to `Net(" : ")`; `call` loops `once` — 2xx returns the body, 401/403 `Auth`, 429 waits `rate_limit_wait` up to `RETRIES` then `RateLimited`, 5xx retried up to `RETRIES` times sleeping `RETRY_5XX`, else `Status` with the first `BODY_KEPT` lossy-UTF-8 chars via a `status_error` helper; `me`/`create_post`/`posts_since`/`direct_channel` build the four calls, `posts_since` and `me`/`direct_channel` reject non-`valid_id` ids as `Json` before sending. All 7 `mm_json` and 10 `mm_rest` tests pass (the latter ~3 s on two deliberate rate-limit waits); `make gate` prints `gate: ok` first run. | ? |