M5c-1 plan: task 02, loopd logs the speaker and sends the turn's number

loopd copies the Turn's speaker onto the User record it logs, as it
came (not checked, not neutralised), and never into a request; after a
finished turn it sends SessionUpdated with the turn's number, the count
of User records in the session's log (Session::turns), so the first
turn is 1, a failed turn that was logged counts, and the count goes on
after a reopen or a restart. Failed turns are still not told.

Given: speaker_turn.rs (8 tests) and new versions of loopd's memory.rs,
turn_broker.rs, end_to_end.rs and support/turn.rs (run_turn gains
speaker, MemorySource::updated gains turn). Task 01 now copies
memory.rs-task01, its own version.

Checks: the reference passes make gate with 1138 tests (1130 + 8); the
task replayed from its text on a fresh ref-01 worktree with its own
target is byte-identical to the reference and passes the gate; /tmp
unchanged around both gate runs; 15 one-line mutations (speaker not
logged or dropped by the channel or neutralised, speaker in the request
at log time or at replay, turn 0-based or +1, counting Assistant
records, all records or only speakers' turns, a global or per-process
counter, told before logging or on failure, MemoryPort sending 0) all
caught; speaker_turn run 60 times under load, no failure; no
chat-template tag in the plan or given files; task lines within 100
characters but tables.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/plans/M5c1/01-wire-speaker-state.md

@@ -69,7 +69,8 @@ This is the complete list. Each one is covered by an edit below or by a given fi
     `crates/brokerd/tests/broker.rs`; `crates/bxctl/tests/chat.rs`;
     `crates/loopd/tests/baseline.rs`, `channel.rs`, `instance.rs`, `memory.rs`,
     `neutralise_replay.rs`, `session.rs`, `turn_done_untrusted.rs`;
-    `crates/memoryd/tests/serve.rs`, `crates/memoryd/tests/support/logs.rs`
+    `crates/memoryd/tests/serve.rs`, `crates/memoryd/tests/support/logs.rs`. Your `loopd`
+    `memory.rs` is the given `memory.rs-task01`: the given `memory.rs` is task 02's.
   - Do not copy anything under `docs/plans/M5c1/files/crates/memoryd/tests/fixtures/honcho/`:
     it is for later tasks.
 - Modify: `crates/bxcore/src/wire.rs`, `crates/bxcore/src/log.rs`, `crates/bxcore/src/memory.rs`,
@@ -429,7 +430,8 @@ becomes
   cp docs/plans/M5c1/files/crates/loopd/tests/baseline.rs crates/loopd/tests/
   cp docs/plans/M5c1/files/crates/loopd/tests/channel.rs crates/loopd/tests/
   cp docs/plans/M5c1/files/crates/loopd/tests/instance.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/memory.rs-task01 \
+    crates/loopd/tests/memory.rs
   cp docs/plans/M5c1/files/crates/loopd/tests/neutralise_replay.rs crates/loopd/tests/
   cp docs/plans/M5c1/files/crates/loopd/tests/session.rs crates/loopd/tests/
   cp docs/plans/M5c1/files/crates/loopd/tests/turn_done_untrusted.rs crates/loopd/tests/

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

@@ -0,0 +1,361 @@
+# 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
+
+```rust
+pub fn run_turn(
+    session: &mut Session,
+    rt: &Runtime<'_>,
+    content: &str,
+    on_event: &mut dyn FnMut(&TurnEvent),
+) -> Result<TurnOutcome, TurnError> {
+```
+
+become
+
+```rust
+/// 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
+
+```rust
+            // Set in M5c-1 task 02, from the turn. `None` is not written: the record's bytes
+            // are as before.
+            speaker: None,
+```
+
+become
+
+```rust
+            // 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
+
+```rust
+    /// The session has new records; best effort, a failure is only logged by the caller.
+    fn updated(&self, session: &SessionId) -> Result<(), String>;
+```
+
+become
+
+```rust
+    /// 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
+
+```rust
+    fn updated(&self, _session: &SessionId) -> Result<(), String> {
+```
+
+becomes
+
+```rust
+    fn updated(&self, _session: &SessionId, _turn: u64) -> Result<(), String> {
+```
+
+**M3.** In `impl MemorySource for MemoryPort`, the lines
+
+```rust
+    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
+
+```rust
+    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
+
+```rust
+    pub fn last_usage(&self) -> Option<bxcore::Usage> {
+```
+
+becomes
+
+```rust
+    /// 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
+
+```rust
+    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
+
+```rust
+    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
+
+```rust
+    // 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
+
+```rust
+    // 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:
+
+  ```sh
+  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.

docs/plans/M5c1/files/crates/loopd/tests/end_to_end.rs

@@ -0,0 +1,169 @@
+//! End to end in two processes: `loopd`'s turn loop, with a `BrokerPort`, against the real
+//! `brokerd` binary on a temporary home, and the fake llama server. Do not edit.
+//!
+//! It cannot be one process: `loopd` may not depend on `brokerd`, even for tests. The path of
+//! the `brokerd` binary comes from `BOXMAKER_BROKERD`; without it the test is ignored, and
+//! `make gate` builds the workspace and runs it with the variable set.
+
+mod support;
+
+use std::os::unix::net::UnixStream;
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+use bxcore::{
+    AuditEvent, AuditRecord, ChainVerifier, DecisionRecord, DenyReason, SessionId, TurnEvent,
+};
+use loopd::baseline::Baseline;
+use loopd::broker_port::BrokerPort;
+use loopd::llama::Client;
+use loopd::session::Session;
+use loopd::tools::Registry;
+use loopd::turn::{Runtime, run_turn};
+use support::{FakeServer, Home, Reply};
+
+const CHAT: &str = "/v1/chat/completions";
+
+/// `brokerd serve`, killed when dropped.
+struct Brokerd(Child);
+
+impl Drop for Brokerd {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+fn start_brokerd(home: &Path) -> (Brokerd, PathBuf) {
+    let binary = std::env::var_os("BOXMAKER_BROKERD")
+        .expect("set BOXMAKER_BROKERD to the brokerd binary; `make gate` does");
+    std::fs::create_dir_all(home.join("grants")).unwrap();
+    let config = home.join("brokerd.toml");
+    let text = format!(
+        "[paths]\nhome = \"{h}\"\ngrants = \"{h}/grants\"\n",
+        h = home.display()
+    );
+    std::fs::write(&config, text).unwrap();
+    let child = Command::new(binary)
+        .args(["serve", "--config"])
+        .arg(&config)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .unwrap();
+    let brokerd = Brokerd(child);
+    let socket = home.join("run/loop-broker/broker.sock");
+    let until = Instant::now() + Duration::from_secs(10);
+    while UnixStream::connect(&socket).is_err() {
+        assert!(
+            Instant::now() < until,
+            "brokerd never listened on {}",
+            socket.display()
+        );
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    (brokerd, socket)
+}
+
+/// Every record under `audit/`, after checking that the chain verifies.
+fn audit(dir: &Path) -> Vec<AuditRecord> {
+    let mut names: Vec<String> = std::fs::read_dir(dir)
+        .unwrap()
+        .map(|e| e.unwrap().file_name().into_string().unwrap())
+        .filter(|n| n.ends_with(".jsonl"))
+        .collect();
+    names.sort();
+    let mut verifier = ChainVerifier::new();
+    let mut records = Vec::new();
+    for name in &names {
+        let bytes = std::fs::read(dir.join(name)).unwrap();
+        verifier.feed(name, &bytes);
+        for line in String::from_utf8(bytes).unwrap().lines() {
+            records.push(serde_json::from_str(line).unwrap());
+        }
+    }
+    let report = verifier.finish();
+    assert!(report.failure.is_none(), "{:?}", report.failure);
+    assert!(report.torn_tail.is_none());
+    assert_eq!(report.records, records.len() as u64);
+    records
+}
+
+#[test]
+#[ignore = "needs the brokerd binary in BOXMAKER_BROKERD; `make gate` runs it"]
+fn a_call_no_grant_allows_reaches_the_model_as_a_denial_and_is_audited() {
+    let home = Home::new();
+    let broker_home = home.dir.join("broker-home");
+    let (_brokerd, socket) = start_brokerd(&broker_home);
+
+    let server = FakeServer::start();
+    // The recorded model calls `read_file` on /etc/hostname, then answers in plain text.
+    server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let cfg = home.config(&server.socket);
+    let client = Client::new(cfg.clone());
+    let port = BrokerPort::new(socket, Duration::from_secs(10));
+    let registry = Registry::m2b();
+    let baseline = Baseline::assemble(&cfg, &registry).unwrap();
+    let id = SessionId::new("e2e").unwrap();
+    let mut session = Session::create(&home.dir, id, baseline, cfg.slots.main).unwrap();
+    let runtime = Runtime {
+        cfg: &cfg,
+        client: &client,
+        port: &port,
+        registry: &registry,
+    };
+    let mut events = Vec::new();
+    let outcome = run_turn(
+        &mut session,
+        &runtime,
+        "what is this host called?",
+        None,
+        &mut |e| events.push(e.clone()),
+    );
+    assert!(
+        outcome.is_ok(),
+        "the turn goes on after a denial: {outcome:?}"
+    );
+    assert!(
+        events.contains(&TurnEvent::ToolDenied {
+            name: "read_file".to_string(),
+            reason: DenyReason::NoGrant,
+        }),
+        "{events:?}"
+    );
+    // The model reads the denial in its next request.
+    let second = server.requests_to(CHAT)[1].json();
+    assert_eq!(
+        second["messages"][3]["content"],
+        "Denied: no grant allows this call."
+    );
+
+    let records = audit(&broker_home.join("audit"));
+    assert_eq!(records.len(), 1, "{records:?}");
+    match &records[0].event {
+        AuditEvent::Decision {
+            session,
+            tool,
+            arguments,
+            outcome,
+            grant,
+            ..
+        } => {
+            assert_eq!(session.as_str(), "e2e");
+            assert_eq!(tool, "read_file");
+            assert_eq!(arguments, r#"{"path":"/etc/hostname"}"#);
+            assert_eq!(
+                *outcome,
+                DecisionRecord::Denied {
+                    reason: DenyReason::NoGrant
+                }
+            );
+            assert_eq!(*grant, None);
+        }
+        other => panic!("{other:?}"),
+    }
+}

docs/plans/M5c1/files/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()

docs/plans/M5c1/files/crates/loopd/tests/memory.rs-task01

@@ -0,0 +1,187 @@
+//! `loopd` and memory: the memory block in a new session's baseline and not in a running one's,
+//! no session without it, `memoryd` told after a finished turn, and the two core memory tools
+//! (M5a spec, section 9). Do not edit.
+
+mod support;
+
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use bxcore::{
+    Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, read_frame, write_frame,
+};
+use loopd::channel::{Context, serve};
+use loopd::llama::Client;
+use loopd::memory::{MemorySource, UNAVAILABLE};
+use loopd::tools::Registry;
+use support::{FakeServer, Home, Reply, ScriptedPort};
+
+const CHAT: &str = "/v1/chat/completions";
+
+/// A `memoryd` stand-in: the block it serves, and the sessions it was told about.
+struct Recorder {
+    block: Mutex<Result<String, String>>,
+    updated: Mutex<Vec<String>>,
+}
+
+/// 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> {
+        self.0.block.lock().unwrap().clone()
+    }
+    fn updated(&self, session: &SessionId) -> Result<(), String> {
+        self.0
+            .updated
+            .lock()
+            .unwrap()
+            .push(session.as_str().to_string());
+        Ok(())
+    }
+}
+
+struct Loopd {
+    server: FakeServer,
+    socket: PathBuf,
+    memory: Arc<Recorder>,
+    // 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,
+}
+
+fn start(block: Result<&str, &str>) -> Loopd {
+    let home = Home::new();
+    let server = FakeServer::start();
+    let cfg = home.config(&server.socket);
+    let socket = cfg.channel_socket();
+    let listener = UnixListener::bind(&socket).unwrap();
+    let memory = Arc::new(Recorder {
+        block: Mutex::new(block.map(str::to_string).map_err(str::to_string)),
+        updated: Mutex::new(Vec::new()),
+    });
+    let ctx = Arc::new(Context::new(
+        cfg.clone(),
+        Client::new(cfg),
+        Box::new(ScriptedPort::new(vec![])),
+        Registry::m2b(),
+        Box::new(Shared(Arc::clone(&memory))),
+    ));
+    std::thread::spawn(move || serve(listener, ctx));
+    Loopd {
+        _home: home,
+        server,
+        socket,
+        memory,
+    }
+}
+
+/// One turn; the last frame.
+fn turn(l: &Loopd, session: &str, resume: bool) -> Message {
+    let mut stream = UnixStream::connect(&l.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: None,
+    });
+    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 {
+            return frame.msg;
+        }
+    }
+}
+
+fn system_prompt(l: &Loopd, n: usize) -> String {
+    let requests = l.server.requests_to(CHAT);
+    requests[n].json()["messages"][0]["content"]
+        .as_str()
+        .unwrap()
+        .to_string()
+}
+
+#[test]
+fn a_new_session_starts_with_the_block_and_keeps_it() {
+    let l = start(Ok("# soul.md\nYou are kind."));
+    l.server
+        .route(CHAT, vec![Reply::fixture("plain"), Reply::fixture("plain")]);
+    assert!(matches!(turn(&l, "a", false), Message::TurnDone(_)));
+    assert_eq!(
+        system_prompt(&l, 0),
+        "You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind."
+    );
+    *l.memory.block.lock().unwrap() = Ok("# soul.md\nYou changed.".to_string());
+    assert!(matches!(turn(&l, "a", true), Message::TurnDone(_)));
+    assert_eq!(
+        system_prompt(&l, 1),
+        "You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind.",
+        "a running session keeps the bytes it started with"
+    );
+}
+
+#[test]
+fn no_session_starts_without_the_block() {
+    let l = start(Err(
+        "memoryd at /x: refused; see docs/runbook.md#memory-unavailable",
+    ));
+    match turn(&l, "b", false) {
+        Message::Error(e) => {
+            assert_eq!(e.code, ErrorCode::Internal);
+            assert!(e.detail.ends_with(UNAVAILABLE), "{}", e.detail);
+        }
+        other => panic!("started without memory: {other:?}"),
+    }
+    assert!(l.server.requests_to(CHAT).is_empty(), "no request was made");
+}
+
+#[test]
+fn memoryd_is_told_after_a_finished_turn_only() {
+    let l = start(Ok(""));
+    l.server.route(
+        CHAT,
+        vec![Reply::fixture("plain"), Reply::fixture("bad_request")],
+    );
+    turn(&l, "c", false);
+    assert!(
+        matches!(turn(&l, "d", false), Message::Error(_)),
+        "a turn that ran and failed"
+    );
+    assert!(matches!(
+        turn(&l, "nobody-made-this", true),
+        Message::Error(_)
+    ));
+    assert_eq!(*l.memory.updated.lock().unwrap(), ["c"]);
+}
+
+#[test]
+fn recall_and_memory_are_core_tools_of_the_m5a_registry() {
+    let names: Vec<String> = Registry::m5a()
+        .core_schemas()
+        .into_iter()
+        .map(|s| s.name)
+        .collect();
+    assert_eq!(
+        names,
+        ["clock", "recall", "memory", "find_tool", "call_tool"]
+    );
+    let m3a: Vec<String> = Registry::m3a()
+        .core_schemas()
+        .into_iter()
+        .map(|s| s.name)
+        .collect();
+    assert_eq!(m3a, ["clock", "find_tool", "call_tool"], "m3a is unchanged");
+    assert!(Registry::m5a().get("read_file").is_some());
+}

docs/plans/M5c1/files/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);
+}

docs/plans/M5c1/files/crates/loopd/tests/support/turn.rs

@@ -0,0 +1,97 @@
+//! A turn-loop setup shared by the `turn` and `limits` tests: a home, a fake server, a scripted
+//! port and a registry. Included with `#[path]` because it needs `loopd::turn`, which the earlier
+//! tasks' tests must not depend on. Do not edit.
+
+#![allow(dead_code)]
+
+use crate::support::{FakeServer, Home, ScriptedPort};
+use bxcore::{LogRecord, SessionId, ToolResponse, TurnEvent};
+use loopd::baseline::Baseline;
+use loopd::llama::Client;
+use loopd::session::Session;
+use loopd::tools::Registry;
+use loopd::turn::{Runtime, TurnError, run_turn};
+
+pub struct Setup {
+    pub home: Home,
+    pub server: FakeServer,
+    pub cfg: loopd::config::Config,
+    pub client: Client,
+    pub port: ScriptedPort,
+    pub registry: Registry,
+}
+
+pub fn setup(replies: Vec<ToolResponse>) -> Setup {
+    let home = Home::new();
+    let server = FakeServer::start();
+    let cfg = home.config(&server.socket);
+    let client = Client::new(cfg.clone());
+    Setup {
+        home,
+        server,
+        cfg,
+        client,
+        port: ScriptedPort::new(replies),
+        registry: Registry::m2b(),
+    }
+}
+
+impl Setup {
+    pub fn session(&self, id: &str) -> Session {
+        let baseline = Baseline::assemble(&self.cfg, &self.registry).unwrap();
+        Session::create(
+            &self.home.dir,
+            SessionId::new(id).unwrap(),
+            baseline,
+            self.cfg.slots.main,
+        )
+        .unwrap()
+    }
+
+    pub fn runtime(&self) -> Runtime<'_> {
+        Runtime {
+            cfg: &self.cfg,
+            client: &self.client,
+            port: &self.port,
+            registry: &self.registry,
+        }
+    }
+
+    pub fn turn(
+        &self,
+        session: &mut Session,
+        text: &str,
+    ) -> (Result<loopd::turn::TurnOutcome, TurnError>, Vec<TurnEvent>) {
+        let mut events = Vec::new();
+        let result = run_turn(session, &self.runtime(), text, None, &mut |e| {
+            events.push(e.clone())
+        });
+        (result, events)
+    }
+}
+
+/// The recordings were made in separate conversations, so their timings do not line up and the
+/// loop rightly reports cache losses between them. Most tests are not about that, so `types`
+/// and `records` leave `CacheLoss` out; one test checks it on purpose.
+pub fn without_cache_loss(records: &[LogRecord]) -> Vec<LogRecord> {
+    records
+        .iter()
+        .filter(|r| !matches!(r, LogRecord::CacheLoss { .. }))
+        .cloned()
+        .collect()
+}
+
+pub fn types(records: &[LogRecord]) -> Vec<&'static str> {
+    without_cache_loss(records)
+        .iter()
+        .map(|r| match r {
+            LogRecord::SessionStart { .. } => "start",
+            LogRecord::User { .. } => "user",
+            LogRecord::Assistant { .. } => "assistant",
+            LogRecord::Usage { .. } => "usage",
+            LogRecord::ToolResult { .. } => "tool_result",
+            LogRecord::CacheLoss { .. } => "cache_loss",
+            LogRecord::EpochEnd { .. } => "epoch_end",
+        })
+        .collect()
+}

docs/plans/M5c1/files/crates/loopd/tests/turn_broker.rs

@@ -0,0 +1,270 @@
+//! Tests for what the turn loop does with the broker's answers: denials, pending approvals, and
+//! a port that misbehaves. Do not edit.
+
+mod support;
+#[path = "support/turn.rs"]
+mod turn_support;
+
+use bxcore::{DataClass, DenyReason, LogRecord, Timestamp, ToolResponse, TurnEvent};
+use support::{Reply, ReturnsPendingPort, ok_result, pending};
+use turn_support::{setup, types};
+
+const CHAT: &str = "/v1/chat/completions";
+
+/// The tool events of a turn, in order, as short strings.
+fn tool_events(events: &[TurnEvent]) -> Vec<String> {
+    events
+        .iter()
+        .filter_map(|e| match e {
+            TurnEvent::ToolCallStarted { name } => Some(format!("started {name}")),
+            TurnEvent::ApprovalPending { approval, tool, .. } => {
+                Some(format!("pending {approval} {tool}"))
+            }
+            TurnEvent::ToolDenied { name, reason } => Some(format!("denied {name} {reason:?}")),
+            TurnEvent::ToolResult { name, .. } => Some(format!("result {name}")),
+            _ => None,
+        })
+        .collect()
+}
+
+#[test]
+fn a_denial_is_a_fixed_sentence_for_the_model_and_an_event_for_the_owner() {
+    let s = setup(vec![ToolResponse::Denied {
+        reason: DenyReason::TaintTooHigh,
+        detail: None,
+    }]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let (result, events) = s.turn(&mut session, "x");
+    assert!(
+        result.is_ok(),
+        "the turn goes on after a denial: {result:?}"
+    );
+    match &session.records()[4] {
+        LogRecord::ToolResult {
+            content,
+            class,
+            untrusted,
+            truncated,
+            ..
+        } => {
+            assert_eq!(
+                content,
+                "Denied: this session has seen data too sensitive for this call."
+            );
+            assert_eq!(
+                (*class, *untrusted, *truncated),
+                (DataClass::Public, false, false)
+            );
+        }
+        other => panic!("{other:?}"),
+    }
+    assert_eq!(
+        tool_events(&events),
+        [
+            "started read_file",
+            "denied read_file TaintTooHigh",
+            "result read_file"
+        ],
+        "the denial comes before the result"
+    );
+    // The model reads the sentence in the next request.
+    let m2 = s.server.requests_to(CHAT)[1].json();
+    assert_eq!(
+        m2["messages"][3]["content"],
+        "Denied: this session has seen data too sensitive for this call."
+    );
+}
+
+/// The broker's detail reaches the model, in the log and in the next request; the owner's event
+/// is the same as for any denial.
+#[test]
+fn an_invalid_call_tells_the_model_why() {
+    let s = setup(vec![ToolResponse::Denied {
+        reason: DenyReason::InvalidArguments,
+        detail: Some("invalid path: n/a".to_string()),
+    }]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let (result, events) = s.turn(&mut session, "x");
+    assert!(result.is_ok(), "{result:?}");
+    let want = "Denied: the arguments are not valid for this tool: invalid path: n/a.";
+    match &session.records()[4] {
+        LogRecord::ToolResult {
+            content,
+            class,
+            untrusted,
+            truncated,
+            ..
+        } => {
+            assert_eq!(content, want);
+            assert_eq!(
+                (*class, *untrusted, *truncated),
+                (DataClass::Public, false, false)
+            );
+        }
+        other => panic!("{other:?}"),
+    }
+    assert_eq!(
+        tool_events(&events),
+        [
+            "started read_file",
+            "denied read_file InvalidArguments",
+            "result read_file"
+        ]
+    );
+    let m2 = s.server.requests_to(CHAT)[1].json();
+    assert_eq!(m2["messages"][3]["content"], want);
+}
+
+#[test]
+fn a_denied_call_tool_names_the_target_tool_in_the_denial() {
+    let s = setup(vec![ToolResponse::Denied {
+        reason: DenyReason::NoGrant,
+        detail: None,
+    }]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("call_tool"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let (result, events) = s.turn(&mut session, "echo box");
+    assert!(result.is_ok(), "{result:?}");
+    assert_eq!(
+        tool_events(&events),
+        [
+            "started call_tool",
+            "denied echo NoGrant",
+            "result call_tool"
+        ],
+        "the owner writes grants for `echo`, not for `call_tool`"
+    );
+    assert!(
+        matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: no grant allows this call.")
+    );
+}
+
+#[test]
+fn a_pending_approval_is_an_event_and_the_answer_after_it_is_the_result() {
+    let s = setup(vec![
+        pending(41, "2026-09-18T12:15:00.000Z"),
+        ok_result("straylight\n"),
+    ]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let (result, events) = s.turn(&mut session, "x");
+    assert!(result.is_ok(), "{result:?}");
+    assert_eq!(
+        tool_events(&events),
+        [
+            "started read_file",
+            "pending 41 read_file",
+            "result read_file"
+        ]
+    );
+    let expires = events.iter().find_map(|e| match e {
+        TurnEvent::ApprovalPending { expires, .. } => Some(*expires),
+        _ => None,
+    });
+    assert_eq!(
+        expires,
+        Some(Timestamp::parse("2026-09-18T12:15:00.000Z").unwrap())
+    );
+    assert_eq!(s.port.calls().len(), 1, "one call, however long it waited");
+    assert_eq!(
+        types(session.records()),
+        [
+            "start",
+            "user",
+            "assistant",
+            "usage",
+            "tool_result",
+            "assistant",
+            "usage"
+        ],
+        "waiting writes nothing to the log"
+    );
+    assert!(
+        matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "straylight\n")
+    );
+}
+
+#[test]
+fn a_pending_approval_that_ends_in_a_refusal() {
+    let s = setup(vec![
+        pending(7, "2026-09-18T12:15:00.000Z"),
+        ToolResponse::Denied {
+            reason: DenyReason::ApprovalRefused,
+            detail: None,
+        },
+    ]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let (result, events) = s.turn(&mut session, "x");
+    assert!(result.is_ok(), "{result:?}");
+    assert_eq!(
+        tool_events(&events),
+        [
+            "started read_file",
+            "pending 7 read_file",
+            "denied read_file ApprovalRefused",
+            "result read_file"
+        ]
+    );
+    assert!(
+        matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: the owner refused this call.")
+    );
+}
+
+#[test]
+fn a_port_that_returns_a_pending_frame_as_its_answer_is_a_failure_not_a_decision() {
+    let s = setup(vec![]);
+    s.server.route(
+        CHAT,
+        vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
+    );
+    let mut session = s.session("a");
+    let runtime = loopd::turn::Runtime {
+        cfg: &s.cfg,
+        client: &s.client,
+        port: &ReturnsPendingPort,
+        registry: &s.registry,
+    };
+    let mut events = Vec::new();
+    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 {
+            content,
+            class,
+            untrusted,
+            ..
+        } => {
+            assert_eq!(
+                content,
+                "The tool failed: the tool broker gave no final answer"
+            );
+            assert_eq!((*class, *untrusted), (DataClass::Public, false));
+        }
+        other => panic!("{other:?}"),
+    }
+    assert_eq!(
+        tool_events(&events),
+        ["started read_file", "result read_file"],
+        "neither pending nor denied: the port said neither"
+    );
+}