M5b plan: task 01, the invalid-arguments detail

Task file and 13 given files (2 new: the wire fixture with a detail and
brokerd/tests/invalid_detail.rs; 11 new versions of existing tests, most
only adding `detail: None`).

Checked against a reference in a scratch worktree with its own target/,
on titan (rustc 1.98.1, clippy 0.1.98): make gate prints gate: ok with
956 tests (944 + 12 new); /tmp has no new entries after the gate. All 21
quoted before-blocks occur once in master, and applying the edits to
master gives the reference byte for byte. The given files are the
reference's, formatted, none over 500 lines; no chat-template tag in
the plan.

Before the edits, strict's envelopes_reject_unknown_keys_at_every_depth
fails and ten test targets do not compile, as step 2 lists.

Mutations: 21 one-line breaks of the reference; 20 caught by the given
tests. The one not caught (dropping `default` from the detail attribute)
cannot change behaviour: serde reads a missing Option field as None.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-26 02:35:35 -07:00
co-authored by Claude Opus 5.5
parent 56b969d55b
commit fa5bcae503
14 changed files with 4338 additions and 0 deletions
@@ -0,0 +1,551 @@
# M5b task 01: an invalid call is denied with its reason
**Branch:** `m5b` (run `git switch m5b`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `bxcore, brokerd, loopd: an invalid-arguments denial says why`
## Goal
`brokerd` refuses a tool call whose arguments do not parse (`invalid_arguments`). Its parser
explains why (`ArgsError`), but `policy::decide` throws the explanation away, so the model reads
only a fixed sentence and has to guess. On the device (2026-09-25) it spent two calls finding that
`note` takes no `file`. After this task the explanation reaches the model (M5b spec, section 4):
| The model's call to `memory` | What it reads back |
|---|---|
| `{"action":"note","file":"notes.md","text":"x"}`, before | `Denied: the arguments are not valid for this tool.` |
| the same, after | ``Denied: the arguments are not valid for this tool: invalid argument shape: unknown field `file`, expected `text`.`` |
Four parts:
1. **`bxcore`**: `ToolResponse::Denied` gains a field `detail: Option<String>`, written only when
it is `Some`. A denial without a detail has exactly the bytes it had before, so every existing
fixture stays valid. One new fixture has a detail.
2. **`brokerd`**: `ArgsError` gains `detail()`: its `Display` text on one line (each control
character, and U+2028 and U+2029, becomes a space), cut to 300 characters. `policy::Denial`
gains `detail`, set by `decide` for `InvalidArguments` only. The ledger passes the whole
`Denial` on (`Decided::Denied(Denial)`), and `broker.sock` sends its detail.
3. **`loopd`**: a new `denial_message(reason, detail)` puts the detail after the sentence, before
its full stop. Without a detail it is exactly today's sentence.
4. **Unchanged**: the audit record (`DecisionRecord::Denied { reason }` has no detail: the record
holds the arguments already), `TurnEvent::ToolDenied`, and every text `gatewayd` and `bxctl`
show the owner. Do not touch `crates/gatewayd`, `crates/bxctl` or `crates/bxcore/src/audit.rs`.
## Every place `ToolResponse::Denied` 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 enum | gains `detail` (edit 1) |
| `crates/brokerd/src/broker.rs`, `handle`, step 3a | sends the `Denial`'s detail (edit B1) |
| `crates/brokerd/src/broker.rs`, `pending`, `Verdict::Denied` and `Err(reason)` | `detail: None`: an approval's denial has no detail (edits B2, B3) |
| `crates/brokerd/src/ledger/answer.rs`, `finish` | a pattern with `..`; no change |
| `crates/loopd/src/turn.rs`, `run_call` | reads `detail` and calls `denial_message` (edit T2) |
| `crates/loopd/src/tools.rs`, `FakeTools::call` | `detail: None` (edit L2) |
| `crates/loopd/src/broker_port.rs` | a pattern with `..`; no change |
| tests in `bxcore`, `brokerd`, `loopd` | new versions are given; `brokerd/tests/broker_sequence.rs` uses `..` and is not changed |
## Files
- Copy, from `docs/plans/M5b/files/` (the same path under it):
- new: `crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json`,
`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`
- 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`,
`docs/implementer-log.md`
## Facts you need
- `serde`: `#[serde(default, skip_serializing_if = "Option::is_none")]` on a field of a variant of
this enum (internally tagged, `deny_unknown_fields`) does what it says: a missing `detail` reads
as `None`, `None` is not written, and an unknown key is still refused. Checked against every
fixture; `strict.rs` walks the new one too.
- `char::is_control(self) -> bool` is true for Unicode's control characters only: U+0000 to
U+001F and U+007F to U+009F (tab, line feed, carriage return, escape and NEL among them). It is
false for U+2028 (line separator) and U+2029 (paragraph separator), so `detail` names those two.
- The 300 is characters (`chars()`), not bytes, so a cut never splits a character.
- `str::strip_suffix(&self, '.') -> Option<&str>`; `Option::as_deref(&self) -> Option<&str>` for an
`Option<String>`.
- `Denial { detail: Some(..), ..Denial::new(reason) }` is Rust's struct update syntax: the other
fields come from `Denial::new`.
- `crates/brokerd/src/ledger/mod.rs` already imports `Denial` and `DenyReason`; keep both.
## The edits, exactly
Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file.
### `crates/bxcore/src/wire.rs`
**1.** In `enum ToolResponse`, the lines
```rust
Denied {
reason: DenyReason,
},
}
```
become
```rust
Denied {
reason: DenyReason,
/// Why the arguments were not valid, for `invalid_arguments` only. Written only when
/// present, so a denial without one has the same bytes as before M5b.
#[serde(default, skip_serializing_if = "Option::is_none")]
detail: Option<String>,
},
}
```
### `crates/brokerd/src/args.rs`
**A1.** The lines
```rust
/// Maximum length, in bytes, of a URL.
pub const MAX_URL: usize = 2048;
```
become
```rust
/// Maximum length, in bytes, of a URL.
pub const MAX_URL: usize = 2048;
/// Maximum length, in characters, of the detail an invalid call is denied with.
pub const MAX_DETAIL: usize = 300;
```
**A2.** The line
```rust
impl std::error::Error for ArgsError {
```
becomes
```rust
impl ArgsError {
/// The error as the model is shown it: the `Display` text on one line, each control character
/// and each Unicode line or paragraph separator replaced by a space, cut to `MAX_DETAIL`
/// characters. The text holds the model's own arguments, which may be long or hold newlines.
pub fn detail(&self) -> String {
self.to_string()
.chars()
.map(|c| {
if c.is_control() || c == '\u{2028}' || c == '\u{2029}' {
' '
} else {
c
}
})
.take(MAX_DETAIL)
.collect()
}
}
impl std::error::Error for ArgsError {
```
### `crates/brokerd/src/policy/mod.rs`
**P1.** In `struct Denial`, the lines
```rust
/// `Some` only for `DeniedByGrant`: the deny grant's file hash.
pub grant_sha256: Option<Hash32>,
}
```
become
```rust
/// `Some` only for `DeniedByGrant`: the deny grant's file hash.
pub grant_sha256: Option<Hash32>,
/// `Some` only for `InvalidArguments`: why, from `ArgsError::detail`. It goes to the model in
/// the `ToolResponse`, never into the audit record, which holds the arguments already.
pub detail: Option<String>,
}
```
**P2.** In `Denial::new`, the lines
```rust
grant: None,
grant_sha256: None,
}
```
become
```rust
grant: None,
grant_sha256: None,
detail: None,
}
```
**P3.** In `decide`, the lines
```rust
// 2. Malformed arguments are refused before matching, whatever the grants say.
let Ok(args) = args::parse(tool, &request.arguments) else {
return Outcome::Denied(Denial::new(DenyReason::InvalidArguments));
};
```
become
```rust
// 2. Malformed arguments are refused before matching, whatever the grants say, and the
// model is told why, so that it can correct the call.
let args = match args::parse(tool, &request.arguments) {
Ok(args) => args,
Err(error) => {
return Outcome::Denied(Denial {
detail: Some(error.detail()),
..Denial::new(DenyReason::InvalidArguments)
});
}
};
```
**P4.** In `decide`, the lines
```rust
Mode::Deny => Outcome::Denied(Denial {
reason: DenyReason::DeniedByGrant,
grant: Some(winner.id.clone()),
grant_sha256: Some(winner.grant_sha256),
}),
```
become
```rust
Mode::Deny => Outcome::Denied(Denial {
reason: DenyReason::DeniedByGrant,
grant: Some(winner.id.clone()),
grant_sha256: Some(winner.grant_sha256),
detail: None,
}),
```
**P5.** In `redecide`, the lines
```rust
Mode::Deny => Err(Denial {
reason: DenyReason::DeniedByGrant,
grant: Some(winner.id.clone()),
grant_sha256: Some(winner.grant_sha256),
}),
```
become
```rust
Mode::Deny => Err(Denial {
reason: DenyReason::DeniedByGrant,
grant: Some(winner.id.clone()),
grant_sha256: Some(winner.grant_sha256),
detail: None,
}),
```
### `crates/brokerd/src/ledger/mod.rs`
**D1.** The lines
```rust
/// How a call was decided.
#[derive(Debug)]
pub enum Decided {
```
become
```rust
/// How a call was decided. A denial carries the policy's `Denial`, so an invalid call keeps its
/// detail on the way to the answer.
#[derive(Debug)]
pub enum Decided {
```
**D2.** At the end of `enum Decided`, the lines
```rust
Denied(DenyReason),
}
```
become
```rust
Denied(Denial),
}
```
**D3.** In `Ledger::decide`, the line
```rust
None => return Decided::Denied(DenyReason::AuditUnavailable),
```
becomes
```rust
None => return Decided::Denied(Denial::new(DenyReason::AuditUnavailable)),
```
**D4.** The line
```rust
RecordPayload::Simple(reason) => Decided::Denied(reason),
```
becomes
```rust
RecordPayload::Simple(reason) => Decided::Denied(Denial::new(reason)),
```
**D5.** The line
```rust
RecordPayload::Denial(denial) => Decided::Denied(denial.reason),
```
becomes
```rust
RecordPayload::Denial(denial) => Decided::Denied(denial),
```
**D6.** The line
```rust
Err(()) => Decided::Denied(DenyReason::AuditUnavailable),
```
becomes
```rust
Err(()) => Decided::Denied(Denial::new(DenyReason::AuditUnavailable)),
```
### `crates/brokerd/src/broker.rs`
**B1.** In `handle`, the line
```rust
Decided::Denied(reason) => Some(ToolResponse::Denied { reason }), // 3a. denied
```
becomes
```rust
Decided::Denied(denial) => Some(ToolResponse::Denied {
reason: denial.reason,
detail: denial.detail,
}), // 3a. denied
```
**B2.** In `pending`, the line
```rust
Verdict::Denied(reason) => Some(ToolResponse::Denied { reason }),
```
becomes
```rust
Verdict::Denied(reason) => Some(ToolResponse::Denied {
reason,
detail: None,
}),
```
**B3.** At the end of `pending`, the line
```rust
Err(reason) => Some(ToolResponse::Denied { reason }),
```
becomes
```rust
Err(reason) => Some(ToolResponse::Denied {
reason,
detail: None,
}),
```
### `crates/loopd/src/tools.rs`
**L1.** The line
```rust
pub struct FakeTools {
```
becomes
```rust
/// What a denial reaches the model as: the reason's sentence, and the broker's detail, if it sent
/// one, after a colon (`Denied: the arguments are not valid for this tool: <detail>.`).
pub fn denial_message(reason: bxcore::DenyReason, detail: Option<&str>) -> String {
let sentence = denial_text(reason);
match detail {
None => sentence.to_string(),
Some(detail) => {
let stem = sentence.strip_suffix('.').unwrap_or(sentence);
format!("{stem}: {detail}.")
}
}
}
pub struct FakeTools {
```
`denial_text` itself does not change.
**L2.** In `FakeTools::call`, the lines
```rust
_ => bxcore::ToolResponse::Denied {
reason: bxcore::DenyReason::NoGrant,
},
```
become
```rust
_ => bxcore::ToolResponse::Denied {
reason: bxcore::DenyReason::NoGrant,
detail: None,
},
```
### `crates/loopd/src/turn.rs`
**T1.** The line
```rust
use crate::tools::{Registry, ToolPort, cap_result, denial_text, dispatch};
```
becomes
```rust
use crate::tools::{Registry, ToolPort, cap_result, denial_message, dispatch};
```
**T2.** In `run_call`, the lines
```rust
ToolResponse::Denied { reason } => {
on_event(&TurnEvent::ToolDenied {
name: request.tool.clone(),
reason,
});
(denial_text(reason).to_string(), DataClass::Public, false)
}
```
become
```rust
ToolResponse::Denied { reason, detail } => {
on_event(&TurnEvent::ToolDenied {
name: request.tool.clone(),
reason,
});
(
denial_message(reason, detail.as_deref()),
DataClass::Public,
false,
)
}
```
The event keeps only the reason: the owner's texts do not change.
## Steps
- [ ] **1. Copy the given files.** From the repository root, thirteen commands:
```sh
cp docs/plans/M5b/files/crates/bxcore/tests/fixtures/wire/tool_response_denied_detail.json \
crates/bxcore/tests/fixtures/wire/
cp docs/plans/M5b/files/crates/bxcore/tests/wire.rs \
crates/bxcore/tests/
cp docs/plans/M5b/files/crates/bxcore/tests/strict.rs \
crates/bxcore/tests/
cp docs/plans/M5b/files/crates/brokerd/tests/invalid_detail.rs \
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_pending.rs \
crates/brokerd/tests/
cp docs/plans/M5b/files/crates/brokerd/tests/courier.rs \
crates/brokerd/tests/
cp docs/plans/M5b/files/crates/brokerd/tests/ledger.rs \
crates/brokerd/tests/
cp docs/plans/M5b/files/crates/brokerd/tests/serve.rs \
crates/brokerd/tests/
cp docs/plans/M5b/files/crates/loopd/tests/broker_port.rs \
crates/loopd/tests/
cp docs/plans/M5b/files/crates/loopd/tests/tools.rs \
crates/loopd/tests/
cp docs/plans/M5b/files/crates/loopd/tests/turn_broker.rs \
crates/loopd/tests/
```
Then `git status --short` lists 11 modified files and 2 new ones, all under `crates/*/tests/`.
- [ ] **2. See them fail.**
- `cargo test -p bxcore --test strict`: one test fails,
`envelopes_reject_unknown_keys_at_every_depth`, with
`tool_response_denied_detail.json: fixture must decode`.
- `cargo test -p bxcore --test wire`: does not compile, with two errors `E0559`
(``variant `bxcore::ToolResponse::Denied` has no field named `detail` ``).
- `cargo build --tests --keep-going -p brokerd -p loopd`: ten test targets do not compile:
`admin`, `broker`, `broker_pending`, `courier`, `invalid_detail`, `ledger` and `serve` in
`brokerd`; `broker_port`, `tools` and `turn_broker` in `loopd`. Every error is about a name
this task adds: `E0559` (no field named `detail` on `ToolResponse::Denied`), `E0609` (no field
`detail` on `Denial`), `E0599` (no method named `detail` for `ArgsError`), `E0432`
(unresolved import `MAX_DETAIL` or `denial_message`), or `E0308` (mismatched types, in
`ledger.rs`, where `Decided::Denied` will hold a `Denial`).
- [ ] **3. `bxcore`.** Make edit 1. `cargo test -p bxcore`: every suite passes.
- [ ] **4. `brokerd`.** Make edits A1, A2, P1 to P5, D1 to D6, and B1 to B3.
`cargo test -p brokerd`: every suite passes, among them the eight tests of `invalid_detail`.
- [ ] **5. `loopd`.** Make edits L1, L2, T1 and T2. `cargo test -p loopd`: every suite passes.
- [ ] **6. Check.** `grep -rn "Decided::Denied(DenyReason" crates/` prints nothing, and
`git diff --stat -- crates/gatewayd crates/bxctl crates/bxcore/src/audit.rs` prints nothing.
- [ ] **7. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms.
Expected last line: `gate: ok`, with 956 tests: 944 before this task, plus 12 new ones (two in
`bxcore/tests/wire.rs`, eight in `brokerd/tests/invalid_detail.rs`, one in
`loopd/tests/tools.rs`, one in `loopd/tests/turn_broker.rs`).
- [ ] **8. Log and commit.**
`git add crates/bxcore crates/brokerd 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 956 tests.
- `grep -rn "Decided::Denied(DenyReason" crates/` prints nothing.
## Stop and report if
- Any quoted "before" block 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`, or an error of another kind.
- After step 3, 4 or 5, a test fails. Do not change a test or a fixture.
- A test in `gatewayd`, `bxctl`, `memoryd` 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 956.
- A comment in an edit does not fit the code under it. Quote it and stop.
@@ -0,0 +1,388 @@
//! One `admin.sock` connection: list, approve, refuse, check the grants; and expiry, which
//! answers an approval the same way. 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::os::unix::net::UnixStream;
use std::sync::{Arc, Barrier};
use std::time::{Duration, Instant};
use brokerd::admin;
use brokerd::broker::{self, Broker};
use brokerd::policy::Label;
use bxcore::{
ApprovalAnswer, Approve, AuditEvent, DataClass, DecisionRecord, DenyReason, Empty, ErrorCode,
Message, Refuse, SessionId, Timestamp, ToolResponse,
};
use client::{Serve, next, open};
use rig::{Rig, grant_text, request};
use runtime::Recording;
fn ask_notes(rig: &Rig) {
rig.grant(
"notes",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
}
/// Sends a call that waits for approval; returns its connection once the pending frame is read.
fn pending(broker: &Arc<Broker>, call: u64, arguments: &str) -> (UnixStream, u64) {
let req = request("s1", call, "read_file", arguments);
let mut stream = open(broker, broker::handle, call, Message::ToolRequest(req));
match next(&mut stream).msg {
Message::ToolResponse(ToolResponse::PendingApproval { approval, .. }) => (stream, approval),
other => panic!("{other:?}"),
}
}
/// One admin request; returns the answer's message after checking its id and `final`.
fn admin(broker: &Arc<Broker>, msg: Message) -> Message {
let mut stream = open(broker, admin::handle, 5, msg);
let env = next(&mut stream);
assert_eq!((env.id, env.r#final), (5, true), "{env:?}");
env.msg
}
fn approve(broker: &Arc<Broker>, approval: u64) -> Message {
admin(broker, Message::Approve(Approve { approval }))
}
fn refuse(broker: &Arc<Broker>, approval: u64, reason: Option<&str>) -> Message {
let reason = reason.map(str::to_string);
admin(broker, Message::Refuse(Refuse { approval, reason }))
}
fn final_answer(stream: &mut UnixStream) -> ToolResponse {
match next(stream).msg {
Message::ToolResponse(r) => r,
other => panic!("{other:?}"),
}
}
fn outcome(msg: Message) -> DecisionRecord {
match msg {
Message::ApproveResult(r) => r.outcome,
other => panic!("{other:?}"),
}
}
fn error_code(msg: &Message) -> Option<ErrorCode> {
match msg {
Message::Error(e) => Some(e.code),
_ => None,
}
}
fn denied(reason: DenyReason) -> ToolResponse {
ToolResponse::Denied {
reason,
detail: None,
}
}
#[test]
fn approvals_lists_the_arguments_as_policy_parsed_them() {
let rig = Rig::new("admin-list");
ask_notes(&rig);
let broker = rig.broker(&Recording::answering("x"));
assert!(matches!(
admin(&broker, Message::Approvals(Empty {})),
Message::ApprovalList(list) if list.items.is_empty()
));
// A backslash-u escape of `/` is `/` in JSON; the owner sees the one spelling policy matched.
let escaped = format!(r#"{{"path":"{}n{}a"}}"#, "\\u002f", "\\u002f");
assert!(
escaped.contains("u002f"),
"the escape must survive: {escaped}"
);
let (_stream, approval) = pending(&broker, 1, &escaped);
match admin(&broker, Message::Approvals(Empty {})) {
Message::ApprovalList(list) => {
assert_eq!(list.items.len(), 1);
assert_eq!(list.items[0].approval, approval);
assert_eq!(list.items[0].arguments, r#"{"path":"/n/a"}"#);
assert_eq!(list.items[0].grant, "notes");
}
other => panic!("{other:?}"),
}
// The audit log keeps the raw string.
match &rig.events()[0] {
AuditEvent::Decision { arguments, .. } => assert_eq!(arguments, &escaped),
other => panic!("{other:?}"),
}
}
#[test]
fn approve_runs_the_call_and_answers_with_the_re_decision() {
let rig = Rig::new("admin-approve");
ask_notes(&rig);
let rt = Recording::answering("the notes");
let broker = rig.broker(&rt);
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
assert_eq!(outcome(approve(&broker, approval)), DecisionRecord::Ask {});
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert_eq!(rt.count(), 1);
match &rig.events()[1] {
AuditEvent::Approval { answer, by, .. } => {
assert_eq!(*answer, ApprovalAnswer::Approved);
assert_eq!(by.as_deref(), Some("bxctl"));
}
other => panic!("{other:?}"),
}
}
#[test]
fn refuse_denies_the_call_and_records_the_reason() {
let rig = Rig::new("admin-refuse");
ask_notes(&rig);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
assert!(matches!(
refuse(&broker, approval, Some("not today")),
Message::Ok(Empty {})
));
assert_eq!(
final_answer(&mut stream),
denied(DenyReason::ApprovalRefused)
);
assert_eq!(rt.count(), 0);
match &rig.events()[1] {
AuditEvent::Approval { answer, reason, .. } => {
assert_eq!(*answer, ApprovalAnswer::Refused);
assert_eq!(reason.as_deref(), Some("not today"));
}
other => panic!("{other:?}"),
}
}
#[test]
fn an_unknown_or_answered_id_is_no_such_approval() {
let rig = Rig::new("admin-unknown");
ask_notes(&rig);
let broker = rig.broker(&Recording::answering("x"));
assert_eq!(
error_code(&approve(&broker, 41)),
Some(ErrorCode::NoSuchApproval)
);
assert_eq!(
error_code(&refuse(&broker, 41, None)),
Some(ErrorCode::NoSuchApproval)
);
let (_stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
refuse(&broker, approval, None);
assert_eq!(
error_code(&approve(&broker, approval)),
Some(ErrorCode::NoSuchApproval)
);
assert_eq!(rig.events().len(), 2, "one decision, one approval");
}
#[test]
fn approve_and_refuse_at_once_give_exactly_one_answer() {
let rig = Rig::new("admin-race");
ask_notes(&rig);
let broker = rig.broker(&Recording::answering("x"));
for round in 0..100u64 {
let (mut stream, approval) = pending(&broker, round, r#"{"path":"/n/a"}"#);
let start = Arc::new(Barrier::new(2));
let racers: Vec<_> = [true, false]
.into_iter()
.map(|approving| {
let (broker, start) = (Arc::clone(&broker), Arc::clone(&start));
std::thread::spawn(move || {
start.wait();
if approving {
approve(&broker, approval)
} else {
refuse(&broker, approval, None)
}
})
})
.collect();
let answers: Vec<Message> = racers.into_iter().map(|r| r.join().unwrap()).collect();
let losers = answers
.iter()
.filter(|m| error_code(m) == Some(ErrorCode::NoSuchApproval))
.count();
assert_eq!(losers, 1, "round {round}: {answers:?}");
final_answer(&mut stream);
let approvals = rig
.events()
.iter()
.filter(|e| matches!(e, AuditEvent::Approval { decision, .. } if *decision == approval))
.count();
assert_eq!(approvals, 1, "round {round}");
}
}
#[test]
fn approve_after_the_grant_file_is_removed_is_denied() {
let rig = Rig::new("admin-removed");
ask_notes(&rig);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
rig.remove_grant("notes");
let reason = DenyReason::NoGrant;
assert_eq!(
outcome(approve(&broker, approval)),
DecisionRecord::Denied { reason }
);
assert_eq!(final_answer(&mut stream), denied(reason));
assert_eq!(rt.count(), 0);
}
#[test]
fn approve_after_the_taint_rose_past_the_grant_is_denied() {
let rig = Rig::new("admin-taint");
let text = grant_text("read_file", "ask", "", "paths = [\"/n\"]")
.replace("max_taint = \"secret\"", "max_taint = \"private\"");
rig.grant("notes", &text);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
let s1 = SessionId::new("s1").unwrap();
let secret = Label {
class: DataClass::Secret,
untrusted: false,
};
rig.state()
.raise(&s1, rig.state().read(&s1).unwrap(), secret)
.unwrap();
let reason = DenyReason::TaintTooHigh;
assert_eq!(
outcome(approve(&broker, approval)),
DecisionRecord::Denied { reason }
);
assert_eq!(final_answer(&mut stream), denied(reason));
assert_eq!(rt.count(), 0);
}
#[test]
fn an_approval_that_cannot_be_recorded_is_denied_on_both_sides() {
let rig = Rig::new("admin-norecord");
ask_notes(&rig);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
rig.switch.fail(true);
let reason = DenyReason::AuditUnavailable;
assert_eq!(
outcome(approve(&broker, approval)),
DecisionRecord::Denied { reason }
);
assert_eq!(final_answer(&mut stream), denied(reason));
assert_eq!(rt.count(), 0);
}
#[test]
fn a_refusal_that_cannot_be_recorded_is_an_error_for_bxctl() {
let rig = Rig::new("admin-norefuse");
ask_notes(&rig);
let broker = rig.broker(&Recording::answering("x"));
let (mut stream, approval) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
rig.switch.fail(true);
match refuse(&broker, approval, None) {
Message::Error(e) => {
assert_eq!(e.code, ErrorCode::Internal);
assert!(
e.detail.ends_with("see docs/runbook.md#audit-unavailable"),
"{}",
e.detail
);
}
other => panic!("{other:?}"),
}
assert_eq!(
final_answer(&mut stream),
denied(DenyReason::AuditUnavailable)
);
}
#[test]
fn expiry_denies_an_approval_nobody_answered() {
let rig = Rig::with_ttl("admin-expire", 100);
ask_notes(&rig);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, _) = pending(&broker, 1, r#"{"path":"/n/a"}"#);
let until = Instant::now() + Duration::from_secs(10);
while admin::expire_due(&broker, Timestamp::now()) == 0 {
assert!(Instant::now() < until, "never expired");
std::thread::sleep(Duration::from_millis(20));
}
assert_eq!(
final_answer(&mut stream),
denied(DenyReason::ApprovalExpired)
);
match &rig.events()[1] {
AuditEvent::Approval { answer, by, .. } => {
assert_eq!((*answer, by.as_deref()), (ApprovalAnswer::Expired, None));
}
other => panic!("{other:?}"),
}
assert_eq!(rt.count(), 0);
}
#[test]
fn check_grants_reports_every_problem_or_none() {
let rig = Rig::new("admin-grants");
ask_notes(&rig);
let broker = rig.broker(&Recording::answering("x"));
match admin(&broker, Message::CheckGrants(Empty {})) {
Message::GrantsReport(r) => assert!(r.problems.is_empty(), "{r:?}"),
other => panic!("{other:?}"),
}
rig.grant("bad", "tool = \"read_file\"\nmdoe = \"auto\"\n");
rig.grant(
"Worse",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
match admin(&broker, Message::CheckGrants(Empty {})) {
Message::GrantsReport(r) => {
let files: Vec<&str> = r.problems.iter().map(|p| p.file.as_str()).collect();
assert!(files.contains(&"bad.toml"), "{r:?}");
assert!(files.contains(&"Worse.toml"), "{r:?}");
}
other => panic!("{other:?}"),
}
}
#[test]
fn every_other_kind_on_admin_sock_is_forbidden() {
let rig = Rig::new("admin-forbidden");
let broker = rig.broker(&Recording::answering("x"));
let req = request("s1", 1, "read_file", r#"{"path":"/n/a"}"#);
for msg in [
Message::ToolRequest(req),
Message::Ok(Empty {}),
Message::ToolResponse(denied(DenyReason::NoGrant)),
] {
assert_eq!(error_code(&admin(&broker, msg)), Some(ErrorCode::Forbidden));
}
let lines = rig.lines.with("on admin.sock");
assert_eq!(lines.len(), 3, "{:?}", rig.lines.all());
assert!(lines[0].contains("tool_request"), "{}", lines[0]);
assert!(
lines
.iter()
.all(|l| l.ends_with("see docs/runbook.md#socket-forbidden"))
);
assert!(
rig.events().is_empty(),
"a refused tool request decides nothing"
);
}
@@ -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<Message> {
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}");
}
@@ -0,0 +1,233 @@
//! A call an `ask` grant matched: the pending frame, the table entry, the wait, and a requester
//! that goes away. The tests answer entries by hand, as `admin` will. 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::os::unix::net::UnixStream;
use std::sync::Arc;
use std::time::{Duration, Instant};
use brokerd::broker::{self, Broker, GONE};
use brokerd::ledger::Answer;
use bxcore::{
AuditEvent, DataClass, DecisionRecord, DenyReason, Message, ResultStatus, Timestamp,
ToolResponse,
};
use client::{Serve, next, open};
use rig::{Rig, grant_text, request};
use runtime::Recording;
fn ask_notes(rig: &Rig, extra: &str) {
rig.grant(
"notes",
&grant_text("read_file", "ask", extra, "paths = [\"/n\"]"),
);
}
/// Sends a call and reads its pending frame. Returns the connection and the frame's values.
fn start(broker: &Arc<Broker>, arguments: &str) -> (UnixStream, u64, Timestamp) {
let req = request("s1", 9, "read_file", arguments);
let mut stream = open(broker, broker::handle, 9, Message::ToolRequest(req));
let env = next(&mut stream);
assert_eq!((env.id, env.r#final), (9, false), "{env:?}");
match env.msg {
Message::ToolResponse(ToolResponse::PendingApproval { approval, expires }) => {
(stream, approval, expires)
}
other => panic!("{other:?}"),
}
}
/// What `admin` does for an entry it took: record the answer, send the verdict.
fn answer_by_hand(broker: &Broker, id: u64, answer: Answer) -> DecisionRecord {
let entry = broker.table().take(id).expect("the entry is pending");
answer_entry(broker, entry, answer)
}
fn answer_entry(
broker: &Broker,
entry: brokerd::approvals::Entry,
answer: Answer,
) -> DecisionRecord {
let grants = broker.grants();
let done = broker.ledger().answer(
entry.ask,
entry.info.approval,
answer,
&grants,
Timestamp::now(),
);
entry.reply.send(done.verdict).unwrap();
done.outcome
}
fn approved() -> Answer {
Answer::Approved {
by: Some("bxctl".to_string()),
post: None,
}
}
fn final_answer(stream: &mut UnixStream) -> ToolResponse {
let env = next(stream);
assert!(env.r#final, "{env:?}");
match env.msg {
Message::ToolResponse(r) => r,
other => panic!("{other:?}"),
}
}
fn eventually(what: &str, mut done: impl FnMut() -> bool) {
let until = Instant::now() + Duration::from_secs(10);
while !done() {
assert!(Instant::now() < until, "never happened: {what}");
std::thread::sleep(Duration::from_millis(20));
}
}
#[test]
fn an_ask_call_waits_is_listed_and_runs_when_approved() {
let rig = Rig::new("pending-approve");
ask_notes(&rig, "");
let rt = Recording::answering("the notes");
let broker = rig.broker(&rt);
let before = Timestamp::now();
// Spaced JSON: the table shows the arguments as policy parsed them, not as sent.
let (mut stream, approval, expires) = start(&broker, r#"{ "path" : "/n/a" }"#);
assert_eq!(approval, 0, "the approval id is the decision's seq");
let items = broker.table().list();
assert_eq!(items.len(), 1);
let item = &items[0];
assert_eq!(item.approval, 0);
assert_eq!((item.session.as_str(), item.call.0), ("s1", 9));
assert_eq!(item.tool, "read_file");
assert_eq!(item.arguments, r#"{"path":"/n/a"}"#);
assert_eq!(
(item.grant.as_str(), item.taint),
("notes", DataClass::Private)
);
assert!(item.created >= before);
assert_eq!(item.expires, expires);
assert_eq!(
expires.unix_millis() - item.created.unix_millis(),
900_000,
"now plus ttl_ms"
);
assert_eq!(rt.count(), 0, "nothing runs while it waits");
assert_eq!(
answer_by_hand(&broker, 0, approved()),
DecisionRecord::Ask {}
);
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert_eq!(rt.count(), 1);
let events = rig.events();
assert!(
matches!(
events.as_slice(),
[
AuditEvent::Decision { .. },
AuditEvent::Approval { decision: 0, .. },
AuditEvent::Result { decision: 0, .. }
]
),
"{events:?}"
);
}
#[test]
fn the_expiry_is_the_grants_when_that_is_earlier() {
let rig = Rig::new("pending-grant-expiry");
let soon = Timestamp::from_unix_millis(Timestamp::now().unix_millis() + 60_000).unwrap();
ask_notes(&rig, &format!("expires = \"{}\"", soon.to_rfc3339()));
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (_stream, _, expires) = start(&broker, r#"{"path":"/n/a"}"#);
assert_eq!(expires, soon);
}
#[test]
fn a_denied_verdict_reaches_the_requester_and_nothing_runs() {
let rig = Rig::new("pending-refuse");
ask_notes(&rig, "");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, approval, _) = start(&broker, r#"{"path":"/n/a"}"#);
let refused = Answer::Refused {
by: Some("bxctl".to_string()),
post: None,
reason: None,
};
answer_by_hand(&broker, approval, refused);
assert_eq!(
final_answer(&mut stream),
ToolResponse::Denied {
reason: DenyReason::ApprovalRefused,
detail: None,
}
);
assert_eq!(rt.count(), 0);
}
#[test]
fn a_requester_that_leaves_takes_its_own_entry_and_nothing_is_written() {
let rig = Rig::new("pending-leave");
ask_notes(&rig, "");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (stream, _, _) = start(&broker, r#"{"path":"/n/a"}"#);
drop(stream);
// The waiting thread looks at its connection every second.
eventually("the entry is removed", || broker.table().list().is_empty());
std::thread::sleep(Duration::from_millis(200));
assert_eq!(rig.events().len(), 1, "only the decision");
let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap();
assert_eq!(report.abandoned, [0]);
assert!(report.failure.is_none());
assert_eq!(rt.count(), 0);
}
#[test]
fn a_requester_gone_at_the_last_look_runs_nothing_and_closes_the_call() {
let rig = Rig::new("pending-lastlook");
ask_notes(&rig, "");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (stream, approval, _) = start(&broker, r#"{"path":"/n/a"}"#);
// Take the entry first, so the waiting thread cannot take it back when it sees the
// connection gone; then approve.
let entry = broker.table().take(approval).unwrap();
drop(stream);
answer_entry(&broker, entry, approved());
eventually("the call is closed", || rig.events().len() == 3);
match &rig.events()[2] {
AuditEvent::Result {
status,
decision,
sha256,
bytes,
..
} => {
assert_eq!((*status, *decision), (ResultStatus::Failed, 0));
assert_eq!(*sha256, bxcore::sha256(GONE.as_bytes()).unwrap());
assert_eq!(*bytes, GONE.len() as u64);
}
other => panic!("{other:?}"),
}
assert_eq!(GONE, "the requester went away");
assert_eq!(rt.count(), 0);
let report = brokerd::audit::verify_dir(&rig.cfg.audit_dir()).unwrap();
assert!(report.unfinished.is_empty() && report.abandoned.is_empty());
}
@@ -0,0 +1,449 @@
//! The courier carries a pending approval to `gatewayd` and its answer back, against a fake
//! `gatewayd` on a real socket (M4b spec, section 5). Do not edit.
#[path = "support/client.rs"]
mod client;
#[path = "support/fake_gateway.rs"]
mod fake_gateway;
#[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::os::unix::net::UnixStream;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::{Duration, Instant};
use brokerd::broker::{self, Broker};
use brokerd::config::Gateway;
use brokerd::courier::{self, Timing};
use brokerd::ledger::Answer;
use bxcore::{
ApprovalAnswer, ApprovalReply, AuditEvent, DenyReason, GatewayOutcome, Message, Timestamp,
ToolResponse,
};
use client::{Serve, next, open};
use fake_gateway::{FakeGateway, KYLE, POST};
use rig::{Rig, grant_text, request};
use runtime::Recording;
const WAIT: Duration = Duration::from_secs(5);
/// A rig whose `[gateway]` is a fake gatewayd in its home, and an `ask` grant for /n.
fn rig_with_gateway(tag: &str) -> (Rig, FakeGateway) {
let mut rig = Rig::new(tag);
let socket = rig.dir.path().join("gw.sock");
let fake = FakeGateway::start(&socket);
rig.cfg.gateway = Some(Gateway { socket });
rig.grant(
"notes",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
(rig, fake)
}
/// A call that waits for approval; its connection and the approval id.
fn ask(broker: &Arc<Broker>) -> (UnixStream, u64) {
let mut stream = open(
broker,
broker::handle,
9,
Message::ToolRequest(request("s1", 9, "read_file", r#"{"path":"/n/a"}"#)),
);
match next(&mut stream).msg {
Message::ToolResponse(ToolResponse::PendingApproval { approval, .. }) => (stream, approval),
other => panic!("{other:?}"),
}
}
fn final_answer(stream: &mut UnixStream) -> ToolResponse {
let env = next(stream);
assert!(env.r#final, "{env:?}");
match env.msg {
Message::ToolResponse(r) => r,
other => panic!("{other:?}"),
}
}
/// The `Approval` record's (answer, by, post, reason).
fn approval_record(
rig: &Rig,
) -> (
ApprovalAnswer,
Option<String>,
Option<String>,
Option<String>,
) {
let events = rig.events();
let found = events.iter().find_map(|e| match e {
AuditEvent::Approval {
answer,
by,
post,
reason,
..
} => Some((*answer, by.clone(), post.clone(), reason.clone())),
_ => None,
});
found.expect("an Approval record")
}
fn some(s: &str) -> Option<String> {
Some(s.to_string())
}
/// Sets the courier's stop flag when dropped, so a failing assertion ends the courier too.
struct StopOnDrop<'a>(&'a AtomicBool);
impl Drop for StopOnDrop<'_> {
fn drop(&mut self) {
self.0.store(true, std::sync::atomic::Ordering::SeqCst);
}
}
/// The courier must end by itself within five seconds.
fn ends_by_itself<T>(courier: &std::thread::ScopedJoinHandle<'_, T>) {
let until = Instant::now() + WAIT;
while !courier.is_finished() {
assert!(Instant::now() < until, "the courier did not stop");
std::thread::sleep(Duration::from_millis(20));
}
}
#[test]
fn approved_in_mattermost_it_runs_and_the_record_names_who_and_where() {
let (rig, fake) = rig_with_gateway("courier-approve");
let rt = Recording::answering("the notes");
let broker = rig.broker(&rt);
let (mut stream, id) = ask(&broker);
let mut conn = fake.next(WAIT).expect("the approval reaches gatewayd");
assert_eq!(
(conn.request.approval, conn.request.tool.as_str()),
(id, "read_file")
);
assert_eq!(conn.request.arguments, r#"{"path":"/n/a"}"#);
conn.posted();
conn.approve();
assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {}));
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert_eq!(rt.count(), 1);
assert_eq!(
approval_record(&rig),
(ApprovalAnswer::Approved, some(KYLE), some(POST), None)
);
}
#[test]
fn refused_in_mattermost_it_is_denied_with_the_reason() {
let (rig, fake) = rig_with_gateway("courier-refuse");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, _) = ask(&broker);
let mut conn = fake.next(WAIT).unwrap();
conn.posted();
conn.refuse("not on a Friday");
assert_eq!(conn.outcome(), Some(GatewayOutcome::Refused {}));
assert_eq!(
final_answer(&mut stream),
ToolResponse::Denied {
reason: DenyReason::ApprovalRefused,
detail: None,
}
);
assert_eq!(rt.count(), 0);
assert_eq!(
approval_record(&rig),
(
ApprovalAnswer::Refused,
some(KYLE),
some(POST),
some("not on a Friday")
)
);
}
#[test]
fn approved_after_the_grant_is_gone_it_is_denied_on_the_recheck() {
let (rig, fake) = rig_with_gateway("courier-recheck");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, _) = ask(&broker);
let mut conn = fake.next(WAIT).unwrap();
conn.posted();
rig.remove_grant("notes");
conn.approve();
assert_eq!(
conn.outcome(),
Some(GatewayOutcome::Denied {
reason: DenyReason::NoGrant
})
);
assert_eq!(
final_answer(&mut stream),
ToolResponse::Denied {
reason: DenyReason::NoGrant,
detail: None,
}
);
assert_eq!(rt.count(), 0);
}
#[test]
fn answered_in_bxctl_the_courier_closes_without_an_outcome() {
let (rig, fake) = rig_with_gateway("courier-bxctl");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, id) = ask(&broker);
let mut conn = fake.next(WAIT).unwrap();
conn.posted();
let entry = broker.table().take(id).unwrap();
brokerd::admin::answer(
&broker,
entry,
Answer::Approved {
by: some("bxctl"),
post: None,
},
Timestamp::now(),
);
assert!(
conn.closed_within(WAIT),
"the courier closes, and sends nothing"
);
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert_eq!(approval_record(&rig).1, some("bxctl"));
}
#[test]
fn an_answer_after_it_ended_is_gone() {
let rig = Rig::new("courier-gone");
rig.grant(
"notes",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (_stream, id) = ask(&broker);
// The courier is run by hand here, so that the answer can arrive after the entry is taken.
let socket = rig.dir.path().join("gw.sock");
let fake = FakeGateway::start(&socket);
let info = broker.table().list().remove(0);
let stop = AtomicBool::new(false);
std::thread::scope(|scope| {
let courier = scope.spawn(|| {
courier::carry(
&broker,
&socket,
&info,
&stop,
Timing {
retry: Duration::from_millis(100),
tick: Duration::from_secs(5),
},
)
});
let _guard = StopOnDrop(&stop);
let mut conn = fake.next(WAIT).unwrap();
conn.posted();
let entry = broker.table().take(id).unwrap();
brokerd::admin::answer(&broker, entry, Answer::Expired, Timestamp::now());
conn.approve();
assert_eq!(conn.outcome(), Some(GatewayOutcome::Gone {}));
ends_by_itself(&courier);
});
}
#[test]
fn if_gatewayd_goes_away_the_approval_is_posted_again() {
let (rig, fake) = rig_with_gateway("courier-again");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, id) = ask(&broker);
let mut first = fake.next(WAIT).unwrap();
first.posted();
first.close();
let mut second = fake.next(Duration::from_secs(10)).expect("posted again");
assert_eq!(second.request.approval, id);
second.posted();
second.approve();
assert_eq!(second.outcome(), Some(GatewayOutcome::Runs {}));
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert!(
rig.lines.all().iter().any(|l| l.contains(&format!(
"approval {id}: gatewayd went away; posting it again"
))),
"{:?}",
rig.lines.all()
);
}
#[test]
fn a_bad_answer_answers_nothing() {
let (rig, fake) = rig_with_gateway("courier-bad");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (_stream, id) = ask(&broker);
for bad in [
ApprovalReply {
approval: id + 1,
approved: true,
by: KYLE.to_string(),
post: POST.to_string(),
reason: None,
},
ApprovalReply {
approval: id,
approved: true,
by: "kyle".to_string(),
post: POST.to_string(),
reason: None,
},
ApprovalReply {
approval: id,
approved: false,
by: KYLE.to_string(),
post: POST.to_string(),
reason: Some("x".repeat(501)),
},
] {
let mut conn = fake.next(Duration::from_secs(10)).unwrap();
conn.posted();
conn.reply(bad);
assert_eq!(conn.outcome(), None, "no outcome for a bad answer");
assert!(broker.table().contains(id), "still pending");
}
assert_eq!(rt.count(), 0);
}
#[test]
fn unreachable_at_first_then_posted_and_said_once() {
let rig = Rig::new("courier-late");
rig.grant(
"notes",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (_stream, id) = ask(&broker);
let socket = rig.dir.path().join("gw.sock");
let info = broker.table().list().remove(0);
let stop = AtomicBool::new(false);
let timing = Timing {
retry: Duration::from_millis(100),
tick: Duration::from_millis(20),
};
std::thread::scope(|scope| {
let courier = scope.spawn(|| courier::carry(&broker, &socket, &info, &stop, timing));
let _guard = StopOnDrop(&stop);
std::thread::sleep(Duration::from_millis(500));
let fake = FakeGateway::start(&socket);
let mut conn = fake.next(WAIT).unwrap();
conn.posted();
let until = Instant::now() + WAIT;
while !rig
.lines
.all()
.iter()
.any(|l| l.contains("posted in Mattermost"))
{
assert!(Instant::now() < until, "{:?}", rig.lines.all());
std::thread::sleep(Duration::from_millis(20));
}
broker.table().take(id);
ends_by_itself(&courier);
});
let lines = rig.lines.all();
let unreachable: Vec<&String> = lines
.iter()
.filter(|l| l.contains("gatewayd is not reachable"))
.collect();
assert_eq!(
unreachable.len(),
1,
"said once, not every retry: {lines:?}"
);
assert!(
unreachable[0].starts_with(&format!(
"brokerd: approval {id}: gatewayd is not reachable at "
)),
"{lines:?}"
);
assert!(
unreachable[0]
.ends_with("; bxctl can still answer it\nsee docs/runbook.md#gateway-unreachable"),
"{lines:?}"
);
assert!(
lines.contains(&format!("brokerd: approval {id} posted in Mattermost")),
"{lines:?}"
);
}
#[test]
fn if_gatewayd_cannot_post_it_is_tried_again() {
let (rig, fake) = rig_with_gateway("courier-cannot");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, _) = ask(&broker);
fake.next(WAIT).unwrap().cannot_post();
let mut conn = fake.next(Duration::from_secs(10)).expect("tried again");
conn.posted();
conn.approve();
assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {}));
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
assert!(
rig.lines
.all()
.iter()
.any(|l| l.contains("gatewayd could not post it: Mattermost is down"))
);
}
#[test]
fn without_a_gateway_nothing_is_carried() {
let rig = Rig::new("courier-none");
rig.grant(
"notes",
&grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
);
let socket = rig.dir.path().join("gw.sock");
let fake = FakeGateway::start(&socket);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let (mut stream, id) = ask(&broker);
assert!(
fake.next(Duration::from_millis(500)).is_none(),
"no [gateway]: no connection"
);
let entry = broker.table().take(id).unwrap();
brokerd::admin::answer(
&broker,
entry,
Answer::Approved {
by: some("bxctl"),
post: None,
},
Timestamp::now(),
);
assert!(matches!(
final_answer(&mut stream),
ToolResponse::Result { .. }
));
}
@@ -0,0 +1,258 @@
//! An invalid call is denied with the parser's reason: `ArgsError::detail` makes it one short
//! line, `policy::decide` keeps it, and `broker.sock` sends it. The audit record is unchanged.
//! Do not edit.
#[path = "support/build.rs"]
mod build;
#[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 brokerd::args::{ArgsError, MAX_DETAIL};
use brokerd::policy::decide;
use build::{denied, fetch, grant, now, private, read, set, shell};
use bxcore::{AuditEvent, DecisionRecord, DenyReason, Envelope, Message, Mode, ToolResponse};
use client::{Serve, call};
use rig::{Rig, grant_text};
use runtime::Recording;
#[test]
fn the_detail_is_the_error_text() {
assert_eq!(MAX_DETAIL, 300);
let cases = [
(
ArgsError::Shape("missing field `path`".to_string()),
"invalid argument shape: missing field `path`",
),
(
ArgsError::Path("notes/a.md".to_string()),
"invalid path: notes/a.md",
),
(
ArgsError::Url("http://example.com/".to_string()),
"invalid url: http://example.com/",
),
];
for (error, want) in cases {
assert_eq!(error.detail(), want);
assert_eq!(
error.detail(),
error.to_string(),
"a short clean text is kept as it is"
);
}
}
/// Every control character, and U+2028 and U+2029 (the line and paragraph separators), becomes
/// one space each; other characters, spaces and non-ASCII letters included, are kept.
#[test]
fn the_detail_is_one_line() {
let error = ArgsError::Path(
"a\nb\r\nc\td\u{7}e\u{1b}[31mf\u{7f}g\u{85}h\u{2028}i\u{2029}j k \u{e9}\u{65e5}"
.to_string(),
);
assert_eq!(
error.detail(),
"invalid path: a b c d e [31mf g h i j k \u{e9}\u{65e5}"
);
let error = ArgsError::Shape("line one\nline two\n".to_string());
assert_eq!(error.detail(), "invalid argument shape: line one line two ");
}
/// At most `MAX_DETAIL` characters, counted as characters, not bytes: a cut never splits one.
#[test]
fn the_detail_is_cut_to_300_characters() {
// "invalid path: " is 14 characters.
let exact = ArgsError::Path("a".repeat(286));
assert_eq!(
exact.detail(),
exact.to_string(),
"300 characters are kept whole"
);
let over = ArgsError::Path("a".repeat(287));
assert_eq!(over.detail(), format!("invalid path: {}", "a".repeat(286)));
let wide = ArgsError::Path("\u{e9}".repeat(1000));
let detail = wide.detail();
assert_eq!(detail.chars().count(), 300);
assert_eq!(detail, format!("invalid path: {}", "\u{e9}".repeat(286)));
// Newlines past the cut do not matter; newlines before it are spaces.
let long = ArgsError::Shape(format!("{}\n{}", "x".repeat(10), "y\n".repeat(500)));
let detail = long.detail();
assert_eq!(detail.chars().count(), 300);
assert!(!detail.contains('\n'), "{detail:?}");
assert!(detail.starts_with("invalid argument shape: xxxxxxxxxx y y"));
}
/// Each kind of `ArgsError`, from the model's own argument text, reaches the denial as its detail,
/// with no grant and with a grant that would cover the call: arguments are checked first. (The
/// memory tools take no grant: their policy is built in.)
#[test]
fn each_kind_of_invalid_arguments_carries_its_detail() {
let covering = set(vec![
grant("r", "read_file", Mode::Auto).paths(&["/n"]),
grant("s", "shell", Mode::Auto),
grant("f", "http_fetch", Mode::Auto).hosts(&["example.com"]),
]);
let none = set(vec![]);
let cases = [
(
build::request(
"memory",
r#"{"action":"note","file":"notes.md","text":"x"}"#,
),
"invalid argument shape: unknown field `file`, expected `text`",
),
(
build::request("read_file", r#"{"path":"/n/a","file":"x"}"#),
"invalid argument shape: unknown field `file`, expected `path` at line 1 column 21",
),
(
build::request("read_file", "not json"),
"invalid argument shape: expected ident at line 1 column 2",
),
(
build::request("read_file", "{}"),
"invalid argument shape: missing field `path` at line 1 column 2",
),
(
build::request("recall", r#"{"query":""}"#),
"invalid argument shape: query must have 1 to 500 characters",
),
(read("n/a"), "invalid path: n/a"),
(shell(Some("a\\nb")), "invalid path: a b"),
(
fetch("http://example.com/"),
"invalid url: http://example.com/",
),
];
for grants in [&none, &covering] {
for (req, want) in &cases {
let text = req.arguments.clone();
let denial = denied(decide(req.clone(), grants, private(), now()));
assert_eq!(denial.reason, DenyReason::InvalidArguments, "{text}");
assert_eq!(denial.detail.as_deref(), Some(*want), "{text}");
assert_eq!(denial.grant, None, "{text}");
}
}
}
#[test]
fn a_long_invalid_path_is_cut_in_the_denial() {
let path = format!("/{}", "a".repeat(5000));
let denial = denied(decide(read(&path), &set(vec![]), private(), now()));
assert_eq!(denial.reason, DenyReason::InvalidArguments);
let detail = denial.detail.unwrap();
assert_eq!(detail.chars().count(), 300);
assert_eq!(detail, format!("invalid path: /{}", "a".repeat(285)));
}
/// Only an invalid-arguments denial has a detail.
#[test]
fn other_denials_have_no_detail() {
let grants = set(vec![
grant("no", "read_file", Mode::Deny).paths(&["/n"]),
grant("old", "shell", Mode::Auto).expires("2026-09-01T00:00:00.000Z"),
]);
let cases = [
(build::request("echo", "not json"), DenyReason::NoGrant),
(fetch("https://example.com/"), DenyReason::NoGrant),
(read("/n/a"), DenyReason::DeniedByGrant),
(shell(None), DenyReason::GrantExpired),
];
for (req, reason) in cases {
let text = req.arguments.clone();
let denial = denied(decide(req, &grants, private(), now()));
assert_eq!(denial.reason, reason, "{text}");
assert_eq!(denial.detail, None, "{text}");
}
}
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:?}"),
}
}
#[test]
fn broker_sock_sends_the_detail_and_the_audit_record_does_not_hold_it() {
let rig = Rig::new("broker-invalid-detail");
rig.grant(
"notes",
&grant_text("read_file", "auto", "", "paths = [\"/n\"]"),
);
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let arguments = r#"{"path":"/n/a","file":"x"}"#;
let frames = call(&broker, rig::request("s1", 5, "read_file", arguments));
let detail =
"invalid argument shape: unknown field `file`, expected `path` at line 1 column 21";
assert_eq!(
only(&frames),
&ToolResponse::Denied {
reason: DenyReason::InvalidArguments,
detail: Some(detail.to_string()),
}
);
assert_eq!(rt.count(), 0, "nothing ran");
match rig.events().as_slice() {
[
AuditEvent::Decision {
outcome,
arguments: recorded,
..
},
] => {
assert_eq!(
*outcome,
DecisionRecord::Denied {
reason: DenyReason::InvalidArguments
}
);
assert_eq!(recorded, arguments, "the record keeps the arguments");
}
other => panic!("{other:?}"),
}
let audit = rig.cfg.audit_dir();
let mut logs = 0;
for entry in std::fs::read_dir(&audit).unwrap() {
let path = entry.unwrap().path();
if path.extension().is_none_or(|e| e != "jsonl") {
continue;
}
logs += 1;
let text = std::fs::read_to_string(&path).unwrap();
assert!(!text.contains("unknown field"), "{text}");
assert!(!text.contains("\"detail\""), "{text}");
}
assert_eq!(logs, 1, "one audit log file");
}
#[test]
fn a_call_no_grant_allows_is_denied_on_broker_sock_without_a_detail() {
let rig = Rig::new("broker-no-detail");
let rt = Recording::answering("x");
let broker = rig.broker(&rt);
let frames = call(
&broker,
rig::request("s1", 6, "read_file", r#"{"path":"/n/a"}"#),
);
assert_eq!(
only(&frames),
&ToolResponse::Denied {
reason: DenyReason::NoGrant,
detail: None,
}
);
}
@@ -0,0 +1,425 @@
//! The ledger's first and third steps: decide and record; raise the state and record the result.
//! And what happens after a failed append or a panic. Do not edit.
#[path = "support/build.rs"]
mod build;
#[path = "support/rig.rs"]
mod rig;
#[path = "support/sink.rs"]
mod sink;
#[path = "support/tmp.rs"]
mod tmp;
use std::os::unix::fs::PermissionsExt;
use std::panic::AssertUnwindSafe;
use brokerd::ledger::{Call, Decided, Grants, NOT_RECORDED, POISONED, STOPPED};
use brokerd::policy::{Denial, Label, SessionState};
use build::{grant, now, read, set};
use bxcore::{
AuditEvent, CallId, DataClass, DecisionRecord, DenyReason, GrantProblem, Mode, ResultStatus,
SessionId, ToolResponse,
};
use rig::Rig;
fn one(grants: Vec<build::Build>) -> Grants {
Ok(set(grants))
}
fn notes(mode: Mode) -> Grants {
one(vec![grant("n", "read_file", mode).paths(&["/n"])])
}
fn call(label: Label) -> Call {
Call {
session: SessionId::new("s1").unwrap(),
call: CallId(1),
decision: 0,
label,
}
}
fn secret() -> Label {
Label {
class: DataClass::Secret,
untrusted: true,
}
}
fn result(content: &str, label: Label) -> ToolResponse {
ToolResponse::Result {
content: content.to_string(),
class: label.class,
untrusted: label.untrusted,
truncated: false,
}
}
#[test]
fn an_allowed_call_is_recorded_in_full() {
let rig = Rig::new("ledger-allowed");
let ledger = rig.ledger();
let request = read("/n/a");
let arguments = request.arguments.clone();
match ledger.decide(request, &notes(Mode::Auto), now()) {
Decided::Allowed { decision, seq } => {
assert_eq!(seq, 0);
assert_eq!(decision.grant(), "n");
}
other => panic!("{other:?}"),
}
let records = rig.records();
assert_eq!(records.len(), 1);
assert_eq!(records[0].time, now());
assert_eq!(
records[0].event,
AuditEvent::Decision {
session: SessionId::new("s1").unwrap(),
call: CallId(1),
tool: "read_file".to_string(),
arguments,
outcome: DecisionRecord::Allowed {},
grant: Some("n".to_string()),
grant_sha256: Some(bxcore::sha256(b"n").unwrap()),
taint: DataClass::Private,
untrusted: false,
}
);
}
#[test]
fn an_ask_is_recorded_as_ask_with_the_state_it_was_decided_at() {
let rig = Rig::new("ledger-ask");
let ledger = rig.ledger();
match ledger.decide(read("/n/a"), &notes(Mode::Ask), now()) {
Decided::Ask { ask, seq, state } => {
assert_eq!((seq, ask.grant()), (0, "n"));
assert_eq!(state, SessionState::default());
}
other => panic!("{other:?}"),
}
match &rig.events()[0] {
AuditEvent::Decision { outcome, grant, .. } => {
assert_eq!(*outcome, DecisionRecord::Ask {});
assert_eq!(grant.as_deref(), Some("n"));
}
other => panic!("{other:?}"),
}
}
#[test]
fn a_denial_names_a_grant_only_when_a_grant_denied() {
let rig = Rig::new("ledger-denied");
let ledger = rig.ledger();
let none = ledger.decide(read("/n/a"), &one(Vec::new()), now());
assert!(
matches!(
none,
Decided::Denied(Denial {
reason: DenyReason::NoGrant,
..
})
),
"{none:?}"
);
let deny = one(vec![grant("d", "read_file", Mode::Deny).paths(&["/n"])]);
let by = ledger.decide(read("/n/a"), &deny, now());
assert!(
matches!(
by,
Decided::Denied(Denial {
reason: DenyReason::DeniedByGrant,
..
})
),
"{by:?}"
);
let events = rig.events();
let named: Vec<(DecisionRecord, Option<String>)> = events
.iter()
.map(|e| match e {
AuditEvent::Decision { outcome, grant, .. } => (outcome.clone(), grant.clone()),
other => panic!("{other:?}"),
})
.collect();
assert_eq!(
named,
[
(
DecisionRecord::Denied {
reason: DenyReason::NoGrant
},
None
),
(
DecisionRecord::Denied {
reason: DenyReason::DeniedByGrant
},
Some("d".to_string())
),
]
);
}
#[test]
fn invalid_grants_come_first_then_an_unreadable_state() {
let rig = Rig::new("ledger-order");
let ledger = rig.ledger();
std::fs::create_dir_all(rig.cfg.state_dir()).unwrap();
std::fs::write(rig.state_file("s1"), "not json").unwrap();
let invalid: Grants = Err(vec![GrantProblem {
file: "x.toml".to_string(),
line: None,
problem: "bad".to_string(),
}]);
let first = ledger.decide(read("/n/a"), &invalid, now());
assert!(
matches!(
first,
Decided::Denied(Denial {
reason: DenyReason::GrantsInvalid,
..
})
),
"{first:?}"
);
let second = ledger.decide(read("/n/a"), &notes(Mode::Auto), now());
assert!(
matches!(
second,
Decided::Denied(Denial {
reason: DenyReason::StateUnreadable,
..
})
),
"{second:?}"
);
// A state brokerd cannot read is recorded as the most sensitive one.
for event in rig.events() {
match event {
AuditEvent::Decision {
taint, untrusted, ..
} => assert_eq!((taint, untrusted), (DataClass::Secret, true)),
other => panic!("{other:?}"),
}
}
let damaged = rig.lines.with("see docs/runbook.md#broker-state-damaged");
assert!(!damaged.is_empty(), "{:?}", rig.lines.all());
}
#[test]
fn a_failed_append_denies_this_call_and_every_later_one() {
let rig = Rig::new("ledger-stop");
let ledger = rig.ledger();
rig.switch.fail(true);
let first = ledger.decide(read("/n/a"), &notes(Mode::Auto), now());
assert!(
matches!(
first,
Decided::Denied(Denial {
reason: DenyReason::AuditUnavailable,
..
})
),
"{first:?}"
);
assert_eq!(rig.switch.attempts(), 1);
// The sink would work again; the ledger does not try it.
rig.switch.fail(false);
let later = ledger.decide(read("/n/a"), &notes(Mode::Auto), now());
assert!(
matches!(
later,
Decided::Denied(Denial {
reason: DenyReason::AuditUnavailable,
..
})
),
"{later:?}"
);
let finished = ledger.finish(&call(secret()), result("x", secret()), now());
assert_eq!(
finished,
ToolResponse::Failed {
message: NOT_RECORDED.to_string()
}
);
assert_eq!(rig.switch.attempts(), 1, "nothing more was written");
assert!(rig.records().is_empty());
assert!(!rig.state_file("s1").exists(), "the state was not raised");
assert!(!rig.lines.with(STOPPED).is_empty());
assert!(
rig.lines
.all()
.iter()
.all(|l| !l.contains("runbook") || l.ends_with("see docs/runbook.md#audit-unavailable")),
"{:?}",
rig.lines.all()
);
}
#[test]
fn a_panic_while_holding_the_ledger_denies_every_later_call() {
let rig = Rig::new("ledger-poison");
let ledger = rig.ledger();
rig.switch.panic_next();
let panicked = std::panic::catch_unwind(AssertUnwindSafe(|| {
ledger.decide(read("/n/a"), &notes(Mode::Auto), now())
}));
assert!(panicked.is_err());
let later = ledger.decide(read("/n/a"), &notes(Mode::Auto), now());
assert!(
matches!(
later,
Decided::Denied(Denial {
reason: DenyReason::AuditUnavailable,
..
})
),
"{later:?}"
);
assert_eq!(rig.switch.attempts(), 1);
assert!(
!rig.lines.with(POISONED).is_empty(),
"{:?}",
rig.lines.all()
);
assert!(POISONED.ends_with("see docs/runbook.md#audit-unavailable"));
}
#[test]
fn a_result_raises_the_state_then_is_recorded() {
let rig = Rig::new("ledger-result");
let ledger = rig.ledger();
let answer = ledger.finish(&call(secret()), result("key", secret()), now());
assert_eq!(answer, result("key", secret()), "passed on unchanged");
let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap();
assert_eq!(
state,
SessionState {
taint: DataClass::Secret,
untrusted: true
}
);
assert_eq!(
rig.events(),
[AuditEvent::Result {
session: SessionId::new("s1").unwrap(),
call: CallId(1),
decision: 0,
status: ResultStatus::Result,
class: DataClass::Secret,
untrusted: true,
truncated: false,
bytes: 3,
sha256: bxcore::sha256(b"key").unwrap(),
taint_after: DataClass::Secret,
}]
);
}
#[test]
fn taint_never_goes_down() {
let rig = Rig::new("ledger-down");
let ledger = rig.ledger();
let public = Label {
class: DataClass::Public,
untrusted: false,
};
ledger.finish(&call(secret()), result("a", secret()), now());
ledger.finish(&call(public), result("b", public), now());
let after: Vec<DataClass> = rig
.events()
.iter()
.map(|e| match e {
AuditEvent::Result { taint_after, .. } => *taint_after,
other => panic!("{other:?}"),
})
.collect();
assert_eq!(after, [DataClass::Secret, DataClass::Secret]);
let state = rig.state().read(&SessionId::new("s1").unwrap()).unwrap();
assert_eq!((state.taint, state.untrusted), (DataClass::Secret, true));
}
#[test]
fn a_failure_changes_no_state_and_is_recorded_by_its_message() {
let rig = Rig::new("ledger-failed");
let ledger = rig.ledger();
let failed = ToolResponse::Failed {
message: "the tool timed out".to_string(),
};
assert_eq!(
ledger.finish(&call(secret()), failed.clone(), now()),
failed
);
assert!(!rig.state_file("s1").exists());
match &rig.events()[0] {
AuditEvent::Result {
status,
class,
bytes,
sha256,
taint_after,
..
} => {
assert_eq!(*status, ResultStatus::Failed);
assert_eq!(
*class,
DataClass::Secret,
"the label the result would have had"
);
assert_eq!(*bytes, 18);
assert_eq!(*sha256, bxcore::sha256(b"the tool timed out").unwrap());
assert_eq!(*taint_after, DataClass::Private, "the taint before");
}
other => panic!("{other:?}"),
}
}
#[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("ledger-ro");
let ledger = rig.ledger();
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 answer = ledger.finish(&call(secret()), result("key", secret()), now());
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700)).unwrap();
assert_eq!(
answer,
ToolResponse::Failed {
message: NOT_RECORDED.to_string()
}
);
assert!(
rig.records().is_empty(),
"no Result for a state not on disk"
);
assert!(
!rig.lines
.with("see docs/runbook.md#broker-state-damaged")
.is_empty()
);
}
#[test]
fn a_result_record_that_cannot_be_written_withholds_the_content() {
let rig = Rig::new("ledger-norecord");
let ledger = rig.ledger();
rig.switch.fail(true);
let answer = ledger.finish(&call(secret()), result("key", secret()), now());
assert_eq!(
answer,
ToolResponse::Failed {
message: NOT_RECORDED.to_string()
}
);
assert!(
!rig.lines
.with("see docs/runbook.md#audit-unavailable")
.is_empty()
);
}
@@ -0,0 +1,411 @@
//! `brokerd serve` as a process: its startup, both sockets, and the expiry thread. Do not edit.
#[path = "support/tmp.rs"]
mod tmp;
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::time::{Duration, Instant};
use brokerd::audit::RECOVERED_NOTICE;
use brokerd::runner::REFUSING;
use bxcore::{
AuditEvent, CallId, DenyReason, Empty, Envelope, ErrorCode, Message, PROTOCOL_VERSION,
ResultStatus, SessionId, ToolRequest, ToolResponse,
};
use tmp::TempDir;
struct Home {
dir: TempDir,
config: PathBuf,
}
impl Home {
fn new(tag: &str, ttl_ms: u64) -> Home {
let dir = TempDir::new(tag);
std::fs::create_dir_all(dir.path().join("grants")).unwrap();
let text = format!(
"[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n[approvals]\nttl_ms = {ttl_ms}\n",
home = dir.path().display()
);
let config = dir.write("brokerd.toml", &text);
Home { dir, config }
}
fn path(&self, relative: &str) -> PathBuf {
self.dir.path().join(relative)
}
fn tools(&self) -> PathBuf {
self.path("run/loop-broker/broker.sock")
}
fn admin(&self) -> PathBuf {
self.path("run/owner-broker/admin.sock")
}
fn grant(&self, id: &str, mode: &str) {
let text = format!(
"tool = \"read_file\"\nmode = \"{mode}\"\nmax_taint = \"secret\"\n\
result_class = \"private\"\nuntrusted = false\n[constraints]\npaths = [\"/n\"]\n"
);
std::fs::write(self.path(&format!("grants/{id}.toml")), text).unwrap();
}
fn command(&self, extra: &[&str]) -> Command {
let mut command = Command::new(env!("CARGO_BIN_EXE_brokerd"));
command
.args(["serve", "--config"])
.arg(&self.config)
.args(extra);
command
}
/// Starts `brokerd serve` and waits until both sockets answer.
fn serve(&self) -> Running {
let child = self
.command(&[])
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let running = Running(Some(child));
let until = Instant::now() + Duration::from_secs(10);
while UnixStream::connect(self.tools()).is_err()
|| UnixStream::connect(self.admin()).is_err()
{
assert!(Instant::now() < until, "brokerd never listened");
std::thread::sleep(Duration::from_millis(20));
}
running
}
/// Runs `brokerd serve` expecting it to exit by itself.
fn run(&self, extra: &[&str]) -> Output {
self.command(extra).output().unwrap()
}
fn events(&self) -> Vec<AuditEvent> {
let dir = self.path("audit");
let mut names: Vec<String> = std::fs::read_dir(&dir)
.unwrap()
.map(|e| e.unwrap().file_name().into_string().unwrap())
.filter(|n| n.ends_with(".jsonl"))
.collect();
names.sort();
names
.iter()
.flat_map(|n| {
let text = std::fs::read_to_string(dir.join(n)).unwrap();
text.lines()
.map(|l| {
serde_json::from_str::<bxcore::AuditRecord>(l)
.unwrap()
.event
})
.collect::<Vec<_>>()
})
.collect()
}
}
/// Kills the daemon when dropped; `stop` returns what it printed.
struct Running(Option<Child>);
impl Running {
fn stop(mut self) -> String {
let mut child = self.0.take().unwrap();
child.kill().unwrap();
let output = child.wait_with_output().unwrap();
String::from_utf8_lossy(&output.stderr).to_string()
}
}
impl Drop for Running {
fn drop(&mut self) {
if let Some(child) = &mut self.0 {
let _ = child.kill();
let _ = child.wait();
}
}
}
fn mode(path: &Path) -> u32 {
std::fs::metadata(path).unwrap().permissions().mode() & 0o777
}
fn exchange(socket: &Path, id: u64, msg: Message) -> Vec<Envelope> {
let mut stream = UnixStream::connect(socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.unwrap();
let env = Envelope {
v: PROTOCOL_VERSION,
id,
r#final: true,
msg,
};
bxcore::write_frame(&mut stream, &env).unwrap();
let mut frames = Vec::new();
loop {
let env = bxcore::read_frame(&mut stream).unwrap();
let last = env.r#final;
frames.push(env);
if last {
return frames;
}
}
}
fn read_notes(call: u64) -> Message {
Message::ToolRequest(ToolRequest {
session: SessionId::new("s1").unwrap(),
call: CallId(call),
tool: "read_file".to_string(),
arguments: r#"{"path":"/n/a"}"#.to_string(),
})
}
fn last_response(frames: &[Envelope]) -> &ToolResponse {
match &frames.last().unwrap().msg {
Message::ToolResponse(r) => r,
other => panic!("{other:?}"),
}
}
fn stderr(output: &Output) -> String {
String::from_utf8_lossy(&output.stderr).to_string()
}
#[test]
fn it_makes_its_directories_0700_and_its_sockets_0600() {
let home = Home::new("serve-modes", 900_000);
// One directory found too open, one made.
std::fs::create_dir_all(home.path("run/owner-broker")).unwrap();
std::fs::set_permissions(
home.path("run/owner-broker"),
std::fs::Permissions::from_mode(0o755),
)
.unwrap();
let running = home.serve();
assert_eq!(mode(&home.path("run/loop-broker")), 0o700);
assert_eq!(mode(&home.path("run/owner-broker")), 0o700);
assert_eq!(mode(&home.tools()), 0o600);
assert_eq!(mode(&home.admin()), 0o600);
assert_eq!(mode(&home.path("audit")), 0o700);
let printed = running.stop();
assert!(printed.contains("serving"), "{printed}");
}
#[test]
fn a_stale_socket_is_replaced() {
let home = Home::new("serve-stale", 900_000);
std::fs::create_dir_all(home.path("run/loop-broker")).unwrap();
drop(UnixListener::bind(home.tools()).unwrap());
assert!(home.tools().exists(), "the stale socket file is there");
let _running = home.serve();
let frames = exchange(&home.tools(), 3, read_notes(3));
assert_eq!(
last_response(&frames),
&ToolResponse::Denied {
reason: DenyReason::NoGrant,
detail: None,
}
);
}
#[test]
fn a_second_brokerd_on_the_same_home_refuses_to_start() {
let home = Home::new("serve-twice", 900_000);
let _running = home.serve();
let second = home.run(&[]);
assert_eq!(second.status.code(), Some(1));
let text = stderr(&second);
assert!(text.contains("brokerd is already running"), "{text}");
assert!(
text.trim_end()
.ends_with("see docs/runbook.md#brokerd-already-running"),
"{text}"
);
// The first still has its sockets.
let frames = exchange(&home.admin(), 1, Message::Approvals(Empty {}));
assert!(matches!(frames[0].msg, Message::ApprovalList(_)));
}
#[test]
fn it_answers_each_socket_and_refuses_the_other_kinds() {
let home = Home::new("serve-kinds", 900_000);
home.grant("notes", "auto");
let running = home.serve();
let frames = exchange(&home.tools(), 7, read_notes(7));
assert_eq!(frames[0].id, 7);
// The production runtime runs nothing.
assert_eq!(
last_response(&frames),
&ToolResponse::Failed {
message: REFUSING.to_string()
}
);
let wrong = exchange(&home.tools(), 8, Message::Approvals(Empty {}));
assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden));
let wrong = exchange(&home.admin(), 9, read_notes(9));
assert!(matches!(&wrong[0].msg, Message::Error(e) if e.code == ErrorCode::Forbidden));
let printed = running.stop();
assert!(
printed.contains("approvals on broker.sock\nsee docs/runbook.md#socket-forbidden"),
"{printed}"
);
assert!(
printed.contains("tool_request on admin.sock\nsee docs/runbook.md#socket-forbidden"),
"{printed}"
);
assert!(matches!(
home.events().as_slice(),
[
AuditEvent::Decision { .. },
AuditEvent::Result {
status: ResultStatus::Failed,
..
}
]
));
}
#[test]
fn an_approval_nobody_answers_expires() {
let home = Home::new("serve-expire", 100);
home.grant("notes", "ask");
let _running = home.serve();
let started = Instant::now();
let frames = exchange(&home.tools(), 2, read_notes(2));
assert_eq!(frames.len(), 2, "{frames:?}");
assert!(matches!(
&frames[0].msg,
Message::ToolResponse(ToolResponse::PendingApproval { approval: 0, .. })
));
assert_eq!(
last_response(&frames),
&ToolResponse::Denied {
reason: DenyReason::ApprovalExpired,
detail: None,
}
);
// The expiry thread looks every second.
assert!(
started.elapsed() < Duration::from_secs(5),
"{:?}",
started.elapsed()
);
}
fn copy_case(home: &Home, case: &str, only: &[&str]) {
let from = format!(
"{}/../bxcore/tests/fixtures/audit/{case}",
env!("CARGO_MANIFEST_DIR")
);
std::fs::create_dir_all(home.path("audit")).unwrap();
for name in only {
std::fs::copy(
format!("{from}/{name}"),
home.path(&format!("audit/{name}")),
)
.unwrap();
}
}
fn snapshot(dir: &Path) -> Vec<(String, Vec<u8>)> {
let mut all: Vec<(String, Vec<u8>)> = std::fs::read_dir(dir)
.unwrap()
.map(|e| e.unwrap())
.filter(|e| e.file_name().to_string_lossy().ends_with(".jsonl"))
.map(|e| {
(
e.file_name().into_string().unwrap(),
std::fs::read(e.path()).unwrap(),
)
})
.collect();
all.sort();
all
}
#[test]
fn a_broken_chain_stops_it_before_any_socket_and_nothing_is_written() {
let home = Home::new("serve-broken", 900_000);
copy_case(&home, "changed-byte", &["2026-09-17.jsonl"]);
let before = snapshot(&home.path("audit"));
let output = home.run(&[]);
assert_eq!(output.status.code(), Some(1));
let text = stderr(&output);
assert!(text.contains("2026-09-17.jsonl:4: "), "{text}");
assert!(
text.trim_end()
.ends_with("see docs/runbook.md#audit-chain-broken"),
"{text}"
);
assert_eq!(snapshot(&home.path("audit")), before);
assert!(!home.tools().exists() && !home.admin().exists());
}
#[test]
fn a_torn_tail_is_recovered_and_it_serves() {
let home = Home::new("serve-torn", 900_000);
copy_case(
&home,
"torn-tail",
&["2026-09-17.jsonl", "2026-09-18.jsonl"],
);
let running = home.serve();
let printed = running.stop();
assert!(printed.contains(RECOVERED_NOTICE), "{printed}");
assert!(RECOVERED_NOTICE.ends_with("see docs/runbook.md#audit-recovered"));
}
#[test]
fn accept_break_with_nothing_to_accept_exits_2() {
let home = Home::new("serve-nothing", 900_000);
let output = home.run(&["--accept-break"]);
assert_eq!(output.status.code(), Some(2));
assert!(
stderr(&output).contains("nothing to accept"),
"{}",
stderr(&output)
);
}
#[test]
fn bad_arguments_and_bad_configs_do_not_start() {
let brokerd = env!("CARGO_BIN_EXE_brokerd");
for args in [
&[][..],
&["serve"][..],
&["serve", "--config"][..],
&["serve", "--config", "a", "--config", "b"][..],
&["serve", "--config", "a", "--loud"][..],
&["run", "--config", "a"][..],
] {
let output = Command::new(brokerd).args(args).output().unwrap();
assert_eq!(output.status.code(), Some(2), "{args:?}");
assert!(
stderr(&output).starts_with("usage: brokerd serve"),
"{args:?}"
);
}
let home = Home::new("serve-config", 900_000);
std::fs::write(&home.config, "[paths]\nhoem = \"/x\"\n").unwrap();
let output = home.run(&[]);
assert_eq!(output.status.code(), Some(1));
assert!(
stderr(&output).contains("brokerd.toml"),
"{}",
stderr(&output)
);
std::fs::remove_file(&home.config).unwrap();
assert_eq!(home.run(&[]).status.code(), Some(1));
assert!(
!home.path("audit").exists(),
"nothing made before the config is read"
);
}
@@ -0,0 +1 @@
{"v":1,"id":9,"final":true,"msg":{"kind":"tool_response","body":{"status":"denied","reason":"invalid_arguments","detail":"invalid argument shape: unknown field `file`, expected `text`"}}}
@@ -0,0 +1,151 @@
//! Every JSON object in every fixture must reject an unknown key. Do not edit.
//!
//! The other test files check unknown fields in a few hand-picked places. This one checks all of
//! them: it walks each fixture, adds one unknown key to one object at a time, at every depth, and
//! requires that the result no longer decodes.
use bxcore::{AuditRecord, Envelope, Grant, LogRecord};
use serde::de::DeserializeOwned;
use serde_json::Value;
/// Every copy of `value` that has exactly one extra key in exactly one object.
fn with_one_unknown_key(value: &Value) -> Vec<Value> {
let mut out = Vec::new();
match value {
Value::Object(map) => {
let mut extended = map.clone();
extended.insert("zz_unknown".to_string(), Value::Bool(true));
out.push(Value::Object(extended));
for (key, child) in map {
for changed in with_one_unknown_key(child) {
let mut copy = map.clone();
copy.insert(key.clone(), changed);
out.push(Value::Object(copy));
}
}
}
Value::Array(items) => {
for (i, child) in items.iter().enumerate() {
for changed in with_one_unknown_key(child) {
let mut copy = items.clone();
copy[i] = changed;
out.push(Value::Array(copy));
}
}
}
_ => {}
}
out
}
/// Returns how many variations were tried, so callers can check the walk reached nested objects.
fn check<T: DeserializeOwned>(what: &str, text: &str) -> usize {
let value: Value = serde_json::from_str(text).unwrap_or_else(|e| panic!("{what}: {e}"));
assert!(
serde_json::from_value::<T>(value.clone()).is_ok(),
"{what}: fixture must decode"
);
let variations = with_one_unknown_key(&value);
for changed in &variations {
assert!(
serde_json::from_value::<T>(changed.clone()).is_err(),
"{what}: accepted an unknown key: {changed}"
);
}
variations.len()
}
fn fixture(path: &str) -> String {
let full = format!("{}/tests/fixtures/{path}", env!("CARGO_MANIFEST_DIR"));
std::fs::read_to_string(&full).unwrap_or_else(|e| panic!("{full}: {e}"))
}
#[test]
fn envelopes_reject_unknown_keys_at_every_depth() {
for name in [
"tool_request.json",
"tool_response_pending.json",
"tool_response_result.json",
"tool_response_failed.json",
"tool_response_denied.json",
"tool_response_denied_detail.json",
"error.json",
"turn.json",
"turn_event_tool_result.json",
"turn_event_content.json",
"turn_event_retrying.json",
"turn_done.json",
"error_session_full.json",
"approvals.json",
"approval_list.json",
"approval_list_empty.json",
"approve.json",
"approve_result_allowed.json",
"approve_result_denied.json",
"refuse.json",
"refuse_no_reason.json",
"ok.json",
"check_grants.json",
"grants_report.json",
"grants_report_ok.json",
"error_forbidden.json",
"error_no_such_approval.json",
"turn_event_approval_pending.json",
"turn_event_tool_denied.json",
] {
// Envelope, msg and body: three objects. Some bodies hold more: turn_done a usage
// object, approval_list one item, approve_result an outcome, grants_report two problems.
let want = match name {
"turn_done.json" | "approval_list.json" => 4,
"approve_result_allowed.json" | "approve_result_denied.json" => 4,
"grants_report.json" => 5,
_ => 3,
};
assert_eq!(
check::<Envelope>(name, &fixture(&format!("wire/{name}"))),
want,
"{name}"
);
}
}
#[test]
fn audit_records_reject_unknown_keys_at_every_depth() {
for (i, line) in fixture("records/audit.jsonl").lines().enumerate() {
// The record and its event: two objects. Decision and approval events also hold an
// outcome object.
let want = if line.contains("\"outcome\"") { 3 } else { 2 };
assert_eq!(
check::<AuditRecord>(&format!("audit.jsonl:{}", i + 1), line),
want
);
}
}
#[test]
fn log_records_reject_unknown_keys_at_every_depth() {
let mut tried = 0;
for (i, line) in fixture("records/session.jsonl").lines().enumerate() {
tried += check::<LogRecord>(&format!("session.jsonl:{}", i + 1), line);
}
// Seven records, plus the one tool call inside the first assistant record.
assert_eq!(tried, 8);
}
#[test]
fn usage_log_records_reject_unknown_keys_at_every_depth() {
let mut tried = 0;
for (i, line) in fixture("records/session_usage.jsonl").lines().enumerate() {
tried += check::<LogRecord>(&format!("session_usage.jsonl:{}", i + 1), line);
}
// Seven records, plus the one tool call inside the first assistant record.
assert_eq!(tried, 8);
}
#[test]
fn grants_reject_unknown_keys_at_every_depth() {
let grant: Grant = toml::from_str(&fixture("grant/full.toml")).unwrap();
let text = serde_json::to_string(&grant).unwrap();
// The grant and its constraints: two objects.
assert_eq!(check::<Grant>("full.toml as JSON", &text), 2);
}
@@ -0,0 +1,219 @@
//! Tests for IPC messages against byte-exact fixtures. Do not edit these or the fixtures.
use bxcore::{
CallId, DataClass, DenyReason, Envelope, ErrorCode, Message, SessionId, Timestamp, ToolRequest,
ToolResponse, WireError,
};
fn fixture(name: &str) -> String {
let path = format!("{}/tests/fixtures/wire/{name}", env!("CARGO_MANIFEST_DIR"));
let text = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("{path}: {e}"));
text.trim_end_matches('\n').to_string()
}
/// The fixture must decode to `want`, and `want` must encode to exactly the fixture's bytes.
fn check(name: &str, want: Envelope) {
let text = fixture(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 response(id: u64, r#final: bool, body: ToolResponse) -> Envelope {
Envelope {
v: 1,
id,
r#final,
msg: Message::ToolResponse(body),
}
}
#[test]
fn tool_request() {
let body = ToolRequest {
session: SessionId::new("mm-thread-42").unwrap(),
call: CallId(3),
tool: "read_file".to_string(),
arguments: r#"{"path":"/etc/hosts"}"#.to_string(),
};
check(
"tool_request.json",
Envelope {
v: 1,
id: 7,
r#final: true,
msg: Message::ToolRequest(body),
},
);
}
#[test]
fn tool_response_pending() {
let body = ToolResponse::PendingApproval {
approval: 41,
expires: Timestamp::parse("2026-09-17T08:35:00.000Z").unwrap(),
};
check("tool_response_pending.json", response(7, false, body));
}
#[test]
fn an_approval_id_is_a_number_not_a_string() {
let good = fixture("tool_response_pending.json");
let bad = good.replacen("\"approval\":41", "\"approval\":\"41\"", 1);
assert_ne!(good, bad);
assert!(serde_json::from_str::<Envelope>(&good).is_ok());
assert!(serde_json::from_str::<Envelope>(&bad).is_err());
let negative = good.replacen("\"approval\":41", "\"approval\":-1", 1);
assert!(serde_json::from_str::<Envelope>(&negative).is_err());
}
#[test]
fn tool_response_result() {
let body = ToolResponse::Result {
content: "127.0.0.1 localhost\n".to_string(),
class: DataClass::Private,
untrusted: true,
truncated: false,
};
check("tool_response_result.json", response(7, true, body));
}
#[test]
fn tool_response_failed() {
let body = ToolResponse::Failed {
message: "exit status 2".to_string(),
};
check("tool_response_failed.json", response(8, true, body));
}
#[test]
fn tool_response_denied() {
let body = ToolResponse::Denied {
reason: DenyReason::TaintTooHigh,
detail: None,
};
check("tool_response_denied.json", response(9, true, body));
}
/// A detail is written only when there is one, so the fixture above has no `detail` key.
#[test]
fn tool_response_denied_with_a_detail() {
let body = ToolResponse::Denied {
reason: DenyReason::InvalidArguments,
detail: Some("invalid argument shape: unknown field `file`, expected `text`".to_string()),
};
check("tool_response_denied_detail.json", response(9, true, body));
}
#[test]
fn a_detail_must_be_a_string() {
let good = fixture("tool_response_denied_detail.json");
assert!(serde_json::from_str::<Envelope>(&good).is_ok());
let at = good.find("\"detail\":").unwrap();
for value in ["7", "[\"a\"]", "{}", "true"] {
let bad = format!("{}\"detail\":{value}}}}}}}", &good[..at]);
assert!(
serde_json::from_str::<Envelope>(&bad).is_err(),
"accepted {bad}"
);
}
}
#[test]
fn error_message() {
let body = WireError {
code: ErrorCode::BadVersion,
detail: "expected 1".to_string(),
};
check(
"error.json",
Envelope {
v: 1,
id: 0,
r#final: true,
msg: Message::Error(body),
},
);
}
#[test]
fn deny_reasons_and_error_codes_are_snake_case() {
let reasons = [
(DenyReason::NoGrant, "no_grant"),
(DenyReason::GrantExpired, "grant_expired"),
(DenyReason::TaintTooHigh, "taint_too_high"),
(DenyReason::DeniedByGrant, "denied_by_grant"),
(DenyReason::ApprovalRefused, "approval_refused"),
(DenyReason::ApprovalExpired, "approval_expired"),
(DenyReason::GrantsInvalid, "grants_invalid"),
(DenyReason::AuditUnavailable, "audit_unavailable"),
(DenyReason::InvalidArguments, "invalid_arguments"),
(DenyReason::StateUnreadable, "state_unreadable"),
];
for (value, text) in reasons {
assert_eq!(
serde_json::to_string(&value).unwrap(),
format!("\"{text}\"")
);
}
let codes = [
(ErrorCode::BadFrame, "bad_frame"),
(ErrorCode::BadVersion, "bad_version"),
(ErrorCode::BadMessage, "bad_message"),
(ErrorCode::Internal, "internal"),
(ErrorCode::Forbidden, "forbidden"),
(ErrorCode::NoSuchApproval, "no_such_approval"),
];
for (value, text) in codes {
assert_eq!(
serde_json::to_string(&value).unwrap(),
format!("\"{text}\"")
);
}
}
#[test]
fn unknown_and_missing_fields_are_rejected() {
let good = fixture("tool_request.json");
assert!(serde_json::from_str::<Envelope>(&good).is_ok());
let bad = [
// extra field in the envelope
good.replacen("{\"v\":1,", "{\"v\":1,\"extra\":0,", 1),
// extra field beside kind and body
good.replacen(
"\"kind\":\"tool_request\",",
"\"kind\":\"tool_request\",\"x\":1,",
1,
),
// extra field in the body
good.replacen("\"call\":3,", "\"call\":3,\"priority\":9,", 1),
// missing field in the body
good.replacen("\"call\":3,", "", 1),
// missing `final`
good.replacen("\"final\":true,", "", 1),
// unknown kind
good.replacen("tool_request", "tool_demand", 1),
// invalid session id inside a message
good.replacen("mm-thread-42", "../../etc", 1),
];
for text in bad {
assert!(
serde_json::from_str::<Envelope>(&text).is_err(),
"accepted {text}"
);
}
}
#[test]
fn unknown_field_in_a_response_variant_is_rejected() {
let good = fixture("tool_response_denied.json");
let bad = good.replacen("\"reason\":", "\"note\":\"x\",\"reason\":", 1);
assert!(serde_json::from_str::<Envelope>(&good).is_ok());
assert!(serde_json::from_str::<Envelope>(&bad).is_err());
let bad_status = good.replacen("denied", "refused", 1);
assert!(serde_json::from_str::<Envelope>(&bad_status).is_err());
}
@@ -0,0 +1,321 @@
//! Tests for `BrokerPort` against a fake `brokerd` on a temporary socket: answers, approvals and
//! timeouts. Frames that break the protocol are in `broker_port_bad.rs`. Do not edit.
#[path = "support/broker.rs"]
mod fake;
use std::io::Write;
use std::os::unix::net::UnixListener;
use std::thread;
use std::time::Duration;
use bxcore::{
DenyReason, Message, PROTOCOL_VERSION, Timestamp, ToolResponse, read_frame, write_frame,
};
use fake::{assert_unavailable, broker, call, frame, in_ms, request, result, send, socket_path};
use loopd::broker_port::{BrokerPort, NoBroker, POINTER, TOO_LARGE};
use loopd::tools::{Pending, ToolPort};
#[test]
fn a_result_comes_back_as_it_was_sent() {
let (socket, broker) = broker(|stream, request| {
send(stream, request.id, true, result("hello\n"));
});
let got = call(socket, 2_000, &request());
assert_eq!(got.response, result("hello\n"));
assert!(got.pending.is_empty());
assert!(got.lines.is_empty(), "nothing to report: {:?}", got.lines);
// What the broker received: one final frame holding exactly the request.
let sent = broker.join().unwrap();
assert_eq!(sent.v, PROTOCOL_VERSION);
assert!(sent.r#final, "a request is a single final frame");
assert_eq!(sent.msg, Message::ToolRequest(request()));
}
#[test]
fn every_denial_and_a_failure_come_back_as_they_were_sent() {
let mut answers: Vec<ToolResponse> = [
DenyReason::NoGrant,
DenyReason::GrantExpired,
DenyReason::TaintTooHigh,
DenyReason::DeniedByGrant,
DenyReason::ApprovalRefused,
DenyReason::ApprovalExpired,
DenyReason::GrantsInvalid,
DenyReason::AuditUnavailable,
DenyReason::InvalidArguments,
DenyReason::StateUnreadable,
]
.into_iter()
.map(|reason| ToolResponse::Denied {
reason,
detail: None,
})
.collect();
answers.push(ToolResponse::Denied {
reason: DenyReason::InvalidArguments,
detail: Some("invalid path: n/a".to_string()),
});
answers.push(ToolResponse::Failed {
message: "the runner arrives in M3b".to_string(),
});
for answer in answers {
let reply = answer.clone();
let (socket, broker) = broker(move |stream, request| {
send(stream, request.id, true, reply);
});
let got = call(socket, 2_000, &request());
assert_eq!(got.response, answer);
assert!(
got.lines.is_empty(),
"a denial is not an outage: {:?}",
got.lines
);
broker.join().unwrap();
}
}
#[test]
fn a_pending_frame_reaches_the_callback_and_the_final_frame_is_the_answer() {
let expires = in_ms(60_000);
let (socket, broker) = broker(move |stream, request| {
let pending = ToolResponse::PendingApproval {
approval: 41,
expires,
};
send(stream, request.id, false, pending);
thread::sleep(Duration::from_millis(150));
send(stream, request.id, true, result("approved"));
});
let got = call(socket, 2_000, &request());
assert_eq!(got.response, result("approved"));
assert_eq!(
got.pending,
[Pending {
approval: 41,
expires
}],
"called once, with the frame's values"
);
assert!(got.lines.is_empty(), "{:?}", got.lines);
broker.join().unwrap();
}
#[test]
fn an_answer_after_expires_but_within_the_timeout_of_it_is_delivered() {
// expires in 100 ms, timeout 1,500 ms: the port waits until 1,600 ms. The answer comes at
// about 400 ms, well after `expires`: an approval given at the last moment still gets the
// whole timeout to run.
let (socket, broker) = broker(|stream, request| {
let pending = ToolResponse::PendingApproval {
approval: 1,
expires: in_ms(100),
};
send(stream, request.id, false, pending);
thread::sleep(Duration::from_millis(400));
send(stream, request.id, true, result("late but good"));
});
let got = call(socket, 1_500, &request());
assert_eq!(got.response, result("late but good"));
assert!(got.lines.is_empty(), "{:?}", got.lines);
broker.join().unwrap();
}
#[test]
fn an_expiry_that_has_already_passed_still_leaves_the_timeout() {
let (socket, broker) = broker(|stream, request| {
let pending = ToolResponse::PendingApproval {
approval: 1,
expires: Timestamp::parse("2020-01-01T00:00:00.000Z").unwrap(),
};
send(stream, request.id, false, pending);
thread::sleep(Duration::from_millis(200));
send(stream, request.id, true, result("fine"));
});
let got = call(socket, 1_500, &request());
assert_eq!(got.response, result("fine"));
broker.join().unwrap();
}
#[test]
fn no_socket_is_unavailable() {
let socket = socket_path();
let _ = std::fs::remove_file(&socket);
let got = call(socket.to_path_buf(), 2_000, &request());
assert_unavailable(&got, "no socket file");
assert!(
got.lines[0].contains(&socket.display().to_string()),
"the line names the socket: {}",
got.lines[0]
);
assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took);
}
#[test]
fn a_broker_that_closes_without_answering_is_unavailable() {
let (socket, broker) = broker(|_, _| {});
let got = call(socket, 2_000, &request());
assert_unavailable(&got, "closed before any frame");
assert!(
got.took < Duration::from_millis(1_500),
"a close is seen at once: {:?}",
got.took
);
broker.join().unwrap();
}
#[test]
fn a_broker_that_closes_while_pending_is_unavailable() {
let (socket, broker) = broker(|stream, request| {
let pending = ToolResponse::PendingApproval {
approval: 3,
expires: in_ms(60_000),
};
send(stream, request.id, false, pending);
// brokerd was restarted: the connection just ends.
});
let got = call(socket, 2_000, &request());
assert_unavailable(&got, "closed while pending");
assert_eq!(got.pending.len(), 1, "the pending frame was reported first");
assert!(got.took < Duration::from_millis(1_500), "{:?}", got.took);
broker.join().unwrap();
}
#[test]
fn a_broker_that_never_answers_is_unavailable_after_the_timeout() {
let (socket, broker) = broker(|_, _| thread::sleep(Duration::from_millis(1_200)));
let got = call(socket, 300, &request());
assert_unavailable(&got, "silence");
assert!(
got.took >= Duration::from_millis(250),
"gave up early: {:?}",
got.took
);
assert!(
got.took < Duration::from_millis(1_100),
"gave up late: {:?}",
got.took
);
broker.join().unwrap();
}
#[test]
fn a_pending_call_nobody_answers_ends_at_expires_plus_the_timeout() {
let (socket, broker) = broker(|stream, request| {
let pending = ToolResponse::PendingApproval {
approval: 3,
expires: in_ms(300),
};
send(stream, request.id, false, pending);
thread::sleep(Duration::from_millis(1_800));
});
let got = call(socket, 300, &request());
assert_unavailable(&got, "silence while pending");
assert!(
got.took >= Duration::from_millis(550),
"it must wait for expires (300) plus the timeout (300): {:?}",
got.took
);
assert!(
got.took < Duration::from_millis(1_700),
"gave up late: {:?}",
got.took
);
broker.join().unwrap();
}
#[test]
fn the_timeout_is_a_deadline_for_the_frame_not_for_each_read() {
// The four header bytes arrive 250 ms apart, the last at 1,000 ms. A port that only sets a
// 600 ms read timeout sees every single read succeed and returns the result; a port with a
// deadline gives up at 600 ms. The answer tells them apart, so no timing is asserted.
let (socket, broker) = broker(|stream, request| {
let mut bytes = Vec::new();
write_frame(
&mut bytes,
&frame(request.id, true, Message::ToolResponse(result("slow"))),
)
.unwrap();
for byte in bytes.iter().take(4) {
thread::sleep(Duration::from_millis(250));
if stream.write_all(&[*byte]).is_err() {
return;
}
}
let _ = stream.write_all(&bytes[4..]);
});
let got = call(socket, 600, &request());
assert_unavailable(&got, "a trickled frame");
broker.join().unwrap();
}
#[test]
fn a_zero_timeout_fails_closed_and_does_not_panic() {
let (socket, _broker) = broker(|stream, request| {
send(stream, request.id, true, result("too late"));
});
let got = call(socket, 0, &request());
assert_unavailable(&got, "timeout_ms = 0");
}
#[test]
fn a_request_too_large_for_a_frame_is_its_own_failure() {
// Nothing is sent, so the fake broker sees a connection that closes or none at all.
let path = socket_path();
let _ = std::fs::remove_file(&path);
let _listener = UnixListener::bind(&path).unwrap();
let mut big = request();
big.arguments = format!(r#"{{"path":"/{}"}}"#, "a".repeat(bxcore::MAX_FRAME));
let got = call(path.to_path_buf(), 1_000, &big);
assert_eq!(
got.response,
ToolResponse::Failed {
message: TOO_LARGE.to_string()
}
);
assert!(
got.lines.is_empty(),
"the broker is fine; this is not an outage: {:?}",
got.lines
);
}
#[test]
fn every_call_is_its_own_connection() {
let path = socket_path();
let _ = std::fs::remove_file(&path);
let listener = UnixListener::bind(&path).unwrap();
let server = thread::spawn(move || {
for n in 0..3u64 {
let (mut stream, _) = listener.accept().unwrap();
let request = read_frame(&mut stream).unwrap();
send(
&mut stream,
request.id,
true,
result(&format!("answer {n}")),
);
}
});
let port = BrokerPort::new(path.to_path_buf(), Duration::from_millis(2_000));
for n in 0..3 {
let got = port.call(&request(), &mut |_| {});
assert_eq!(got, result(&format!("answer {n}")));
}
server.join().unwrap();
}
#[test]
fn without_a_broker_every_call_fails_in_words_the_model_can_repeat() {
let mut seen = 0;
let got = NoBroker.call(&request(), &mut |_| seen += 1);
assert_eq!(
got,
ToolResponse::Failed {
message: "no tool broker is configured".to_string()
}
);
assert_eq!(seen, 0);
assert_eq!(POINTER, "see docs/runbook.md#broker-unavailable");
}
@@ -0,0 +1,335 @@
//! Tests for the registry, dispatch, the denial sentences, the result cap and the fake tools.
//! Do not edit.
mod support;
use bxcore::{CallId, DenyReason, SessionId, ToolRequest, ToolResponse};
use loopd::tools::{
Dispatch, FakeTools, Pending, Registry, ToolPort, cap_result, denial_message, denial_text,
dispatch,
};
fn req(tool: &str, arguments: &str) -> ToolRequest {
ToolRequest {
session: SessionId::new("s").unwrap(),
call: CallId(1),
tool: tool.to_string(),
arguments: arguments.to_string(),
}
}
#[test]
fn the_core_schemas_are_the_fixed_tools_array() {
let names: Vec<String> = Registry::m2b()
.core_schemas()
.into_iter()
.map(|s| s.name)
.collect();
assert_eq!(names, ["clock", "find_tool", "call_tool"]);
let find = &Registry::m2b().core_schemas()[1];
assert_eq!(find.parameters["required"], serde_json::json!(["query"]));
let call = &Registry::m2b().core_schemas()[2];
assert_eq!(
call.parameters["required"],
serde_json::json!(["name", "arguments"])
);
}
#[test]
fn find_matches_name_or_description_case_insensitively() {
let r = Registry::m2b();
let names = |q: &str| r.find(q).iter().map(|s| s.name.clone()).collect::<Vec<_>>();
assert_eq!(names("echo"), ["echo"]);
assert_eq!(names("ECHO"), ["echo"]);
assert_eq!(names("unchanged"), ["echo"], "matches the description too");
assert_eq!(names("time"), ["clock"]);
assert!(names("weather").is_empty());
assert!(
names("").is_empty(),
"an empty query matches nothing, not everything"
);
assert!(names(" ").is_empty());
}
#[test]
fn find_tool_is_answered_locally() {
let r = Registry::m2b();
match dispatch(&r, "find_tool", r#"{"query":"echo"}"#) {
Dispatch::Local(text) => {
assert!(text.starts_with("1 tool(s) match:\n"), "{text}");
assert!(text.contains("\"name\":\"echo\""), "{text}");
assert!(
text.contains("\"parameters\""),
"the schema is included: {text}"
);
assert!(text.ends_with("Call it with call_tool."), "{text}");
}
other => panic!("{other:?}"),
}
assert_eq!(
dispatch(&r, "find_tool", r#"{"query":"weather"}"#),
Dispatch::Local("No tool matches.".to_string())
);
assert!(
matches!(dispatch(&r, "find_tool", "not json"), Dispatch::Local(t) if t == "No tool matches.")
);
}
#[test]
fn call_tool_unwraps_a_known_non_core_tool_and_nothing_else() {
let r = Registry::m2b();
let args = r#"{"name":"echo","arguments":{"text": "box"}}"#;
match dispatch(&r, "call_tool", args) {
Dispatch::Port { tool, arguments } => {
assert_eq!(tool, "echo");
assert_eq!(
serde_json::from_str::<serde_json::Value>(&arguments).unwrap(),
serde_json::json!({"text": "box"})
);
}
other => panic!("{other:?}"),
}
let cases = [
(
r#"{"name":"weather","arguments":{}}"#,
"No tool named \"weather\"",
),
(r#"{"name":"clock","arguments":{}}"#, "core tool"),
(r#"{"arguments":{}}"#, "No tool named \"\""),
("not json", "needs a JSON object"),
];
for (args, want) in cases {
match dispatch(&r, "call_tool", args) {
Dispatch::Local(text) => assert!(text.contains(want), "{args}: {text}"),
other => panic!("{args}: {other:?}, must never reach the port"),
}
}
}
#[test]
fn any_other_tool_goes_to_the_port_as_it_is() {
let r = Registry::m2b();
// Even one the registry does not know: the port (brokerd) decides, not loopd.
let want = Dispatch::Port {
tool: "read_file".to_string(),
arguments: r#"{"path":"/x"}"#.to_string(),
};
assert_eq!(dispatch(&r, "read_file", r#"{"path":"/x"}"#), want);
let want = Dispatch::Port {
tool: "weather".to_string(),
arguments: "not json".to_string(),
};
assert_eq!(dispatch(&r, "weather", "not json"), want);
}
#[test]
fn the_clock_is_answered_locally_whatever_its_arguments() {
for registry in [Registry::m2b(), Registry::m3a(), Registry::new(vec![])] {
for arguments in ["{}", r#"{"zone":"UTC"}"#, "not json", ""] {
let before = bxcore::Timestamp::now();
match dispatch(&registry, "clock", arguments) {
Dispatch::Local(text) => {
let time = bxcore::Timestamp::parse(&text)
.unwrap_or_else(|e| panic!("an RFC 3339 time, got {text:?}: {e:?}"));
assert!(time >= before, "{text}");
assert!(text.ends_with('Z'), "UTC: {text}");
}
other => panic!("{arguments:?}: {other:?}, the clock must not reach the port"),
}
}
}
}
#[test]
fn the_m3a_registry_has_the_same_core_and_the_four_broker_tools() {
let r = Registry::m3a();
assert_eq!(
r.core_schemas(),
Registry::m2b().core_schemas(),
"the tools array is part of the baseline: it must not change"
);
let names = |q: &str| r.find(q).iter().map(|s| s.name.clone()).collect::<Vec<_>>();
assert_eq!(names("file"), ["read_file", "write_file"]);
assert_eq!(names("shell"), ["shell"]);
assert_eq!(names("fetch"), ["http_fetch"]);
assert_eq!(names("https"), ["http_fetch"]);
assert!(names("echo").is_empty(), "echo is a test tool only");
// The argument schemas are section 3's table: exactly these properties, all strings.
let table: [(&str, &[&str], &[&str]); 4] = [
("read_file", &["path"], &["path"]),
("write_file", &["content", "path"], &["path", "content"]),
("shell", &["command", "cwd"], &["command"]),
("http_fetch", &["url"], &["url"]),
];
for (name, properties, required) in table {
let entry = r.get(name).unwrap_or_else(|| panic!("{name} is missing"));
assert!(!entry.core, "{name} is found with find_tool, not declared");
let p = &entry.schema.parameters;
assert_eq!(p["type"], "object", "{name}");
let mut got: Vec<&str> = p["properties"]
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
got.sort_unstable();
assert_eq!(got, properties, "{name}: properties");
for property in properties {
assert_eq!(
p["properties"][property]["type"], "string",
"{name}.{property}"
);
assert!(
p["properties"][property]["description"].is_string(),
"{name}.{property} needs a description"
);
}
assert_eq!(
p["required"],
serde_json::json!(required),
"{name}: required"
);
}
// call_tool lets the four through to the port and nothing else.
match dispatch(
&r,
"call_tool",
r#"{"name":"shell","arguments":{"command":"ls"}}"#,
) {
Dispatch::Port { tool, arguments } => {
assert_eq!(tool, "shell");
assert_eq!(arguments, r#"{"command":"ls"}"#);
}
other => panic!("{other:?}"),
}
assert!(matches!(
dispatch(&r, "call_tool", r#"{"name":"echo","arguments":{}}"#),
Dispatch::Local(t) if t.contains("No tool named \"echo\"")
));
}
#[test]
fn every_deny_reason_has_its_sentence() {
let table = [
(DenyReason::NoGrant, "Denied: no grant allows this call."),
(
DenyReason::GrantExpired,
"Denied: the grant for this call has expired.",
),
(
DenyReason::TaintTooHigh,
"Denied: this session has seen data too sensitive for this call.",
),
(
DenyReason::DeniedByGrant,
"Denied: a grant forbids this call.",
),
(
DenyReason::ApprovalRefused,
"Denied: the owner refused this call.",
),
(
DenyReason::ApprovalExpired,
"Denied: the approval request expired without an answer.",
),
(
DenyReason::InvalidArguments,
"Denied: the arguments are not valid for this tool.",
),
(
DenyReason::GrantsInvalid,
"Denied: the grant files have an error; the owner has been told.",
),
(
DenyReason::AuditUnavailable,
"Denied: the audit log cannot be written; the owner has been told.",
),
(
DenyReason::StateUnreadable,
"Denied: this session's broker state is damaged; the owner has been told.",
),
];
for (reason, want) in table {
assert_eq!(denial_text(reason), want, "{reason:?}");
assert_eq!(
denial_message(reason, None),
want,
"{reason:?}: without a detail, the sentence alone"
);
}
}
/// The broker's detail goes after the sentence, before its full stop.
#[test]
fn a_denial_with_a_detail_appends_it_to_the_sentence() {
assert_eq!(
denial_message(
DenyReason::InvalidArguments,
Some("invalid argument shape: unknown field `file`, expected `text`")
),
"Denied: the arguments are not valid for this tool: invalid argument shape: unknown field `file`, expected `text`."
);
assert_eq!(
denial_message(DenyReason::InvalidArguments, Some("invalid path: n/a")),
"Denied: the arguments are not valid for this tool: invalid path: n/a."
);
assert_eq!(
denial_message(DenyReason::NoGrant, Some("x")),
"Denied: no grant allows this call: x.",
"any reason's detail is shown the same way"
);
}
#[test]
fn results_are_cut_on_a_character_boundary_and_marked() {
assert_eq!(cap_result("short", 100), ("short".to_string(), false));
assert_eq!(cap_result("exactly", 7), ("exactly".to_string(), false));
let (text, truncated) = cap_result("abcdefghij", 4);
assert_eq!(text, "abcd\n[truncated]");
assert!(truncated);
// 3 ASCII bytes, then 2-byte characters: byte 4 is inside a character.
let (text, truncated) = cap_result("abc\u{e9}\u{e9}\u{e9}", 4);
assert_eq!(text, "abc\n[truncated]");
assert!(truncated);
assert_eq!(cap_result("", 0), (String::new(), false));
assert_eq!(cap_result("x", 0), ("\n[truncated]".to_string(), true));
}
#[test]
fn fake_tools_answer_echo_deny_the_rest_and_record_calls() {
let fake = FakeTools::new();
let mut seen = 0;
let mut on_pending = |_: &Pending| seen += 1;
match fake.call(&req("echo", r#"{"text":"box"}"#), &mut on_pending) {
ToolResponse::Result {
content,
class,
untrusted,
truncated,
} => {
assert_eq!(content, "box");
assert_eq!(class, bxcore::DataClass::Public);
assert!(!untrusted && !truncated);
}
other => panic!("{other:?}"),
}
assert!(matches!(
fake.call(&req("echo", r#"{"tex":"box"}"#), &mut on_pending),
ToolResponse::Failed { .. }
));
for tool in ["weather", "read_file", "clock"] {
assert_eq!(
fake.call(&req(tool, "{}"), &mut on_pending),
ToolResponse::Denied {
reason: DenyReason::NoGrant,
detail: None,
},
"{tool}: the clock is loopd's own now, not the port's"
);
}
assert_eq!(seen, 0, "the fake never asks for approval");
assert_eq!(fake.calls().len(), 5);
assert_eq!(fake.calls()[0].tool, "echo");
}
@@ -0,0 +1,269 @@
//! Tests for what the turn loop does with the broker's answers: denials, pending approvals, and
//! a port that misbehaves. Do not edit.
mod support;
#[path = "support/turn.rs"]
mod turn_support;
use bxcore::{DataClass, DenyReason, LogRecord, Timestamp, ToolResponse, TurnEvent};
use support::{Reply, ReturnsPendingPort, ok_result, pending};
use turn_support::{setup, types};
const CHAT: &str = "/v1/chat/completions";
/// The tool events of a turn, in order, as short strings.
fn tool_events(events: &[TurnEvent]) -> Vec<String> {
events
.iter()
.filter_map(|e| match e {
TurnEvent::ToolCallStarted { name } => Some(format!("started {name}")),
TurnEvent::ApprovalPending { approval, tool, .. } => {
Some(format!("pending {approval} {tool}"))
}
TurnEvent::ToolDenied { name, reason } => Some(format!("denied {name} {reason:?}")),
TurnEvent::ToolResult { name, .. } => Some(format!("result {name}")),
_ => None,
})
.collect()
}
#[test]
fn a_denial_is_a_fixed_sentence_for_the_model_and_an_event_for_the_owner() {
let s = setup(vec![ToolResponse::Denied {
reason: DenyReason::TaintTooHigh,
detail: None,
}]);
s.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let (result, events) = s.turn(&mut session, "x");
assert!(
result.is_ok(),
"the turn goes on after a denial: {result:?}"
);
match &session.records()[4] {
LogRecord::ToolResult {
content,
class,
untrusted,
truncated,
..
} => {
assert_eq!(
content,
"Denied: this session has seen data too sensitive for this call."
);
assert_eq!(
(*class, *untrusted, *truncated),
(DataClass::Public, false, false)
);
}
other => panic!("{other:?}"),
}
assert_eq!(
tool_events(&events),
[
"started read_file",
"denied read_file TaintTooHigh",
"result read_file"
],
"the denial comes before the result"
);
// The model reads the sentence in the next request.
let m2 = s.server.requests_to(CHAT)[1].json();
assert_eq!(
m2["messages"][3]["content"],
"Denied: this session has seen data too sensitive for this call."
);
}
/// The broker's detail reaches the model, in the log and in the next request; the owner's event
/// is the same as for any denial.
#[test]
fn an_invalid_call_tells_the_model_why() {
let s = setup(vec![ToolResponse::Denied {
reason: DenyReason::InvalidArguments,
detail: Some("invalid path: n/a".to_string()),
}]);
s.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let (result, events) = s.turn(&mut session, "x");
assert!(result.is_ok(), "{result:?}");
let want = "Denied: the arguments are not valid for this tool: invalid path: n/a.";
match &session.records()[4] {
LogRecord::ToolResult {
content,
class,
untrusted,
truncated,
..
} => {
assert_eq!(content, want);
assert_eq!(
(*class, *untrusted, *truncated),
(DataClass::Public, false, false)
);
}
other => panic!("{other:?}"),
}
assert_eq!(
tool_events(&events),
[
"started read_file",
"denied read_file InvalidArguments",
"result read_file"
]
);
let m2 = s.server.requests_to(CHAT)[1].json();
assert_eq!(m2["messages"][3]["content"], want);
}
#[test]
fn a_denied_call_tool_names_the_target_tool_in_the_denial() {
let s = setup(vec![ToolResponse::Denied {
reason: DenyReason::NoGrant,
detail: None,
}]);
s.server.route(
CHAT,
vec![Reply::fixture("call_tool"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let (result, events) = s.turn(&mut session, "echo box");
assert!(result.is_ok(), "{result:?}");
assert_eq!(
tool_events(&events),
[
"started call_tool",
"denied echo NoGrant",
"result call_tool"
],
"the owner writes grants for `echo`, not for `call_tool`"
);
assert!(
matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: no grant allows this call.")
);
}
#[test]
fn a_pending_approval_is_an_event_and_the_answer_after_it_is_the_result() {
let s = setup(vec![
pending(41, "2026-09-18T12:15:00.000Z"),
ok_result("straylight\n"),
]);
s.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let (result, events) = s.turn(&mut session, "x");
assert!(result.is_ok(), "{result:?}");
assert_eq!(
tool_events(&events),
[
"started read_file",
"pending 41 read_file",
"result read_file"
]
);
let expires = events.iter().find_map(|e| match e {
TurnEvent::ApprovalPending { expires, .. } => Some(*expires),
_ => None,
});
assert_eq!(
expires,
Some(Timestamp::parse("2026-09-18T12:15:00.000Z").unwrap())
);
assert_eq!(s.port.calls().len(), 1, "one call, however long it waited");
assert_eq!(
types(session.records()),
[
"start",
"user",
"assistant",
"usage",
"tool_result",
"assistant",
"usage"
],
"waiting writes nothing to the log"
);
assert!(
matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "straylight\n")
);
}
#[test]
fn a_pending_approval_that_ends_in_a_refusal() {
let s = setup(vec![
pending(7, "2026-09-18T12:15:00.000Z"),
ToolResponse::Denied {
reason: DenyReason::ApprovalRefused,
detail: None,
},
]);
s.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let (result, events) = s.turn(&mut session, "x");
assert!(result.is_ok(), "{result:?}");
assert_eq!(
tool_events(&events),
[
"started read_file",
"pending 7 read_file",
"denied read_file ApprovalRefused",
"result read_file"
]
);
assert!(
matches!(&session.records()[4], LogRecord::ToolResult { content, .. } if content == "Denied: the owner refused this call.")
);
}
#[test]
fn a_port_that_returns_a_pending_frame_as_its_answer_is_a_failure_not_a_decision() {
let s = setup(vec![]);
s.server.route(
CHAT,
vec![Reply::fixture("tool_call"), Reply::fixture("plain")],
);
let mut session = s.session("a");
let runtime = loopd::turn::Runtime {
cfg: &s.cfg,
client: &s.client,
port: &ReturnsPendingPort,
registry: &s.registry,
};
let mut events = Vec::new();
let result =
loopd::turn::run_turn(&mut session, &runtime, "x", &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"
);
}