gatewayd: a turn's speaker, the username of the one person who wrote its burst

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

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 <path>: <e>")).
@@ -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: <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| {
+    let outcome = run_turn(socket, batch, speaker, &mut |event| {
         if let TurnEvent::ApprovalPending { approval, .. } = event
             && !notice.arrives(*approval)
         {

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), &notice);
+            deliver(
+                client.as_ref(),
+                &socket,
+                &batch,
+                speaker.as_deref(),
+                &|line| log(line),
+                &notice,
+            );
             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<String> {
+        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> {

crates/gatewayd/src/serve/mod.rs

@@ -121,6 +121,8 @@ pub(crate) struct Gateway {
     askers: HashMap<String, approvals::Asker>,
     /// 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<String, String>,
 }
 
 /// 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(),
         }
     }
 

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<String>,
+}
+
+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<String>,
+    /// The author of each message in `waiting`, in the same order.
+    senders: Vec<String>,
 }
 
 /// 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);

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<String> {
         poster,
         &dir.path().join("loop.sock"),
         batch,
+        None,
         &|line| log.lock().unwrap().push(line.to_string()),
         &notice,
     );

crates/gatewayd/tests/deliver_links.rs

@@ -68,6 +68,7 @@ fn posts(tag: &str, reply: fn(&str) -> Reply, content: &'static str) -> Vec<Stri
         },
         resume: false,
         text: "hi".to_string(),
+        senders: vec!["k0000000000000000000000000".to_string()],
     };
     let never = |_: u64| false;
     let notice = Notice {
@@ -80,6 +81,7 @@ fn posts(tag: &str, reply: fn(&str) -> Reply, content: &'static str) -> Vec<Stri
         &record,
         &socket,
         &batch,
+        None,
         &|line| log.lock().unwrap().push(line.to_string()),
         &notice,
     );

crates/gatewayd/tests/deliver_notice.rs

@@ -54,10 +54,18 @@ fn turn(tag: &str, posted: &dyn Fn(u64) -> bool, wait: Duration) -> (Vec<String>
         },
         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)
 }

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<Vec<String>>);
+
+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<fake_loop::Reply>,
+) -> (Vec<Turn>, Vec<String>) {
+    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()),
+        &notice,
+    );
+    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<fake_loop::Reply> {
+    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"]);
+}

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::<Vec<_>>(),
+        [&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());
+}

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");
+}

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/<id>` (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<Inner>, ws_tx: &mpsc::Sender<WsPeer>) {
@@ -367,6 +373,9 @@ fn rest(inner: &Mutex<Inner>, 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 {

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<String,String>`; 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<String>` (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 |