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

crates/buzz-agent/src/agent.rs

@@ -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 {

crates/buzz-agent/src/config.rs

@@ -813,6 +813,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).
@@ -956,6 +968,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,
@@ -1002,6 +1019,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,
@@ -1063,6 +1082,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());
         }

crates/buzz-agent/src/lib.rs

@@ -8,6 +8,7 @@ mod handoff;
 mod hints;
 mod llm;
 mod mcp;
+mod permission;
 pub mod types;
 mod wire;
 
@@ -31,6 +32,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};
@@ -53,6 +55,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
@@ -180,28 +193,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>(
@@ -234,7 +272,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
         }
@@ -249,7 +290,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();
@@ -290,7 +331,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,
@@ -302,6 +343,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(
@@ -730,6 +777,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,

crates/buzz-agent/src/llm.rs

@@ -2603,6 +2603,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,

crates/buzz-agent/src/mcp.rs

@@ -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.