M5b plan: task 09, TurnDone.untrusted from the session's log
bxcore's TurnDone gains untrusted: bool, always written and required when read (no serde default: a missing field would read as a clean session, the wrong way to fail). loopd's new Session::untrusted is true once any ToolResult in the log is labelled untrusted, and channel::handle puts it in the TurnDone; the log is the one source, so it holds across resumes and restarts. gatewayd, bxctl and brokerd sources are unchanged; their tests that build a TurnDone get new versions. Task 01's broker.rs is kept as broker.rs-task01, and task 01 now copies it. Checks: the given tests fail at ref 08 as the task says (strict's turn_done.json, and 13 test targets with E0560/E0609 only); the task's edits applied from its text to a fresh ref 08 tree are byte-identical to the reference; the gate passes with 1058 tests; 12 mutations caught; /private/tmp and TMPDIR unchanged around the runs; no tag in plan or given files. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -165,6 +165,7 @@ fn every_kind_but_tool_request() -> Vec<Message> {
|
||||
Message::TurnDone(TurnDone {
|
||||
content: String::new(),
|
||||
usage,
|
||||
untrusted: false,
|
||||
}),
|
||||
Message::Approvals(Empty {}),
|
||||
Message::ApprovalList(ApprovalList { items: Vec::new() }),
|
||||
|
||||
@@ -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 @@
|
||||
{"v":1,"id":3,"final":true,"msg":{"kind":"turn_done","body":{"content":"It is noon.","usage":{"cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false},"untrusted":false}}}
|
||||
+1
@@ -0,0 +1 @@
|
||||
{"v":1,"id":4,"final":true,"msg":{"kind":"turn_done","body":{"content":"The page says to run the script.","usage":{"cache_n":566,"prompt_n":412,"predicted_n":31,"reasoning_tokens":12,"thinking_capped":false},"untrusted":true}}}
|
||||
@@ -0,0 +1,320 @@
|
||||
//! Tests for the channel messages and the usage record, against byte-exact fixtures. Do not edit.
|
||||
|
||||
use bxcore::{
|
||||
CallId, DataClass, DenyReason, Envelope, Epoch, ErrorCode, Hash32, LogRecord, Message,
|
||||
SessionId, Timestamp, ToolCall, Turn, TurnDone, TurnEvent, Usage, WireError,
|
||||
};
|
||||
|
||||
fn fixture(kind: &str, name: &str) -> String {
|
||||
let path = format!(
|
||||
"{}/tests/fixtures/{kind}/{name}",
|
||||
env!("CARGO_MANIFEST_DIR")
|
||||
);
|
||||
std::fs::read_to_string(&path)
|
||||
.unwrap_or_else(|e| panic!("{path}: {e}"))
|
||||
.trim_end_matches('\n')
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn check(name: &str, want: Envelope) {
|
||||
let text = fixture("wire", name);
|
||||
let got: Envelope = serde_json::from_str(&text).unwrap_or_else(|e| panic!("{name}: {e}"));
|
||||
assert_eq!(got, want, "{name}: decoded value");
|
||||
assert_eq!(
|
||||
serde_json::to_string(&want).unwrap(),
|
||||
text,
|
||||
"{name}: encoded bytes"
|
||||
);
|
||||
}
|
||||
|
||||
fn env(id: u64, r#final: bool, msg: Message) -> Envelope {
|
||||
Envelope {
|
||||
v: 1,
|
||||
id,
|
||||
r#final,
|
||||
msg,
|
||||
}
|
||||
}
|
||||
|
||||
fn usage() -> Usage {
|
||||
Usage {
|
||||
cache_n: 539,
|
||||
prompt_n: 27,
|
||||
predicted_n: 24,
|
||||
reasoning_tokens: 8,
|
||||
thinking_capped: false,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn turn() {
|
||||
let body = Turn {
|
||||
session: SessionId::new("chat-1789700000-42").unwrap(),
|
||||
content: "What time is it?".to_string(),
|
||||
resume: false,
|
||||
};
|
||||
check("turn.json", env(3, true, Message::Turn(body)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn turn_events() {
|
||||
check(
|
||||
"turn_event_tool_result.json",
|
||||
env(
|
||||
3,
|
||||
false,
|
||||
Message::TurnEvent(TurnEvent::ToolResult {
|
||||
name: "clock".to_string(),
|
||||
class: DataClass::Public,
|
||||
truncated: false,
|
||||
}),
|
||||
),
|
||||
);
|
||||
check(
|
||||
"turn_event_content.json",
|
||||
env(
|
||||
3,
|
||||
false,
|
||||
Message::TurnEvent(TurnEvent::Content {
|
||||
text: "It is ".to_string(),
|
||||
}),
|
||||
),
|
||||
);
|
||||
check(
|
||||
"turn_event_retrying.json",
|
||||
env(
|
||||
3,
|
||||
false,
|
||||
Message::TurnEvent(TurnEvent::Retrying {
|
||||
attempt: 2,
|
||||
after_ms: 1500,
|
||||
error: "the server went silent".to_string(),
|
||||
}),
|
||||
),
|
||||
);
|
||||
check(
|
||||
"turn_event_approval_pending.json",
|
||||
env(
|
||||
3,
|
||||
false,
|
||||
Message::TurnEvent(TurnEvent::ApprovalPending {
|
||||
approval: 41,
|
||||
tool: "shell".to_string(),
|
||||
expires: Timestamp::parse("2026-09-18T08:20:00.000Z").unwrap(),
|
||||
}),
|
||||
),
|
||||
);
|
||||
check(
|
||||
"turn_event_tool_denied.json",
|
||||
env(
|
||||
3,
|
||||
false,
|
||||
Message::TurnEvent(TurnEvent::ToolDenied {
|
||||
name: "read_file".to_string(),
|
||||
reason: DenyReason::NoGrant,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn turn_done_and_the_new_error_codes() {
|
||||
check(
|
||||
"turn_done.json",
|
||||
env(
|
||||
3,
|
||||
true,
|
||||
Message::TurnDone(TurnDone {
|
||||
content: "It is noon.".to_string(),
|
||||
usage: usage(),
|
||||
untrusted: false,
|
||||
}),
|
||||
),
|
||||
);
|
||||
let body = WireError {
|
||||
code: ErrorCode::SessionFull,
|
||||
detail: "this conversation is full; start a new one".to_string(),
|
||||
};
|
||||
check(
|
||||
"error_session_full.json",
|
||||
env(3, true, Message::Error(body)),
|
||||
);
|
||||
let codes = [
|
||||
(ErrorCode::SessionFull, "session_full"),
|
||||
(ErrorCode::TurnLimit, "turn_limit"),
|
||||
(ErrorCode::SessionBusy, "session_busy"),
|
||||
(ErrorCode::NoSuchSession, "no_such_session"),
|
||||
(ErrorCode::SessionExists, "session_exists"),
|
||||
(ErrorCode::Inference, "inference"),
|
||||
];
|
||||
for (value, text) in codes {
|
||||
assert_eq!(
|
||||
serde_json::to_string(&value).unwrap(),
|
||||
format!("\"{text}\"")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// A reply from a session that has read untrusted content says so, and `untrusted` is written
|
||||
/// whether it is true or false (M5b spec, section 9).
|
||||
#[test]
|
||||
fn turn_done_from_an_untrusted_session() {
|
||||
check(
|
||||
"turn_done_untrusted.json",
|
||||
env(
|
||||
4,
|
||||
true,
|
||||
Message::TurnDone(TurnDone {
|
||||
content: "The page says to run the script.".to_string(),
|
||||
usage: Usage {
|
||||
cache_n: 566,
|
||||
prompt_n: 412,
|
||||
predicted_n: 31,
|
||||
reasoning_tokens: 12,
|
||||
thinking_capped: false,
|
||||
},
|
||||
untrusted: true,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/// `untrusted` has no default: a `turn_done` without it, as a `loopd` from before M5b sent, is
|
||||
/// refused, never read as a reply from a clean session.
|
||||
#[test]
|
||||
fn a_turn_done_without_untrusted_is_refused() {
|
||||
let old = r#"{"v":1,"id":3,"final":true,"msg":{"kind":"turn_done","body":{"content":"It is noon.","usage":{"cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false}}}}"#;
|
||||
let error = serde_json::from_str::<Envelope>(old).unwrap_err();
|
||||
assert!(
|
||||
error.to_string().contains("missing field `untrusted`"),
|
||||
"{error}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_turn_event_kind_round_trips() {
|
||||
let all = vec![
|
||||
TurnEvent::Queued { ahead: 1 },
|
||||
TurnEvent::Waiting { slot_busy: true },
|
||||
TurnEvent::Progress {
|
||||
total: 100,
|
||||
cache: 50,
|
||||
processed: 75,
|
||||
},
|
||||
TurnEvent::Reasoning {
|
||||
text: "hm".to_string(),
|
||||
},
|
||||
TurnEvent::Content {
|
||||
text: "hi".to_string(),
|
||||
},
|
||||
TurnEvent::ToolCallStarted {
|
||||
name: "clock".to_string(),
|
||||
},
|
||||
TurnEvent::ToolResult {
|
||||
name: "clock".to_string(),
|
||||
class: DataClass::Secret,
|
||||
truncated: true,
|
||||
},
|
||||
TurnEvent::ThinkingCapped { tokens: 4096 },
|
||||
TurnEvent::Retrying {
|
||||
attempt: 2,
|
||||
after_ms: 10,
|
||||
error: "x".to_string(),
|
||||
},
|
||||
TurnEvent::CacheLoss {
|
||||
expected: 500,
|
||||
got: 20,
|
||||
},
|
||||
TurnEvent::ApprovalPending {
|
||||
approval: u64::MAX,
|
||||
tool: "http_fetch".to_string(),
|
||||
expires: Timestamp::parse("2026-09-18T08:20:00.000Z").unwrap(),
|
||||
},
|
||||
TurnEvent::ToolDenied {
|
||||
name: "shell".to_string(),
|
||||
reason: DenyReason::StateUnreadable,
|
||||
},
|
||||
];
|
||||
for event in all {
|
||||
let text = serde_json::to_string(&event).unwrap();
|
||||
assert!(text.starts_with("{\"event\":\""), "{text}");
|
||||
assert_eq!(serde_json::from_str::<TurnEvent>(&text).unwrap(), event);
|
||||
}
|
||||
assert!(serde_json::from_str::<TurnEvent>(r#"{"event":"content","text":"x","zz":1}"#).is_err());
|
||||
assert!(serde_json::from_str::<TurnEvent>(r#"{"event":"dance"}"#).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn usage_records() {
|
||||
let text = fixture("records", "session_usage.jsonl");
|
||||
let lines: Vec<&str> = text.lines().collect();
|
||||
let want = [
|
||||
LogRecord::SessionStart {
|
||||
time: Timestamp::parse("2026-09-18T08:05:00.000Z").unwrap(),
|
||||
session: SessionId::new("chat-1789700000-42").unwrap(),
|
||||
epoch: Epoch(0),
|
||||
slot: 0,
|
||||
baseline: Hash32::from_hex(
|
||||
"000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f",
|
||||
)
|
||||
.unwrap(),
|
||||
},
|
||||
LogRecord::User {
|
||||
time: Timestamp::parse("2026-09-18T08:05:01.000Z").unwrap(),
|
||||
content: "What time is it?".to_string(),
|
||||
},
|
||||
LogRecord::Assistant {
|
||||
time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(),
|
||||
content: None,
|
||||
reasoning_content: None,
|
||||
tool_calls: vec![ToolCall {
|
||||
id: "call_a1".to_string(),
|
||||
name: "clock".to_string(),
|
||||
arguments: "{}".to_string(),
|
||||
}],
|
||||
},
|
||||
LogRecord::Usage {
|
||||
time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(),
|
||||
cache_n: 539,
|
||||
prompt_n: 27,
|
||||
predicted_n: 24,
|
||||
reasoning_tokens: 8,
|
||||
thinking_capped: false,
|
||||
},
|
||||
LogRecord::ToolResult {
|
||||
time: Timestamp::parse("2026-09-18T08:05:03.100Z").unwrap(),
|
||||
call: CallId(1),
|
||||
tool_call_id: "call_a1".to_string(),
|
||||
content: "2026-09-18T08:05:03.000Z".to_string(),
|
||||
class: DataClass::Public,
|
||||
untrusted: false,
|
||||
truncated: false,
|
||||
},
|
||||
LogRecord::Assistant {
|
||||
time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(),
|
||||
content: Some("It is five past eight.".to_string()),
|
||||
reasoning_content: None,
|
||||
tool_calls: vec![],
|
||||
},
|
||||
LogRecord::Usage {
|
||||
time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(),
|
||||
cache_n: 589,
|
||||
prompt_n: 40,
|
||||
predicted_n: 64,
|
||||
reasoning_tokens: 27,
|
||||
thinking_capped: true,
|
||||
},
|
||||
];
|
||||
assert_eq!(lines.len(), want.len());
|
||||
for (i, (line, want)) in lines.iter().zip(&want).enumerate() {
|
||||
let got: LogRecord =
|
||||
serde_json::from_str(line).unwrap_or_else(|e| panic!("line {}: {e}", i + 1));
|
||||
assert_eq!(&got, want, "line {}", i + 1);
|
||||
assert_eq!(
|
||||
&serde_json::to_string(want).unwrap(),
|
||||
line,
|
||||
"line {}: bytes",
|
||||
i + 1
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,469 @@
|
||||
//! Tests for `bxctl chat`, against a fake `loopd` that speaks the channel protocol. Do not edit.
|
||||
|
||||
#[path = "support/temp.rs"]
|
||||
mod temp;
|
||||
use bxcore::{
|
||||
DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnDone,
|
||||
TurnEvent, Usage, WireError, read_frame, write_frame,
|
||||
};
|
||||
use bxctl::chat::{ChatError, Printer, new_session_id, run_turn};
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread;
|
||||
|
||||
/// What the fake `loopd` sends back for one turn, after the events: done or an error.
|
||||
#[derive(Clone)]
|
||||
enum End {
|
||||
Done(TurnDone),
|
||||
Error(ErrorCode, &'static str),
|
||||
}
|
||||
|
||||
struct FakeLoopd {
|
||||
socket: PathBuf,
|
||||
turns: Arc<Mutex<Vec<Turn>>>,
|
||||
_dir: temp::TempDir,
|
||||
}
|
||||
|
||||
fn usage() -> Usage {
|
||||
Usage {
|
||||
cache_n: 10,
|
||||
prompt_n: 5,
|
||||
predicted_n: 7,
|
||||
reasoning_tokens: 3,
|
||||
thinking_capped: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Serves every connection with the same script: the given events, then `end`. Records the
|
||||
/// turns it received.
|
||||
fn fake_loopd(events: Vec<TurnEvent>, end: End) -> FakeLoopd {
|
||||
let dir = temp::TempDir::new("bxctl-test");
|
||||
let socket = dir.join("loop.sock");
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let turns = Arc::new(Mutex::new(Vec::new()));
|
||||
let seen = Arc::clone(&turns);
|
||||
thread::spawn(move || {
|
||||
for stream in listener.incoming() {
|
||||
let mut stream = stream.unwrap();
|
||||
let request = read_frame(&mut stream).unwrap();
|
||||
let Message::Turn(turn) = request.msg else {
|
||||
panic!("not a turn")
|
||||
};
|
||||
seen.lock().unwrap().push(turn.clone());
|
||||
let id = request.id;
|
||||
let end = if turn.resume && turn.content == "trigger-no-such-session" {
|
||||
End::Error(ErrorCode::NoSuchSession, "session x does not exist")
|
||||
} else if turn.content == "trigger-turn-limit" {
|
||||
End::Error(ErrorCode::TurnLimit, "the turn hit a limit")
|
||||
} else {
|
||||
end.clone()
|
||||
};
|
||||
for e in &events {
|
||||
write_frame(
|
||||
&mut stream,
|
||||
&Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id,
|
||||
r#final: false,
|
||||
msg: Message::TurnEvent(e.clone()),
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
let last = match end {
|
||||
End::Done(done) => Message::TurnDone(done),
|
||||
End::Error(code, detail) => Message::Error(WireError {
|
||||
code,
|
||||
detail: detail.to_string(),
|
||||
}),
|
||||
};
|
||||
write_frame(
|
||||
&mut stream,
|
||||
&Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id,
|
||||
r#final: true,
|
||||
msg: last,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
});
|
||||
FakeLoopd {
|
||||
socket,
|
||||
turns,
|
||||
_dir: dir,
|
||||
}
|
||||
}
|
||||
|
||||
fn events() -> Vec<TurnEvent> {
|
||||
vec![
|
||||
TurnEvent::Queued { ahead: 1 },
|
||||
TurnEvent::Waiting { slot_busy: true },
|
||||
TurnEvent::Progress {
|
||||
total: 100,
|
||||
cache: 50,
|
||||
processed: 75,
|
||||
},
|
||||
TurnEvent::Reasoning {
|
||||
text: "let me ".to_string(),
|
||||
},
|
||||
TurnEvent::Reasoning {
|
||||
text: "think".to_string(),
|
||||
},
|
||||
TurnEvent::ToolCallStarted {
|
||||
name: "clock".to_string(),
|
||||
},
|
||||
TurnEvent::ToolResult {
|
||||
name: "clock".to_string(),
|
||||
class: DataClass::Public,
|
||||
truncated: true,
|
||||
},
|
||||
TurnEvent::ThinkingCapped { tokens: 4096 },
|
||||
TurnEvent::Retrying {
|
||||
attempt: 2,
|
||||
after_ms: 1500,
|
||||
error: "the server went silent".to_string(),
|
||||
},
|
||||
TurnEvent::CacheLoss {
|
||||
expected: 500,
|
||||
got: 20,
|
||||
},
|
||||
TurnEvent::Content {
|
||||
text: "It is ".to_string(),
|
||||
},
|
||||
TurnEvent::Content {
|
||||
text: "noon.".to_string(),
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
fn done() -> End {
|
||||
End::Done(TurnDone {
|
||||
content: "It is noon.".to_string(),
|
||||
usage: usage(),
|
||||
untrusted: false,
|
||||
})
|
||||
}
|
||||
|
||||
fn id(s: &str) -> SessionId {
|
||||
SessionId::new(s).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_turn_delivers_every_event_in_order_then_the_answer() {
|
||||
let fake = fake_loopd(events(), done());
|
||||
let mut seen = Vec::new();
|
||||
let done = run_turn(
|
||||
&fake.socket,
|
||||
&id("s1"),
|
||||
"what time is it?",
|
||||
false,
|
||||
&mut |e| seen.push(e.clone()),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(seen, events());
|
||||
assert_eq!(done.content, "It is noon.");
|
||||
assert_eq!(done.usage, usage());
|
||||
let turns = fake.turns.lock().unwrap();
|
||||
assert_eq!(turns.len(), 1);
|
||||
assert_eq!(
|
||||
turns[0],
|
||||
Turn {
|
||||
session: id("s1"),
|
||||
content: "what time is it?".to_string(),
|
||||
resume: false
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_error_frame_is_refused_with_its_code_and_detail() {
|
||||
let fake = fake_loopd(
|
||||
vec![TurnEvent::Content {
|
||||
text: "partial".to_string(),
|
||||
}],
|
||||
End::Error(ErrorCode::SessionFull, "this conversation is full"),
|
||||
);
|
||||
let mut seen = 0;
|
||||
let err = run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| seen += 1).unwrap_err();
|
||||
assert_eq!(seen, 1, "events before the error are still delivered");
|
||||
match err {
|
||||
ChatError::Refused(w) => {
|
||||
assert_eq!(w.code, ErrorCode::SessionFull);
|
||||
assert_eq!(w.detail, "this conversation is full");
|
||||
}
|
||||
other => panic!("{other:?}"),
|
||||
}
|
||||
let e: Box<dyn std::error::Error> =
|
||||
Box::new(run_turn(&fake.socket, &id("s1"), "x", true, &mut |_| {}).unwrap_err());
|
||||
assert!(e.to_string().contains("session full"), "{e}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_loopd_is_a_connect_error() {
|
||||
let missing = std::env::temp_dir().join("bxctl-no-such-loopd.sock");
|
||||
assert!(matches!(
|
||||
run_turn(&missing, &id("s1"), "x", false, &mut |_| {}),
|
||||
Err(ChatError::Connect(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn new_session_ids_are_valid_and_distinct() {
|
||||
let a = new_session_id();
|
||||
let b = new_session_id();
|
||||
assert!(a.as_str().starts_with("chat-"));
|
||||
assert_ne!(a, b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_printer_formats_each_event_kind() {
|
||||
let mut out = Vec::new();
|
||||
let mut p = Printer::new(true, false);
|
||||
for e in events() {
|
||||
p.event(&mut out, &e).unwrap();
|
||||
}
|
||||
p.end_reasoning(&mut out).unwrap();
|
||||
let text = String::from_utf8(out).unwrap();
|
||||
assert!(
|
||||
text.contains("\x1b[2mlet me think\x1b[0m\n"),
|
||||
"reasoning dimmed, joined, and ended once: {text:?}"
|
||||
);
|
||||
assert!(text.contains("[tool clock]\n"), "{text:?}");
|
||||
assert!(text.contains("[clock: Public, truncated]\n"), "{text:?}");
|
||||
assert!(text.contains("[waiting: slot busy]\n"), "{text:?}");
|
||||
assert!(
|
||||
text.contains("[retrying: attempt 2 in 1500 ms: the server went silent]\n"),
|
||||
"{text:?}"
|
||||
);
|
||||
assert!(
|
||||
text.contains("[thinking capped at 4096 tokens]\n"),
|
||||
"{text:?}"
|
||||
);
|
||||
assert!(text.contains("[cache loss: 20 of 500]\n"), "{text:?}");
|
||||
assert!(
|
||||
text.ends_with("It is noon."),
|
||||
"content streams as it is: {text:?}"
|
||||
);
|
||||
assert!(
|
||||
!text.contains("Queued") && !text.contains("Progress"),
|
||||
"queued and progress are silent: {text:?}"
|
||||
);
|
||||
|
||||
let mut out = Vec::new();
|
||||
let mut p = Printer::new(false, false);
|
||||
for e in events() {
|
||||
p.event(&mut out, &e).unwrap();
|
||||
}
|
||||
let text = String::from_utf8(out).unwrap();
|
||||
assert!(
|
||||
!text.contains("let me"),
|
||||
"--no-thinking hides reasoning: {text:?}"
|
||||
);
|
||||
assert!(
|
||||
!text.contains("\x1b["),
|
||||
"and no escape codes are left: {text:?}"
|
||||
);
|
||||
|
||||
let mut out = Vec::new();
|
||||
let mut p = Printer::new(true, true);
|
||||
for e in events() {
|
||||
p.event(&mut out, &e).unwrap();
|
||||
}
|
||||
let text = String::from_utf8(out).unwrap();
|
||||
let lines: Vec<&str> = text.lines().collect();
|
||||
assert_eq!(
|
||||
lines.len(),
|
||||
events().len(),
|
||||
"--json: one line per event, none skipped"
|
||||
);
|
||||
let first: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
|
||||
assert_eq!(first, serde_json::json!({"event": "queued", "ahead": 1}));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn say_prints_only_the_answer_on_stdout() {
|
||||
let fake = fake_loopd(events(), done());
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.args(["--session", "scripted-1", "--say", "what time is it?"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(output.status.success());
|
||||
assert_eq!(String::from_utf8_lossy(&output.stdout), "It is noon.\n");
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
assert!(
|
||||
stderr.contains("[tool clock]"),
|
||||
"events go to stderr: {stderr}"
|
||||
);
|
||||
let turns = fake.turns.lock().unwrap();
|
||||
assert_eq!(turns.len(), 1);
|
||||
assert_eq!(turns[0].session, id("scripted-1"));
|
||||
assert!(
|
||||
turns[0].resume,
|
||||
"a session given on the command line is resumed"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn say_creates_a_named_session_that_does_not_exist_yet() {
|
||||
let fake = fake_loopd(vec![], done());
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.args(["--session", "fresh", "--say", "trigger-no-such-session"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
let turns = fake.turns.lock().unwrap();
|
||||
assert_eq!(turns.len(), 2, "resume was refused, so it was created");
|
||||
assert!(turns[0].resume && !turns[1].resume);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn say_reports_an_error_on_stderr_with_status_1() {
|
||||
let fake = fake_loopd(
|
||||
vec![],
|
||||
End::Error(ErrorCode::TurnLimit, "the turn hit a limit"),
|
||||
);
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.args(["--say", "x"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert_eq!(output.status.code(), Some(1));
|
||||
assert_eq!(String::from_utf8_lossy(&output.stdout), "");
|
||||
assert!(String::from_utf8_lossy(&output.stderr).contains("turn limit: the turn hit a limit"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_mode_prints_frames_as_json_lines() {
|
||||
let fake = fake_loopd(events(), done());
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.args(["--session", "j", "--json", "--say", "x"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(output.status.success());
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
let lines: Vec<&str> = stderr.lines().collect();
|
||||
assert_eq!(
|
||||
lines.len(),
|
||||
events().len() + 1,
|
||||
"every event, then the done frame: {stderr}"
|
||||
);
|
||||
let last: serde_json::Value = serde_json::from_str(lines[lines.len() - 1]).unwrap();
|
||||
assert_eq!(last["content"], "It is noon.");
|
||||
assert_eq!(last["usage"]["cache_n"], 10);
|
||||
assert_eq!(last["untrusted"], false, "the done frame is printed whole");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn interactive_mode_reads_lines_until_quit() {
|
||||
let fake = fake_loopd(
|
||||
vec![TurnEvent::Content {
|
||||
text: "ok".to_string(),
|
||||
}],
|
||||
End::Done(TurnDone {
|
||||
content: "ok".to_string(),
|
||||
usage: usage(),
|
||||
untrusted: false,
|
||||
}),
|
||||
);
|
||||
let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.stdin(std::process::Stdio::piped())
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.spawn()
|
||||
.unwrap();
|
||||
{
|
||||
let mut stdin = child.stdin.take().unwrap();
|
||||
std::io::Write::write_all(&mut stdin, b"first\n\nsecond\n/quit\nnever sent\n").unwrap();
|
||||
}
|
||||
let output = child.wait_with_output().unwrap();
|
||||
assert!(output.status.success());
|
||||
let stdout = String::from_utf8_lossy(&output.stdout);
|
||||
assert!(stdout.starts_with("session chat-"), "{stdout}");
|
||||
assert_eq!(stdout.matches("ok").count(), 2, "{stdout}");
|
||||
let turns = fake.turns.lock().unwrap();
|
||||
assert_eq!(
|
||||
turns.len(),
|
||||
2,
|
||||
"a blank line sends nothing, and /quit stops"
|
||||
);
|
||||
assert!(
|
||||
!turns[0].resume && turns[1].resume,
|
||||
"the first turn creates, the second resumes"
|
||||
);
|
||||
assert_eq!(turns[0].session, turns[1].session);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bad_arguments_print_usage() {
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--session", "Not Valid!"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert_eq!(output.status.code(), Some(2));
|
||||
assert!(String::from_utf8_lossy(&output.stderr).contains("usage"));
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["dance"])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert_eq!(output.status.code(), Some(2));
|
||||
}
|
||||
|
||||
/// A failed turn is reported, and the conversation goes on: the session still exists and the
|
||||
/// next line is a new turn on it.
|
||||
#[test]
|
||||
fn interactive_mode_survives_a_failed_turn() {
|
||||
let fake = fake_loopd(
|
||||
vec![TurnEvent::Content {
|
||||
text: "ok".to_string(),
|
||||
}],
|
||||
End::Done(TurnDone {
|
||||
content: "ok".to_string(),
|
||||
usage: usage(),
|
||||
untrusted: false,
|
||||
}),
|
||||
);
|
||||
let mut child = Command::new(env!("CARGO_BIN_EXE_bxctl"))
|
||||
.args(["chat", "--socket"])
|
||||
.arg(&fake.socket)
|
||||
.stdin(std::process::Stdio::piped())
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.spawn()
|
||||
.unwrap();
|
||||
{
|
||||
let mut stdin = child.stdin.take().unwrap();
|
||||
std::io::Write::write_all(&mut stdin, b"first\ntrigger-turn-limit\nthird\n/quit\n")
|
||||
.unwrap();
|
||||
}
|
||||
let output = child.wait_with_output().unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"a failed turn does not end the chat"
|
||||
);
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
assert!(
|
||||
stderr.contains("turn limit: the turn hit a limit"),
|
||||
"{stderr}"
|
||||
);
|
||||
let turns = fake.turns.lock().unwrap();
|
||||
assert_eq!(turns.len(), 3, "the turn after the failure was sent");
|
||||
assert!(turns[2].resume, "and it resumed the same session");
|
||||
assert_eq!(turns[2].session, turns[0].session);
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
//! Scripted stand-ins for `brokerd`'s admin socket and for `loopd`, for tests. Do not edit.
|
||||
|
||||
#![allow(dead_code)] // each test file uses a different part of this module
|
||||
|
||||
use bxcore::{
|
||||
ApprovalList, CallId, DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION,
|
||||
PendingApproval, SessionId, Timestamp, TurnDone, TurnEvent, Usage, WireError, read_frame,
|
||||
write_frame,
|
||||
};
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread;
|
||||
|
||||
static NEXT: AtomicU32 = AtomicU32::new(0);
|
||||
|
||||
/// A socket path in a fresh temporary directory, which is removed when the value is dropped
|
||||
/// (tip T44).
|
||||
pub struct TempSocket {
|
||||
dir: PathBuf,
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
impl std::ops::Deref for TempSocket {
|
||||
type Target = std::path::Path;
|
||||
fn deref(&self) -> &std::path::Path {
|
||||
&self.path
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRef<std::path::Path> for TempSocket {
|
||||
fn as_ref(&self) -> &std::path::Path {
|
||||
&self.path
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TempSocket {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_dir_all(&self.dir);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn temp_socket(name: &str) -> TempSocket {
|
||||
let n = NEXT.fetch_add(1, Ordering::SeqCst);
|
||||
let dir = std::env::temp_dir().join(format!("bxctl-m3a-{}-{n}", std::process::id()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let path = dir.join(name);
|
||||
TempSocket { dir, path }
|
||||
}
|
||||
|
||||
pub fn ts(text: &str) -> Timestamp {
|
||||
Timestamp::parse(text).unwrap()
|
||||
}
|
||||
|
||||
pub struct FakeBrokerd {
|
||||
pub socket: TempSocket,
|
||||
/// Every request message received, in order.
|
||||
pub requests: Arc<Mutex<Vec<Message>>>,
|
||||
}
|
||||
|
||||
impl FakeBrokerd {
|
||||
pub fn requests(&self) -> Vec<Message> {
|
||||
self.requests.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
/// Serves `admin.sock`: one request per connection, answered with the frames `answer` returns,
|
||||
/// written exactly as given (so a test can send a wrong id or a frame that is not final). An
|
||||
/// empty list closes the connection without an answer.
|
||||
pub fn fake_brokerd_frames(
|
||||
answer: impl Fn(&Envelope) -> Vec<Envelope> + Send + 'static,
|
||||
) -> FakeBrokerd {
|
||||
let socket = temp_socket("admin.sock");
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let requests = Arc::new(Mutex::new(Vec::new()));
|
||||
let seen = Arc::clone(&requests);
|
||||
thread::spawn(move || {
|
||||
for stream in listener.incoming() {
|
||||
let Ok(mut stream) = stream else { continue };
|
||||
let Ok(request) = read_frame(&mut stream) else {
|
||||
continue;
|
||||
};
|
||||
seen.lock().unwrap().push(request.msg.clone());
|
||||
for frame in answer(&request) {
|
||||
if write_frame(&mut stream, &frame).is_err() {
|
||||
break; // the client went away; the next connection is still served
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
FakeBrokerd { socket, requests }
|
||||
}
|
||||
|
||||
/// The usual case: one final frame with the request's id.
|
||||
pub fn fake_brokerd(answer: impl Fn(&Message) -> Message + Send + 'static) -> FakeBrokerd {
|
||||
fake_brokerd_frames(move |request| {
|
||||
vec![Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: request.id,
|
||||
r#final: true,
|
||||
msg: answer(&request.msg),
|
||||
}]
|
||||
})
|
||||
}
|
||||
|
||||
pub fn wire_error(code: ErrorCode, detail: &str) -> Message {
|
||||
Message::Error(WireError {
|
||||
code,
|
||||
detail: detail.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
/// A pending approval created at 12:00:00 that expires at 12:15:00 on 2026-09-18.
|
||||
pub fn pending(approval: u64, tool: &str, arguments: &str) -> PendingApproval {
|
||||
PendingApproval {
|
||||
approval,
|
||||
session: SessionId::new("chat-1758196800-123456789").unwrap(),
|
||||
call: CallId(7),
|
||||
tool: tool.to_string(),
|
||||
arguments: arguments.to_string(),
|
||||
grant: "shell-scratch".to_string(),
|
||||
taint: DataClass::Private,
|
||||
created: ts("2026-09-18T12:00:00.000Z"),
|
||||
expires: ts("2026-09-18T12:15:00.000Z"),
|
||||
}
|
||||
}
|
||||
|
||||
/// A `brokerd` with these approvals pending. `approve` is answered with `outcome`, `refuse` with
|
||||
/// `ok`; both answer `no_such_approval` for an id that is not in the list.
|
||||
pub fn brokerd_with(items: Vec<PendingApproval>, outcome: bxcore::DecisionRecord) -> FakeBrokerd {
|
||||
fake_brokerd(move |msg| {
|
||||
let known = |id: u64| items.iter().any(|item| item.approval == id);
|
||||
match msg {
|
||||
Message::Approvals(_) => Message::ApprovalList(ApprovalList {
|
||||
items: items.clone(),
|
||||
}),
|
||||
Message::Approve(a) if known(a.approval) => {
|
||||
Message::ApproveResult(bxcore::ApproveResult {
|
||||
outcome: outcome.clone(),
|
||||
})
|
||||
}
|
||||
Message::Refuse(r) if known(r.approval) => Message::Ok(bxcore::Empty {}),
|
||||
Message::Approve(_) | Message::Refuse(_) => {
|
||||
wire_error(ErrorCode::NoSuchApproval, "no such approval")
|
||||
}
|
||||
_ => wire_error(ErrorCode::BadMessage, "not an admin request"),
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
pub struct FakeLoopd {
|
||||
pub socket: TempSocket,
|
||||
/// The content of every turn received, in order.
|
||||
pub turns: Arc<Mutex<Vec<String>>>,
|
||||
}
|
||||
|
||||
pub fn usage() -> Usage {
|
||||
Usage {
|
||||
cache_n: 10,
|
||||
prompt_n: 5,
|
||||
predicted_n: 7,
|
||||
reasoning_tokens: 3,
|
||||
thinking_capped: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Serves `loop.sock`: every turn gets the same events, then `turn_done` with `answer`. It does
|
||||
/// not wait for an approval to be answered: `bxctl` deals with each event before it reads the
|
||||
/// next frame, so the order of what it does is fixed all the same.
|
||||
pub fn fake_loopd(events: Vec<TurnEvent>, answer: &str) -> FakeLoopd {
|
||||
let socket = temp_socket("loop.sock");
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let turns = Arc::new(Mutex::new(Vec::new()));
|
||||
let seen = Arc::clone(&turns);
|
||||
let answer = answer.to_string();
|
||||
thread::spawn(move || {
|
||||
for stream in listener.incoming() {
|
||||
let Ok(mut stream) = stream else { continue };
|
||||
let Ok(request) = read_frame(&mut stream) else {
|
||||
continue;
|
||||
};
|
||||
let Message::Turn(turn) = request.msg else {
|
||||
continue;
|
||||
};
|
||||
seen.lock().unwrap().push(turn.content);
|
||||
let mut frames: Vec<(bool, Message)> = events
|
||||
.iter()
|
||||
.map(|e| (false, Message::TurnEvent(e.clone())))
|
||||
.collect();
|
||||
frames.push((
|
||||
true,
|
||||
Message::TurnDone(TurnDone {
|
||||
content: answer.clone(),
|
||||
usage: usage(),
|
||||
untrusted: false,
|
||||
}),
|
||||
));
|
||||
for (last, msg) in frames {
|
||||
let frame = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: request.id,
|
||||
r#final: last,
|
||||
msg,
|
||||
};
|
||||
if write_frame(&mut stream, &frame).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
FakeLoopd { socket, turns }
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
//! A fake `loopd` on a Unix socket: it records each turn and answers with the frames the test's
|
||||
//! function gives for it. Do not edit.
|
||||
|
||||
#![allow(dead_code)] // each test file uses a different part of this module
|
||||
|
||||
use std::io::Write;
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::path::Path;
|
||||
use std::sync::mpsc;
|
||||
|
||||
use bxcore::{
|
||||
Envelope, ErrorCode, Message, PROTOCOL_VERSION, Turn, TurnDone, TurnEvent, Usage, WireError,
|
||||
read_frame, write_frame,
|
||||
};
|
||||
|
||||
const USAGE: Usage = Usage {
|
||||
cache_n: 0,
|
||||
prompt_n: 1,
|
||||
predicted_n: 1,
|
||||
reasoning_tokens: 0,
|
||||
thinking_capped: false,
|
||||
};
|
||||
|
||||
/// What the fake sends back for a turn.
|
||||
pub enum Reply {
|
||||
Frame(Envelope),
|
||||
/// Raw bytes, for broken frames.
|
||||
Bytes(Vec<u8>),
|
||||
/// Stop answering this connection (the caller sees it close).
|
||||
Close,
|
||||
}
|
||||
|
||||
pub fn event(e: TurnEvent) -> Reply {
|
||||
Reply::Frame(Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 1,
|
||||
r#final: false,
|
||||
msg: Message::TurnEvent(e),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn done(content: &str) -> Reply {
|
||||
let msg = Message::TurnDone(TurnDone {
|
||||
content: content.to_string(),
|
||||
usage: USAGE,
|
||||
untrusted: false,
|
||||
});
|
||||
Reply::Frame(Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 1,
|
||||
r#final: true,
|
||||
msg,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn error(code: ErrorCode, detail: &str) -> Reply {
|
||||
let msg = Message::Error(WireError {
|
||||
code,
|
||||
detail: detail.to_string(),
|
||||
});
|
||||
Reply::Frame(Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 1,
|
||||
r#final: true,
|
||||
msg,
|
||||
})
|
||||
}
|
||||
|
||||
/// Listen on `socket`; `script(n, &turn)` gives the replies to the n-th turn (from 0).
|
||||
pub fn serve_loop<F>(socket: &Path, script: F) -> mpsc::Receiver<Turn>
|
||||
where
|
||||
F: Fn(usize, &Turn) -> Vec<Reply> + Send + 'static,
|
||||
{
|
||||
let listener = UnixListener::bind(socket).unwrap();
|
||||
let (tx, rx) = mpsc::channel();
|
||||
std::thread::spawn(move || {
|
||||
for (n, stream) in listener.incoming().enumerate() {
|
||||
let Ok(mut stream) = stream else { continue };
|
||||
let Ok(Envelope {
|
||||
msg: Message::Turn(turn),
|
||||
..
|
||||
}) = read_frame(&mut stream)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
// Reported before the script runs: a script may wait for the test.
|
||||
let _ = tx.send(turn.clone());
|
||||
let replies = script(n, &turn);
|
||||
for reply in replies {
|
||||
let ok = match reply {
|
||||
Reply::Frame(env) => write_frame(&mut stream, &env).is_ok(),
|
||||
Reply::Bytes(b) => stream.write_all(&b).is_ok(),
|
||||
Reply::Close => false,
|
||||
};
|
||||
if !ok {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
rx
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
//! `TurnDone.untrusted` follows the session's labels (M5b spec, sections 8 and 9): false while
|
||||
//! every tool result in the session's log is labelled trusted, and true from the turn in which one
|
||||
//! is labelled untrusted (a fetched page, or a recalled memory-tainted note) and in every turn
|
||||
//! after it, also after `loopd` restarts. A denial or a failure is never untrusted. Do not edit.
|
||||
|
||||
mod support;
|
||||
#[path = "support/tags.rs"]
|
||||
mod tags;
|
||||
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use bxcore::{
|
||||
DataClass, DenyReason, Envelope, Message, PROTOCOL_VERSION, SessionId, ToolResponse, Turn,
|
||||
read_frame, write_frame,
|
||||
};
|
||||
use loopd::channel::{Context, serve};
|
||||
use loopd::llama::Client;
|
||||
use loopd::memory::NoMemory;
|
||||
use loopd::tools::Registry;
|
||||
use support::{FakeServer, Home, Reply, ScriptedPort};
|
||||
use tags::{call, stream};
|
||||
|
||||
const CHAT: &str = "/v1/chat/completions";
|
||||
|
||||
/// A fake inference server and a home that one `loopd` or several, one after the other, serve
|
||||
/// from.
|
||||
struct Setup {
|
||||
server: FakeServer,
|
||||
// Declared last so it is dropped last: fields drop in order, and the home must outlive
|
||||
// everything that may still write into it (tip T44).
|
||||
home: Home,
|
||||
}
|
||||
|
||||
impl Setup {
|
||||
/// The server answers the completions in `chat` in order; the last one repeats.
|
||||
fn new(chat: Vec<Reply>) -> Setup {
|
||||
let home = Home::new();
|
||||
let server = FakeServer::start();
|
||||
server.route(CHAT, chat);
|
||||
Setup { server, home }
|
||||
}
|
||||
|
||||
/// Starts a channel server on the home, as a new `loopd` process would: its own context and
|
||||
/// tool port, nothing carried over but the files. `name` names its socket; the port answers
|
||||
/// the tool calls with `replies`, in order.
|
||||
fn loopd(&self, name: &str, replies: Vec<ToolResponse>) -> PathBuf {
|
||||
let mut cfg = self.home.config(&self.server.socket);
|
||||
cfg.channel.socket = self.home.dir.join(name);
|
||||
let socket = cfg.channel_socket();
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let ctx = Arc::new(Context::new(
|
||||
cfg.clone(),
|
||||
Client::new(cfg),
|
||||
Box::new(ScriptedPort::new(replies)),
|
||||
Registry::m2b(),
|
||||
Box::new(NoMemory),
|
||||
));
|
||||
thread::spawn(move || serve(listener, ctx));
|
||||
socket
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends one turn and returns its `TurnDone`'s `untrusted`. Any other final frame fails the test.
|
||||
fn untrusted(socket: &Path, session: &str, content: &str, resume: bool) -> bool {
|
||||
let mut stream = UnixStream::connect(socket).unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(10)))
|
||||
.unwrap();
|
||||
let msg = Message::Turn(Turn {
|
||||
session: SessionId::new(session).unwrap(),
|
||||
content: content.to_string(),
|
||||
resume,
|
||||
});
|
||||
let turn = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 7,
|
||||
r#final: true,
|
||||
msg,
|
||||
};
|
||||
write_frame(&mut stream, &turn).unwrap();
|
||||
loop {
|
||||
let frame = read_frame(&mut stream).unwrap();
|
||||
if !frame.r#final {
|
||||
continue;
|
||||
}
|
||||
match frame.msg {
|
||||
Message::TurnDone(done) => return done.untrusted,
|
||||
other => panic!("the turn did not end with turn_done: {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A result as `brokerd` answers it, with its label.
|
||||
fn result(content: &str, untrusted: bool) -> ToolResponse {
|
||||
ToolResponse::Result {
|
||||
content: content.to_string(),
|
||||
class: DataClass::Private,
|
||||
untrusted,
|
||||
truncated: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// The model calls `recall`; the fake port stands for `brokerd` and `memoryd` behind it.
|
||||
fn recall() -> Reply {
|
||||
stream(
|
||||
&[call("call-r", "recall", r#"{"query":"deploy"}"#)],
|
||||
"tool_calls",
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_clean_session_s_turns_carry_false() {
|
||||
let s = Setup::new(vec![Reply::fixture("turn1"), Reply::fixture("turn2")]);
|
||||
let socket = s.loopd("loop.sock", vec![]);
|
||||
assert!(!untrusted(&socket, "a", "hello", false), "turn 1");
|
||||
assert!(!untrusted(&socket, "a", "and again", true), "turn 2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trusted_results_denials_and_failures_keep_it_false() {
|
||||
let s = Setup::new(vec![
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("plain"),
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("plain"),
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("plain"),
|
||||
]);
|
||||
let socket = s.loopd(
|
||||
"loop.sock",
|
||||
vec![
|
||||
result("straylight\n", false),
|
||||
ToolResponse::Denied {
|
||||
reason: DenyReason::NoGrant,
|
||||
detail: None,
|
||||
},
|
||||
ToolResponse::Failed {
|
||||
message: "the container could not start".to_string(),
|
||||
},
|
||||
],
|
||||
);
|
||||
assert!(
|
||||
!untrusted(&socket, "a", "hostname?", false),
|
||||
"a trusted result"
|
||||
);
|
||||
assert!(!untrusted(&socket, "a", "again?", true), "a denial");
|
||||
assert!(!untrusted(&socket, "a", "once more?", true), "a failure");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_untrusted_result_makes_its_turn_and_every_later_one_true() {
|
||||
let s = Setup::new(vec![
|
||||
Reply::fixture("plain"),
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("plain"),
|
||||
Reply::fixture("tool_call"),
|
||||
Reply::fixture("plain"),
|
||||
]);
|
||||
let socket = s.loopd(
|
||||
"loop.sock",
|
||||
vec![
|
||||
result("a fetched page\n", true),
|
||||
result("straylight\n", false),
|
||||
],
|
||||
);
|
||||
assert!(!untrusted(&socket, "a", "hello", false), "before the page");
|
||||
assert!(untrusted(&socket, "a", "read it", true), "the page's turn");
|
||||
assert!(
|
||||
untrusted(&socket, "a", "hostname?", true),
|
||||
"a later turn whose own result is trusted"
|
||||
);
|
||||
assert!(untrusted(&socket, "a", "thanks", true), "a later turn");
|
||||
assert!(
|
||||
!untrusted(&socket, "b", "hello", false),
|
||||
"another session is not touched"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_recalled_memory_tainted_note_makes_it_true() {
|
||||
let s = Setup::new(vec![recall(), Reply::fixture("plain")]);
|
||||
let hit = "[1] memory-tainted · note, 2026-09-25 14:02, session chat-7 · private\n\
|
||||
the deploy runs from ci.example.org";
|
||||
let socket = s.loopd("loop.sock", vec![result(hit, true)]);
|
||||
assert!(
|
||||
untrusted(&socket, "a", "how do we deploy?", false),
|
||||
"the recall's turn"
|
||||
);
|
||||
assert!(untrusted(&socket, "a", "thanks", true), "a later turn");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_recall_of_trusted_memory_keeps_it_false() {
|
||||
let s = Setup::new(vec![recall(), Reply::fixture("plain")]);
|
||||
let hit = "[1] identity · moved 2026-09-25 from memory.md · private\n\
|
||||
deploys go through ci.example.org";
|
||||
let socket = s.loopd("loop.sock", vec![result(hit, false)]);
|
||||
assert!(!untrusted(&socket, "a", "how do we deploy?", false));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_restarted_loopd_reads_it_from_the_log() {
|
||||
let s = Setup::new(vec![Reply::fixture("tool_call"), Reply::fixture("plain")]);
|
||||
let first = s.loopd("first.sock", vec![result("a fetched page\n", true)]);
|
||||
assert!(untrusted(&first, "a", "read it", false), "the first loopd");
|
||||
let second = s.loopd("second.sock", vec![]);
|
||||
assert!(
|
||||
untrusted(&second, "a", "thanks", true),
|
||||
"the second loopd, on the same session"
|
||||
);
|
||||
assert!(
|
||||
!untrusted(&second, "b", "hello", false),
|
||||
"the second loopd, on a new session"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user