jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(buzz-agent): gate LLM tool calls on session/request_permission (#5712)

Browse Source 
## Summary

buzz-agent now authorizes every LLM-issued MCP tool call through
`session/request_permission` before it executes, instead of running
tools unconditionally. The agent **always asks**; the client applies
`BUZZ_ACP_PERMISSION_POLICY` and answers. buzz-agent never reads the
policy — this keeps the policy decision on the client side, matching the
layering of the other ACP harnesses, and avoids duplicating policy logic
that would drift.

### Broker

A crate-local `PermissionBroker`
(`crates/buzz-agent/src/permission.rs`), owned by `App` for the
connection lifetime, owns the full request-correlation lifecycle:

- **Process-wide admission.** A global `Semaphore`
(`BUZZ_AGENT_MAX_PENDING_PERMISSIONS`, default 32, validated `>= 1`) is
acquired *before* any correlation entry is inserted. This is the
security bound: the per-turn tool semaphore is constructed fresh per
turn and `max_sessions` is unbounded by default, so neither bounds
simultaneously-outstanding asks process-wide.
- **Abort-safe cleanup.** A successful admission returns a
`PendingPermission` lease that owns the admission permit and the
correlation id. Its `Drop` synchronously removes the still-pending entry
and releases the slot, covering task abort/panic that bypasses the
normal `run_prompt` tail.
- **At-most-once resolution.** `deliver` claims (removes) the entry
*before* waking the waiter, so each id resolves once and a later lease
`Drop` is a no-op. Unknown/late ids are logged and dropped; only ids the
broker minted (`perm-<n>`) are recognized.
- **Single absolute deadline.** Admission, request enqueue, and response
wait all share one deadline (`BUZZ_AGENT_PERMISSION_TIMEOUT_SECS`,
default 330s, validated `>= 1`), so a saturated call cannot outlive one
timeout window even when a stalled writer blocks the enqueue.
Cancellation races inside every wait — resolution never depends on the
outer abort drain. A writer that dies mid-connection (stdout closed, or
a blocking write that only surfaces its error at flush) is
connection-fatal: it cancels all sessions, which resolves any ask
waiting on a reply that can never be written.

### Wire

`request_permission_params` (`crates/buzz-agent/src/wire.rs`) is
version-aware, keyed on the protocol version negotiated at `initialize`
and stored on `App` for the connection lifetime (never derived from a
later mutable session field). v2 nests the tool call under `subject:
{type: "tool_call", toolCall}` with top-level `title`/`options`; v1 uses
the legacy top-level `toolCall`. No hybrid shape. Both offered options
(`allow_once`, `reject_once`) carry `optionId == kind`, so the client's
`kind`-based selection and this side's `optionId`-based predicate agree
without a lookup table.

### Gate

In each spawned tool task (`crates/buzz-agent/src/agent.rs`) the
sequence is: acquire per-turn permit → argument-shape validation →
broker admission + request + wait → cancellation recheck →
`emit_in_progress` → `mcp.call`. Argument-shape validation is hoisted
out of `mcp.rs::do_call` into `validate_arg_shape` so a malformed
non-object argument is rejected locally without prompting for a call
that could never execute.

Authorization is fail-closed, stated once in `evaluate`: execute IFF
`outcome.outcome == "selected"` **and** the selected `optionId` equals
the offered allow option. Every other shape (reject, cancelled, JSON-RPC
error, malformed, unknown outcome, wrong/unknown option, timeout,
wire-channel closure) denies with a synthetic tool error, and the turn
continues.

### Scope

Only LLM-issued MCP calls are gated. The built-in `load_skill` tool and
`call_hooks` lifecycle calls (`_Stop`, `_PostCompact`) are exempt — they
are not model-issued. `readOnlyHint` is never treated as a security
boundary.

First cut ships `allow_once`/`reject_once` only; session-scoped grants
are deliberately out of scope.

### Related issue

Part of #4938. This PR and
[#5106](https://github.com/block/buzz/pull/5106) jointly implement the
feature: #5106 is the client-side policy engine and permission cards;
this PR is buzz-agent's asking side (`session/request_permission`).
Neither closes #4938 alone.

---------

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: Hayt <41ea58f1e64c243627e8acde7c89be667052ee6e17d8f021c1195be4324ebf04@buzz.block.builderlab.xyz>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
This commit is contained in:
Will Pfleger
2026-08-25 11:49:07 -04:00
committed by GitHub
parent 931747c9c4
commit a1219070fa
12 changed files with 2494 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
crates/buzz-agent/src/agent.rs
View File

@@ -14,6 +14,7 @@ use crate::hints::SkillEntry;
use crate::llm::Llm;
use crate::mcp::McpRegistry;
use crate::mcp::ResultBudget;
use crate::permission::PermissionDecision;
use crate::types::{
AgentError, CacheTotalState, ContentBlock, HistoryItem, PricingIdentity, ProviderStop,
@@ -142,6 +143,14 @@ pub struct RunCtx<'a> {
pub system_prompt: &'a str,
pub llm: &'a Llm,
pub mcp: &'a Arc<McpRegistry>,
/// Process-wide permission broker (owned by `App`). Every LLM-issued MCP
/// tool call asks the client to authorize it through this broker before
/// executing. Shared across all sessions so the global admission cap bounds
/// simultaneously-outstanding asks process-wide.
pub permissions: &'a Arc<crate::permission::PermissionBroker>,
/// ACP protocol version negotiated at `initialize`, fixed for the
/// connection. Selects the `session/request_permission` wire shape.
pub protocol_version: u32,
/// Skills discovered at session creation; used by the built-in `load_skill` tool.
pub skills: &'a [SkillEntry],
pub wire: &'a WireSender,
@@ -882,8 +891,10 @@ impl RunCtx<'_> {
total: MAX_TOOL_RESULT_BYTES,
text: self.cfg.max_tool_result_text_bytes,
};
let cancel = self.cancel.clone();
let mut cancel = self.cancel.clone();
let sem = Arc::clone(&sem);
let permissions = Arc::clone(self.permissions);
let protocol_version = self.protocol_version;
set.spawn(async move {
// Acquire a permit; if the semaphore is closed (cancel),
// emit a terminal wire update and skip the call.
@@ -894,6 +905,37 @@ impl RunCtx<'_> {
return (i, InvokeOutcome::Failed("cancelled".into()));
}
};
// Argument-shape validation BEFORE the ask: a malformed
// non-object argument can never execute, so reject it locally
// without prompting the user to approve a doomed call.
if let Err(e) = crate::mcp::validate_arg_shape(&call.name, &call.arguments) {
let msg = e.to_string();
emit_failed(&wire, &session_id, &call, &msg).await;
return (i, InvokeOutcome::Failed(msg));
}
// Ask the client to authorize this call. The broker owns the
// full correlation lifecycle and races cancellation internally;
// every non-authorizing outcome fails closed.
match permissions
.request_permission(&wire, protocol_version, &session_id, &call, &mut cancel)
.await
{
PermissionDecision::Allowed => {}
PermissionDecision::Denied(msg) => {
emit_failed(&wire, &session_id, &call, msg).await;
return (i, InvokeOutcome::Failed(msg.into()));
}
PermissionDecision::Cancelled => {
emit_failed(&wire, &session_id, &call, "cancelled").await;
return (i, InvokeOutcome::Failed("cancelled".into()));
}
}
// Cancellation recheck: a cancel may have landed while we
// waited for approval. Do not start the call in that case.
if *cancel.borrow() {
emit_failed(&wire, &session_id, &call, "cancelled").await;
return (i, InvokeOutcome::Failed("cancelled".into()));
}
emit_in_progress(&wire, &session_id, &call).await;
let outcome = invoke_tool_inner(&mcp, &call, timeout, budget, cancel).await;
match &outcome {

25
crates/buzz-agent/src/config.rs
View File

@@ -479,6 +479,18 @@ pub struct Config {
/// Set via `BUZZ_AGENT_MAX_HANDOFFS`. Default 10.
pub max_handoffs: usize,
pub max_parallel_tools: usize,
/// Process-wide cap on simultaneously-outstanding `session/request_permission`
/// asks. Bounds the [`PermissionBroker`](crate::permission::PermissionBroker)
/// correlation map independently of the per-turn tool semaphore (which is
/// fresh per turn) and of `max_sessions` (unbounded by default). Default 32.
/// Set via `BUZZ_AGENT_MAX_PENDING_PERMISSIONS`; validated `>= 1`.
pub max_pending_permissions: usize,
/// Single absolute deadline for a permission ask — shared by broker
/// admission and the response wait, so a saturated call cannot live for two
/// full timeout windows. Default 330s, chosen to outlast the client's 300s
/// auto-deny so the answer (or auto-deny) lands first. Set via
/// `BUZZ_AGENT_PERMISSION_TIMEOUT_SECS`; validated `>= 1`.
pub permission_timeout: Duration,
pub hook_timeout: Duration,
/// Maximum `_Stop` rejections per prompt. Default 3. Set to 0 to
/// disable `_Stop` hooks entirely (agent always honors end_turn).
@@ -622,6 +634,11 @@ impl Config {
max_context_tokens: parse_env("BUZZ_AGENT_MAX_CONTEXT_TOKENS", 200_000u64)?,
max_handoffs: parse_env("BUZZ_AGENT_MAX_HANDOFFS", 10)?,
max_parallel_tools: parse_env("BUZZ_AGENT_MAX_PARALLEL_TOOLS", 8usize)?,
max_pending_permissions: parse_env("BUZZ_AGENT_MAX_PENDING_PERMISSIONS", 32usize)?,
permission_timeout: Duration::from_secs(parse_env(
"BUZZ_AGENT_PERMISSION_TIMEOUT_SECS",
330u64,
)?),
hook_timeout: Duration::from_millis(parse_env("BUZZ_AGENT_HOOK_TIMEOUT_MS", 2500u64)?),
stop_max_rejections: parse_env("BUZZ_AGENT_STOP_MAX_REJECTIONS", 3u32)?,
require_reply: parse_env("BUZZ_AGENT_REQUIRE_REPLY", 0u8)? != 0,
@@ -668,6 +685,8 @@ impl Config {
max_context_tokens: 200_001,
max_handoffs: 0,
max_parallel_tools: 1,
max_pending_permissions: 32,
permission_timeout: Duration::from_secs(330),
hook_timeout: Duration::from_secs(1),
stop_max_rejections: 0,
require_reply: false,
@@ -729,6 +748,12 @@ impl Config {
if self.max_parallel_tools < 1 {
return Err("config: BUZZ_AGENT_MAX_PARALLEL_TOOLS must be >= 1".into());
}
if self.max_pending_permissions < 1 {
return Err("config: BUZZ_AGENT_MAX_PENDING_PERMISSIONS must be >= 1".into());
}
if self.permission_timeout < MIN_TIMEOUT {
return Err("config: BUZZ_AGENT_PERMISSION_TIMEOUT_SECS must be >= 1".into());
}
if self.mcp_max_restart_attempts < 1 {
return Err("config: BUZZ_AGENT_MCP_RESTART_MAX_ATTEMPTS must be >= 1".into());
}

77
crates/buzz-agent/src/lib.rs
View File

@@ -9,6 +9,7 @@ mod hints;
mod llm;
mod mcp;
pub mod model_capabilities;
mod permission;
pub mod types;
mod wire;
@@ -32,6 +33,7 @@ pub const WINDOWS_SHELL_RESOLUTION_ENV: &[&str] = &[
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use serde_json::{json, Value};
@@ -54,6 +56,17 @@ struct App {
cfg: Config,
llm: Arc<Llm>,
sessions: Mutex<HashMap<String, Session>>,
/// ACP protocol version negotiated at `initialize`, stored for the whole
/// connection lifetime. The `session/request_permission` wire shape derives
/// from this value — never from a later mutable session field — so a strict
/// client always receives exactly the shape it negotiated. Defaults to
/// [`PROTOCOL_VERSION`] before `initialize`; no prompt (and thus no
/// permission ask) can run before then.
negotiated_version: AtomicU32,
/// Owns the entire `session/request_permission` correlation lifecycle:
/// process-wide admission, id allocation, response delivery, and abort-safe
/// cleanup. See [`permission::PermissionBroker`].
permissions: Arc<permission::PermissionBroker>,
/// Cached model catalog for Databricks providers. Populated lazily on the
/// first successful `session/new` discovery call. Failed discovery is never
/// cached: static-token authentication errors reject session creation, while
@@ -181,28 +194,53 @@ async fn async_main() {
let cfg = Config::from_env().unwrap_or_else(|e| die(e));
let llm = Arc::new(Llm::new(&cfg).unwrap_or_else(|e| die(e.to_string())));
let max_line = cfg.max_line_bytes;
let permissions = Arc::new(permission::PermissionBroker::new(
cfg.max_pending_permissions,
cfg.permission_timeout,
));
let app = Arc::new(App {
cfg,
llm,
sessions: Mutex::new(HashMap::new()),
negotiated_version: AtomicU32::new(PROTOCOL_VERSION),
permissions,
models_cache: tokio::sync::OnceCell::new(),
});
let (wire_tx, wire_rx) = mpsc::channel::<WireMsg>(64);
let writer = tokio::spawn(wire::writer_task(wire_rx));
if let Err(e) = read_loop(
BufReader::new(tokio::io::stdin()),
app.clone(),
wire_tx,
max_line,
)
.await
{
tracing::error!("io: reader: {e}");
let mut writer = tokio::spawn(wire::writer_task(wire_rx));
// Whichever ends first drives shutdown. The reader ending is the normal
// path (stdin EOF/error). The writer ending while the reader still runs
// means stdout is closed/broken: no reply can ever be written, so we must
// stop reading and cancel every session rather than leave the process
// reading input while outstanding permission asks wait out their full
// deadline for a response that can never arrive.
tokio::select! {
r = read_loop(
BufReader::new(tokio::io::stdin()),
app.clone(),
wire_tx,
max_line,
) => {
if let Err(e) = r {
tracing::error!("io: reader: {e}");
}
cancel_all_sessions(&app).await;
let _ = writer.await;
}
_ = &mut writer => {
tracing::error!("io: writer exited (stdout closed); shutting down connection");
cancel_all_sessions(&app).await;
}
}
}
/// Signal every live session to cancel. Run on connection teardown so in-flight
/// prompts — including any waiting on a `session/request_permission` response —
/// resolve promptly instead of waiting out their deadline.
async fn cancel_all_sessions(app: &Arc<App>) {
for session in app.sessions.lock().await.values() {
let _ = session.cancel_tx.send(true);
}
let _ = writer.await;
}
async fn read_loop<R: tokio::io::AsyncBufRead + Unpin>(
@@ -235,7 +273,10 @@ async fn dispatch(app: &Arc<App>, msg: Value, wire_tx: &WireSender) {
handle_request(app, id, method, params, wire_tx).await
}
Inbound::Notification { method, params } => handle_notification(app, &method, params).await,
Inbound::Ignored => {}
// Client's answer to a `session/request_permission` we issued. The
// broker matches it to a live correlation id (waking that waiter) or
// ignores an unknown/late id.
Inbound::Response { id, result } => app.permissions.deliver(&id, result),
Inbound::Invalid { id, code, message } => {
wire::send(wire_tx, wire::err(id, code, &message)).await
}
@@ -250,7 +291,7 @@ async fn handle_request(
wire_tx: &WireSender,
) {
match method.as_str() {
"initialize" => initialize(id, params, wire_tx).await,
"initialize" => initialize(app, id, params, wire_tx).await,
"session/new" => {
let app = app.clone();
let wire_tx = wire_tx.clone();
@@ -291,7 +332,7 @@ async fn handle_notification(app: &Arc<App>, method: &str, params: Value) {
}
}
async fn initialize(id: Value, params: Value, wire_tx: &WireSender) {
async fn initialize(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSender) {
let p: InitializeParams = match decode(params, "initialize") {
Ok(p) => p,
Err(m) => return reject(wire_tx, id, INVALID_PARAMS, &m).await,
@@ -303,6 +344,12 @@ async fn initialize(id: Value, params: Value, wire_tx: &WireSender) {
// RFD. Revisit when that RFD merges; otherwise a genuine upstream-v2 agent
// would silently lose `[Base]`.
let negotiated_version = p.protocol_version.min(PROTOCOL_VERSION);
// Store the negotiated version for the connection lifetime: the
// `session/request_permission` wire shape derives from this value, never
// from a later mutable session field, so a strict client always receives
// exactly the shape it negotiated at `initialize`.
app.negotiated_version
.store(negotiated_version, Ordering::Relaxed);
wire::send(
wire_tx,
wire::ok(
@@ -734,6 +781,8 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
system_prompt: &effective_system_prompt,
llm: &app.llm,
mcp: &mcp,
permissions: &app.permissions,
protocol_version: app.negotiated_version.load(Ordering::Relaxed),
skills: &skills,
wire: &wire_tx,
cancel: &mut cancel_rx,

2
crates/buzz-agent/src/llm.rs
View File

@@ -2609,6 +2609,8 @@ mod tests {
max_context_tokens: 200_000,
max_handoffs: 1,
max_parallel_tools: 1,
max_pending_permissions: 32,
permission_timeout: Duration::from_secs(330),
hook_timeout: Duration::from_secs(1),
stop_max_rejections: 0,
require_reply: false,

33
crates/buzz-agent/src/mcp.rs
View File

@@ -594,15 +594,7 @@ impl McpRegistry {
budget: ResultBudget,
cancel: &mut watch::Receiver<bool>,
) -> Result<ToolResult, AgentError> {
let arg_obj = match arguments {
Value::Object(m) => Some(m.clone()),
Value::Null => None,
_ => {
return Err(AgentError::Mcp(format!(
"tool {qname} arguments must be a JSON object"
)))
}
};
let arg_obj = validate_arg_shape(qname, arguments)?;
let mut params = CallToolRequestParams::default();
params.name = bare.to_owned().into();
params.arguments = arg_obj;
@@ -812,6 +804,29 @@ async fn spawn_one(
Ok((client, pgid, names, tools))
}
/// Validate that tool-call arguments are a shape the MCP transport can carry:
/// a JSON object (`Some(map)`) or absent (`None`). Any other JSON type is a
/// malformed call that the transport would reject.
///
/// Hoisted out of `do_call` so the permission gate can run it *before* asking
/// the user: a malformed non-object argument is rejected locally without
/// prompting for approval of a call that could never execute. `do_call` runs
/// it again as the single authoritative shape check — the duplicate is a cheap
/// idempotent match, and keeping it here means no code path can reach the
/// transport with an unvalidated shape.
pub fn validate_arg_shape(
qname: &str,
arguments: &Value,
) -> Result<Option<Map<String, Value>>, AgentError> {
match arguments {
Value::Object(m) => Ok(Some(m.clone())),
Value::Null => Ok(None),
_ => Err(AgentError::Mcp(format!(
"tool {qname} arguments must be a JSON object"
))),
}
}
/// Send `notifications/cancelled` to the MCP server, fire-and-forget.
/// Per MCP spec, cancellation notifications are best-effort; we never
/// block the agent on slow server stdio.

1047
crates/buzz-agent/src/permission.rs Normal file
View File

File diff suppressed because it is too large Load Diff

327
crates/buzz-agent/src/wire.rs
View File

@@ -1,6 +1,6 @@
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt};
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::mpsc;
use crate::types::{ContentBlock, McpServerStdio};
@@ -27,7 +27,19 @@ pub enum Inbound {
method: String,
params: Value,
},
Ignored,
/// A bare JSON-RPC response (id present, no method) — the client's answer
/// to a request buzz-agent issued. Today the only such request is
/// `session/request_permission`. `result` carries the JSON-RPC `result`
/// field ONLY when the frame is a structurally valid response — no `method`
/// member and exactly one of `result`/`error`. Any malformed shape (present
/// non-string `method`, both `result` and `error`, or neither) is normalized
/// to `Null` so a possibly-`selected` payload is never laundered into an
/// approval; every non-`selected` shape fails the broker's authorization
/// predicate and denies.
Response {
id: Value,
result: Value,
},
Invalid {
id: Value,
code: i32,
@@ -109,9 +121,27 @@ pub fn classify(msg: &Value) -> Inbound {
params,
},
(Some(m), None) => Inbound::Notification { method: m, params },
// Bare responses (id present, no method) are unexpected — buzz-agent
// does not issue requests to the client. Ignore silently.
(None, Some(_)) => Inbound::Ignored,
// Bare responses (id present, no method) answer a request buzz-agent
// issued — today only `session/request_permission`. Route to the
// permission broker, which matches a live correlation id or ignores an
// unknown one. Forward the `result` ONLY when the frame is a
// structurally valid response — the exactly-one-of invariant: no
// `method` member at all, and `result` present with `error` absent. A
// present non-string `method` (which `as_str` above collapsed to
// `None`), both `result` and `error`, or neither is malformed; forward
// `Null` so the broker fails closed (deny) rather than laundering a
// possibly-`selected` payload into an approval.
(None, Some(id)) => {
let well_formed = msg.get("method").is_none()
&& msg.get("result").is_some()
&& msg.get("error").is_none();
let result = if well_formed {
msg.get("result").cloned().unwrap_or(Value::Null)
} else {
Value::Null
};
Inbound::Response { id, result }
}
(None, None) => Inbound::Invalid {
id: Value::Null,
code: INVALID_REQUEST,
@@ -120,6 +150,79 @@ pub fn classify(msg: &Value) -> Inbound {
}
}
/// `optionId`/`kind` of the single allow option offered on every
/// `session/request_permission`. buzz-acp's answering side selects the option
/// whose `kind == "allow_once"` (never by hardcoded `optionId`), and the
/// authorization predicate on this side requires the returned `optionId` to
/// equal exactly this value. Keeping option id and kind identical means both
/// sides agree without a separate lookup table.
pub const ALLOW_OPTION_ID: &str = "allow_once";
/// The two options offered on every permission request: allow-once and
/// reject-once. First cut ships only these (no session-scoped grant), so every
/// offered option is already in the desktop card's exact actionable allowlist.
fn permission_options() -> Value {
json!([
{ "optionId": ALLOW_OPTION_ID, "name": "Allow", "kind": ALLOW_OPTION_ID },
{ "optionId": "reject_once", "name": "Deny", "kind": "reject_once" },
])
}
/// Build `session/request_permission` params for the negotiated protocol
/// version. No hybrid shapes — the request must match exactly what the client
/// negotiated at `initialize`, or a strict client can reject it before policy
/// is applied.
///
/// - **v2** (what buzz-agent negotiates with current buzz-acp): tool context
/// lives under `subject: {type: "tool_call", toolCall}` with top-level
/// `title` and `options`.
/// - **v1** (still negotiated when a client requests it): the legacy shape with
/// `toolCall` (carrying `kind`) directly at the params level.
pub fn request_permission_params(
version: u32,
session_id: &str,
tool_call_id: &str,
title: &str,
raw_input: &Value,
) -> Value {
if version >= 2 {
json!({
"sessionId": session_id,
"title": title,
"subject": {
"type": "tool_call",
"toolCall": {
"toolCallId": tool_call_id,
"title": title,
"rawInput": raw_input,
},
},
"options": permission_options(),
})
} else {
json!({
"sessionId": session_id,
"toolCall": {
"toolCallId": tool_call_id,
"title": title,
"kind": "other",
"rawInput": raw_input,
},
"options": permission_options(),
})
}
}
/// Build an outbound JSON-RPC request `session/request_permission` frame.
pub fn request_permission(id: Value, params: Value) -> Value {
json!({
"jsonrpc": "2.0",
"id": id,
"method": "session/request_permission",
"params": params,
})
}
pub fn ok(id: Value, result: Value) -> Value {
json!({ "jsonrpc": "2.0", "id": id, "result": result })
}
@@ -267,7 +370,18 @@ pub fn session_update_with_goose_meta(sid: &str, update: Value, goose_meta: Valu
}
pub async fn send(wire: &WireSender, msg: Value) {
let _ = wire.send(WireMsg::Notify(msg)).await;
let _ = send_checked(wire, msg).await;
}
/// Enqueue a frame, reporting whether the writer accepted it. Unlike mpsc's
/// non-blocking `try_send`, this awaits channel capacity; it fails only when
/// the writer task has dropped its receiver, which happens exactly when the
/// writer has exited because stdout is closed/broken. A frame that fails here
/// will never be written, so callers that correlate a response — the
/// permission broker — must fail closed immediately rather than wait out a
/// deadline for a reply that can never arrive.
pub async fn send_checked(wire: &WireSender, msg: Value) -> Result<(), ()> {
wire.send(WireMsg::Notify(msg)).await.map_err(|_| ())
}
pub async fn read_bounded_line<R: AsyncBufRead + Unpin>(
@@ -316,8 +430,25 @@ pub async fn read_bounded_line<R: AsyncBufRead + Unpin>(
}
}
pub async fn writer_task(mut rx: mpsc::Receiver<WireMsg>) {
let mut stdout = tokio::io::stdout();
pub async fn writer_task(rx: mpsc::Receiver<WireMsg>) {
write_frames(rx, tokio::io::stdout()).await;
}
/// Drain `rx`, writing each frame to `out` as a newline-terminated JSON line.
/// Generic over the sink so tests can inject an `AsyncWrite` that fails on
/// flush; production passes stdout.
///
/// Both `write_all` and `flush` failure are connection-fatal: they return,
/// dropping `rx` so `async_main`'s writer-death arm cancels every session.
/// Flush must be fatal too — a blocking stdout can report `Ok` from
/// `write_all` when it only schedules the underlying write and surface the
/// real error at `flush`, so ignoring flush failure would leave a dead stdout
/// undetected and strand any correlated ask waiting for a reply that can never
/// be written.
pub(crate) async fn write_frames<W: AsyncWrite + Unpin>(
mut rx: mpsc::Receiver<WireMsg>,
mut out: W,
) {
while let Some(msg) = rx.recv().await {
let WireMsg::Notify(v) = msg;
let mut s = match serde_json::to_string(&v) {
@@ -328,10 +459,9 @@ pub async fn writer_task(mut rx: mpsc::Receiver<WireMsg>) {
}
};
s.push('\n');
if stdout.write_all(s.as_bytes()).await.is_err() {
if out.write_all(s.as_bytes()).await.is_err() || out.flush().await.is_err() {
return;
}
let _ = stdout.flush().await;
}
}
@@ -534,4 +664,181 @@ mod tests {
assert_eq!(payload["accumulatedInputTokens"], serde_json::json!(1000));
assert_eq!(payload["accumulatedOutputTokens"], serde_json::json!(200));
}
// ── request_permission_params: version-aware wire shape ──────────────────
/// v2 (what buzz-agent negotiates with current buzz-acp): tool context is
/// nested under `subject: {type: "tool_call", toolCall}` with top-level
/// `title` and `options`, matching the ACP v2 `RequestPermissionRequest`.
#[test]
fn request_permission_params_v2_nests_tool_call_under_subject() {
let raw = json!({ "command": "ls" });
let p = request_permission_params(2, "ses_1", "fake__shell", "fake__shell", &raw);
assert_eq!(p["sessionId"], "ses_1");
assert_eq!(p["title"], "fake__shell");
assert_eq!(p["subject"]["type"], "tool_call");
assert_eq!(p["subject"]["toolCall"]["toolCallId"], "fake__shell");
assert_eq!(p["subject"]["toolCall"]["title"], "fake__shell");
assert_eq!(p["subject"]["toolCall"]["rawInput"], raw);
// No hybrid: v2 must NOT carry a top-level `toolCall`.
assert!(p.get("toolCall").is_none(), "v2 must not use the v1 shape");
assert_options(&p["options"]);
}
/// v1 (still negotiated when a client requests it): the legacy shape with
/// `toolCall` (carrying `kind`) directly at the params level, no `subject`.
#[test]
fn request_permission_params_v1_uses_legacy_top_level_tool_call() {
let raw = json!({ "command": "ls" });
let p = request_permission_params(1, "ses_1", "fake__shell", "fake__shell", &raw);
assert_eq!(p["sessionId"], "ses_1");
assert_eq!(p["toolCall"]["toolCallId"], "fake__shell");
assert_eq!(p["toolCall"]["title"], "fake__shell");
assert_eq!(p["toolCall"]["kind"], "other");
assert_eq!(p["toolCall"]["rawInput"], raw);
// No hybrid: v1 must NOT carry the v2 `subject` or top-level `title`.
assert!(p.get("subject").is_none(), "v1 must not use the v2 shape");
assert!(p.get("title").is_none(), "v1 has no top-level title");
assert_options(&p["options"]);
}
/// Both offered options are exactly allow-once and reject-once, with
/// `optionId == kind` so buzz-acp's `kind`-based selector and this side's
/// `optionId`-based predicate agree without a lookup table.
fn assert_options(options: &Value) {
let opts = options.as_array().expect("options is an array");
assert_eq!(opts.len(), 2, "first cut offers exactly two options");
assert_eq!(opts[0]["optionId"], ALLOW_OPTION_ID);
assert_eq!(opts[0]["kind"], ALLOW_OPTION_ID);
assert_eq!(opts[0]["name"], "Allow");
assert_eq!(opts[1]["optionId"], "reject_once");
assert_eq!(opts[1]["kind"], "reject_once");
assert_eq!(opts[1]["name"], "Deny");
}
/// The outbound frame wraps params in a JSON-RPC request whose id echoes
/// back verbatim so the broker can correlate the response.
#[test]
fn request_permission_frame_is_a_correlatable_jsonrpc_request() {
let params = request_permission_params(2, "ses_1", "t", "t", &json!({}));
let frame = request_permission(json!("perm-7"), params);
assert_eq!(frame["jsonrpc"], "2.0");
assert_eq!(frame["id"], "perm-7");
assert_eq!(frame["method"], "session/request_permission");
assert_eq!(frame["params"]["sessionId"], "ses_1");
}
// ── classify: bare responses route to the broker ─────────────────────────
/// A bare JSON-RPC response (id, no method) is the client's answer to a
/// request buzz-agent issued; it routes to the broker with its `result`.
#[test]
fn classify_bare_response_routes_to_broker() {
let msg = json!({
"jsonrpc": "2.0",
"id": "perm-3",
"result": { "outcome": { "outcome": "selected", "optionId": ALLOW_OPTION_ID } },
});
match classify(&msg) {
Inbound::Response { id, result } => {
assert_eq!(id, json!("perm-3"));
assert_eq!(result["outcome"]["outcome"], "selected");
}
other => panic!("expected Response, got {other:?}"),
}
}
/// A JSON-RPC error response (id, `error`, no `result`) still routes to the
/// broker but with `result == Null`, which the authorization predicate
/// fails closed. buzz-agent never leaves the waiter hanging on an error.
#[test]
fn classify_error_response_routes_with_null_result() {
let msg = json!({
"jsonrpc": "2.0",
"id": "perm-3",
"error": { "code": -32601, "message": "method not found" },
});
match classify(&msg) {
Inbound::Response { id, result } => {
assert_eq!(id, json!("perm-3"));
assert_eq!(result, Value::Null, "error/absent result → Null → deny");
}
other => panic!("expected Response, got {other:?}"),
}
}
/// Carl's frame #1: a response carrying BOTH `result` and `error` is
/// structurally ambiguous and must NOT deliver the `result`, even when that
/// `result` is a well-formed `selected`/`allow_once` payload. The wire layer
/// normalizes it to `Null` so the broker denies instead of the frame
/// laundering an approval upstream of every fail-closed check.
#[test]
fn classify_response_with_both_result_and_error_denies() {
let msg = json!({
"jsonrpc": "2.0",
"id": "perm-3",
"result": { "outcome": { "outcome": "selected", "optionId": ALLOW_OPTION_ID } },
"error": { "code": -32603, "message": "internal" },
});
match classify(&msg) {
Inbound::Response { id, result } => {
assert_eq!(id, json!("perm-3"));
assert_eq!(
result,
Value::Null,
"result+error is malformed → Null → deny, never forward the allow payload"
);
}
other => panic!("expected Response, got {other:?}"),
}
}
/// Carl's frame #2: a present but non-string `method` is NOT "method
/// absent". `as_str` collapses `method: 7` to `None`, which lands the frame
/// in the response arm, but it is not a valid response and must not forward
/// its `result` (a well-formed `selected` payload here). The structural
/// check sees the present `method` member and normalizes to `Null` → deny.
#[test]
fn classify_response_with_non_string_method_denies() {
let msg = json!({
"jsonrpc": "2.0",
"id": "perm-3",
"method": 7,
"result": { "outcome": { "outcome": "selected", "optionId": ALLOW_OPTION_ID } },
});
match classify(&msg) {
Inbound::Response { id, result } => {
assert_eq!(id, json!("perm-3"));
assert_eq!(
result,
Value::Null,
"present non-string method → not a valid response → Null → deny"
);
}
other => panic!("expected Response, got {other:?}"),
}
}
// ── send_checked: observable wire closure ────────────────────────────────
/// `send_checked` reports `Ok` while the writer's receiver is alive and
/// `Err` once it is gone (writer task exited on closed/broken stdout). This
/// is the contract the permission broker relies on to fail an undeliverable
/// ask closed immediately instead of waiting out its deadline for a reply
/// that can never be written.
#[tokio::test]
async fn send_checked_reports_closure_when_writer_gone() {
let (tx, rx) = mpsc::channel::<WireMsg>(4);
assert!(
send_checked(&tx, json!({ "ok": 1 })).await.is_ok(),
"send succeeds while the writer receiver is alive"
);
drop(rx); // writer exited → receiver dropped
assert!(
send_checked(&tx, json!({ "ok": 2 })).await.is_err(),
"send reports failure once the writer is gone"
);
}
}

43
crates/buzz-agent/tests/bin/fake_mcp.rs
View File

@@ -23,6 +23,12 @@
//! tree dies on timeout.
//! FAKE_MCP_GRANDCHILD_PID_FILE=path
//! — path to write the grandchild PID to.
//! FAKE_MCP_CANCEL_LOG=path — append each `notifications/cancelled` frame to
//! `path` (one JSON line per notification).
//! FAKE_MCP_CALL_LOG=path — append the tool name of each `tools/call` to
//! `path` (one name per line). Lets a test assert
//! a tool was invoked exactly once, or never — the
//! permission gate's core proof.
//! FAKE_MCP_STOP_HOOK=1 — expose a `_Stop` hook tool
//! FAKE_MCP_STOP_TEXT=text — `_Stop` returns this text (default: "keep going")
//! FAKE_MCP_STOP_DELAY=N — `_Stop` sleeps N seconds before replying
@@ -39,6 +45,11 @@
//! `command` string. Lets a test drive the
//! reply guard's recognition of a real,
//! registered shell tool.
//! FAKE_MCP_NAMED_TOOLS=a,b — expose one no-arg tool per comma-separated bare
//! name (each registered as `<server>__<name>`), in
//! addition to any `FAKE_MCP_TOOL_COUNT` tools. Lets
//! a test issue parallel calls to distinctly named
//! tools and tell them apart in `FAKE_MCP_CALL_LOG`.
use std::io::{BufRead, Write};
@@ -83,6 +94,7 @@ fn make_tools(
include_stop_hook: bool,
include_post_compact_hook: bool,
include_shell_tool: bool,
named_tools: &[String],
) -> Vec<Value> {
let mut tools: Vec<Value> = (0..count)
.map(|i| {
@@ -93,6 +105,13 @@ fn make_tools(
})
})
.collect();
for name in named_tools {
tools.push(json!({
"name": name,
"description": "named test tool",
"inputSchema": { "type": "object", "properties": {} },
}));
}
if include_stop_hook {
tools.push(json!({
"name": "_Stop",
@@ -156,6 +175,14 @@ fn main() {
let post_compact_hook = env_flag("FAKE_MCP_POSTCOMPACT_HOOK");
let shell_tool = env_flag("FAKE_MCP_SHELL_TOOL");
let post_compact_text = std::env::var("FAKE_MCP_POSTCOMPACT_TEXT").unwrap_or_default();
// One extra no-arg tool per comma-separated bare name.
let named_tools: Vec<String> = std::env::var("FAKE_MCP_NAMED_TOOLS")
.unwrap_or_default()
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_owned)
.collect();
// Use a channel-based stdin reader so notifications (which carry no id)
// are captured even while the main thread is sleeping during a tool call.
@@ -231,6 +258,7 @@ fn main() {
stop_hook,
post_compact_hook,
shell_tool,
&named_tools,
)
}),
);
@@ -248,6 +276,21 @@ fn main() {
.and_then(|p| p.get("name"))
.and_then(Value::as_str)
.unwrap_or("");
// Append every invoked tool name so a test can prove a call
// reached the server exactly once (or never). This fires for
// ALL tools/call, including `_Stop`/`_PostCompact` hooks, so a
// test can also prove hooks are NOT permission-gated by
// observing they still reach the server without an ask.
if let Ok(path) = std::env::var("FAKE_MCP_CALL_LOG") {
use std::io::Write as _;
if let Ok(mut f) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
{
let _ = writeln!(f, "{called_name}");
}
}
// Optionally spawn a long-sleeping grandchild so the test
// can verify process-group killing reaches the whole tree.
if env_flag("FAKE_MCP_SPAWN_GRANDCHILD") {

279
crates/buzz-agent/tests/common/mod.rs Normal file
View File

@@ -0,0 +1,279 @@
//! Shared subprocess test harness for the buzz-agent ACP integration suites.
//!
//! Every integration test file is its own crate, so this module is included
//! with `mod common;` in each and compiles once per binary — each binary uses
//! only the subset it needs, hence the module-wide `dead_code` allow.
//!
//! It drives a real `buzz-agent` child over the ACP wire against a fake LLM
//! (`CapturingLlm`, which records each request body) and answers the
//! `session/request_permission` surface (`approve_permission`, selecting the
//! offered `allow_once` option by `kind`, never a hardcoded `optionId`).
#![allow(dead_code)]
use std::collections::VecDeque;
use std::process::Stdio;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;
use tokio::sync::Mutex;
pub struct CapturingLlm {
pub url: String,
pub captured: Arc<Mutex<Vec<Value>>>,
}
pub async fn spawn_capturing_llm(responses: Vec<Value>) -> CapturingLlm {
spawn_capturing_llm_with_status(responses.into_iter().map(|v| (200u16, v)).collect()).await
}
/// Like `spawn_capturing_llm` but each canned response carries its own HTTP
/// status, so a test can serve a real provider rejection (e.g. a context-window
/// 400) instead of only success bodies.
pub async fn spawn_capturing_llm_with_status(responses: Vec<(u16, Value)>) -> CapturingLlm {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let queue = Arc::new(Mutex::new(VecDeque::from(responses)));
let captured: Arc<Mutex<Vec<Value>>> = Arc::new(Mutex::new(Vec::new()));
let cap2 = captured.clone();
tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
let queue = queue.clone();
let captured = cap2.clone();
tokio::spawn(async move {
let mut buf = Vec::new();
let mut tmp = [0u8; 8192];
// Read until headers complete.
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(n) => buf.extend_from_slice(&tmp[..n]),
}
if buf.len() > 4_000_000 {
return;
}
}
// Parse Content-Length and read body.
let header_end = buf.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
let headers = &buf[..header_end];
let mut body_len = 0usize;
for line in headers.split(|b| *b == b'\n') {
let line = std::str::from_utf8(line).unwrap_or("");
if let Some(rest) = line.to_ascii_lowercase().strip_prefix("content-length:") {
body_len = rest.trim().trim_end_matches('\r').parse().unwrap_or(0);
}
}
while buf.len() < header_end + body_len {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(n) => buf.extend_from_slice(&tmp[..n]),
}
}
if let Ok(req) = serde_json::from_slice::<Value>(&buf[header_end..]) {
captured.lock().await.push(req);
}
let (status, body) = queue
.lock()
.await
.pop_front()
.unwrap_or_else(|| (200, json!({ "error": "no canned response" })));
let body_s = serde_json::to_string(&body).unwrap();
let reason = match status {
200 => "OK",
400 => "Bad Request",
_ => "Error",
};
let resp = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body_s.len(), body_s,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
});
}
});
CapturingLlm { url, captured }
}
pub struct Harness {
child: tokio::process::Child,
stdin: tokio::process::ChildStdin,
stdout: BufReader<tokio::process::ChildStdout>,
stderr: Arc<StdMutex<String>>,
next_id: i64,
}
impl Harness {
pub async fn spawn_with_env(base_url: &str, extra: &[(&str, &str)]) -> Self {
let bin = env!("CARGO_BIN_EXE_buzz-agent");
let mut cmd = tokio::process::Command::new(bin);
cmd.env("BUZZ_AGENT_PROVIDER", "openai")
.env("OPENAI_COMPAT_API_KEY", "test")
.env("OPENAI_COMPAT_MODEL", "fake-model")
.env("OPENAI_COMPAT_BASE_URL", base_url)
.env("BUZZ_AGENT_LLM_TIMEOUT_SECS", "5")
.env("BUZZ_AGENT_TOOL_TIMEOUT_SECS", "5")
.env("BUZZ_AGENT_MAX_ROUNDS", "8")
.env("BUZZ_AGENT_MCP_INIT_TIMEOUT_SECS", "2");
for (k, v) in extra {
cmd.env(k, v);
}
cmd.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
let mut child = cmd.spawn().expect("spawn buzz-agent");
let stdin = child.stdin.take().unwrap();
let stdout = BufReader::new(child.stdout.take().unwrap());
let stderr = child.stderr.take().unwrap();
let stderr_buf = Arc::new(StdMutex::new(String::new()));
let stderr_out = Arc::clone(&stderr_buf);
tokio::spawn(async move {
let mut reader = BufReader::new(stderr);
let mut line = String::new();
loop {
line.clear();
let n = match reader.read_line(&mut line).await {
Ok(n) => n,
Err(_) => break,
};
if n == 0 {
break;
}
if let Ok(mut out) = stderr_out.lock() {
out.push_str(&line);
}
}
});
Self {
child,
stdin,
stdout,
stderr: stderr_buf,
next_id: 1,
}
}
pub async fn spawn(base_url: &str) -> Self {
Self::spawn_with_env(base_url, &[]).await
}
pub async fn send(&mut self, method: &str, params: Value) -> i64 {
let id = self.next_id;
self.next_id += 1;
self.write(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }))
.await;
id
}
pub async fn notify(&mut self, method: &str, params: Value) {
self.write(json!({ "jsonrpc": "2.0", "method": method, "params": params }))
.await;
}
pub async fn write(&mut self, msg: Value) {
let mut s = serde_json::to_string(&msg).unwrap();
s.push('\n');
self.stdin.write_all(s.as_bytes()).await.unwrap();
self.stdin.flush().await.unwrap();
}
pub async fn recv(&mut self) -> Value {
let mut line = String::new();
let n = tokio::time::timeout(Duration::from_secs(15), self.stdout.read_line(&mut line))
.await
.expect("recv timeout")
.expect("read line");
assert!(n > 0, "agent EOF; stderr={}", self.stderr_text());
serde_json::from_str(&line).expect("non-JSON line")
}
pub async fn recv_until<F: FnMut(&Value) -> bool>(&mut self, mut pred: F) -> Value {
loop {
let v = self.recv().await;
if pred(&v) {
return v;
}
}
}
/// Like `recv_until`, but auto-approves any `session/request_permission`
/// seen while waiting. Tests that exercise tool execution, not the
/// permission boundary (that lives in `permission_boundary.rs`), must
/// approve a model-issued tool call so it reaches the server.
pub async fn recv_until_approving<F: FnMut(&Value) -> bool>(&mut self, mut pred: F) -> Value {
loop {
let v = self.recv().await;
if v.get("method") == Some(&json!("session/request_permission")) {
let resp = approve_permission(&v);
self.write(resp).await;
continue;
}
if pred(&v) {
return v;
}
}
}
pub async fn shutdown(mut self) {
drop(self.stdin);
let _ = tokio::time::timeout(Duration::from_secs(2), self.child.wait()).await;
let _ = self.child.start_kill();
}
pub fn stderr_text(&self) -> String {
self.stderr.lock().map(|s| s.clone()).unwrap_or_default()
}
}
pub fn openai_text(content: &str) -> Value {
json!({
"id": "cc-1", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": { "role": "assistant", "content": content },
"finish_reason": "stop",
}],
})
}
pub fn openai_tool_call(id: &str, name: &str, args: Value) -> Value {
json!({
"id": "cc-2", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": {
"role": "assistant", "content": null,
"tool_calls": [{
"id": id, "type": "function",
"function": { "name": name, "arguments": args.to_string() },
}],
},
"finish_reason": "tool_calls",
}],
})
}
/// Select the offered option whose `kind == "allow_once"` and return the
/// `session/request_permission` response. Mirrors buzz-acp's answering side,
/// which selects by `kind`, never by a hardcoded `optionId`. Centralizing this
/// means a future option-id rename can't silently turn allow into a denial.
pub fn approve_permission(request: &Value) -> Value {
let option_id = request["params"]["options"]
.as_array()
.and_then(|opts| opts.iter().find(|o| o["kind"] == "allow_once"))
.and_then(|o| o["optionId"].as_str())
.expect("request must offer an allow_once option");
json!({
"jsonrpc": "2.0",
"id": request["id"],
"result": { "outcome": { "outcome": "selected", "optionId": option_id } },
})
}

10
crates/buzz-agent/tests/fake_llm.rs
View File

@@ -16,6 +16,9 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;
use tokio::sync::Mutex;
mod common;
use common::approve_permission;
async fn spawn_fake_llm(responses: Vec<Value>) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
@@ -396,12 +399,7 @@ async fn unsupported_image_response_recovers_without_replaying_image() {
loop {
let message = h.recv().await;
if message.get("method") == Some(&json!("session/request_permission")) {
h.write(json!({
"jsonrpc": "2.0",
"id": message["id"],
"result": { "outcome": { "outcome": "selected", "optionId": "allow" } },
}))
.await;
h.write(approve_permission(&message)).await;
} else if message["id"] == json!(prompt_id) {
assert_eq!(message["result"]["stopReason"], "end_turn");
break;

631
crates/buzz-agent/tests/permission_boundary.rs Normal file
View File

@@ -0,0 +1,631 @@
//! Production authorization-boundary tests for the `session/request_permission`
//! surface.
//!
//! These drive a real `buzz-agent` subprocess against a fake MCP server and a
//! capturing LLM, and prove the security invariant end to end: an LLM-issued
//! MCP tool call reaches the server IFF the client selected the offered
//! allow-once option, and every other outcome fails closed without invoking the
//! tool. `fake_mcp` appends each invoked *bare* tool name to `FAKE_MCP_CALL_LOG`
//! (fired for `_Stop`/`_PostCompact` too), so "reached the tool exactly once"
//! and "never reached the tool" are both directly observable from disk.
//!
//! Timeout/abort/multi-session-cap state invariants live in the broker-seam unit
//! tests (`src/permission.rs`), which use an injectable deadline and inspect the
//! private correlation map — neither of which a subprocess can do.
//!
//! The subprocess `Harness`, capturing LLM, and `approve_permission` helper are
//! shared with the other integration suites via `mod common`.
use std::time::Duration;
use serde_json::{json, Value};
mod common;
use common::{approve_permission, spawn_capturing_llm, Harness};
// ─────────────────────────────────────────────────────────────────────────────
// LLM response builders
// ─────────────────────────────────────────────────────────────────────────────
fn openai_text(content: &str) -> Value {
json!({
"id": "cc-1", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": { "role": "assistant", "content": content },
"finish_reason": "stop",
}],
})
}
/// One assistant turn issuing `calls`, each `(id, qualified_name, arguments)`.
fn openai_tool_calls(calls: &[(&str, &str, Value)]) -> Value {
let tool_calls: Vec<Value> = calls
.iter()
.map(|(id, name, args)| {
json!({
"id": id, "type": "function",
"function": { "name": name, "arguments": args.to_string() },
})
})
.collect();
json!({
"id": "cc-tc", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": { "role": "assistant", "content": null, "tool_calls": tool_calls },
"finish_reason": "tool_calls",
}],
})
}
fn shell_call(id: &str) -> Value {
openai_tool_calls(&[(id, "fake__shell", json!({ "command": "ls" }))])
}
// ─────────────────────────────────────────────────────────────────────────────
// Permission-response builders + drivers
// ─────────────────────────────────────────────────────────────────────────────
/// Bare JSON-RPC response selecting `option_id`.
fn resp_selected(id: &Value, option_id: &str) -> Value {
json!({
"jsonrpc": "2.0",
"id": id.clone(),
"result": { "outcome": { "outcome": "selected", "optionId": option_id } },
})
}
/// Bare JSON-RPC response carrying an arbitrary `result` shape.
fn resp_result(id: &Value, result: Value) -> Value {
json!({ "jsonrpc": "2.0", "id": id.clone(), "result": result })
}
/// Bare JSON-RPC *error* response (id present, `error`, no `result`).
fn resp_error(id: &Value) -> Value {
json!({
"jsonrpc": "2.0",
"id": id.clone(),
"error": { "code": -32601, "message": "method not found" },
})
}
/// The tool title carried by a permission request, across both wire shapes.
fn perm_title(req: &Value) -> String {
let p = &req["params"];
p["title"]
.as_str()
.or_else(|| p["toolCall"]["title"].as_str())
.or_else(|| p["subject"]["toolCall"]["title"].as_str())
.unwrap_or("")
.to_owned()
}
/// Drive one prompt to completion. For each `session/request_permission`,
/// `decide(&req)` returns `Some(response)` to answer or `None` to leave it
/// unanswered. Returns the final prompt response and every request seen.
async fn drive(
h: &mut Harness,
sid: &str,
prompt: &str,
mut decide: impl FnMut(&Value) -> Option<Value>,
) -> (Value, Vec<Value>) {
let p = h
.send(
"session/prompt",
json!({ "sessionId": sid, "prompt": [{ "type": "text", "text": prompt }] }),
)
.await;
let mut requests: Vec<Value> = Vec::new();
loop {
let v = h.recv().await;
if v.get("method") == Some(&json!("session/request_permission")) {
requests.push(v.clone());
if let Some(resp) = decide(&v) {
h.write(resp).await;
}
continue;
}
if v["id"] == json!(p) {
return (v, requests);
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Session init + call-log helpers
// ─────────────────────────────────────────────────────────────────────────────
fn call_log_path(tag: &str) -> String {
let p = std::env::temp_dir().join(format!(
"buzz_perm_calllog_{tag}_{}_{:x}.log",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
));
let s = p.to_string_lossy().to_string();
let _ = std::fs::remove_file(&s);
s
}
/// Invoked tool names recorded by fake_mcp (bare names, one per `tools/call`).
/// A missing file means zero invocations.
fn call_log_lines(path: &str) -> Vec<String> {
std::fs::read_to_string(path)
.unwrap_or_default()
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.map(str::to_owned)
.collect()
}
/// Initialize + create a session with a single fake MCP server named `fake`,
/// negotiating `protocol_version` and passing `mcp_env` to the server. `cwd`
/// controls skill discovery (`.agents/skills`).
async fn init(
h: &mut Harness,
protocol_version: u32,
cwd: &str,
mcp_env: &[(&str, &str)],
) -> String {
let fake_mcp = env!("CARGO_BIN_EXE_fake-mcp");
let env: Vec<Value> = mcp_env
.iter()
.map(|(k, v)| json!({ "name": k, "value": v }))
.collect();
h.send(
"initialize",
json!({ "protocolVersion": protocol_version, "clientCapabilities": {} }),
)
.await;
let _ = h.recv().await;
let servers = if mcp_env.is_empty() {
json!([])
} else {
json!([{ "name": "fake", "command": fake_mcp, "args": [], "env": env }])
};
h.send("session/new", json!({ "cwd": cwd, "mcpServers": servers }))
.await;
let r = h
.recv_until(|v| v.get("result").is_some() || v.get("error").is_some())
.await;
r["result"]["sessionId"]
.as_str()
.unwrap_or_else(|| panic!("session/new failed: {r}, stderr={}", h.stderr_text()))
.to_owned()
}
fn stop_reason(resp: &Value) -> String {
resp["result"]["stopReason"]
.as_str()
.unwrap_or("")
.to_owned()
}
// ═════════════════════════════════════════════════════════════════════════════
// The authorization boundary
// ═════════════════════════════════════════════════════════════════════════════
/// Exact allow reaches the tool exactly once — and never *before* approval.
/// The pre-approval check proves the gate precedes the MCP call, not just that
/// the tally ends at one.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_allow_reaches_tool_exactly_once_and_not_before_approval() {
let log = call_log_path("allow");
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("done")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
1,
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let log_for_check = log.clone();
let (resp, requests) = drive(&mut h, &sid, "go", |req| {
// Before answering, the tool must not have run.
assert!(
call_log_lines(&log_for_check).is_empty(),
"tool invoked BEFORE approval"
);
Some(approve_permission(req))
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert_eq!(requests.len(), 1, "exactly one call → exactly one ask");
assert_eq!(
call_log_lines(&log),
vec!["shell"],
"approved tool reached MCP exactly once"
);
h.shutdown().await;
}
/// Selecting the offered reject option never invokes the tool, and the model
/// sees a permission-denied tool error (the turn continues to `end_turn`).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_reject_option_never_invokes_tool() {
let log = call_log_path("reject");
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("understood")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
1,
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let (resp, requests) = drive(&mut h, &sid, "go", |req| {
Some(resp_selected(&req["id"], "reject_once"))
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert_eq!(requests.len(), 1);
assert!(
call_log_lines(&log).is_empty(),
"rejected tool must never reach MCP"
);
h.shutdown().await;
}
/// Every non-authorizing response shape fails closed: the tool never runs.
/// One subprocess per shape keeps the failure attributable.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_adversarial_outcomes_never_invoke_tool() {
// (tag, response-for-request builder) — the full fail-closed matrix.
type Shape = (&'static str, fn(&Value) -> Value);
let shapes: Vec<Shape> = vec![
("cancelled", |req| {
resp_result(&req["id"], json!({ "outcome": { "outcome": "cancelled" } }))
}),
("jsonrpc_error", |req| resp_error(&req["id"])),
("missing_outcome", |req| resp_result(&req["id"], json!({}))),
("empty_outcome", |req| {
resp_result(&req["id"], json!({ "outcome": {} }))
}),
("selected_no_option", |req| {
resp_result(&req["id"], json!({ "outcome": { "outcome": "selected" } }))
}),
("unknown_outcome", |req| {
resp_result(
&req["id"],
json!({ "outcome": { "outcome": "banana", "optionId": "allow_once" } }),
)
}),
("wrong_option_id", |req| {
resp_selected(&req["id"], "not_an_offered_option")
}),
// Structurally malformed *frames* that each still carry a well-formed
// `selected`/`allow_once` result — the payload would authorize, so only
// the frame-structure check (wire::classify) stands between them and the
// tool. These mirror the two broker-seam unit tests
// (`src/permission.rs::test_frame_with_{result_and_error,non_string_method}_denies_tool`)
// at the live MCP-log seam, proving the frame gate denies end to end.
("result_and_error", |req| {
let mut frame = approve_permission(req);
frame["error"] = json!({ "code": -32603, "message": "internal" });
frame
}),
("non_string_method", |req| {
let mut frame = approve_permission(req);
frame["method"] = json!(7);
frame
}),
];
for (tag, build) in shapes {
let log = call_log_path(tag);
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("ok")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
1,
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let (resp, requests) = drive(&mut h, &sid, "go", |req| Some(build(req))).await;
assert_eq!(
stop_reason(&resp),
"end_turn",
"shape {tag}: turn should continue"
);
assert_eq!(requests.len(), 1, "shape {tag}: exactly one ask");
assert!(
call_log_lines(&log).is_empty(),
"shape {tag}: non-authorizing outcome must never reach MCP"
);
h.shutdown().await;
}
}
/// A stale/unknown/foreign response id is ignored and does not unblock the live
/// waiter; the correct id then authorizes exactly once.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_stale_id_ignored_then_real_id_authorizes() {
let log = call_log_path("staleid");
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("done")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
1,
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let p = h
.send(
"session/prompt",
json!({ "sessionId": sid, "prompt": [{ "type": "text", "text": "go" }] }),
)
.await;
// Wait for the ask.
let req = h
.recv_until(|v| v.get("method") == Some(&json!("session/request_permission")))
.await;
// Feed several ignorable responses first: a minted-but-never-issued id, a
// foreign numeric id, and a null id. None may unblock the real waiter.
h.write(resp_selected(&json!("perm-9999"), "allow_once"))
.await;
h.write(resp_selected(&json!(7), "allow_once")).await;
h.write(resp_selected(&Value::Null, "allow_once")).await;
// Give the agent a beat to (wrongly) act on any of them.
tokio::time::sleep(Duration::from_millis(150)).await;
assert!(
call_log_lines(&log).is_empty(),
"stale/foreign ids must not authorize the pending call"
);
// The real id resolves it exactly once.
h.write(approve_permission(&req)).await;
let resp = h.recv_until(|v| v["id"] == json!(p)).await;
assert_eq!(stop_reason(&resp), "end_turn");
assert_eq!(
call_log_lines(&log),
vec!["shell"],
"only the correct id authorizes, exactly once"
);
h.shutdown().await;
}
/// `session/cancel` while a permission ask is outstanding terminates the turn
/// promptly, executes nothing, and needs no `cancelled` permission response —
/// a Buzz client always answers, but a non-Buzz ACP client may violate the spec
/// by staying silent, and cancellation must not depend on that answer.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_cancel_while_waiting_executes_nothing_without_client_answer() {
let log = call_log_path("cancelwait");
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("done")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
1,
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let p = h
.send(
"session/prompt",
json!({ "sessionId": sid, "prompt": [{ "type": "text", "text": "go" }] }),
)
.await;
// Wait for the ask, then cancel WITHOUT ever answering the permission.
let _req = h
.recv_until(|v| v.get("method") == Some(&json!("session/request_permission")))
.await;
h.notify("session/cancel", json!({ "sessionId": sid }))
.await;
let resp = h.recv_until(|v| v["id"] == json!(p)).await;
assert_eq!(
stop_reason(&resp),
"cancelled",
"cancel resolves the turn without a client permission answer"
);
assert!(
call_log_lines(&log).is_empty(),
"a cancelled ask must never reach the tool"
);
h.shutdown().await;
}
/// Two parallel calls each get their own ask (distinct ids); crossed decisions —
/// deny the first-asked, allow the second-asked — authorize only the allowed
/// call. Serial admission (`max_parallel_tools=1`) serializes the asks: the
/// second ask fires only after the first resolves. Which tool is admitted first
/// is a tokio scheduling detail, so the test denies whichever is asked first and
/// proves only the allowed (second) call reached MCP — order-agnostic.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_crossed_parallel_decisions_authorize_only_matching_call() {
let log = call_log_path("crossed");
// Two distinct registered tools so the call log distinguishes them by name.
let llm = spawn_capturing_llm(vec![
openai_tool_calls(&[
("tc-alpha", "fake__alpha", json!({})),
("tc-bravo", "fake__bravo", json!({})),
]),
openai_text("done"),
])
.await;
let mut h = Harness::spawn_with_env(&llm.url, &[("BUZZ_AGENT_MAX_PARALLEL_TOOLS", "1")]).await;
let sid = init(
&mut h,
1,
"/tmp",
&[
("FAKE_MCP_NAMED_TOOLS", "alpha,bravo"),
("FAKE_MCP_CALL_LOG", &log),
],
)
.await;
// Deny whichever tool is asked first; allow the second. Ids are distinct.
let mut seen_ids: Vec<Value> = Vec::new();
let mut allowed_title = String::new();
let (resp, requests) = drive(&mut h, &sid, "go", |req| {
assert!(
!seen_ids.contains(&req["id"]),
"each parallel call must get a distinct request id"
);
seen_ids.push(req["id"].clone());
if seen_ids.len() == 1 {
Some(resp_selected(&req["id"], "reject_once")) // deny the first-asked
} else {
allowed_title = perm_title(req);
Some(approve_permission(req)) // allow the second-asked
}
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert_eq!(requests.len(), 2, "one ask per parallel call");
// The call log records bare tool names; the title carries the qualified
// `<server>__<name>`. Only the allowed (second-asked) call reached MCP.
let allowed_bare = allowed_title
.strip_prefix("fake__")
.expect("qualified title")
.to_owned();
assert_eq!(
call_log_lines(&log),
vec![allowed_bare],
"only the allowed (second-asked) call reached MCP; the denied one did not"
);
h.shutdown().await;
}
/// The built-in `load_skill` tool is not an MCP call and is exempt from the
/// permission boundary: it executes with no `session/request_permission`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_load_skill_emits_no_permission_request() {
let tmp = tempfile::TempDir::new().unwrap();
let cwd = tmp.path();
let skill_dir = cwd.join(".agents/skills/my-skill");
std::fs::create_dir_all(&skill_dir).unwrap();
std::fs::write(
skill_dir.join("SKILL.md"),
"---\nname: my-skill\ndescription: A skill\n---\nSKILL_BODY_77\n",
)
.unwrap();
let llm = spawn_capturing_llm(vec![
openai_tool_calls(&[("tc-ls", "load_skill", json!({ "name": "my-skill" }))]),
openai_text("done"),
])
.await;
let mut h = Harness::spawn(&llm.url).await;
// No MCP server: `load_skill` is a built-in, and skills come from `cwd`.
let sid = init(&mut h, 1, cwd.to_str().unwrap(), &[]).await;
let (resp, requests) = drive(&mut h, &sid, "use my-skill", |_| {
panic!("load_skill must not trigger a permission request")
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert!(requests.is_empty(), "built-in load_skill is exempt");
h.shutdown().await;
}
/// Lifecycle hooks (`_Stop`, `_PostCompact`) invoke MCP through `call_hooks`,
/// not the model-issued tool path, so they are exempt: the hook reaches the
/// server (call log records it) with no permission ask. Here the `_Stop` hook
/// objects once, forcing a hook invocation the test can observe.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_stop_hook_reaches_mcp_without_permission_ask() {
let log = call_log_path("stophook");
// Text turn triggers the _Stop gate → hook objects once → agent loops →
// second text turn, hook silent → end_turn. No model-issued tool call.
let llm = spawn_capturing_llm(vec![
openai_text("premature"),
openai_text("really done"),
openai_text("unexpected"),
])
.await;
let mut h = Harness::spawn_with_env(
&llm.url,
&[
("MCP_HOOK_SERVERS", "fake"),
("BUZZ_AGENT_STOP_MAX_REJECTIONS", "10"),
],
)
.await;
let sid = init(
&mut h,
1,
"/tmp",
&[
("FAKE_MCP_TOOL_COUNT", "1"),
("FAKE_MCP_STOP_HOOK", "1"),
("FAKE_MCP_STOP_TEXT", "you have open work"),
("FAKE_MCP_STOP_COUNT", "1"),
("FAKE_MCP_CALL_LOG", &log),
],
)
.await;
let (resp, requests) = drive(&mut h, &sid, "go", |_| {
panic!("a lifecycle hook must not trigger a permission request")
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert!(requests.is_empty(), "_Stop hook is exempt from the ask");
assert!(
call_log_lines(&log).contains(&"_Stop".to_owned()),
"the _Stop hook still reached MCP without an ask; log={:?}",
call_log_lines(&log)
);
h.shutdown().await;
}
/// Under a v2-negotiated connection, the emitted `session/request_permission`
/// carries the v2 shape (`subject.toolCall`, top-level `title`), never the v1
/// legacy top-level `toolCall`. Complements the pure v1/v2 builder unit tests
/// with an end-to-end proof that the negotiated version reaches the wire.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_v2_negotiation_emits_v2_request_shape() {
let log = call_log_path("v2shape");
let llm = spawn_capturing_llm(vec![shell_call("tc1"), openai_text("done")]).await;
let mut h = Harness::spawn(&llm.url).await;
let sid = init(
&mut h,
2, // negotiate v2
"/tmp",
&[("FAKE_MCP_SHELL_TOOL", "1"), ("FAKE_MCP_CALL_LOG", &log)],
)
.await;
let (resp, requests) = drive(&mut h, &sid, "go", |req| {
let params = &req["params"];
// v2: tool context under `subject.toolCall`, with top-level `title`.
assert_eq!(params["subject"]["type"], "tool_call", "v2 uses subject");
assert_eq!(params["subject"]["toolCall"]["title"], "fake__shell");
assert_eq!(params["title"], "fake__shell", "v2 has top-level title");
assert!(
params.get("toolCall").is_none(),
"v2 must not carry the v1 top-level toolCall"
);
Some(approve_permission(req))
})
.await;
assert_eq!(stop_reason(&resp), "end_turn");
assert_eq!(requests.len(), 1);
assert_eq!(call_log_lines(&log), vec!["shell"]);
h.shutdown().await;
}

275
crates/buzz-agent/tests/regressions.rs
View File

@@ -5,220 +5,16 @@
//! - cancellation leaves history valid for the next prompt
//! - empty-content assistant turn doesn't poison OpenAI history
use std::collections::VecDeque;
use std::process::Stdio;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, Instant};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;
use tokio::sync::Mutex;
struct CapturingLlm {
url: String,
captured: Arc<Mutex<Vec<Value>>>,
}
async fn spawn_capturing_llm(responses: Vec<Value>) -> CapturingLlm {
spawn_capturing_llm_with_status(responses.into_iter().map(|v| (200u16, v)).collect()).await
}
/// Like `spawn_capturing_llm` but each canned response carries its own HTTP
/// status, so a test can serve a real provider rejection (e.g. a context-window
/// 400) instead of only success bodies.
async fn spawn_capturing_llm_with_status(responses: Vec<(u16, Value)>) -> CapturingLlm {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let queue = Arc::new(Mutex::new(VecDeque::from(responses)));
let captured: Arc<Mutex<Vec<Value>>> = Arc::new(Mutex::new(Vec::new()));
let cap2 = captured.clone();
tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
let queue = queue.clone();
let captured = cap2.clone();
tokio::spawn(async move {
let mut buf = Vec::new();
let mut tmp = [0u8; 8192];
// Read until headers complete.
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(n) => buf.extend_from_slice(&tmp[..n]),
}
if buf.len() > 4_000_000 {
return;
}
}
// Parse Content-Length and read body.
let header_end = buf.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
let headers = &buf[..header_end];
let mut body_len = 0usize;
for line in headers.split(|b| *b == b'\n') {
let line = std::str::from_utf8(line).unwrap_or("");
if let Some(rest) = line.to_ascii_lowercase().strip_prefix("content-length:") {
body_len = rest.trim().trim_end_matches('\r').parse().unwrap_or(0);
}
}
while buf.len() < header_end + body_len {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(n) => buf.extend_from_slice(&tmp[..n]),
}
}
if let Ok(req) = serde_json::from_slice::<Value>(&buf[header_end..]) {
captured.lock().await.push(req);
}
let (status, body) = queue
.lock()
.await
.pop_front()
.unwrap_or_else(|| (200, json!({ "error": "no canned response" })));
let body_s = serde_json::to_string(&body).unwrap();
let reason = match status {
200 => "OK",
400 => "Bad Request",
_ => "Error",
};
let resp = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body_s.len(), body_s,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
});
}
});
CapturingLlm { url, captured }
}
struct Harness {
child: tokio::process::Child,
stdin: tokio::process::ChildStdin,
stdout: BufReader<tokio::process::ChildStdout>,
stderr: Arc<StdMutex<String>>,
next_id: i64,
}
impl Harness {
async fn spawn_with_env(base_url: &str, extra: &[(&str, &str)]) -> Self {
let bin = env!("CARGO_BIN_EXE_buzz-agent");
let mut cmd = tokio::process::Command::new(bin);
cmd.env("BUZZ_AGENT_PROVIDER", "openai")
.env("OPENAI_COMPAT_API_KEY", "test")
.env("OPENAI_COMPAT_MODEL", "fake-model")
.env("OPENAI_COMPAT_BASE_URL", base_url)
.env("BUZZ_AGENT_LLM_TIMEOUT_SECS", "5")
.env("BUZZ_AGENT_TOOL_TIMEOUT_SECS", "5")
.env("BUZZ_AGENT_MAX_ROUNDS", "8")
.env("BUZZ_AGENT_MCP_INIT_TIMEOUT_SECS", "2");
for (k, v) in extra {
cmd.env(k, v);
}
cmd.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
let mut child = cmd.spawn().expect("spawn buzz-agent");
let stdin = child.stdin.take().unwrap();
let stdout = BufReader::new(child.stdout.take().unwrap());
let stderr = child.stderr.take().unwrap();
let stderr_buf = Arc::new(StdMutex::new(String::new()));
let stderr_out = Arc::clone(&stderr_buf);
tokio::spawn(async move {
let mut reader = BufReader::new(stderr);
let mut line = String::new();
loop {
line.clear();
let n = match reader.read_line(&mut line).await {
Ok(n) => n,
Err(_) => break,
};
if n == 0 {
break;
}
if let Ok(mut out) = stderr_out.lock() {
out.push_str(&line);
}
}
});
Self {
child,
stdin,
stdout,
stderr: stderr_buf,
next_id: 1,
}
}
async fn spawn(base_url: &str) -> Self {
Self::spawn_with_env(base_url, &[]).await
}
async fn send(&mut self, method: &str, params: Value) -> i64 {
let id = self.next_id;
self.next_id += 1;
self.write(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }))
.await;
id
}
async fn notify(&mut self, method: &str, params: Value) {
self.write(json!({ "jsonrpc": "2.0", "method": method, "params": params }))
.await;
}
async fn write(&mut self, msg: Value) {
let mut s = serde_json::to_string(&msg).unwrap();
s.push('\n');
self.stdin.write_all(s.as_bytes()).await.unwrap();
self.stdin.flush().await.unwrap();
}
async fn recv(&mut self) -> Value {
let mut line = String::new();
let n = tokio::time::timeout(Duration::from_secs(15), self.stdout.read_line(&mut line))
.await
.expect("recv timeout")
.expect("read line");
assert!(n > 0, "agent EOF");
serde_json::from_str(&line).expect("non-JSON line")
}
async fn recv_until<F: FnMut(&Value) -> bool>(&mut self, mut pred: F) -> Value {
loop {
let v = self.recv().await;
if pred(&v) {
return v;
}
}
}
async fn shutdown(mut self) {
drop(self.stdin);
let _ = tokio::time::timeout(Duration::from_secs(2), self.child.wait()).await;
let _ = self.child.start_kill();
}
fn stderr_text(&self) -> String {
self.stderr.lock().map(|s| s.clone()).unwrap_or_default()
}
}
fn openai_text(content: &str) -> Value {
json!({
"id": "cc-1", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": { "role": "assistant", "content": content },
"finish_reason": "stop",
}],
})
}
mod common;
use common::{
approve_permission, openai_text, openai_tool_call, spawn_capturing_llm,
spawn_capturing_llm_with_status, Harness,
};
/// Like [`openai_text`] but attaches a `usage` block so tests can drive the
/// token-based handoff gate. `prompt_tokens` is the input-token count the
@@ -253,23 +49,6 @@ fn openai_max_tokens(content: &str, tool_calls: Value) -> Value {
})
}
fn openai_tool_call(id: &str, name: &str, args: Value) -> Value {
json!({
"id": "cc-2", "object": "chat.completion", "model": "fake-model",
"choices": [{
"index": 0,
"message": {
"role": "assistant", "content": null,
"tool_calls": [{
"id": id, "type": "function",
"function": { "name": name, "arguments": args.to_string() },
}],
},
"finish_reason": "tool_calls",
}],
})
}
async fn init_session(h: &mut Harness, mcp_servers: Value) -> String {
h.send(
"initialize",
@@ -676,13 +455,7 @@ async fn per_turn_tool_call_cap_enforced() {
loop {
let v = h.recv().await;
if v.get("method") == Some(&json!("session/request_permission")) {
let id = v["id"].clone();
h.write(json!({
"jsonrpc": "2.0",
"id": id,
"result": { "outcome": { "outcome": "selected", "optionId": "allow" } },
}))
.await;
h.write(approve_permission(&v)).await;
continue;
}
if v.get("method") == Some(&json!("session/update"))
@@ -858,7 +631,7 @@ async fn hook_stop_blocks_premature_end() {
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
)
.await;
let r = h.recv_until(|v| v["id"] == json!(p)).await;
let r = h.recv_until_approving(|v| v["id"] == json!(p)).await;
assert!(r.get("result").is_some(), "errored: {r}");
assert_eq!(r["result"]["stopReason"], "end_turn");
@@ -936,7 +709,7 @@ async fn hook_stop_budget_exhausted() {
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
)
.await;
let r = h.recv_until(|v| v["id"] == json!(p)).await;
let r = h.recv_until_approving(|v| v["id"] == json!(p)).await;
assert!(r.get("result").is_some(), "errored: {r}");
assert_eq!(r["result"]["stopReason"], "end_turn");
@@ -1517,7 +1290,7 @@ async fn stale_usage_plus_history_growth_triggers_handoff() {
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
)
.await;
let _ = h.recv_until(|v| v["id"] == json!(p)).await;
let _ = h.recv_until_approving(|v| v["id"] == json!(p)).await;
// req1 (tool_call) + summarize (handoff) + req2 (done) = 3. Without the
// growth estimate we'd see only 2 (stale 8500 < 9000, no handoff).
let captured = llm.captured.lock().await.len();
@@ -1788,7 +1561,7 @@ async fn cancel_sends_notifications_cancelled_to_any_mcp_server() {
.await;
// Wait for tool call to be in-progress.
h.recv_until(|v| {
h.recv_until_approving(|v| {
v.get("params")
.and_then(|p| p.get("update"))
.and_then(|u| u.get("status"))
@@ -1903,13 +1676,7 @@ async fn prompt_to_completion(h: &mut Harness, sid: &str) -> Value {
loop {
let v = h.recv().await;
if v.get("method") == Some(&json!("session/request_permission")) {
let id = v["id"].clone();
h.write(json!({
"jsonrpc": "2.0",
"id": id,
"result": { "outcome": { "outcome": "selected", "optionId": "allow" } },
}))
.await;
h.write(approve_permission(&v)).await;
continue;
}
if v["id"] == json!(p) {
@@ -2664,7 +2431,9 @@ async fn max_tokens_recovery_can_proceed_to_tool_call() {
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
)
.await;
let reply = h.recv_until(|v| v["id"] == json!(prompt_id)).await;
let reply = h
.recv_until_approving(|v| v["id"] == json!(prompt_id))
.await;
assert_eq!(reply["result"]["stopReason"], "end_turn", "{reply}");
let requests = llm.captured.lock().await;
assert_eq!(requests.len(), 3);
@@ -3642,13 +3411,7 @@ async fn handoff_cap_binds_within_a_single_turn() {
}
if v.get("method") == Some(&json!("session/request_permission")) {
let id = v["id"].clone();
h.write(json!({
"jsonrpc": "2.0",
"id": id,
"result": { "outcome": { "outcome": "selected", "optionId": "allow" } },
}))
.await;
h.write(approve_permission(&v)).await;
continue;
}
if v["id"] == json!(p2) {
@@ -3795,13 +3558,7 @@ async fn failed_summarize_burns_handoff_attempt_budget() {
loop {
let v = h.recv().await;
if v.get("method") == Some(&json!("session/request_permission")) {
let id = v["id"].clone();
h.write(json!({
"jsonrpc": "2.0",
"id": id,
"result": { "outcome": { "outcome": "selected", "optionId": "allow" } },
}))
.await;
h.write(approve_permission(&v)).await;
continue;
}
if v["id"] == json!(p2) {