diff --git a/docs/plans/M5b/01-invalid-arguments-detail.md b/docs/plans/M5b/01-invalid-arguments-detail.md index 50fb8b3..6ed6a80 100644 --- a/docs/plans/M5b/01-invalid-arguments-detail.md +++ b/docs/plans/M5b/01-invalid-arguments-detail.md @@ -52,7 +52,8 @@ This is the complete list. Each one is covered by an edit below or by a given fi `crates/brokerd/tests/invalid_detail.rs` - **new versions** of: `crates/bxcore/tests/wire.rs`, `crates/bxcore/tests/strict.rs`; `crates/brokerd/tests/admin.rs`, `broker.rs`, `broker_pending.rs`, `courier.rs`, `ledger.rs`, - `serve.rs`; `crates/loopd/tests/broker_port.rs`, `tools.rs`, `turn_broker.rs` + `serve.rs`; `crates/loopd/tests/broker_port.rs`, `tools.rs`, `turn_broker.rs`. Your + `broker.rs` is the given `broker.rs-task01`: the given `broker.rs` is task 09's. - Modify: `crates/bxcore/src/wire.rs`, `crates/brokerd/src/args.rs`, `crates/brokerd/src/policy/mod.rs`, `crates/brokerd/src/ledger/mod.rs`, `crates/brokerd/src/broker.rs`, `crates/loopd/src/tools.rs`, `crates/loopd/src/turn.rs`, @@ -489,8 +490,8 @@ The event keeps only the reason: the owner's texts do not change. crates/brokerd/tests/ cp docs/plans/M5b/files/crates/brokerd/tests/admin.rs \ crates/brokerd/tests/ - cp docs/plans/M5b/files/crates/brokerd/tests/broker.rs \ - crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/broker.rs-task01 \ + crates/brokerd/tests/broker.rs cp docs/plans/M5b/files/crates/brokerd/tests/broker_pending.rs \ crates/brokerd/tests/ cp docs/plans/M5b/files/crates/brokerd/tests/courier.rs \ diff --git a/docs/plans/M5b/09-turn-done-untrusted.md b/docs/plans/M5b/09-turn-done-untrusted.md new file mode 100644 index 0000000..18da866 --- /dev/null +++ b/docs/plans/M5b/09-turn-done-untrusted.md @@ -0,0 +1,250 @@ +# M5b task 09: `TurnDone` says whether the session has read untrusted content + +**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop) +**Commit subject:** `bxcore, loopd: TurnDone says whether the session has read untrusted content` + +## Goal + +`gatewayd` will post the links of a reply as code when the reply comes from a session that has +read untrusted content, a fetched page or a recalled memory-tainted note (M5b spec, section 9; +that is task 10). It learns that from `loopd`, in the frame that ends each turn. So `TurnDone` +gains a field (M5b spec, sections 8 and 9): + +```json +{"kind":"turn_done","body":{"content":"…","usage":{…},"untrusted":true}} +``` + +- **`bxcore`**: `TurnDone` gains `untrusted: bool`, last. It is always written, and it is + **required** when read: a frame without it is refused, never read as a reply from a clean + session (see "Facts you need"). The fixture `turn_done.json` changes, and a second one, + `turn_done_untrusted.json`, holds `true`. +- **`loopd`**: a new method, `Session::untrusted`, is true when any tool result in the session's + log is labelled untrusted. `channel::handle` puts it in the `TurnDone`. The log is the one + source: it already holds each tool result with the label `brokerd` gave it, it is read back + whole when a session is reopened, and records are only ever appended, so once a session has + read untrusted content every later `TurnDone` of it says so, also after `loopd` restarts. +- **Unchanged**: every source file of `gatewayd`, `bxctl` and `brokerd`. Only their tests build a + `TurnDone`, and new versions of those are given. + +| The session's log holds | `TurnDone.untrusted` | +|---|---| +| no tool result, or only results labelled trusted | `false` | +| a denial, or a failure (they are logged trusted) | `false` | +| a result labelled untrusted, in this turn or any earlier one | `true` | +| a `recall` result with a memory-tainted hit (`brokerd` labels it untrusted) | `true` | + +## Every place `TurnDone` appears + +This is the complete list. Each one is covered by an edit below or by a given file. + +| Place | What happens there | +|---|---| +| `crates/bxcore/src/wire.rs`, the struct | gains `untrusted` (edit W1) | +| `crates/loopd/src/session.rs` | new `Session::untrusted` (edit S1) | +| `crates/loopd/src/channel.rs`, `handle` | sets `untrusted` from the session (edit C1) | +| `crates/gatewayd/src/deliver.rs` | reads `content` only; no change (task 10 reads the flag) | +| `crates/bxctl/src/chat.rs`, `crates/bxctl/src/main.rs` | pass the `TurnDone` on; `--json` prints it whole with `serde_json`, so the new field shows there; no change | +| `crates/brokerd/src/broker.rs` | `Message::TurnDone(_)`, a pattern; no change | +| `crates/loopd/tests/channel.rs`, `memory.rs` | patterns and field reads; no change | +| tests that build a `TurnDone` in `bxcore`, `brokerd`, `bxctl`, `gatewayd` | new versions are given | + +## Files + +- Copy, from `docs/plans/M5b/files/` (the same path under it): + - new: `crates/bxcore/tests/fixtures/wire/turn_done_untrusted.json`, + `crates/loopd/tests/turn_done_untrusted.rs` + - **new versions** of: `crates/bxcore/tests/fixtures/wire/turn_done.json`, + `crates/bxcore/tests/turn_wire.rs`; `crates/brokerd/tests/broker.rs`; + `crates/bxctl/tests/chat.rs`, `crates/bxctl/tests/support/mod.rs`; + `crates/gatewayd/tests/support/fake_loop.rs` (they replace the files in the tree). Do not copy + `broker.rs-task01`: it is the older version. +- Modify: `crates/bxcore/src/wire.rs` (W1), `crates/loopd/src/session.rs` (S1), + `crates/loopd/src/channel.rs` (C1), `docs/implementer-log.md`. +- Do not touch: any other source file, any `Cargo.toml`, `crates/bxcore/tests/strict.rs`. + +## Facts you need + +- **Why the field is required.** `serde` treats a `bool` field without `#[serde(default)]` as + required: decoding a `turn_done` without it fails with ``missing field `untrusted` ``. A default + would read a missing field as `false`, "clean", which is the wrong way to fail: a reply from an + untrusted session would keep its links live. `loopd`, `gatewayd` and `bxctl` are built from + this one workspace and deployed together, and `TurnDone` already refuses unknown fields, so an + older reader refuses the new frame anyway. So: no `#[serde(default)]`, no + `skip_serializing_if`. `PROTOCOL_VERSION` does not change (task 01 added a field without + changing it too). +- **Where the labels come from.** `brokerd` labels every result it answers with + (`ToolResponse::Result { untrusted, .. }`): from the grant, and for `recall` also from + `memoryd`, which marks a result holding a memory-tainted hit untrusted; `brokerd` keeps the + stricter of the two. `run_turn` (`crates/loopd/src/turn.rs`, unchanged) logs each tool result + as `LogRecord::ToolResult { untrusted, .. }` with that label. A denial, a failure and + `loopd`'s own tools (`clock`, `find_tool`, `call_tool`'s answers) are logged with + `untrusted: false`, as `brokerd`'s own state does: a failure changes no state. +- `Session::open` reads every record of the log back into `records`, so `Session::untrusted` on a + reopened session sees every earlier turn's results. Every turn after a session's first reopens + it (`resume`), in the same `loopd` or a restarted one. +- In `channel::handle`, `session` is still in scope after `run_turn` returns: `run_turn` borrows + it only for the call. `session.untrusted()` borrows it again, immutably. +- The calls the edits make: + + | Call | Signature, and what it gives | + |---|---| + | `Iterator::any` | `self.records.iter().any(f) -> bool`: true when `f` is true for some record | + | `matches!` | `matches!(record, LogRecord::ToolResult { untrusted: true, .. })`: true for a tool result whose field `untrusted` is `true` | + | `Session::untrusted` | `pub fn untrusted(&self) -> bool` (new, S1) | + +## The edits, exactly + +Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. + +### `crates/bxcore/src/wire.rs` + +**W1.** The lines + +```rust +// JSON: {"kind":"turn_done","body":{"content":"…","usage":{…}}} +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TurnDone { + pub content: String, + pub usage: Usage, +} +``` + +become + +```rust +// JSON: {"kind":"turn_done","body":{"content":"…","usage":{…},"untrusted":false}} +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TurnDone { + pub content: String, + pub usage: Usage, + /// True once the session has read untrusted content, in this turn or an earlier one (M5b + /// spec, section 9). It has no default: a frame without it is refused, never read as clean. + pub untrusted: bool, +} +``` + +### `crates/loopd/src/session.rs` + +**S1.** The line + +```rust + pub fn last_usage(&self) -> Option { +``` + +becomes + +```rust + /// True once any tool result in the log is labelled untrusted: a result under an untrusted + /// grant, or a recalled memory-tainted note, which `brokerd` labels so. The labels are in + /// the log, so this holds after the session is reopened, and it never becomes false again. + pub fn untrusted(&self) -> bool { + self.records.iter().any(|record| { + matches!( + record, + LogRecord::ToolResult { + untrusted: true, + .. + } + ) + }) + } + + 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 + msg: Message::TurnDone(TurnDone { + content: outcome.content, + usage: outcome.usage, + }), +``` + +become + +```rust + msg: Message::TurnDone(TurnDone { + content: outcome.content, + usage: outcome.usage, + // The whole log, this turn's results included (M5b spec, section 8). + untrusted: session.untrusted(), + }), +``` + +## Steps + +- [ ] **1. Copy the given files.** From the repository root, eight commands: + + ```sh + cp docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done.json \ + crates/bxcore/tests/fixtures/wire/ + cp docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done_untrusted.json \ + crates/bxcore/tests/fixtures/wire/ + cp docs/plans/M5b/files/crates/bxcore/tests/turn_wire.rs crates/bxcore/tests/ + cp docs/plans/M5b/files/crates/brokerd/tests/broker.rs crates/brokerd/tests/ + cp docs/plans/M5b/files/crates/bxctl/tests/chat.rs crates/bxctl/tests/ + cp docs/plans/M5b/files/crates/bxctl/tests/support/mod.rs crates/bxctl/tests/support/ + cp docs/plans/M5b/files/crates/gatewayd/tests/support/fake_loop.rs \ + crates/gatewayd/tests/support/ + cp docs/plans/M5b/files/crates/loopd/tests/turn_done_untrusted.rs crates/loopd/tests/ + ``` + + Then `git status --short` lists 6 modified files and 2 new ones, all under `crates/*/tests/`. +- [ ] **2. See them fail.** Two commands: + - `cargo test -p bxcore --test strict`: one test fails, + `envelopes_reject_unknown_keys_at_every_depth`, with `turn_done.json: fixture must decode`. + - `cargo build --tests --keep-going -p bxcore -p brokerd -p loopd -p bxctl -p gatewayd`: + thirteen test targets do not compile: `turn_wire` in `bxcore`; `broker` in `brokerd`; + `turn_done_untrusted` in `loopd`; `admin`, `chat` and `chat_approvals` in `bxctl`; + `deliver`, `deliver_notice`, `serve`, `serve_answers`, `serve_approvals`, `serve_redirect` + and `serve_restart` in `gatewayd`. Every error is about the field this task adds: `E0560` + (``struct `bxcore::TurnDone` has no field named `untrusted` ``) or, once, `E0609` + (``no field `untrusted` on type `bxcore::TurnDone` ``). +- [ ] **3. `bxcore`.** Make edit W1. `cargo test -p bxcore`: every suite passes, among them the + seven tests of `turn_wire`. +- [ ] **4. `loopd`.** Make edits S1 and C1. `cargo check -p loopd`: no error. Then + `cargo test -p loopd --test turn_done_untrusted`: 6 passed, and `cargo test -p loopd`: every + suite passes. +- [ ] **5. The other crates.** `cargo test -p brokerd -p bxctl -p gatewayd`: every suite passes. + You changed nothing in them but the copied tests. +- [ ] **6. Check.** `git diff --stat -- crates/gatewayd/src crates/bxctl/src crates/brokerd/src` + prints nothing, and `grep -n 'serde(default' crates/bxcore/src/wire.rs` prints only the line + of `ToolResponse::Denied`'s `detail` (task 01), not one for `untrusted`. +- [ ] **7. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. + Expected last line: `gate: ok`, with 1058 tests: 1050 before this task, plus 8 new ones (two in + `bxcore/tests/turn_wire.rs`, six in `loopd/tests/turn_done_untrusted.rs`; the new versions of + the other files keep their counts). +- [ ] **8. Log and commit.** `git status --short` lists the three sources, the six replaced test + files and your log as modified, and `turn_done_untrusted.json` and `turn_done_untrusted.rs` as + new. Stage with + `git add crates/bxcore crates/loopd crates/brokerd crates/bxctl crates/gatewayd 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 1058 tests. +- `git diff HEAD~1 --stat` lists only `crates/bxcore/src/wire.rs`, `crates/loopd/src/session.rs`, + `crates/loopd/src/channel.rs`, the eight copied files and `docs/implementer-log.md`. + +## Stop and report if + +- A "before" block of W1, S1 or C1 is not in its file exactly as shown, or is there more than + once. +- Step 2 shows a different result: another test failing in `strict`, other test targets, or an + error of another kind. +- After step 3, 4 or 5, a test fails. Do not change a test or a fixture: stop, and give the + failing tests' names and their messages. +- The fix seems to need a change anywhere else: in `turn.rs`, in `gatewayd`, `bxctl` or `brokerd` + sources, or a default for the field. It does not; stop and say why you think so. +- A test in `memoryd`, `inferproxy` or `toolkit` 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 1058. +- A comment in an edit does not fit the code under it. Quote it and stop. diff --git a/docs/plans/M5b/files/crates/brokerd/tests/broker.rs b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs index fc31dcb..72e87cd 100644 --- a/docs/plans/M5b/files/crates/brokerd/tests/broker.rs +++ b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs @@ -165,6 +165,7 @@ fn every_kind_but_tool_request() -> Vec { Message::TurnDone(TurnDone { content: String::new(), usage, + untrusted: false, }), Message::Approvals(Empty {}), Message::ApprovalList(ApprovalList { items: Vec::new() }), diff --git a/docs/plans/M5b/files/crates/brokerd/tests/broker.rs-task01 b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs-task01 new file mode 100644 index 0000000..fc31dcb --- /dev/null +++ b/docs/plans/M5b/files/crates/brokerd/tests/broker.rs-task01 @@ -0,0 +1,327 @@ +//! One `broker.sock` connection: decision, record, run, answer; and every way it fails closed. +//! Do not edit. + +#[path = "support/client.rs"] +mod client; +#[path = "support/rig.rs"] +mod rig; +#[path = "support/runtime.rs"] +mod runtime; +#[path = "support/sink.rs"] +mod sink; +#[path = "support/tmp.rs"] +mod tmp; + +use std::io::Write; +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::UnixStream; +use std::sync::Arc; + +use brokerd::broker; +use brokerd::ledger::NOT_RECORDED; +use brokerd::runner::{REFUSING, RunError}; +use bxcore::{ + ApprovalList, Approve, ApproveResult, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, + Envelope, ErrorCode, GrantsReport, Message, Refuse, ResultStatus, SessionId, ToolResponse, + Turn, TurnDone, TurnEvent, Usage, WireError, +}; +use client::{Serve, call, next, open}; +use rig::{Rig, grant_text, request}; +use runtime::Recording; + +const NOTES: &str = r#"{"path":"/n/a"}"#; + +fn read_notes(call_id: u64) -> bxcore::ToolRequest { + request("s1", call_id, "read_file", NOTES) +} + +fn only(frames: &[Envelope]) -> &ToolResponse { + assert_eq!(frames.len(), 1, "{frames:?}"); + assert!(frames[0].r#final); + match &frames[0].msg { + Message::ToolResponse(r) => r, + other => panic!("{other:?}"), + } +} + +fn denied(reason: DenyReason) -> ToolResponse { + ToolResponse::Denied { + reason, + detail: None, + } +} + +fn auto_notes(rig: &Rig, class: &str) { + let text = grant_text("read_file", "auto", "", "paths = [\"/n\"]").replace( + "result_class = \"private\"", + &format!("result_class = \"{class}\""), + ); + rig.grant("notes", &text); +} + +#[test] +fn a_call_no_grant_allows_is_denied_and_recorded() { + let rig = Rig::new("broker-nogrant"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(41)); + assert_eq!(frames[0].id, 41, "the answer carries the request's id"); + assert_eq!(only(&frames), &denied(DenyReason::NoGrant)); + assert_eq!(rt.count(), 0); + match rig.events().as_slice() { + [AuditEvent::Decision { outcome, .. }] => assert_eq!( + *outcome, + DecisionRecord::Denied { + reason: DenyReason::NoGrant + } + ), + other => panic!("{other:?}"), + } +} + +#[test] +fn an_allowed_call_runs_and_its_result_raises_the_state() { + let rig = Rig::new("broker-allowed"); + auto_notes(&rig, "secret"); + let rt = Recording::answering("the notes"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(7)); + assert_eq!( + only(&frames), + &ToolResponse::Result { + content: "the notes".to_string(), + class: DataClass::Secret, + untrusted: false, + truncated: false, + } + ); + assert_eq!(rt.count(), 1); + let events = rig.events(); + assert!(matches!( + events.as_slice(), + [ + AuditEvent::Decision { .. }, + AuditEvent::Result { + decision: 0, + status: ResultStatus::Result, + taint_after: DataClass::Secret, + .. + } + ] + )); + let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap(); + assert_eq!(state.taint, DataClass::Secret); +} + +#[test] +fn a_runtime_failure_is_passed_on_and_recorded_as_failed() { + let rig = Rig::new("broker-refusing"); + auto_notes(&rig, "secret"); + let rt = Recording::with(Err(RunError::Unavailable(REFUSING.to_string()))); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + assert_eq!( + only(&frames), + &ToolResponse::Failed { + message: REFUSING.to_string() + } + ); + match &rig.events()[1] { + AuditEvent::Result { + status, + taint_after, + .. + } => assert_eq!( + (*status, *taint_after), + (ResultStatus::Failed, DataClass::Private) + ), + other => panic!("{other:?}"), + } + assert!(!rig.state_file("s1").exists(), "a failure changes no state"); +} + +fn every_kind_but_tool_request() -> Vec { + let usage = Usage { + cache_n: 0, + prompt_n: 0, + predicted_n: 0, + reasoning_tokens: 0, + thinking_capped: false, + }; + vec![ + Message::ToolResponse(denied(DenyReason::NoGrant)), + Message::Error(WireError { + code: ErrorCode::Internal, + detail: String::new(), + }), + Message::Turn(Turn { + session: SessionId::new("s1").unwrap(), + content: String::new(), + resume: false, + }), + Message::TurnEvent(TurnEvent::Content { + text: String::new(), + }), + Message::TurnDone(TurnDone { + content: String::new(), + usage, + }), + Message::Approvals(Empty {}), + Message::ApprovalList(ApprovalList { items: Vec::new() }), + Message::Approve(Approve { approval: 0 }), + Message::ApproveResult(ApproveResult { + outcome: DecisionRecord::Allowed {}, + }), + Message::Refuse(Refuse { + approval: 0, + reason: None, + }), + Message::Ok(Empty {}), + Message::CheckGrants(Empty {}), + Message::GrantsReport(GrantsReport { + problems: Vec::new(), + }), + ] +} + +#[test] +fn every_other_kind_on_broker_sock_is_forbidden() { + let rig = Rig::new("broker-forbidden"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let kinds = every_kind_but_tool_request(); + assert_eq!(kinds.len(), 13, "every Message variant but tool_request"); + for (i, msg) in kinds.into_iter().enumerate() { + let id = 100 + i as u64; + let mut stream = open(&broker, broker::handle, id, msg); + let env = next(&mut stream); + assert_eq!((env.id, env.r#final), (id, true)); + match env.msg { + Message::Error(e) => assert_eq!(e.code, ErrorCode::Forbidden), + other => panic!("{other:?}"), + } + } + let lines = rig.lines.with("broker.sock"); + assert_eq!(lines.len(), 13, "{:?}", rig.lines.all()); + for line in &lines { + assert!( + line.ends_with("\nsee docs/runbook.md#socket-forbidden"), + "{line}" + ); + } + assert_eq!(rig.lines.with("kind approve on broker.sock").len(), 1); + assert!( + rig.events().is_empty(), + "nothing is recorded for a refused kind" + ); +} + +#[test] +fn a_frame_that_is_not_json_is_answered_with_bad_message() { + let rig = Rig::new("broker-badframe"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let (mut client, server) = UnixStream::pair().unwrap(); + let b = Arc::clone(&broker); + std::thread::spawn(move || broker::handle(server, &b)); + client.write_all(&3u32.to_be_bytes()).unwrap(); + client.write_all(b"{{{").unwrap(); + let env = next(&mut client); + assert_eq!((env.id, env.r#final), (0, true)); + match env.msg { + Message::Error(e) => assert_eq!(e.code, ErrorCode::BadMessage), + other => panic!("{other:?}"), + } + assert!(rig.events().is_empty()); +} + +#[test] +fn one_invalid_grant_file_denies_a_call_a_valid_file_would_allow() { + let rig = Rig::new("broker-invalid"); + auto_notes(&rig, "private"); + rig.grant("bad", "tool = \"read_file\"\nmdoe = \"auto\"\n"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + for id in [1, 2] { + let frames = call(&broker, read_notes(id)); + assert_eq!(only(&frames), &denied(DenyReason::GrantsInvalid)); + } + assert_eq!(rt.count(), 0); + // Printed once for the two calls, with the pointer. + let printed = rig.lines.with("bad.toml"); + assert_eq!(printed.len(), 1, "{:?}", rig.lines.all()); + assert!( + printed[0].ends_with("see docs/runbook.md#grants-invalid"), + "{}", + printed[0] + ); + rig.remove_grant("bad"); + let frames = call(&broker, read_notes(3)); + assert!(matches!(only(&frames), ToolResponse::Result { .. })); + assert_eq!(rt.count(), 1); +} + +#[test] +fn a_damaged_session_state_is_state_unreadable() { + let rig = Rig::new("broker-state"); + auto_notes(&rig, "private"); + std::fs::create_dir_all(rig.cfg.state_dir()).unwrap(); + std::fs::write(rig.state_file("s1"), "{\"taint\":\"loud\"}").unwrap(); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + assert_eq!(only(&frames), &denied(DenyReason::StateUnreadable)); + assert_eq!(rt.count(), 0); + let lines = rig.lines.with("s1.json"); + assert!( + lines + .iter() + .any(|l| l.ends_with("see docs/runbook.md#broker-state-damaged")), + "{lines:?}" + ); +} + +#[test] +fn a_failed_audit_write_runs_nothing_now_or_later() { + let rig = Rig::new("broker-audit"); + auto_notes(&rig, "private"); + let rt = Recording::answering("x"); + let broker = rig.broker(&rt); + rig.switch.fail(true); + let frames = call(&broker, read_notes(1)); + assert_eq!(only(&frames), &denied(DenyReason::AuditUnavailable)); + rig.switch.fail(false); + let frames = call(&broker, read_notes(2)); + assert_eq!(only(&frames), &denied(DenyReason::AuditUnavailable)); + assert_eq!(rt.count(), 0); + assert!( + !rig.lines + .with("see docs/runbook.md#audit-unavailable") + .is_empty() + ); +} + +#[test] +fn a_state_that_cannot_be_written_withholds_the_content() { + if tmp::running_as_root("a_state_that_cannot_be_written_withholds_the_content") { + return; + } + let rig = Rig::new("broker-ro"); + auto_notes(&rig, "secret"); + let dir = rig.cfg.state_dir(); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o500)).unwrap(); + let rt = Recording::answering("the secret"); + let broker = rig.broker(&rt); + let frames = call(&broker, read_notes(1)); + std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap(); + assert_eq!( + only(&frames), + &ToolResponse::Failed { + message: NOT_RECORDED.to_string() + } + ); + assert_eq!(rt.count(), 1, "it ran; its content is what is withheld"); + let text = format!("{frames:?}"); + assert!(!text.contains("the secret"), "{text}"); +} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done.json b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done.json new file mode 100644 index 0000000..24cc4de --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done.json @@ -0,0 +1 @@ +{"v":1,"id":3,"final":true,"msg":{"kind":"turn_done","body":{"content":"It is noon.","usage":{"cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false},"untrusted":false}}} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done_untrusted.json b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done_untrusted.json new file mode 100644 index 0000000..942ecbd --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/turn_done_untrusted.json @@ -0,0 +1 @@ +{"v":1,"id":4,"final":true,"msg":{"kind":"turn_done","body":{"content":"The page says to run the script.","usage":{"cache_n":566,"prompt_n":412,"predicted_n":31,"reasoning_tokens":12,"thinking_capped":false},"untrusted":true}}} diff --git a/docs/plans/M5b/files/crates/bxcore/tests/turn_wire.rs b/docs/plans/M5b/files/crates/bxcore/tests/turn_wire.rs new file mode 100644 index 0000000..cffce50 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxcore/tests/turn_wire.rs @@ -0,0 +1,320 @@ +//! Tests for the channel messages and the usage record, against byte-exact fixtures. Do not edit. + +use bxcore::{ + CallId, DataClass, DenyReason, Envelope, Epoch, ErrorCode, Hash32, LogRecord, Message, + SessionId, Timestamp, ToolCall, Turn, TurnDone, TurnEvent, Usage, WireError, +}; + +fn fixture(kind: &str, name: &str) -> String { + let path = format!( + "{}/tests/fixtures/{kind}/{name}", + env!("CARGO_MANIFEST_DIR") + ); + std::fs::read_to_string(&path) + .unwrap_or_else(|e| panic!("{path}: {e}")) + .trim_end_matches('\n') + .to_string() +} + +fn check(name: &str, want: Envelope) { + let text = fixture("wire", name); + let got: Envelope = serde_json::from_str(&text).unwrap_or_else(|e| panic!("{name}: {e}")); + assert_eq!(got, want, "{name}: decoded value"); + assert_eq!( + serde_json::to_string(&want).unwrap(), + text, + "{name}: encoded bytes" + ); +} + +fn env(id: u64, r#final: bool, msg: Message) -> Envelope { + Envelope { + v: 1, + id, + r#final, + msg, + } +} + +fn usage() -> Usage { + Usage { + cache_n: 539, + prompt_n: 27, + predicted_n: 24, + reasoning_tokens: 8, + thinking_capped: false, + } +} + +#[test] +fn turn() { + let body = Turn { + session: SessionId::new("chat-1789700000-42").unwrap(), + content: "What time is it?".to_string(), + resume: false, + }; + check("turn.json", env(3, true, Message::Turn(body))); +} + +#[test] +fn turn_events() { + check( + "turn_event_tool_result.json", + env( + 3, + false, + Message::TurnEvent(TurnEvent::ToolResult { + name: "clock".to_string(), + class: DataClass::Public, + truncated: false, + }), + ), + ); + check( + "turn_event_content.json", + env( + 3, + false, + Message::TurnEvent(TurnEvent::Content { + text: "It is ".to_string(), + }), + ), + ); + check( + "turn_event_retrying.json", + env( + 3, + false, + Message::TurnEvent(TurnEvent::Retrying { + attempt: 2, + after_ms: 1500, + error: "the server went silent".to_string(), + }), + ), + ); + check( + "turn_event_approval_pending.json", + env( + 3, + false, + Message::TurnEvent(TurnEvent::ApprovalPending { + approval: 41, + tool: "shell".to_string(), + expires: Timestamp::parse("2026-09-18T08:20:00.000Z").unwrap(), + }), + ), + ); + check( + "turn_event_tool_denied.json", + env( + 3, + false, + Message::TurnEvent(TurnEvent::ToolDenied { + name: "read_file".to_string(), + reason: DenyReason::NoGrant, + }), + ), + ); +} + +#[test] +fn turn_done_and_the_new_error_codes() { + check( + "turn_done.json", + env( + 3, + true, + Message::TurnDone(TurnDone { + content: "It is noon.".to_string(), + usage: usage(), + untrusted: false, + }), + ), + ); + let body = WireError { + code: ErrorCode::SessionFull, + detail: "this conversation is full; start a new one".to_string(), + }; + check( + "error_session_full.json", + env(3, true, Message::Error(body)), + ); + let codes = [ + (ErrorCode::SessionFull, "session_full"), + (ErrorCode::TurnLimit, "turn_limit"), + (ErrorCode::SessionBusy, "session_busy"), + (ErrorCode::NoSuchSession, "no_such_session"), + (ErrorCode::SessionExists, "session_exists"), + (ErrorCode::Inference, "inference"), + ]; + for (value, text) in codes { + assert_eq!( + serde_json::to_string(&value).unwrap(), + format!("\"{text}\"") + ); + } +} + +/// A reply from a session that has read untrusted content says so, and `untrusted` is written +/// whether it is true or false (M5b spec, section 9). +#[test] +fn turn_done_from_an_untrusted_session() { + check( + "turn_done_untrusted.json", + env( + 4, + true, + Message::TurnDone(TurnDone { + content: "The page says to run the script.".to_string(), + usage: Usage { + cache_n: 566, + prompt_n: 412, + predicted_n: 31, + reasoning_tokens: 12, + thinking_capped: false, + }, + untrusted: true, + }), + ), + ); +} + +/// `untrusted` has no default: a `turn_done` without it, as a `loopd` from before M5b sent, is +/// refused, never read as a reply from a clean session. +#[test] +fn a_turn_done_without_untrusted_is_refused() { + let old = r#"{"v":1,"id":3,"final":true,"msg":{"kind":"turn_done","body":{"content":"It is noon.","usage":{"cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false}}}}"#; + let error = serde_json::from_str::(old).unwrap_err(); + assert!( + error.to_string().contains("missing field `untrusted`"), + "{error}" + ); +} + +#[test] +fn every_turn_event_kind_round_trips() { + let all = vec![ + TurnEvent::Queued { ahead: 1 }, + TurnEvent::Waiting { slot_busy: true }, + TurnEvent::Progress { + total: 100, + cache: 50, + processed: 75, + }, + TurnEvent::Reasoning { + text: "hm".to_string(), + }, + TurnEvent::Content { + text: "hi".to_string(), + }, + TurnEvent::ToolCallStarted { + name: "clock".to_string(), + }, + TurnEvent::ToolResult { + name: "clock".to_string(), + class: DataClass::Secret, + truncated: true, + }, + TurnEvent::ThinkingCapped { tokens: 4096 }, + TurnEvent::Retrying { + attempt: 2, + after_ms: 10, + error: "x".to_string(), + }, + TurnEvent::CacheLoss { + expected: 500, + got: 20, + }, + TurnEvent::ApprovalPending { + approval: u64::MAX, + tool: "http_fetch".to_string(), + expires: Timestamp::parse("2026-09-18T08:20:00.000Z").unwrap(), + }, + TurnEvent::ToolDenied { + name: "shell".to_string(), + reason: DenyReason::StateUnreadable, + }, + ]; + for event in all { + let text = serde_json::to_string(&event).unwrap(); + assert!(text.starts_with("{\"event\":\""), "{text}"); + assert_eq!(serde_json::from_str::(&text).unwrap(), event); + } + assert!(serde_json::from_str::(r#"{"event":"content","text":"x","zz":1}"#).is_err()); + assert!(serde_json::from_str::(r#"{"event":"dance"}"#).is_err()); +} + +#[test] +fn usage_records() { + let text = fixture("records", "session_usage.jsonl"); + let lines: Vec<&str> = text.lines().collect(); + let want = [ + LogRecord::SessionStart { + time: Timestamp::parse("2026-09-18T08:05:00.000Z").unwrap(), + session: SessionId::new("chat-1789700000-42").unwrap(), + epoch: Epoch(0), + slot: 0, + baseline: Hash32::from_hex( + "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f", + ) + .unwrap(), + }, + LogRecord::User { + time: Timestamp::parse("2026-09-18T08:05:01.000Z").unwrap(), + content: "What time is it?".to_string(), + }, + LogRecord::Assistant { + time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(), + content: None, + reasoning_content: None, + tool_calls: vec![ToolCall { + id: "call_a1".to_string(), + name: "clock".to_string(), + arguments: "{}".to_string(), + }], + }, + LogRecord::Usage { + time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(), + cache_n: 539, + prompt_n: 27, + predicted_n: 24, + reasoning_tokens: 8, + thinking_capped: false, + }, + LogRecord::ToolResult { + time: Timestamp::parse("2026-09-18T08:05:03.100Z").unwrap(), + call: CallId(1), + tool_call_id: "call_a1".to_string(), + content: "2026-09-18T08:05:03.000Z".to_string(), + class: DataClass::Public, + untrusted: false, + truncated: false, + }, + LogRecord::Assistant { + time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(), + content: Some("It is five past eight.".to_string()), + reasoning_content: None, + tool_calls: vec![], + }, + LogRecord::Usage { + time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(), + cache_n: 589, + prompt_n: 40, + predicted_n: 64, + reasoning_tokens: 27, + thinking_capped: true, + }, + ]; + assert_eq!(lines.len(), want.len()); + for (i, (line, want)) in lines.iter().zip(&want).enumerate() { + let got: LogRecord = + serde_json::from_str(line).unwrap_or_else(|e| panic!("line {}: {e}", i + 1)); + assert_eq!(&got, want, "line {}", i + 1); + assert_eq!( + &serde_json::to_string(want).unwrap(), + line, + "line {}: bytes", + i + 1 + ); + } +} diff --git a/docs/plans/M5b/files/crates/bxctl/tests/chat.rs b/docs/plans/M5b/files/crates/bxctl/tests/chat.rs new file mode 100644 index 0000000..5687c38 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxctl/tests/chat.rs @@ -0,0 +1,469 @@ +//! Tests for `bxctl chat`, against a fake `loopd` that speaks the channel protocol. Do not edit. + +#[path = "support/temp.rs"] +mod temp; +use bxcore::{ + DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone, + TurnEvent, Usage, WireError, read_frame, write_frame, +}; +use bxctl::chat::{ChatError, Printer, new_session_id, run_turn}; +use std::os::unix::net::UnixListener; +use std::path::PathBuf; +use std::process::Command; +use std::sync::{Arc, Mutex}; +use std::thread; + +/// What the fake `loopd` sends back for one turn, after the events: done or an error. +#[derive(Clone)] +enum End { + Done(TurnDone), + Error(ErrorCode, &'static str), +} + +struct FakeLoopd { + socket: PathBuf, + turns: Arc>>, + _dir: temp::TempDir, +} + +fn usage() -> Usage { + Usage { + cache_n: 10, + prompt_n: 5, + predicted_n: 7, + reasoning_tokens: 3, + thinking_capped: false, + } +} + +/// Serves every connection with the same script: the given events, then `end`. Records the +/// turns it received. +fn fake_loopd(events: Vec, end: End) -> FakeLoopd { + let dir = temp::TempDir::new("bxctl-test"); + let socket = dir.join("loop.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let turns = Arc::new(Mutex::new(Vec::new())); + let seen = Arc::clone(&turns); + thread::spawn(move || { + for stream in listener.incoming() { + let mut stream = stream.unwrap(); + let request = read_frame(&mut stream).unwrap(); + let Message::Turn(turn) = request.msg else { + panic!("not a turn") + }; + seen.lock().unwrap().push(turn.clone()); + let id = request.id; + let end = if turn.resume && turn.content == "trigger-no-such-session" { + End::Error(ErrorCode::NoSuchSession, "session x does not exist") + } else if turn.content == "trigger-turn-limit" { + End::Error(ErrorCode::TurnLimit, "the turn hit a limit") + } else { + end.clone() + }; + for e in &events { + write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id, + r#final: false, + msg: Message::TurnEvent(e.clone()), + }, + ) + .unwrap(); + } + let last = match end { + End::Done(done) => Message::TurnDone(done), + End::Error(code, detail) => Message::Error(WireError { + code, + detail: detail.to_string(), + }), + }; + write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id, + r#final: true, + msg: last, + }, + ) + .unwrap(); + } + }); + FakeLoopd { + socket, + turns, + _dir: dir, + } +} + +fn events() -> Vec { + vec![ + TurnEvent::Queued { ahead: 1 }, + TurnEvent::Waiting { slot_busy: true }, + TurnEvent::Progress { + total: 100, + cache: 50, + processed: 75, + }, + TurnEvent::Reasoning { + text: "let me ".to_string(), + }, + TurnEvent::Reasoning { + text: "think".to_string(), + }, + TurnEvent::ToolCallStarted { + name: "clock".to_string(), + }, + TurnEvent::ToolResult { + name: "clock".to_string(), + class: DataClass::Public, + truncated: true, + }, + TurnEvent::ThinkingCapped { tokens: 4096 }, + TurnEvent::Retrying { + attempt: 2, + after_ms: 1500, + error: "the server went silent".to_string(), + }, + TurnEvent::CacheLoss { + expected: 500, + got: 20, + }, + TurnEvent::Content { + text: "It is ".to_string(), + }, + TurnEvent::Content { + text: "noon.".to_string(), + }, + ] +} + +fn done() -> End { + End::Done(TurnDone { + content: "It is noon.".to_string(), + usage: usage(), + untrusted: false, + }) +} + +fn id(s: &str) -> SessionId { + SessionId::new(s).unwrap() +} + +#[test] +fn run_turn_delivers_every_event_in_order_then_the_answer() { + let fake = fake_loopd(events(), done()); + let mut seen = Vec::new(); + let done = run_turn( + &fake.socket, + &id("s1"), + "what time is it?", + false, + &mut |e| seen.push(e.clone()), + ) + .unwrap(); + assert_eq!(seen, events()); + assert_eq!(done.content, "It is noon."); + assert_eq!(done.usage, usage()); + let turns = fake.turns.lock().unwrap(); + assert_eq!(turns.len(), 1); + assert_eq!( + turns[0], + Turn { + session: id("s1"), + content: "what time is it?".to_string(), + resume: false + } + ); +} + +#[test] +fn an_error_frame_is_refused_with_its_code_and_detail() { + let fake = fake_loopd( + vec![TurnEvent::Content { + text: "partial".to_string(), + }], + End::Error(ErrorCode::SessionFull, "this conversation is full"), + ); + let mut seen = 0; + let err = run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| seen += 1).unwrap_err(); + assert_eq!(seen, 1, "events before the error are still delivered"); + match err { + ChatError::Refused(w) => { + assert_eq!(w.code, ErrorCode::SessionFull); + assert_eq!(w.detail, "this conversation is full"); + } + other => panic!("{other:?}"), + } + let e: Box = + Box::new(run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| {}).unwrap_err()); + assert!(e.to_string().contains("session full"), "{e}"); +} + +#[test] +fn no_loopd_is_a_connect_error() { + let missing = std::env::temp_dir().join("bxctl-no-such-loopd.sock"); + assert!(matches!( + run_turn(&missing, &id("s1"), "x", false, &mut |_| {}), + Err(ChatError::Connect(_)) + )); +} + +#[test] +fn new_session_ids_are_valid_and_distinct() { + let a = new_session_id(); + let b = new_session_id(); + assert!(a.as_str().starts_with("chat-")); + assert_ne!(a, b); +} + +#[test] +fn the_printer_formats_each_event_kind() { + let mut out = Vec::new(); + let mut p = Printer::new(true, false); + for e in events() { + p.event(&mut out, &e).unwrap(); + } + p.end_reasoning(&mut out).unwrap(); + let text = String::from_utf8(out).unwrap(); + assert!( + text.contains("\x1b[2mlet me think\x1b[0m\n"), + "reasoning dimmed, joined, and ended once: {text:?}" + ); + assert!(text.contains("[tool clock]\n"), "{text:?}"); + assert!(text.contains("[clock: Public, truncated]\n"), "{text:?}"); + assert!(text.contains("[waiting: slot busy]\n"), "{text:?}"); + assert!( + text.contains("[retrying: attempt 2 in 1500 ms: the server went silent]\n"), + "{text:?}" + ); + assert!( + text.contains("[thinking capped at 4096 tokens]\n"), + "{text:?}" + ); + assert!(text.contains("[cache loss: 20 of 500]\n"), "{text:?}"); + assert!( + text.ends_with("It is noon."), + "content streams as it is: {text:?}" + ); + assert!( + !text.contains("Queued") && !text.contains("Progress"), + "queued and progress are silent: {text:?}" + ); + + let mut out = Vec::new(); + let mut p = Printer::new(false, false); + for e in events() { + p.event(&mut out, &e).unwrap(); + } + let text = String::from_utf8(out).unwrap(); + assert!( + !text.contains("let me"), + "--no-thinking hides reasoning: {text:?}" + ); + assert!( + !text.contains("\x1b["), + "and no escape codes are left: {text:?}" + ); + + let mut out = Vec::new(); + let mut p = Printer::new(true, true); + for e in events() { + p.event(&mut out, &e).unwrap(); + } + let text = String::from_utf8(out).unwrap(); + let lines: Vec<&str> = text.lines().collect(); + assert_eq!( + lines.len(), + events().len(), + "--json: one line per event, none skipped" + ); + let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(first, serde_json::json!({"event": "queued", "ahead": 1})); +} + +#[test] +fn say_prints_only_the_answer_on_stdout() { + let fake = fake_loopd(events(), done()); + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .args(["--session", "scripted-1", "--say", "what time is it?"]) + .output() + .unwrap(); + assert!(output.status.success()); + assert_eq!(String::from_utf8_lossy(&output.stdout), "It is noon.\n"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("[tool clock]"), + "events go to stderr: {stderr}" + ); + let turns = fake.turns.lock().unwrap(); + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].session, id("scripted-1")); + assert!( + turns[0].resume, + "a session given on the command line is resumed" + ); +} + +#[test] +fn say_creates_a_named_session_that_does_not_exist_yet() { + let fake = fake_loopd(vec![], done()); + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .args(["--session", "fresh", "--say", "trigger-no-such-session"]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let turns = fake.turns.lock().unwrap(); + assert_eq!(turns.len(), 2, "resume was refused, so it was created"); + assert!(turns[0].resume && !turns[1].resume); +} + +#[test] +fn say_reports_an_error_on_stderr_with_status_1() { + let fake = fake_loopd( + vec![], + End::Error(ErrorCode::TurnLimit, "the turn hit a limit"), + ); + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .args(["--say", "x"]) + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(1)); + assert_eq!(String::from_utf8_lossy(&output.stdout), ""); + assert!(String::from_utf8_lossy(&output.stderr).contains("turn limit: the turn hit a limit")); +} + +#[test] +fn json_mode_prints_frames_as_json_lines() { + let fake = fake_loopd(events(), done()); + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .args(["--session", "j", "--json", "--say", "x"]) + .output() + .unwrap(); + assert!(output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + let lines: Vec<&str> = stderr.lines().collect(); + assert_eq!( + lines.len(), + events().len() + 1, + "every event, then the done frame: {stderr}" + ); + let last: serde_json::Value = serde_json::from_str(lines[lines.len() - 1]).unwrap(); + assert_eq!(last["content"], "It is noon."); + assert_eq!(last["usage"]["cache_n"], 10); + assert_eq!(last["untrusted"], false, "the done frame is printed whole"); +} + +#[test] +fn interactive_mode_reads_lines_until_quit() { + let fake = fake_loopd( + vec![TurnEvent::Content { + text: "ok".to_string(), + }], + End::Done(TurnDone { + content: "ok".to_string(), + usage: usage(), + untrusted: false, + }), + ); + let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap(); + { + let mut stdin = child.stdin.take().unwrap(); + std::io::Write::write_all(&mut stdin, b"first\n\nsecond\n/quit\nnever sent\n").unwrap(); + } + let output = child.wait_with_output().unwrap(); + assert!(output.status.success()); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.starts_with("session chat-"), "{stdout}"); + assert_eq!(stdout.matches("ok").count(), 2, "{stdout}"); + let turns = fake.turns.lock().unwrap(); + assert_eq!( + turns.len(), + 2, + "a blank line sends nothing, and /quit stops" + ); + assert!( + !turns[0].resume && turns[1].resume, + "the first turn creates, the second resumes" + ); + assert_eq!(turns[0].session, turns[1].session); +} + +#[test] +fn bad_arguments_print_usage() { + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--session", "Not Valid!"]) + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert!(String::from_utf8_lossy(&output.stderr).contains("usage")); + let output = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["dance"]) + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); +} + +/// A failed turn is reported, and the conversation goes on: the session still exists and the +/// next line is a new turn on it. +#[test] +fn interactive_mode_survives_a_failed_turn() { + let fake = fake_loopd( + vec![TurnEvent::Content { + text: "ok".to_string(), + }], + End::Done(TurnDone { + content: "ok".to_string(), + usage: usage(), + untrusted: false, + }), + ); + let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl")) + .args(["chat", "--socket"]) + .arg(&fake.socket) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap(); + { + let mut stdin = child.stdin.take().unwrap(); + std::io::Write::write_all(&mut stdin, b"first\ntrigger-turn-limit\nthird\n/quit\n") + .unwrap(); + } + let output = child.wait_with_output().unwrap(); + assert!( + output.status.success(), + "a failed turn does not end the chat" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("turn limit: the turn hit a limit"), + "{stderr}" + ); + let turns = fake.turns.lock().unwrap(); + assert_eq!(turns.len(), 3, "the turn after the failure was sent"); + assert!(turns[2].resume, "and it resumed the same session"); + assert_eq!(turns[2].session, turns[0].session); +} diff --git a/docs/plans/M5b/files/crates/bxctl/tests/support/mod.rs b/docs/plans/M5b/files/crates/bxctl/tests/support/mod.rs new file mode 100644 index 0000000..5fb8e66 --- /dev/null +++ b/docs/plans/M5b/files/crates/bxctl/tests/support/mod.rs @@ -0,0 +1,213 @@ +//! Scripted stand-ins for `brokerd`'s admin socket and for `loopd`, for tests. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use bxcore::{ + ApprovalList, CallId, DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, + PendingApproval, SessionId, Timestamp, TurnDone, TurnEvent, Usage, WireError, read_frame, + write_frame, +}; +use std::os::unix::net::UnixListener; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; + +static NEXT: AtomicU32 = AtomicU32::new(0); + +/// A socket path in a fresh temporary directory, which is removed when the value is dropped +/// (tip T44). +pub struct TempSocket { + dir: PathBuf, + path: PathBuf, +} + +impl std::ops::Deref for TempSocket { + type Target = std::path::Path; + fn deref(&self) -> &std::path::Path { + &self.path + } +} + +impl AsRef for TempSocket { + fn as_ref(&self) -> &std::path::Path { + &self.path + } +} + +impl Drop for TempSocket { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.dir); + } +} + +pub fn temp_socket(name: &str) -> TempSocket { + let n = NEXT.fetch_add(1, Ordering::SeqCst); + let dir = std::env::temp_dir().join(format!("bxctl-m3a-{}-{n}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join(name); + TempSocket { dir, path } +} + +pub fn ts(text: &str) -> Timestamp { + Timestamp::parse(text).unwrap() +} + +pub struct FakeBrokerd { + pub socket: TempSocket, + /// Every request message received, in order. + pub requests: Arc>>, +} + +impl FakeBrokerd { + pub fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } +} + +/// Serves `admin.sock`: one request per connection, answered with the frames `answer` returns, +/// written exactly as given (so a test can send a wrong id or a frame that is not final). An +/// empty list closes the connection without an answer. +pub fn fake_brokerd_frames( + answer: impl Fn(&Envelope) -> Vec + Send + 'static, +) -> FakeBrokerd { + let socket = temp_socket("admin.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let seen = Arc::clone(&requests); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let Ok(request) = read_frame(&mut stream) else { + continue; + }; + seen.lock().unwrap().push(request.msg.clone()); + for frame in answer(&request) { + if write_frame(&mut stream, &frame).is_err() { + break; // the client went away; the next connection is still served + } + } + } + }); + FakeBrokerd { socket, requests } +} + +/// The usual case: one final frame with the request's id. +pub fn fake_brokerd(answer: impl Fn(&Message) -> Message + Send + 'static) -> FakeBrokerd { + fake_brokerd_frames(move |request| { + vec![Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: true, + msg: answer(&request.msg), + }] + }) +} + +pub fn wire_error(code: ErrorCode, detail: &str) -> Message { + Message::Error(WireError { + code, + detail: detail.to_string(), + }) +} + +/// A pending approval created at 12:00:00 that expires at 12:15:00 on 2026-09-18. +pub fn pending(approval: u64, tool: &str, arguments: &str) -> PendingApproval { + PendingApproval { + approval, + session: SessionId::new("chat-1758196800-123456789").unwrap(), + call: CallId(7), + tool: tool.to_string(), + arguments: arguments.to_string(), + grant: "shell-scratch".to_string(), + taint: DataClass::Private, + created: ts("2026-09-18T12:00:00.000Z"), + expires: ts("2026-09-18T12:15:00.000Z"), + } +} + +/// A `brokerd` with these approvals pending. `approve` is answered with `outcome`, `refuse` with +/// `ok`; both answer `no_such_approval` for an id that is not in the list. +pub fn brokerd_with(items: Vec, outcome: bxcore::DecisionRecord) -> FakeBrokerd { + fake_brokerd(move |msg| { + let known = |id: u64| items.iter().any(|item| item.approval == id); + match msg { + Message::Approvals(_) => Message::ApprovalList(ApprovalList { + items: items.clone(), + }), + Message::Approve(a) if known(a.approval) => { + Message::ApproveResult(bxcore::ApproveResult { + outcome: outcome.clone(), + }) + } + Message::Refuse(r) if known(r.approval) => Message::Ok(bxcore::Empty {}), + Message::Approve(_) | Message::Refuse(_) => { + wire_error(ErrorCode::NoSuchApproval, "no such approval") + } + _ => wire_error(ErrorCode::BadMessage, "not an admin request"), + } + }) +} + +pub struct FakeLoopd { + pub socket: TempSocket, + /// The content of every turn received, in order. + pub turns: Arc>>, +} + +pub fn usage() -> Usage { + Usage { + cache_n: 10, + prompt_n: 5, + predicted_n: 7, + reasoning_tokens: 3, + thinking_capped: false, + } +} + +/// Serves `loop.sock`: every turn gets the same events, then `turn_done` with `answer`. It does +/// not wait for an approval to be answered: `bxctl` deals with each event before it reads the +/// next frame, so the order of what it does is fixed all the same. +pub fn fake_loopd(events: Vec, answer: &str) -> FakeLoopd { + let socket = temp_socket("loop.sock"); + let listener = UnixListener::bind(&socket).unwrap(); + let turns = Arc::new(Mutex::new(Vec::new())); + let seen = Arc::clone(&turns); + let answer = answer.to_string(); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let Ok(request) = read_frame(&mut stream) else { + continue; + }; + let Message::Turn(turn) = request.msg else { + continue; + }; + seen.lock().unwrap().push(turn.content); + let mut frames: Vec<(bool, Message)> = events + .iter() + .map(|e| (false, Message::TurnEvent(e.clone()))) + .collect(); + frames.push(( + true, + Message::TurnDone(TurnDone { + content: answer.clone(), + usage: usage(), + untrusted: false, + }), + )); + for (last, msg) in frames { + let frame = Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: last, + msg, + }; + if write_frame(&mut stream, &frame).is_err() { + break; + } + } + } + }); + FakeLoopd { socket, turns } +} diff --git a/docs/plans/M5b/files/crates/gatewayd/tests/support/fake_loop.rs b/docs/plans/M5b/files/crates/gatewayd/tests/support/fake_loop.rs new file mode 100644 index 0000000..3c4abc5 --- /dev/null +++ b/docs/plans/M5b/files/crates/gatewayd/tests/support/fake_loop.rs @@ -0,0 +1,102 @@ +//! A fake `loopd` on a Unix socket: it records each turn and answers with the frames the test's +//! function gives for it. Do not edit. + +#![allow(dead_code)] // each test file uses a different part of this module + +use std::io::Write; +use std::os::unix::net::UnixListener; +use std::path::Path; +use std::sync::mpsc; + +use bxcore::{ + Envelope, ErrorCode, Message, PROTOCOL_VERSION, Turn, TurnDone, TurnEvent, Usage, WireError, + read_frame, write_frame, +}; + +const USAGE: Usage = Usage { + cache_n: 0, + prompt_n: 1, + predicted_n: 1, + reasoning_tokens: 0, + thinking_capped: false, +}; + +/// What the fake sends back for a turn. +pub enum Reply { + Frame(Envelope), + /// Raw bytes, for broken frames. + Bytes(Vec), + /// Stop answering this connection (the caller sees it close). + Close, +} + +pub fn event(e: TurnEvent) -> Reply { + Reply::Frame(Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: false, + msg: Message::TurnEvent(e), + }) +} + +pub fn done(content: &str) -> Reply { + let msg = Message::TurnDone(TurnDone { + content: content.to_string(), + usage: USAGE, + untrusted: false, + }); + Reply::Frame(Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg, + }) +} + +pub fn error(code: ErrorCode, detail: &str) -> Reply { + let msg = Message::Error(WireError { + code, + detail: detail.to_string(), + }); + Reply::Frame(Envelope { + v: PROTOCOL_VERSION, + id: 1, + r#final: true, + msg, + }) +} + +/// Listen on `socket`; `script(n, &turn)` gives the replies to the n-th turn (from 0). +pub fn serve_loop(socket: &Path, script: F) -> mpsc::Receiver +where + F: Fn(usize, &Turn) -> Vec + Send + 'static, +{ + let listener = UnixListener::bind(socket).unwrap(); + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + for (n, stream) in listener.incoming().enumerate() { + let Ok(mut stream) = stream else { continue }; + let Ok(Envelope { + msg: Message::Turn(turn), + .. + }) = read_frame(&mut stream) + else { + continue; + }; + // Reported before the script runs: a script may wait for the test. + let _ = tx.send(turn.clone()); + let replies = script(n, &turn); + for reply in replies { + let ok = match reply { + Reply::Frame(env) => write_frame(&mut stream, &env).is_ok(), + Reply::Bytes(b) => stream.write_all(&b).is_ok(), + Reply::Close => false, + }; + if !ok { + break; + } + } + } + }); + rx +} diff --git a/docs/plans/M5b/files/crates/loopd/tests/turn_done_untrusted.rs b/docs/plans/M5b/files/crates/loopd/tests/turn_done_untrusted.rs new file mode 100644 index 0000000..b26e342 --- /dev/null +++ b/docs/plans/M5b/files/crates/loopd/tests/turn_done_untrusted.rs @@ -0,0 +1,219 @@ +//! `TurnDone.untrusted` follows the session's labels (M5b spec, sections 8 and 9): false while +//! every tool result in the session's log is labelled trusted, and true from the turn in which one +//! is labelled untrusted (a fetched page, or a recalled memory-tainted note) and in every turn +//! after it, also after `loopd` restarts. A denial or a failure is never untrusted. 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::Arc; +use std::thread; +use std::time::Duration; + +use bxcore::{ + DataClass, DenyReason, Envelope, Message, PROTOCOL_VERSION, SessionId, ToolResponse, Turn, + read_frame, write_frame, +}; +use loopd::channel::{Context, serve}; +use loopd::llama::Client; +use loopd::memory::NoMemory; +use loopd::tools::Registry; +use support::{FakeServer, Home, Reply, ScriptedPort}; +use tags::{call, stream}; + +const CHAT: &str = "/v1/chat/completions"; + +/// 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 and + /// tool port, nothing carried over but the files. `name` names its socket; the port answers + /// the tool calls with `replies`, in order. + fn loopd(&self, name: &str, replies: Vec) -> PathBuf { + 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 ctx = Arc::new(Context::new( + cfg.clone(), + Client::new(cfg), + Box::new(ScriptedPort::new(replies)), + Registry::m2b(), + Box::new(NoMemory), + )); + thread::spawn(move || serve(listener, ctx)); + socket + } +} + +/// Sends one turn and returns its `TurnDone`'s `untrusted`. Any other final frame fails the test. +fn untrusted(socket: &Path, session: &str, content: &str, resume: bool) -> 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: content.to_string(), + resume, + }); + let turn = Envelope { + v: PROTOCOL_VERSION, + id: 7, + r#final: true, + msg, + }; + write_frame(&mut stream, &turn).unwrap(); + loop { + let frame = read_frame(&mut stream).unwrap(); + if !frame.r#final { + continue; + } + match frame.msg { + Message::TurnDone(done) => return done.untrusted, + other => panic!("the turn did not end with turn_done: {other:?}"), + } + } +} + +/// A result as `brokerd` answers it, with its label. +fn result(content: &str, untrusted: bool) -> ToolResponse { + ToolResponse::Result { + content: content.to_string(), + class: DataClass::Private, + untrusted, + truncated: false, + } +} + +/// The model calls `recall`; the fake port stands for `brokerd` and `memoryd` behind it. +fn recall() -> Reply { + stream( + &[call("call-r", "recall", r#"{"query":"deploy"}"#)], + "tool_calls", + ) +} + +#[test] +fn a_clean_session_s_turns_carry_false() { + let s = Setup::new(vec![Reply::fixture("turn1"), Reply::fixture("turn2")]); + let socket = s.loopd("loop.sock", vec![]); + assert!(!untrusted(&socket, "a", "hello", false), "turn 1"); + assert!(!untrusted(&socket, "a", "and again", true), "turn 2"); +} + +#[test] +fn trusted_results_denials_and_failures_keep_it_false() { + let s = Setup::new(vec![ + Reply::fixture("tool_call"), + Reply::fixture("plain"), + Reply::fixture("tool_call"), + Reply::fixture("plain"), + Reply::fixture("tool_call"), + Reply::fixture("plain"), + ]); + let socket = s.loopd( + "loop.sock", + vec![ + result("straylight\n", false), + ToolResponse::Denied { + reason: DenyReason::NoGrant, + detail: None, + }, + ToolResponse::Failed { + message: "the container could not start".to_string(), + }, + ], + ); + assert!( + !untrusted(&socket, "a", "hostname?", false), + "a trusted result" + ); + assert!(!untrusted(&socket, "a", "again?", true), "a denial"); + assert!(!untrusted(&socket, "a", "once more?", true), "a failure"); +} + +#[test] +fn an_untrusted_result_makes_its_turn_and_every_later_one_true() { + let s = Setup::new(vec![ + Reply::fixture("plain"), + Reply::fixture("tool_call"), + Reply::fixture("plain"), + Reply::fixture("tool_call"), + Reply::fixture("plain"), + ]); + let socket = s.loopd( + "loop.sock", + vec![ + result("a fetched page\n", true), + result("straylight\n", false), + ], + ); + assert!(!untrusted(&socket, "a", "hello", false), "before the page"); + assert!(untrusted(&socket, "a", "read it", true), "the page's turn"); + assert!( + untrusted(&socket, "a", "hostname?", true), + "a later turn whose own result is trusted" + ); + assert!(untrusted(&socket, "a", "thanks", true), "a later turn"); + assert!( + !untrusted(&socket, "b", "hello", false), + "another session is not touched" + ); +} + +#[test] +fn a_recalled_memory_tainted_note_makes_it_true() { + let s = Setup::new(vec![recall(), Reply::fixture("plain")]); + let hit = "[1] memory-tainted · note, 2026-09-25 14:02, session chat-7 · private\n\ + the deploy runs from ci.example.org"; + let socket = s.loopd("loop.sock", vec![result(hit, true)]); + assert!( + untrusted(&socket, "a", "how do we deploy?", false), + "the recall's turn" + ); + assert!(untrusted(&socket, "a", "thanks", true), "a later turn"); +} + +#[test] +fn a_recall_of_trusted_memory_keeps_it_false() { + let s = Setup::new(vec![recall(), Reply::fixture("plain")]); + let hit = "[1] identity · moved 2026-09-25 from memory.md · private\n\ + deploys go through ci.example.org"; + let socket = s.loopd("loop.sock", vec![result(hit, false)]); + assert!(!untrusted(&socket, "a", "how do we deploy?", false)); +} + +#[test] +fn a_restarted_loopd_reads_it_from_the_log() { + let s = Setup::new(vec![Reply::fixture("tool_call"), Reply::fixture("plain")]); + let first = s.loopd("first.sock", vec![result("a fetched page\n", true)]); + assert!(untrusted(&first, "a", "read it", false), "the first loopd"); + let second = s.loopd("second.sock", vec![]); + assert!( + untrusted(&second, "a", "thanks", true), + "the second loopd, on the same session" + ); + assert!( + !untrusted(&second, "b", "hello", false), + "the second loopd, on a new session" + ); +}