From fae12c3fe23dcdb5b25c382c71b2c0a405e66a74 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 28 Sep 2026 10:51:17 -0700 Subject: [PATCH] gatewayd: a turn's speaker, the username of the one person who wrote its burst Co-Authored-By: ornith-1.5-35b-a3b --- crates/gatewayd/src/deliver.rs | 27 ++-- crates/gatewayd/src/serve/handle.rs | 33 +++- crates/gatewayd/src/serve/mod.rs | 3 + crates/gatewayd/src/sessions.rs | 26 ++++ crates/gatewayd/tests/deliver.rs | 3 + crates/gatewayd/tests/deliver_links.rs | 2 + crates/gatewayd/tests/deliver_notice.rs | 10 +- crates/gatewayd/tests/deliver_speaker.rs | 121 +++++++++++++++ crates/gatewayd/tests/serve_speaker.rs | 188 +++++++++++++++++++++++ crates/gatewayd/tests/sessions.rs | 95 +++++++++++- crates/gatewayd/tests/support/fake_mm.rs | 9 ++ docs/implementer-log.md | 1 + 12 files changed, 503 insertions(+), 15 deletions(-) create mode 100644 crates/gatewayd/tests/deliver_speaker.rs create mode 100644 crates/gatewayd/tests/serve_speaker.rs diff --git a/crates/gatewayd/src/deliver.rs b/crates/gatewayd/src/deliver.rs index 3062ad6..59943af 100644 --- a/crates/gatewayd/src/deliver.rs +++ b/crates/gatewayd/src/deliver.rs @@ -151,6 +151,7 @@ fn one_turn( socket: &Path, batch: &Batch, resume: bool, + speaker: Option<&str>, on_event: &mut dyn FnMut(&TurnEvent), ) -> Outcome { // 1. Connect to the socket (else LoopDown("cannot connect to : ")). @@ -167,8 +168,8 @@ fn one_turn( session: batch.session.clone(), content: batch.text.clone(), resume, - // Set in M5c-1 task 03. Until then no turn has a speaker, and none is synced. - speaker: None, + // On the turn only, never in its text: `memoryd` reads it from the log, the model never. + speaker: speaker.map(str::to_string), }; if let Err(e) = write_frame( &mut stream, @@ -207,14 +208,20 @@ fn one_turn( } } -/// 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 { +/// A turn for a batch, spoken by `speaker` (a Mattermost username) when one person wrote all of +/// it. 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, + speaker: Option<&str>, + 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) { + match one_turn(socket, batch, batch.resume, speaker, on_event) { Outcome::Refused(err) if err.code == ErrorCode::NoSuchSession && batch.resume => { - one_turn(socket, batch, false, on_event) + one_turn(socket, batch, false, speaker, on_event) } outcome => outcome, } @@ -230,11 +237,13 @@ fn post(poster: &dyn Poster, thread: &Thread, text: String, log: &dyn Fn(&str)) } } -/// Run a batch's turn and post what comes of it in its thread. A post that fails is logged. +/// Run a batch's turn, spoken by `speaker`, 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, + speaker: Option<&str>, log: &dyn Fn(&str), notice: &Notice, ) { @@ -244,7 +253,7 @@ pub fn deliver( // 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| { + let outcome = run_turn(socket, batch, speaker, &mut |event| { if let TurnEvent::ApprovalPending { approval, .. } = event && !notice.arrives(*approval) { diff --git a/crates/gatewayd/src/serve/handle.rs b/crates/gatewayd/src/serve/handle.rs index f0eafff..8355d0c 100644 --- a/crates/gatewayd/src/serve/handle.rs +++ b/crates/gatewayd/src/serve/handle.rs @@ -158,13 +158,21 @@ impl Gateway { self.done_tx.clone(), ); let (session, thread) = (batch.session.clone(), batch.thread.clone()); + let speaker = self.speaker(&batch); let spawned = std::thread::Builder::new().spawn(move || { let posted = |id: u64| registry.was_posted(id); let notice = Notice { posted: &posted, wait, }; - deliver(client.as_ref(), &socket, &batch, &|line| log(line), ¬ice); + deliver( + client.as_ref(), + &socket, + &batch, + speaker.as_deref(), + &|line| log(line), + ¬ice, + ); let _ = done.send(batch.session); }); if let Err(e) = spawned { @@ -178,6 +186,29 @@ impl Gateway { Ok(()) } + /// The turn's speaker: the username of the one person who wrote all of the batch, as + /// Mattermost gives it (M5c spec, section 11). `None` when two or more people did, or when + /// the name cannot be found: never the user id, never a guess. A name found is kept; a + /// failure is logged and not kept, so the next turn looks again. + fn speaker(&mut self, batch: &Batch) -> Option { + let user = batch.sender()?; + if let Some(name) = self.usernames.get(user) { + return Some(name.clone()); + } + match self.client.username(user) { + Ok(name) => { + self.usernames.insert(user.to_string(), name.clone()); + Some(name) + } + Err(e) => { + (self.log)(&format!( + "gatewayd: no username for {user}: {e}; the turn has no speaker" + )); + None + } + } + } + /// Turns that ended: out of flight, who asked forgotten, and the next batch of each session /// started. Never blocks. pub(super) fn finished(&mut self) -> Result<(), Stop> { diff --git a/crates/gatewayd/src/serve/mod.rs b/crates/gatewayd/src/serve/mod.rs index 4c18c7f..50e3ed2 100644 --- a/crates/gatewayd/src/serve/mod.rs +++ b/crates/gatewayd/src/serve/mod.rs @@ -121,6 +121,8 @@ pub(crate) struct Gateway { askers: HashMap, /// How long a turn waits for its approval to come through `gateway.sock` before the notice. notice: Duration, + /// Mattermost usernames by user id, each looked up once (M5c spec, section 11). + usernames: HashMap, } /// The wait before attempt `n` (from 0) after a loss. @@ -271,6 +273,7 @@ impl Gateway { authors: HashMap::new(), askers: HashMap::new(), notice, + usernames: HashMap::new(), } } diff --git a/crates/gatewayd/src/sessions.rs b/crates/gatewayd/src/sessions.rs index 4e18490..88131fd 100644 --- a/crates/gatewayd/src/sessions.rs +++ b/crates/gatewayd/src/sessions.rs @@ -39,6 +39,8 @@ pub struct Message { pub text: String, /// A thread in a channel or group message that this Boxmaker now takes part in. pub joins_thread: bool, + /// Who wrote it: the post's `user_id`, a Mattermost user id. + pub user: String, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -245,6 +247,7 @@ impl Router { resume: !post.root_id.is_empty(), text: format!("!{rest}"), joins_thread: channel_type != "D", + user: post.user_id.clone(), }); } if let Some(command) = approval_command(message) { @@ -269,6 +272,7 @@ impl Router { resume: !post.root_id.is_empty(), text: message.to_string(), joins_thread: channel_type != "D", + user: post.user_id.clone(), }) } } @@ -280,6 +284,22 @@ pub struct Batch { pub thread: Thread, pub resume: bool, pub text: String, + /// Who wrote each message of `text`, in order: Mattermost user ids. + pub senders: Vec, +} + +impl Batch { + /// The one user who wrote every message of the batch. `None` when two or more people did, or + /// when it names nobody: a burst joined from several people has no speaker (M5c spec, + /// section 4). + pub fn sender(&self) -> Option<&str> { + let first = self.senders.first()?; + if self.senders.iter().all(|user| user == first) { + Some(first.as_str()) + } else { + None + } + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -295,6 +315,8 @@ pub enum Pushed { struct Pending { thread: Thread, waiting: Vec, + /// The author of each message in `waiting`, in the same order. + senders: Vec, } /// The sessions with a turn running, and the messages waiting for each. @@ -323,6 +345,7 @@ impl Queues { return Pushed::Full(pending.thread.clone()); } pending.waiting.push(message.text.clone()); + pending.senders.push(message.user.clone()); return Pushed::Waiting; } let batch = Batch { @@ -330,12 +353,14 @@ impl Queues { thread: message.thread.clone(), resume: message.resume, text: message.text.clone(), + senders: vec![message.user.clone()], }; self.running.insert( message.session, Pending { thread: message.thread.clone(), waiting: Vec::new(), + senders: Vec::new(), }, ); Pushed::Start(batch) @@ -354,6 +379,7 @@ impl Queues { thread: pending.thread.clone(), resume: true, text: pending.waiting.join("\n\n"), + senders: std::mem::take(&mut pending.senders), }; pending.waiting.clear(); self.running.insert(session.clone(), pending); diff --git a/crates/gatewayd/tests/deliver.rs b/crates/gatewayd/tests/deliver.rs index 7d63146..b4baf9a 100644 --- a/crates/gatewayd/tests/deliver.rs +++ b/crates/gatewayd/tests/deliver.rs @@ -19,6 +19,7 @@ use tmp::TempDir; const DM: &str = "d0000000000000000000000000"; const ROOT: &str = "r0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; #[derive(Default)] struct Record { @@ -59,6 +60,7 @@ fn batch(resume: bool, text: &str) -> Batch { }, resume, text: text.to_string(), + senders: vec![KYLE.to_string()], } } @@ -73,6 +75,7 @@ fn run(dir: &TempDir, poster: &Record, batch: &Batch) -> Vec { poster, &dir.path().join("loop.sock"), batch, + None, &|line| log.lock().unwrap().push(line.to_string()), ¬ice, ); diff --git a/crates/gatewayd/tests/deliver_links.rs b/crates/gatewayd/tests/deliver_links.rs index 81690eb..09bcfc0 100644 --- a/crates/gatewayd/tests/deliver_links.rs +++ b/crates/gatewayd/tests/deliver_links.rs @@ -68,6 +68,7 @@ fn posts(tag: &str, reply: fn(&str) -> Reply, content: &'static str) -> Vec Reply, content: &'static str) -> Vec bool, wait: Duration) -> (Vec }, resume: false, text: "go".to_string(), + senders: vec!["k0000000000000000000000000".to_string()], }; let poster = Record::default(); let started = Instant::now(); - deliver(&poster, &socket, &batch, &|_| {}, &Notice { posted, wait }); + deliver( + &poster, + &socket, + &batch, + None, + &|_| {}, + &Notice { posted, wait }, + ); let took = started.elapsed(); (poster.0.into_inner().unwrap(), took) } diff --git a/crates/gatewayd/tests/deliver_speaker.rs b/crates/gatewayd/tests/deliver_speaker.rs new file mode 100644 index 0000000..9b4ff2c --- /dev/null +++ b/crates/gatewayd/tests/deliver_speaker.rs @@ -0,0 +1,121 @@ +//! The speaker `deliver` is given goes on the turn it sends to `loopd`, and nowhere else: not into +//! the turn's text (M5c spec, section 11). 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 bxcore::{ErrorCode, SessionId, Turn}; +use fake_loop::{done, error, serve_loop}; +use gatewayd::deliver::{Notice, Poster, deliver}; +use gatewayd::mm::MmError; +use gatewayd::sessions::{Batch, Thread}; +use tmp::TempDir; + +const DM: &str = "d0000000000000000000000000"; +const ROOT: &str = "r0000000000000000000000000"; +const KYLE: &str = "k0000000000000000000000000"; +const WAIT: Duration = Duration::from_secs(5); + +#[derive(Default)] +struct Record(Mutex>); + +impl Poster for Record { + fn post(&self, _: &str, _: &str, text: &str) -> Result<(), MmError> { + self.0.lock().unwrap().push(text.to_string()); + Ok(()) + } +} + +/// One `deliver` of "one\n\ntwo" with `speaker`, against a fake `loopd` whose n-th turn gets +/// `script(n)`; the turns it received and what was posted. +fn deliver_with( + tag: &str, + speaker: Option<&str>, + script: fn(usize) -> Vec, +) -> (Vec, Vec) { + let dir = TempDir::new(tag); + let socket = dir.path().join("loop.sock"); + let turns = serve_loop(&socket, move |n, _| script(n)); + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: true, + text: "one\n\ntwo".to_string(), + senders: vec![KYLE.to_string(), KYLE.to_string()], + }; + let never = |_: u64| false; + let notice = Notice { + posted: &never, + wait: Duration::ZERO, + }; + let poster = Record::default(); + let log = Mutex::new(Vec::new()); + deliver( + &poster, + &socket, + &batch, + speaker, + &|line| log.lock().unwrap().push(line.to_string()), + ¬ice, + ); + assert!(log.lock().unwrap().is_empty(), "{:?}", log.lock().unwrap()); + // The fake reports each turn before it answers, so every turn is in by now. + let mut got = vec![turns.recv_timeout(WAIT).unwrap()]; + while let Ok(turn) = turns.try_recv() { + got.push(turn); + } + (got, poster.0.into_inner().unwrap()) +} + +fn answer(_: usize) -> Vec { + vec![done("fine")] +} + +#[test] +fn the_speaker_goes_on_the_turn_and_not_into_its_text() { + let (turns, posts) = deliver_with("speaker-some", Some("first.last"), answer); + assert_eq!(turns.len(), 1); + let turn = &turns[0]; + assert_eq!(turn.speaker.as_deref(), Some("first.last")); + assert_eq!( + (turn.content.as_str(), turn.resume), + ("one\n\ntwo", true), + "the text is the batch's, unchanged" + ); + assert_eq!(posts, ["fine"]); +} + +#[test] +fn without_a_speaker_the_turn_has_none() { + let (turns, posts) = deliver_with("speaker-none", None, answer); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].speaker, None, "never the user id"); + assert_eq!(turns[0].content, "one\n\ntwo"); + assert_eq!(posts, ["fine"]); +} + +#[test] +fn the_turn_sent_again_for_an_unknown_session_keeps_the_speaker() { + let unknown_first = |n: usize| { + if n == 0 { + vec![error(ErrorCode::NoSuchSession, "no such session")] + } else { + vec![done("hello")] + } + }; + let (turns, posts) = deliver_with("speaker-again", Some("kyle"), unknown_first); + let seen: Vec<(bool, Option<&str>)> = turns + .iter() + .map(|t| (t.resume, t.speaker.as_deref())) + .collect(); + assert_eq!(seen, [(true, Some("kyle")), (false, Some("kyle"))]); + assert_eq!(posts, ["hello"]); +} diff --git a/crates/gatewayd/tests/serve_speaker.rs b/crates/gatewayd/tests/serve_speaker.rs new file mode 100644 index 0000000..e1dfce9 --- /dev/null +++ b/crates/gatewayd/tests/serve_speaker.rs @@ -0,0 +1,188 @@ +//! `gatewayd` end to end: each turn's speaker is the Mattermost username of the one person who +//! wrote every message of its burst, and there is none when two people did or the name cannot be +//! found (M5c spec, sections 4 and 11). Do not edit. + +#[path = "support/fake_loop.rs"] +mod fake_loop; +#[path = "support/fake_mm.rs"] +mod fake_mm; +#[path = "support/gateway.rs"] +mod gateway; +#[path = "support/tmp.rs"] +mod tmp; + +use std::sync::{Mutex, mpsc}; +use std::time::{Duration, Instant}; + +use fake_loop::{done, serve_loop}; +use fake_mm::{DM, EVE, FakeMm, KYLE, id, post}; +use gateway::{SHARED, WAIT, loop_dir, read_state, start, up}; +use gatewayd::config::Config; +use tmp::TempDir; + +/// Kyle and Eve both allowed, in the shared channel. +fn two_people(home: &TempDir, url: &str) -> Config { + let text = format!( + r#" +[mattermost] +url = "{url}" +[secrets.mattermost_token] +env = "NOT_READ_BY_RUN" +[allow] +users = ["{KYLE}", "{EVE}"] +channels = ["{SHARED}"] +[paths] +home = "{}" +[limits] +typing_every_ms = 100 +"#, + home.path().display() + ); + Config::parse(&text).unwrap() +} + +/// Wait until `gatewayd` has handled `post`: its id is in the state file's `recent`. The post is +/// queued by then, since the event loop finishes a post before it looks at ended turns. +fn wait_handled(home: &TempDir, post: &str) { + let until = Instant::now() + WAIT; + loop { + let state = read_state(home); + if state["recent"] + .as_array() + .is_some_and(|r| r.iter().any(|p| p == post)) + { + return; + } + assert!(Instant::now() < until, "{post} not handled: {state}"); + std::thread::sleep(Duration::from_millis(10)); + } +} + +fn lookups(mm: &FakeMm, user: &str) -> usize { + let path = format!("/api/v4/users/{user}"); + mm.calls() + .iter() + .filter(|(m, p)| m == "GET" && *p == path) + .count() +} + +#[test] +fn one_persons_message_has_their_username() { + let home = TempDir::new("speaker-one"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "hello there", 5), "D"); + let turn = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (turn.content.as_str(), turn.speaker.as_deref()), + ("hello there", Some("kyle")) + ); + assert_eq!(mm.wait_posts(1, WAIT).len(), 1); +} + +#[test] +fn a_username_is_looked_up_once() { + let home = TempDir::new("speaker-cache"); + let (mm, _running, turns, mut ws) = up(&home, Duration::ZERO); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "first", 5), "D"); + ws.posted(&post(&id('p', 2), KYLE, DM, "", "second", 6), "D"); + let speakers = [ + turns.recv_timeout(WAIT).unwrap().speaker, + turns.recv_timeout(WAIT).unwrap().speaker, + ]; + assert_eq!( + speakers, + [Some("kyle".to_string()), Some("kyle".to_string())] + ); + assert_eq!(lookups(&mm, KYLE), 1, "{:?}", mm.calls()); +} + +#[test] +fn bursts_have_a_speaker_only_when_one_person_wrote_them() { + let home = TempDir::new("speaker-burst"); + loop_dir(&home); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let release = Mutex::new(release_rx); + let turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |n, turn| { + if n <= 1 { + let _ = release.lock().unwrap().recv_timeout(WAIT); + } + vec![done(&format!("answer to {}", turn.content))] + }); + let mm = FakeMm::start(); + let _running = start(two_people(&home, &mm.url())); + let mut ws = mm.next_ws(WAIT); + let p1 = id('p', 1); + ws.posted( + &post(&p1, KYLE, SHARED, "", "@boxmaker-straylight start", 5), + "O", + ); + let first = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (first.content.as_str(), first.speaker.as_deref()), + ("start", Some("kyle")) + ); + + // Eve and Kyle while the first turn runs: one turn, from two people. + ws.posted(&post(&id('p', 2), EVE, SHARED, &p1, "two", 6), "O"); + ws.posted(&post(&id('p', 3), KYLE, SHARED, &p1, "three", 7), "O"); + wait_handled(&home, &id('p', 3)); + release_tx.send(()).unwrap(); + let second = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (second.content.as_str(), second.resume), + ("two\n\nthree", true), + "the text is joined as before" + ); + assert_eq!(second.speaker, None, "two people: no speaker"); + + // Eve alone while the second runs: her name, and nothing left over from the burst before. + ws.posted(&post(&id('p', 4), EVE, SHARED, &p1, "four", 8), "O"); + wait_handled(&home, &id('p', 4)); + release_tx.send(()).unwrap(); + let third = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (third.content.as_str(), third.speaker.as_deref()), + ("four", Some("user-e000")) + ); + assert_eq!(mm.wait_posts(3, WAIT).len(), 3); +} + +#[test] +fn a_name_that_cannot_be_found_leaves_the_turn_without_a_speaker() { + let home = TempDir::new("speaker-missing"); + let (mm, running, turns, mut ws) = up(&home, Duration::ZERO); + mm.refuse_users(true); + ws.posted(&post(&id('p', 1), KYLE, DM, "", "hello", 5), "D"); + let turn = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (turn.content.as_str(), turn.speaker), + ("hello", None), + "never the user id, never a guess" + ); + assert_eq!( + mm.wait_posts(1, WAIT).len(), + 1, + "the turn is answered all the same" + ); + let line = format!( + "gatewayd: no username for {KYLE}: status 404: \"{{\\\"message\\\":\\\"not found\\\"}}\"; \ + the turn has no speaker" + ); + let log = running.wait_log("no username for"); + assert_eq!( + log.iter() + .filter(|l| l.contains("no username")) + .collect::>(), + [&line] + ); + + // A failure is not remembered: the next turn looks again. + mm.refuse_users(false); + ws.posted(&post(&id('p', 2), KYLE, DM, "", "again", 6), "D"); + let again = turns.recv_timeout(WAIT).unwrap(); + assert_eq!( + (again.content.as_str(), again.speaker.as_deref()), + ("again", Some("kyle")) + ); + assert_eq!(lookups(&mm, KYLE), 2, "{:?}", mm.calls()); +} diff --git a/crates/gatewayd/tests/sessions.rs b/crates/gatewayd/tests/sessions.rs index 8f0f3b9..3868b7b 100644 --- a/crates/gatewayd/tests/sessions.rs +++ b/crates/gatewayd/tests/sessions.rs @@ -1,5 +1,6 @@ //! Which posts become turns, in which session, and how messages wait for a running turn (M4a spec, -//! section 7, including its table of examples). Do not edit. +//! section 7, including its table of examples), with the author of each (M5c spec, section 11). +//! Do not edit. use bxcore::SessionId; use gatewayd::mm::Post; @@ -179,6 +180,7 @@ fn sessions_and_threads() { ); assert!(!top.resume && !top.joins_thread); assert_eq!(top.text, "hello"); + assert_eq!(top.user, KYLE); let reply = queued(r.route(&post(DM, ROOT, "more"), "D", &unknown)); assert_eq!(reply.session.as_str(), format!("mm-{ROOT}")); assert_eq!(reply.thread.root, ROOT); @@ -244,6 +246,10 @@ fn commands() { } fn message(root: &str, resume: bool, text: &str) -> Message { + message_by(KYLE, root, resume, text) +} + +fn message_by(user: &str, root: &str, resume: bool, text: &str) -> Message { Message { session: SessionId::new(&format!("mm-{root}")).unwrap(), thread: Thread { @@ -253,9 +259,14 @@ fn message(root: &str, resume: bool, text: &str) -> Message { resume, text: text.to_string(), joins_thread: false, + user: user.to_string(), } } +fn senders(batch: &Batch) -> Vec<&str> { + batch.senders.iter().map(String::as_str).collect() +} + #[test] fn messages_wait_for_the_running_turn_and_go_together() { let mut q = Queues::new(3); @@ -269,7 +280,8 @@ fn messages_wait_for_the_running_turn_and_go_together() { session: first.session.clone(), thread: first.thread.clone(), resume: false, - text: "one".to_string() + text: "one".to_string(), + senders: vec![KYLE.to_string()], } ); assert_eq!(q.push(message(ROOT, true, "two")), Pushed::Waiting); @@ -305,10 +317,85 @@ fn a_full_queue_drops_the_message() { assert_eq!(q.push(message(ROOT, true, "a")), Pushed::Waiting); assert_eq!(q.push(message(ROOT, true, "b")), Pushed::Waiting); assert_eq!( - q.push(message(ROOT, true, "c")), + q.push(message_by(EVE, ROOT, true, "c")), Pushed::Full(m.thread.clone()) ); - assert_eq!(q.finish(&m.session).unwrap().text, "a\n\nb"); + let next = q.finish(&m.session).unwrap(); + assert_eq!(next.text, "a\n\nb"); + assert_eq!( + senders(&next), + [KYLE, KYLE], + "the dropped message's author is not kept" + ); + assert_eq!(next.sender(), Some(KYLE)); assert!(!BUSY.is_empty()); assert_eq!(q.finish(&SessionId::new("mm-never").unwrap()), None); } + +#[test] +fn a_queued_message_names_its_author() { + let r = Router::new( + BOT, + "boxmaker-straylight", + &[KYLE.to_string(), EVE.to_string()], + &[SHARED.to_string()], + ); + for (text, expected) in [ + ("hello", "hello"), + ("!!approve is a word", "!approve is a word"), + ] { + let mut from_eve = post(DM, ROOT, text); + from_eve.user_id = EVE.to_string(); + let m = queued(r.route(&from_eve, "D", &unknown)); + assert_eq!( + (m.user.as_str(), m.text.as_str()), + (EVE, expected), + "{text}" + ); + } +} + +#[test] +fn a_batch_carries_the_author_of_each_message() { + let mut q = Queues::new(5); + let Pushed::Start(first) = q.push(message(ROOT, false, "one")) else { + panic!("not started") + }; + assert_eq!((senders(&first), first.sender()), (vec![KYLE], Some(KYLE))); + assert_eq!(q.push(message_by(EVE, ROOT, true, "two")), Pushed::Waiting); + assert_eq!(q.push(message(ROOT, true, "three")), Pushed::Waiting); + let session = SessionId::new(&format!("mm-{ROOT}")).unwrap(); + let mixed = q.finish(&session).unwrap(); + assert_eq!(mixed.text, "two\n\nthree", "the text is joined as before"); + assert_eq!(senders(&mixed), [EVE, KYLE]); + assert_eq!(mixed.sender(), None, "two people: no one sender"); + assert_eq!(q.push(message_by(EVE, ROOT, true, "four")), Pushed::Waiting); + assert_eq!(q.push(message_by(EVE, ROOT, true, "five")), Pushed::Waiting); + let one = q.finish(&session).unwrap(); + assert_eq!( + (senders(&one), one.sender()), + (vec![EVE, EVE], Some(EVE)), + "each batch has only its own messages' authors" + ); + assert_eq!(q.finish(&session), None); +} + +#[test] +fn a_batch_with_no_authors_has_no_sender() { + let batch = Batch { + session: SessionId::new(&format!("mm-{ROOT}")).unwrap(), + thread: Thread { + channel: DM.to_string(), + root: ROOT.to_string(), + }, + resume: false, + text: "orphan".to_string(), + senders: Vec::new(), + }; + assert_eq!(batch.sender(), None); + let three = Batch { + senders: vec![KYLE.to_string(), KYLE.to_string(), EVE.to_string()], + ..batch + }; + assert_eq!(three.sender(), None, "the last one differs"); +} diff --git a/crates/gatewayd/tests/support/fake_mm.rs b/crates/gatewayd/tests/support/fake_mm.rs index 879ea90..bf05d97 100644 --- a/crates/gatewayd/tests/support/fake_mm.rs +++ b/crates/gatewayd/tests/support/fake_mm.rs @@ -34,6 +34,8 @@ struct Inner { edits: Vec<(String, String)>, /// When set, `POST /posts` is refused with 400. refuse_posts: bool, + /// When set, `GET /users/` (not `users/me`) answers 404. + refuse_users: bool, } pub struct FakeMm { @@ -249,6 +251,10 @@ impl FakeMm { pub fn refuse_posts(&self, refuse: bool) { self.inner.lock().unwrap().refuse_posts = refuse; } + + pub fn refuse_users(&self, refuse: bool) { + self.inner.lock().unwrap().refuse_users = refuse; + } } fn connection(mut stream: TcpStream, inner: &Mutex, ws_tx: &mpsc::Sender) { @@ -367,6 +373,9 @@ fn rest(inner: &Mutex, method: &str, path: &str, body: &[u8]) -> (u16, Va inner.edits.push((post_id.clone(), message.clone())); (200, post(&post_id, BOT, DM, "", &message, 1)) } + ("GET", p) if p.starts_with("/api/v4/users/") && inner.refuse_users => { + (404, json!({"message": "not found"})) + } ("GET", p) if p.starts_with("/api/v4/users/") => { let user = p.trim_start_matches("/api/v4/users/"); let name = if user == KYLE { diff --git a/docs/implementer-log.md b/docs/implementer-log.md index d2d118a..676ee32 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 | |---|---|---|---|---|---|---|---| +| M5c1/03-gatewayd-speaker | 2026-09-28 | done | 2 | pass | none | Copied the seven given files from `docs/plans/M5c1/files/`: `deliver_speaker.rs` and `serve_speaker.rs` (new) and new versions of `sessions.rs`, `deliver.rs`, `deliver_links.rs`, `deliver_notice.rs` and `support/fake_mm.rs` (it gains `refuse_users`). Step 2 showed exactly the named failures: `deliver`, `deliver_links`, `deliver_notice`, `deliver_speaker` and `sessions` did not compile (E0560 for `Batch.senders` / `Message.user`, E0061 for `deliver` taking 5 args, E0282 on the closure), and `serve_speaker` compiled with its four tests failing (three found `None` where a speaker was expected, the fourth had no `"no username for"` log line). Made the edits exactly as the task gives: S1 `Message` gains `user` (the post's `user_id`); S2/S3 set `user` on both `Route::Queue` paths; S4 `Batch` gains `senders` plus `Batch::sender` (`first()?` then `all(|u| u == first)`, `None` for two or more people or for none); S5 `Pending` gains `senders`; S6 `push` pushes `message.user` to `pending.senders` on the `Waiting` path; S7 the `Start` path is `senders: vec![message.user.clone()]` with the new `Pending`'s `senders: Vec::new()`; S8 `finish` takes `senders` with `std::mem::take`. D1 `one_turn` gains `speaker: Option<&str>`; D2 `speaker.map(str::to_string)` on the `Turn` (its comment rewritten); D3 `run_turn` gains `speaker` threaded through both `one_turn` calls; D4 `deliver` gains `speaker`; D5 `deliver` passes `speaker` to `run_turn`. G1 `Gateway` gains `usernames: HashMap`; G2 `new()` inits it. H1 `start` computes `speaker = self.speaker(&batch)` and passes `speaker.as_deref()` to `deliver`. H2 added `Gateway::speaker` (`batch.sender()?`; a cache hit returns it; else `client.username`, `Ok` inserts and returns, `Err` logs `"gatewayd: no username for {user}: {e}; the turn has no speaker"` and returns `None`). `approvals.rs` and `mm/rest.rs` untouched, the text still joins as before and the speaker never enters it. `cargo check -p gatewayd` clean after sessions; after deliver it fails only in `serve/handle.rs` at the `deliver` call (E0061/E0277) as the task predicts, then clean after serve. `sessions` 11 passed, `deliver_speaker` 3, `serve_speaker` 4, every gatewayd suite passes. Step 7: the grep for `speaker: None` prints nothing, `git diff --stat -- crates/gatewayd/src` lists exactly `deliver.rs`, `serve/handle.rs`, `serve/mod.rs`, `sessions.rs`, the other-crates diff is empty. `make gate` printed `gate: ok` with 1148 tests (1137 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately with `BOXMAKER_BROKERD`; 1138 before this task plus the 10 new: 3 in `sessions.rs`, 3 in `deliver_speaker.rs`, 4 in `serve_speaker.rs`). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5c1/02-loopd-speaker-turn | 2026-09-28 | done | 2 | pass | none | Copied five files from `docs/plans/M5c1/files/`: `speaker_turn.rs` (new) and new versions of `memory.rs`, `turn_broker.rs`, `end_to_end.rs` and `support/turn.rs` (skipped the older `memory.rs-task01`). Step 2 showed exactly the named failures: eight non-compiling test targets (end_to_end, limits, memory, neutralise, neutralise_replay, speaker_turn, turn, turn_broker) with E0061 for `run_turn` and `updated`, E0282 on the closure, E0050 and E0599 for `turns`. Made the edits exactly as the task gives: S1 `Session::turns` counting `User` records in `records` via `matches!` and `u64::try_from(...).unwrap_or(u64::MAX)`, placed before `last_usage`; M1 the trait `updated` gains `turn: u64`, M2 `NoMemory` and M3 `MemoryPort` pass it (the `turn: 0` placeholder and its comment dropped for `turn`); T1 `run_turn` gains `speaker: Option<&str>` after `content` with the doc comment, T2 logs `speaker: speaker.map(str::to_string)` (as it came, not checked, not neutralised, `None` not written); C1 passes `turn.speaker.as_deref()` to `run_turn`, C2 passes `session.turns()` to `updated` (the comment now says the number is this turn's, a failed turn not told but its logged message still counts). `baseline::messages` untouched, the request is the same with and without a speaker. Step 7 all pass: the grep for `turn: 0`/`speaker: None` prints nothing, `git diff --stat -- crates/loopd/src` lists exactly channel.rs/memory.rs/session.rs/turn.rs (baseline.rs absent), the other-crates diff is empty. After the edits `cargo check -p loopd` is clean, `speaker_turn` 8 passed, every loopd suite passes. `make gate` printed `gate: ok` with 1138 tests (1130 before plus the 8 of `speaker_turn.rs`). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5c1/01-wire-speaker-state | 2026-09-28 | done | 1 | pass | none | Changed the types in `bxcore` only, as the task says: `Turn` and `LogRecord::User` gain `speaker: Option` (optional, `serde(default, skip_serializing_if = "Option::is_none")`, so a frame or log line without it reads `None` and `None` is not written); `SessionUpdated` gains `turn: u64` (required, no default), plus the two new messages `SessionState { session }` and `SessionStateReply { class, untrusted }` (both fields required). Copied the 20 given files from `docs/plans/M5c1/files/` (4 new fixtures, 16 replaced test files, the loopd `memory.rs` and bxctl `chat.rs` being the `-task01` versions). Step 2 showed exactly the named failures: `strict` `envelopes_reject_unknown_keys_at_every_depth` on `turn_speaker.json` and `speaker_log_records_reject_unknown_keys_at_every_depth` on `session_speaker.jsonl:2`, and 15 non-compiling test targets with E0559/E0560/E0432/E0599. Made every edit exactly as the task gives (W1-W3, L1, M1, X1 in bxcore; B1-B2 in brokerd; G1, C1, T1, U1, I1 in gatewayd, bxctl, loopd twice, memoryd), the placeholders `speaker: None` / `turn: 0` left in place. After the edits `cargo test -p bxcore` every suite passes (turn_wire 9, memory_wire 5, records 4, strict 6) and `cargo test -p brokerd -p gatewayd -p bxctl -p loopd -p memoryd` every suite passes. Step 5 checks all pass: `grep -c 'serde(default' crates/bxcore/src/wire.rs` prints 2, `git diff --stat -- crates/gatewayd/tests crates/inferproxy crates/toolkit` prints nothing, `git diff --stat -- crates/*/src` lists exactly the ten source files. `make gate` printed `gate: ok` with 1130 tests (1129 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately with `BOXMAKER_BROKERD` set; the first gate run's `end_to_end` was the only flake and it passes once the env var is set). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b | | M5b/14-runbook | 2026-09-27 | done | 1 | pass | none | Only `docs/runbook.md` changed. R1 in `memory-write-refused` under **Fix.**: the two-line sentence about narrowing the rule replaced with the same sentence pointing at [`sanitizer-rules`](#sanitizer-rules) instead of "the M5b spec's section 6.5". R2 in `memory-write-flagged` under **Fix.**: the two-line "narrow it as the M5b spec's section 6.5 says" replaced with the same sentence pointing at [`sanitizer-rules`](#sanitizer-rules). Verified before appending: `grep -c "section 6.5"` printed 0. Appended `docs/plans/M5b/files/runbook-sanitizer-rules.md` verbatim with `{ echo; cat ... } >> docs/runbook.md`; `grep -c '^## sanitizer-rules$'` printed 1 and `grep -c '(#sanitizer-rules)'` printed 2. `cargo fmt --all`; `make gate` printed `gate: ok` with 1118 tests (1117 in `cargo test --workspace`, plus the ignored `end_to_end` the gate runs). No code, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |