loopd: a turn's speaker in the log, its number in SessionUpdated

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

crates/loopd/src/channel.rs

@@ -178,25 +178,32 @@ pub fn handle(stream: UnixStream, ctx: Arc<Context>) {
         }
     };
 
-    let outcome = run_turn(&mut session, &runtime, &turn.content, &mut |event| {
-        let _ = write_frame(
-            &mut stream,
-            &Envelope {
-                v: PROTOCOL_VERSION,
-                id: request.id,
-                r#final: false,
-                msg: Message::TurnEvent(event.clone()),
-            },
-        );
-    });
+    let outcome = run_turn(
+        &mut session,
+        &runtime,
+        &turn.content,
+        turn.speaker.as_deref(),
+        &mut |event| {
+            let _ = write_frame(
+                &mut stream,
+                &Envelope {
+                    v: PROTOCOL_VERSION,
+                    id: request.id,
+                    r#final: false,
+                    msg: Message::TurnEvent(event.clone()),
+                },
+            );
+        },
+    );
 
     // 5. Release the session before the final frame, so the next turn does not find it busy.
     drop(held);
 
     // A finished turn's records are for `memoryd` to index; best effort, since `memoryd` catches
-    // up at its next start.
+    // up at its next start. The number is this turn's: its `User` record is the last one counted.
+    // A failed turn is not told, as before, though its logged message still counts.
     if outcome.is_ok()
-        && let Err(e) = ctx.memory.updated(&turn.session)
+        && let Err(e) = ctx.memory.updated(&turn.session, session.turns())
     {
         eprintln!(
             "loopd: memoryd was not told about {}: {e}",

crates/loopd/src/memory.rs

@@ -22,8 +22,9 @@ pub const MEMORY_WAIT: Duration = Duration::from_secs(10);
 pub trait MemorySource: Send + Sync {
     /// The block for a new session. `Err` ends in `UNAVAILABLE`: no session starts without it.
     fn block(&self) -> Result<String, String>;
-    /// The session has new records; best effort, a failure is only logged by the caller.
-    fn updated(&self, session: &SessionId) -> Result<(), String>;
+    /// The session's turn number `turn` has ended, and its records are in the log; best effort,
+    /// a failure is only logged by the caller.
+    fn updated(&self, session: &SessionId, turn: u64) -> Result<(), String>;
 }
 
 /// No memory: for tests and the self-test, never for `loopd serve`.
@@ -33,7 +34,7 @@ impl MemorySource for NoMemory {
     fn block(&self) -> Result<String, String> {
         Ok(String::new())
     }
-    fn updated(&self, _session: &SessionId) -> Result<(), String> {
+    fn updated(&self, _session: &SessionId, _turn: u64) -> Result<(), String> {
         Ok(())
     }
 }
@@ -91,11 +92,10 @@ impl MemorySource for MemoryPort {
         }
     }
 
-    fn updated(&self, session: &SessionId) -> Result<(), String> {
+    fn updated(&self, session: &SessionId, turn: u64) -> Result<(), String> {
         match self.ask(Message::SessionUpdated(SessionUpdated {
             session: session.clone(),
-            // Set in M5c-1 task 02. `memoryd` does not read it before M5c-1 task 09.
-            turn: 0,
+            turn,
         }))? {
             Message::Ok(_) => Ok(()),
             Message::Error(e) => Err(format!("memoryd: {}", e.detail)),

crates/loopd/src/session.rs

@@ -208,6 +208,19 @@ impl Session {
         })
     }
 
+    /// How many turns the session has had: its `User` records, one per turn that reached the
+    /// log, a failed turn's too. Counted from `records`, which `open` reads back whole, so the
+    /// count goes on after a reopen or a restart; right after a turn it is that turn's number,
+    /// the first being 1. Epoch 0 is a session's only log until compaction is built.
+    pub fn turns(&self) -> u64 {
+        let users = self
+            .records
+            .iter()
+            .filter(|record| matches!(record, LogRecord::User { .. }))
+            .count();
+        u64::try_from(users).unwrap_or(u64::MAX)
+    }
+
     pub fn last_usage(&self) -> Option<bxcore::Usage> {
         for record in self.records.iter().rev() {
             if let LogRecord::Usage {

crates/loopd/src/turn.rs

@@ -113,19 +113,23 @@ fn map_event(event: &ChatEvent, on_event: &mut dyn FnMut(&TurnEvent)) {
     }
 }
 
+/// One turn of `session`: `content` is the user's message, and `speaker` who spoke it, when one
+/// person did. The speaker is logged on the `User` record, for `memoryd`, and never sent to the
+/// model: the request is the same with a speaker and without one (M5c spec, section 11).
 pub fn run_turn(
     session: &mut Session,
     rt: &Runtime<'_>,
     content: &str,
+    speaker: Option<&str>,
     on_event: &mut dyn FnMut(&TurnEvent),
 ) -> Result<TurnOutcome, TurnError> {
     session
         .append(LogRecord::User {
             time: Timestamp::now(),
             content: neutralise(content).into_owned(),
-            // Set in M5c-1 task 02, from the turn. `None` is not written: the record's bytes
-            // are as before.
-            speaker: None,
+            // As it came: not checked and not neutralised, since it never reaches the model.
+            // `None` is not written, so the record's bytes are as before M5c.
+            speaker: speaker.map(str::to_string),
         })
         .map_err(TurnError::Session)?;
 

crates/loopd/tests/end_to_end.rs

@@ -121,6 +121,7 @@ fn a_call_no_grant_allows_reaches_the_model_as_a_denial_and_is_audited() {
         &mut session,
         &runtime,
         "what is this host called?",
+        None,
         &mut |e| events.push(e.clone()),
     );
     assert!(

crates/loopd/tests/memory.rs

@@ -33,7 +33,7 @@ impl MemorySource for Shared {
     fn block(&self) -> Result<String, String> {
         self.0.block.lock().unwrap().clone()
     }
-    fn updated(&self, session: &SessionId) -> Result<(), String> {
+    fn updated(&self, session: &SessionId, _turn: u64) -> Result<(), String> {
         self.0
             .updated
             .lock()

crates/loopd/tests/speaker_turn.rs

@@ -0,0 +1,309 @@
+//! A turn's speaker and its number (M5c spec, sections 5 and 11). `loopd` logs the speaker on the
+//! turn's `User` record as it came, and never puts it in a request: the model sees the same bytes
+//! with a speaker and without one. After a finished turn it tells `memoryd` the turn's number: the
+//! count of `User` records in the session's log, so the first turn is 1, a failed turn that was
+//! logged counts too, and the count goes on after a reopen or a restart. Do not edit.
+
+mod support;
+#[path = "support/tags.rs"]
+mod tags;
+
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::{Path, PathBuf};
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+
+use bxcore::{
+    Empty, Envelope, LogRecord, Message, PROTOCOL_VERSION, SessionId, SessionUpdated, Timestamp,
+    Turn, read_frame, write_frame,
+};
+use loopd::baseline::Baseline;
+use loopd::channel::{Context, serve};
+use loopd::llama::Client;
+use loopd::memory::{MemoryPort, MemorySource};
+use loopd::session::Session;
+use loopd::tools::Registry;
+use support::{FakeServer, Home, Reply, ScriptedPort};
+use tags::{holds_a_tag, same};
+
+const CHAT: &str = "/v1/chat/completions";
+
+/// A `memoryd` stand-in that records each `(session, turn)` it is told about.
+#[derive(Default)]
+struct Recorder(Mutex<Vec<(String, u64)>>);
+
+impl Recorder {
+    fn told(&self) -> Vec<(String, u64)> {
+        self.0.lock().unwrap().clone()
+    }
+}
+
+/// Lets the test keep its `Arc<Recorder>` while the context owns a `Box<dyn MemorySource>`.
+struct Shared(Arc<Recorder>);
+
+impl MemorySource for Shared {
+    fn block(&self) -> Result<String, String> {
+        Ok(String::new())
+    }
+    fn updated(&self, session: &SessionId, turn: u64) -> Result<(), String> {
+        let entry = (session.as_str().to_string(), turn);
+        self.0.0.lock().unwrap().push(entry);
+        Ok(())
+    }
+}
+
+/// A fake inference server and a home that one `loopd` or several, one after the other, serve
+/// from.
+struct Setup {
+    server: FakeServer,
+    // Declared last so it is dropped last: fields drop in order, and the home must outlive
+    // everything that may still write into it (tip T44).
+    home: Home,
+}
+
+impl Setup {
+    /// The server answers the completions in `chat` in order; the last one repeats.
+    fn new(chat: Vec<Reply>) -> Setup {
+        let home = Home::new();
+        let server = FakeServer::start();
+        server.route(CHAT, chat);
+        Setup { server, home }
+    }
+
+    /// Starts a channel server on the home, as a new `loopd` process would: its own context,
+    /// nothing carried over but the files. `name` names its socket, under the home; the returned
+    /// recorder hears what this `loopd` tells `memoryd`.
+    fn loopd(&self, name: &str) -> (PathBuf, Arc<Recorder>) {
+        let mut cfg = self.home.config(&self.server.socket);
+        cfg.channel.socket = self.home.dir.join(name);
+        let socket = cfg.channel_socket();
+        let listener = UnixListener::bind(&socket).unwrap();
+        let memory = Arc::new(Recorder::default());
+        let ctx = Arc::new(Context::new(
+            cfg.clone(),
+            Client::new(cfg),
+            Box::new(ScriptedPort::new(vec![])),
+            Registry::m2b(),
+            Box::new(Shared(Arc::clone(&memory))),
+        ));
+        thread::spawn(move || serve(listener, ctx));
+        (socket, memory)
+    }
+
+    /// The `User` records of a session's log, in order.
+    fn users(&self, session: &str) -> Vec<LogRecord> {
+        let records = self.home.records(session);
+        let is_user = |r: &LogRecord| matches!(r, LogRecord::User { .. });
+        records.into_iter().filter(is_user).collect()
+    }
+
+    /// The bodies of the completion requests so far, as sent, in order.
+    fn bodies(&self) -> Vec<String> {
+        let requests = self.server.requests_to(CHAT);
+        let body = |r: &support::Recorded| String::from_utf8(r.body.clone()).unwrap();
+        requests.iter().map(body).collect()
+    }
+}
+
+/// Sends one turn and returns true if it ended with `turn_done`, false if with an error.
+fn turn(socket: &Path, session: &str, resume: bool, speaker: Option<&str>) -> bool {
+    let mut stream = UnixStream::connect(socket).unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_secs(10)))
+        .unwrap();
+    let msg = Message::Turn(Turn {
+        session: SessionId::new(session).unwrap(),
+        content: "hello".to_string(),
+        resume,
+        speaker: speaker.map(str::to_string),
+    });
+    let env = Envelope {
+        v: PROTOCOL_VERSION,
+        id: 7,
+        r#final: true,
+        msg,
+    };
+    write_frame(&mut stream, &env).unwrap();
+    loop {
+        let frame = read_frame(&mut stream).unwrap();
+        if !frame.r#final {
+            continue;
+        }
+        match frame.msg {
+            Message::TurnDone(_) => return true,
+            Message::Error(_) => return false,
+            other => panic!("the turn ended with neither turn_done nor error: {other:?}"),
+        }
+    }
+}
+
+fn told(session: &str, turn: u64) -> (String, u64) {
+    (session.to_string(), turn)
+}
+
+/// The speaker of a `User` record, or a panic for any other record.
+fn speaker(record: &LogRecord) -> Option<String> {
+    match record {
+        LogRecord::User { speaker, .. } => speaker.clone(),
+        other => panic!("not a user record: {other:?}"),
+    }
+}
+
+#[test]
+fn the_speaker_is_logged_on_the_user_record_as_it_came() {
+    let s = Setup::new(vec![Reply::fixture("plain")]);
+    let (socket, _) = s.loopd("loop.sock");
+    // Not a valid Honcho peer name, and one that holds one of the model's tags: `loopd` neither
+    // checks nor neutralises a speaker, since it never reaches the model.
+    let tagged = format!("x{}y", bxcore::sanitize::TAGS[0].text);
+    assert!(turn(&socket, "a", false, Some("first.last")));
+    assert!(turn(&socket, "a", true, None));
+    assert!(turn(&socket, "a", true, Some(&tagged)));
+    let users = s.users("a");
+    assert_eq!(users.len(), 3, "one user record per turn");
+    assert_eq!(speaker(&users[0]), Some("first.last".to_string()));
+    assert_eq!(speaker(&users[1]), None, "a turn without a speaker");
+    same(&speaker(&users[2]), &Some(tagged), "the third speaker");
+    let log = s.home.read("sessions/a/0.jsonl");
+    assert_eq!(
+        log.matches("\"speaker\":").count(),
+        2,
+        "no speaker key for the turn without one"
+    );
+}
+
+#[test]
+fn the_requests_are_the_same_with_and_without_a_speaker() {
+    let s = Setup::new(vec![Reply::fixture("plain")]);
+    let (socket, _) = s.loopd("loop.sock");
+    let tagged = format!("x{}y", bxcore::sanitize::TAGS[0].text);
+    assert!(turn(&socket, "with", false, Some("Zelda_Q")));
+    assert!(turn(&socket, "with", true, Some(&tagged)));
+    assert!(turn(&socket, "without", false, None));
+    assert!(turn(&socket, "without", true, None));
+    let bodies = s.bodies();
+    assert_eq!(bodies.len(), 4, "one request per turn");
+    // Each session's second request replays its first turn from the log: a speaker there would
+    // change the replayed bytes too.
+    assert!(bodies[0] == bodies[2], "the first requests differ");
+    assert!(bodies[1] == bodies[3], "the second requests differ");
+    for (n, body) in bodies.iter().enumerate() {
+        assert!(!body.contains("Zelda_Q"), "request {n} holds the speaker");
+        assert!(!holds_a_tag(body), "request {n} holds the tagged speaker");
+    }
+}
+
+#[test]
+fn the_turn_number_starts_at_one_in_each_session() {
+    let s = Setup::new(vec![Reply::fixture("plain")]);
+    let (socket, memory) = s.loopd("loop.sock");
+    assert!(turn(&socket, "a", false, None));
+    assert!(turn(&socket, "a", true, Some("kyle")));
+    assert!(turn(&socket, "b", false, None));
+    assert!(turn(&socket, "a", true, None));
+    assert_eq!(
+        memory.told(),
+        [told("a", 1), told("a", 2), told("b", 1), told("a", 3)]
+    );
+}
+
+#[test]
+fn a_turn_with_tool_calls_is_one_turn() {
+    // The first turn asks for a tool, then answers: two assistant records, one user record.
+    let s = Setup::new(vec![Reply::fixture("tool_call"), Reply::fixture("plain")]);
+    let (socket, memory) = s.loopd("loop.sock");
+    assert!(turn(&socket, "a", false, None));
+    assert!(turn(&socket, "a", true, None));
+    assert_eq!(memory.told(), [told("a", 1), told("a", 2)]);
+}
+
+#[test]
+fn a_failed_turn_is_not_told_but_is_counted() {
+    let s = Setup::new(vec![
+        Reply::fixture("plain"),
+        Reply::fixture("bad_request"),
+        Reply::fixture("plain"),
+    ]);
+    let (socket, memory) = s.loopd("loop.sock");
+    assert!(turn(&socket, "a", false, None));
+    assert!(!turn(&socket, "a", true, None), "the second turn fails");
+    assert!(!turn(&socket, "nobody", true, None), "no such session");
+    assert!(turn(&socket, "a", true, None));
+    assert_eq!(s.users("a").len(), 3, "the failed turn's message is logged");
+    assert_eq!(memory.told(), [told("a", 1), told("a", 3)]);
+}
+
+#[test]
+fn a_restarted_loopd_goes_on_counting_from_the_log() {
+    let s = Setup::new(vec![Reply::fixture("plain")]);
+    let (first, before) = s.loopd("first.sock");
+    assert!(turn(&first, "a", false, None));
+    assert!(turn(&first, "a", true, None));
+    let (second, after) = s.loopd("second.sock");
+    assert!(turn(&second, "a", true, None));
+    assert!(turn(&second, "b", false, None));
+    assert_eq!(before.told(), [told("a", 1), told("a", 2)]);
+    assert_eq!(after.told(), [told("a", 3), told("b", 1)]);
+}
+
+#[test]
+fn session_turns_counts_the_user_records_of_the_log() {
+    let home = Home::new();
+    let cfg = home.config(Path::new("/tmp/unused.sock"));
+    let baseline = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
+    let id = SessionId::new("a").unwrap();
+    let mut session = Session::create(&home.dir, id.clone(), baseline, 0).unwrap();
+    assert_eq!(session.turns(), 0, "a new session has had no turn");
+    let user = |speaker: Option<&str>| LogRecord::User {
+        time: Timestamp::now(),
+        content: "hello".to_string(),
+        speaker: speaker.map(str::to_string),
+    };
+    let assistant = || LogRecord::Assistant {
+        time: Timestamp::now(),
+        content: Some("hi".to_string()),
+        reasoning_content: None,
+        tool_calls: vec![],
+    };
+    session.append(user(None)).unwrap();
+    session.append(assistant()).unwrap();
+    assert_eq!(session.turns(), 1);
+    session.append(user(Some("kyle"))).unwrap();
+    session.append(assistant()).unwrap();
+    session.append(assistant()).unwrap();
+    assert_eq!(session.turns(), 2);
+    drop(session);
+    let session = Session::open(&home.dir, id).unwrap();
+    assert_eq!(session.turns(), 2, "counted again from the log");
+}
+
+#[test]
+fn the_memory_port_sends_the_turn_number() {
+    let home = Home::new();
+    let socket = home.dir.join("memory.sock");
+    let listener = UnixListener::bind(&socket).unwrap();
+    let (tx, rx) = mpsc::channel();
+    thread::spawn(move || {
+        let (mut stream, _) = listener.accept().unwrap();
+        let env = read_frame(&mut stream).unwrap();
+        let reply = Envelope {
+            v: PROTOCOL_VERSION,
+            id: env.id,
+            r#final: true,
+            msg: Message::Ok(Empty {}),
+        };
+        write_frame(&mut stream, &reply).unwrap();
+        let _ = tx.send(env.msg);
+    });
+    let port = MemoryPort::new(socket, Duration::from_secs(5));
+    let id = SessionId::new("a").unwrap();
+    assert_eq!(port.updated(&id, 7), Ok(()));
+    let sent = rx.recv_timeout(Duration::from_secs(10)).unwrap();
+    let want = Message::SessionUpdated(SessionUpdated {
+        session: id,
+        turn: 7,
+    });
+    assert_eq!(sent, want);
+}

crates/loopd/tests/support/turn.rs

@@ -63,7 +63,7 @@ impl Setup {
         text: &str,
     ) -> (Result<loopd::turn::TurnOutcome, TurnError>, Vec<TurnEvent>) {
         let mut events = Vec::new();
-        let result = run_turn(session, &self.runtime(), text, &mut |e| {
+        let result = run_turn(session, &self.runtime(), text, None, &mut |e| {
             events.push(e.clone())
         });
         (result, events)

crates/loopd/tests/turn_broker.rs

@@ -243,8 +243,9 @@ fn a_port_that_returns_a_pending_frame_as_its_answer_is_a_failure_not_a_decision
         registry: &s.registry,
     };
     let mut events = Vec::new();
-    let result =
-        loopd::turn::run_turn(&mut session, &runtime, "x", &mut |e| events.push(e.clone()));
+    let result = loopd::turn::run_turn(&mut session, &runtime, "x", None, &mut |e| {
+        events.push(e.clone())
+    });
     assert!(result.is_ok(), "{result:?}");
     match &session.records()[4] {
         LogRecord::ToolResult {