diff --git a/docs/plans/M5c1/01-wire-speaker-state.md b/docs/plans/M5c1/01-wire-speaker-state.md index e1f8ea8..002d7eb 100644 --- a/docs/plans/M5c1/01-wire-speaker-state.md +++ b/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/ diff --git a/docs/plans/M5c1/02-loopd-speaker-turn.md b/docs/plans/M5c1/02-loopd-speaker-turn.md new file mode 100644 index 0000000..8c839aa --- /dev/null +++ b/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` to `Option<&str>` | + | `Option::map` | `speaker.map(str::to_string)`: `Option<&str>` to `Option` | + | `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` (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 { +``` + +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 { +``` + +**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 { +``` + +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 { +``` + +`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. diff --git a/docs/plans/M5c1/files/crates/loopd/tests/end_to_end.rs b/docs/plans/M5c1/files/crates/loopd/tests/end_to_end.rs new file mode 100644 index 0000000..2d29c25 --- /dev/null +++ b/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 { + let mut names: Vec = 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, ®istry).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: ®istry, + }; + 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:?}"), + } +} diff --git a/docs/plans/M5c1/files/crates/loopd/tests/memory.rs b/docs/plans/M5c1/files/crates/loopd/tests/memory.rs index 2e4b5f4..27aae1f 100644 --- a/docs/plans/M5c1/files/crates/loopd/tests/memory.rs +++ b/docs/plans/M5c1/files/crates/loopd/tests/memory.rs @@ -33,7 +33,7 @@ impl MemorySource for Shared { fn block(&self) -> Result { 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() diff --git a/docs/plans/M5c1/files/crates/loopd/tests/memory.rs-task01 b/docs/plans/M5c1/files/crates/loopd/tests/memory.rs-task01 new file mode 100644 index 0000000..2e4b5f4 --- /dev/null +++ b/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>, + updated: Mutex>, +} + +/// Lets the test keep its `Arc` while the context owns a `Box`. +struct Shared(Arc); + +impl MemorySource for Shared { + fn block(&self) -> Result { + 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, + // 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 = Registry::m5a() + .core_schemas() + .into_iter() + .map(|s| s.name) + .collect(); + assert_eq!( + names, + ["clock", "recall", "memory", "find_tool", "call_tool"] + ); + let m3a: Vec = 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()); +} diff --git a/docs/plans/M5c1/files/crates/loopd/tests/speaker_turn.rs b/docs/plans/M5c1/files/crates/loopd/tests/speaker_turn.rs new file mode 100644 index 0000000..543ce7b --- /dev/null +++ b/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>); + +impl Recorder { + fn told(&self) -> Vec<(String, u64)> { + self.0.lock().unwrap().clone() + } +} + +/// Lets the test keep its `Arc` while the context owns a `Box`. +struct Shared(Arc); + +impl MemorySource for Shared { + fn block(&self) -> Result { + 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) -> 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) { + 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 { + 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 { + 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 { + 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); +} diff --git a/docs/plans/M5c1/files/crates/loopd/tests/support/turn.rs b/docs/plans/M5c1/files/crates/loopd/tests/support/turn.rs new file mode 100644 index 0000000..e65b458 --- /dev/null +++ b/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) -> 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, Vec) { + 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 { + 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() +} diff --git a/docs/plans/M5c1/files/crates/loopd/tests/turn_broker.rs b/docs/plans/M5c1/files/crates/loopd/tests/turn_broker.rs new file mode 100644 index 0000000..c916408 --- /dev/null +++ b/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 { + 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" + ); +}