boxmaker/docs/plans/M5c1/02-loopd-speaker-turn.md

M5c-1 task 02: loopd logs a turn's speaker and tells memoryd the turn's number

Branch: m5c1 (run git switch m5c1; git status --short must be empty, otherwise stop) Commit subject: loopd: a turn's speaker in the log, its number in SessionUpdated Log row: the Task cell is M5c1/02-loopd-speaker-turn.

Goal

Task 01 added two fields to the messages and left placeholders for them in loopd: the User record of every turn is logged with speaker: None, and SessionUpdated, which loopd sends to memoryd after each finished turn, carries turn: 0. This task puts the real values there (M5c spec, sections 5 and 11). memoryd will use both, from task 09, to send finished turns to Honcho.

1. The speaker. loopd copies the Turn's speaker onto the User record it logs for that turn, as it came: no check on the name, and not neutralised. It is not text for the model: the speaker never goes into a request. The request's user message stays exactly as today, the turn's text. A given test compares the requests of a session with a speaker and a session without one, byte for byte, including the second turn, which replays the first from the log.
2. The turn's number. SessionUpdated.turn is the number of the turn that ended: the count of User records in the session's log, the new one included. It is counted from the log, so it is right after a reopen and after loopd restarts.

The session's log, after the turn	SessionUpdated.turn
the first turn of a new session	1
a resumed session with two earlier turns	3
a turn with tool calls (several Assistant records, one User)	one more than before: it is one turn
an earlier turn failed after its message was logged	that turn counts: its User record is in the log
after loopd restarts, the session resumed	the count goes on from the log

What does not change: which turns are told. Today channel::handle calls updated only when the turn finished (outcome.is_ok()); a turn that fails, whether before its User record is logged (no such session, no memory block) or after it (an inference error), sends nothing. That stays as it is. A failed turn that was logged still counts in the numbers of later turns.

Every place that changes

This is the complete list.

Place	What happens there
crates/loopd/src/turn.rs, run_turn	a new parameter speaker: Option<&str>, after content; logged on the User record (edits T1, T2)
crates/loopd/src/memory.rs, MemorySource::updated and its two implementations	a new parameter turn: u64; MemoryPort sends it (edits M1 to M3)
crates/loopd/src/session.rs	new Session::turns (edit S1)
crates/loopd/src/channel.rs, handle	passes the turn's speaker to run_turn, and session.turns() to updated (edits C1, C2)
crates/loopd/src/baseline.rs, messages	no change: it makes the user message from content only
tests that call run_turn or implement MemorySource	new versions are given

Files

• Copy, from docs/plans/M5c1/files/ (the same path under it):
  • new: crates/loopd/tests/speaker_turn.rs
  • new versions of (they replace the files in the tree): crates/loopd/tests/memory.rs, turn_broker.rs, end_to_end.rs and support/turn.rs. Do not copy memory.rs-task01: it is the older version.
• Modify: crates/loopd/src/turn.rs, crates/loopd/src/memory.rs, crates/loopd/src/session.rs, crates/loopd/src/channel.rs, docs/implementer-log.md.
• Do not touch: any other source file (baseline.rs in particular), any file of another crate, any Cargo.toml.

Facts you need

• Why the speaker is not checked or neutralised here. Only memoryd uses it, and it checks the name against Honcho's pattern in a later task (M5c spec, section 3). Neutralising exists for text that reaches the model, and the speaker never does. A given test sends first.last and a name that holds one of the model's tags, and expects both back in the log unchanged.
• The tag in the tests. speaker_turn.rs builds that name from bxcore::sanitize::TAGS, so the file holds no tag, and no assertion prints it. Do not print it yourself (no dbg!, no println! of a speaker or of a request body in any test run).
• Why the request cannot change. loopd replays each session's log to the inference server, and the server reuses its cache only while every request extends the one before byte for byte. baseline::messages turns a User record into a user message from its content alone, and stays so.
• None is not written. Task 01 put #[serde(default, skip_serializing_if = "Option::is_none")] on the record's speaker, so a turn without a speaker is logged with the same bytes as before M5c.
• Where the count comes from. Session::open reads every record of the log into records, and append adds each new one there too, so counting User records in records counts every turn the log holds. Epoch 0 is a session's only log until compaction is built (a later milestone). In channel::handle, session is still in scope after run_turn returns, and the turn's User record is already in it: run_turn appends it before its first request.
• memoryd does not read turn yet (task 09 does). memoryd's tests are unchanged.
• The calls the edits use:

  Call	Signature, and what it gives
  Option::as_deref	turn.speaker.as_deref(): Option<String> to Option<&str>
  Option::map	speaker.map(str::to_string): Option<&str> to Option<String>
  Iterator::filter, count	self.records.iter().filter(f).count() -> usize
  matches!	matches!(record, LogRecord::User { .. }): true for a User record
  u64::try_from	u64::try_from(n).unwrap_or(u64::MAX): a usize as u64, no as cast
  Session::turns	pub fn turns(&self) -> u64 (new, S1)
  MemorySource::updated	fn updated(&self, session: &SessionId, turn: u64) -> Result<(), String> (M1)
  run_turn	pub fn run_turn(session: &mut Session, rt: &Runtime<'_>, content: &str, speaker: Option<&str>, on_event: &mut dyn FnMut(&TurnEvent)) -> Result<TurnOutcome, TurnError> (T1)

The edits, exactly

Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file.

crates/loopd/src/turn.rs

T1. The lines

pub fn run_turn(
    session: &mut Session,
    rt: &Runtime<'_>,
    content: &str,
    on_event: &mut dyn FnMut(&TurnEvent),
) -> Result<TurnOutcome, TurnError> {

become

/// 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> {

T2. In run_turn, the lines

            // Set in M5c-1 task 02, from the turn. `None` is not written: the record's bytes
            // are as before.
            speaker: None,

become

            // 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),

crates/loopd/src/memory.rs

M1. In trait MemorySource, the lines

    /// The session has new records; best effort, a failure is only logged by the caller.
    fn updated(&self, session: &SessionId) -> Result<(), String>;

become

    /// 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>;

M2. In impl MemorySource for NoMemory, the line

    fn updated(&self, _session: &SessionId) -> Result<(), String> {

becomes

    fn updated(&self, _session: &SessionId, _turn: u64) -> Result<(), String> {

M3. In impl MemorySource for MemoryPort, the lines

    fn updated(&self, session: &SessionId) -> 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,
        }))? {

become

    fn updated(&self, session: &SessionId, turn: u64) -> Result<(), String> {
        match self.ask(Message::SessionUpdated(SessionUpdated {
            session: session.clone(),
            turn,
        }))? {

crates/loopd/src/session.rs

S1. The line

    pub fn last_usage(&self) -> Option<bxcore::Usage> {

becomes

    /// 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> {

LogRecord is imported at the top of the file already.

crates/loopd/src/channel.rs

C1. In handle, the lines

    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()),
            },
        );
    });

become

    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()),
                },
            );
        },
    );

(That is how cargo fmt lays the call out once it has five arguments.)

C2. In handle, the lines

    // A finished turn's records are for `memoryd` to index; best effort, since `memoryd` catches
    // up at its next start.
    if outcome.is_ok()
        && let Err(e) = ctx.memory.updated(&turn.session)
    {

become

    // A finished turn's records are for `memoryd` to index; best effort, since `memoryd` catches
    // 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, session.turns())
    {

Steps

• 1. Copy the given files. From the repository root, five commands:

  cp docs/plans/M5c1/files/crates/loopd/tests/speaker_turn.rs crates/loopd/tests/
  cp docs/plans/M5c1/files/crates/loopd/tests/memory.rs crates/loopd/tests/
  cp docs/plans/M5c1/files/crates/loopd/tests/turn_broker.rs crates/loopd/tests/
  cp docs/plans/M5c1/files/crates/loopd/tests/end_to_end.rs crates/loopd/tests/
  cp docs/plans/M5c1/files/crates/loopd/tests/support/turn.rs crates/loopd/tests/support/
  

  Then git status --short lists 4 modified files and 1 new one, all under crates/loopd/tests/.

• 2. See them fail. cargo build --tests --keep-going -p loopd: eight test targets do not compile: end_to_end, limits, memory, neutralise, neutralise_replay, speaker_turn, turn and turn_broker (four of them include support/turn.rs). Every error is about a name this task changes: E0061 (this function takes 4 arguments but 5 arguments were supplied, for run_turn; this method takes 1 argument but 2 arguments were supplied, for updated), E0282 (type annotations needed, on the closure passed to run_turn), E0050 (method `updated` has 3 parameters but the declaration in trait `updated` has 2) or E0599 (no method named `turns` found).

• 3. session.rs. Make edit S1. cargo check -p loopd: no error.

• 4. memory.rs and turn.rs. Make edits M1, M2, M3, T1 and T2. cargo check -p loopd now fails in channel.rs only, with three errors: E0061 twice (this function takes 5 arguments but 4 arguments were supplied, for run_turn; this method takes 2 arguments but 1 argument was supplied, for updated) and E0282 (type annotations needed, on the closure passed to run_turn).

• 5. channel.rs. Make edits C1 and C2. cargo check -p loopd: no error.

• 6. See them pass. cargo test -p loopd --test speaker_turn: 8 passed. Then cargo test -p loopd: every suite passes. The eight:

  Test	What it checks
  the_speaker_is_logged_on_the_user_record_as_it_came	T2 and C1: each User record's speaker, unchanged
  the_requests_are_the_same_with_and_without_a_speaker	the speaker is in no request
  the_turn_number_starts_at_one_in_each_session	C2 and S1: 1, 2, then 1 for another session
  a_turn_with_tool_calls_is_one_turn	only User records are counted
  a_failed_turn_is_not_told_but_is_counted	no updated for a failed turn; the next is 3
  a_restarted_loopd_goes_on_counting_from_the_log	a second loopd on the same home goes on at 3
  session_turns_counts_the_user_records_of_the_log	S1 directly, before and after Session::open
  the_memory_port_sends_the_turn_number	M3: the frame MemoryPort sends holds turn

• 7. Check. Three commands:

  • grep -rn 'turn: 0\|speaker: None' crates/loopd/src prints nothing: both placeholders are gone.
  • git diff --stat -- crates/loopd/src lists exactly channel.rs, memory.rs, session.rs and turn.rs; baseline.rs is not among them.
  • git diff --stat -- crates/bxcore crates/memoryd crates/brokerd crates/gatewayd crates/bxctl prints nothing.

• 8. Run the gate. cargo fmt --all, then make gate, with a shell timeout of 600000 ms. Expected last line: gate: ok, with 1138 tests: 1130 before this task, plus the 8 of speaker_turn.rs (the new versions of the other files keep their counts).

• 9. Log and commit. Add your row to docs/implementer-log.md; its Task cell is M5c1/02-loopd-speaker-turn. git status --short lists the four sources, the four replaced test files and your log as modified, and speaker_turn.rs as new. Stage with git add crates/loopd docs/implementer-log.md Cargo.lock, then commit as AGENTS.md says. Cargo.lock does not change in this task; the git add is harmless.

Done when

• make gate prints gate: ok with 1138 tests.
• git diff HEAD~1 --stat lists only the four source files of "Modify", the five copied files and docs/implementer-log.md.

Stop and report if

• A "before" block of any edit is not in its file exactly as shown, or is there more than once.
• Step 2 shows a different result: other test targets, or an error of another kind.
• Step 4's cargo check fails anywhere other than channel.rs, or with other errors.
• After step 6, a test fails. Do not change a test: stop, and give the failing tests' names and their messages.
• The fix seems to need a change anywhere else: in baseline.rs, in another crate, a check on the speaker's name, or the speaker in the request. It does not; stop and say why you think so.
• A test in inferproxy, toolkit or memoryd fails in the gate: check once whether it fails without your change, then stop and report both results.
• The gate passes with a count other than 1138.
• A comment in an edit does not fit the code under it. Quote it and stop.