Expand hosted agent controls and message projection

AGENTS.md

@@ -99,6 +99,15 @@ inside the background PowerShell process.
 
 See CONTRIBUTING.md for full setup details and dependency requirements.
 
+### Agent changes are cross-surface changes
+
+Before changing an agent name, avatar, model, access rule, channel membership,
+mention behavior, or historical message presentation, read
+[`docs/agent-surface-map.md`](docs/agent-surface-map.md). It inventories the
+relay events, precedence rules, write paths, caches, desktop/web routes, UI
+consumers, and required tests. Update the map in the same change when a route,
+source of truth, consumer, or invalidation boundary changes.
+
 ---
 
 ## Quality Gates

Cargo.toml

@@ -136,6 +136,18 @@ buzz-sdk = { path = "crates/buzz-sdk" }
 buzz-ws-client = { path = "crates/buzz-ws-client" }
 buzz-relay-mesh = { path = "crates/buzz-relay-mesh" }
 
+# Dev profile — workspace crates keep full debug info; dependencies carry none.
+# Rust's default `debug = true` emits DWARF/PDB for all ~800 dependency crates,
+# and that debug info is the majority of a debug `target/` by size. We almost
+# never step into a dependency, so dropping it costs nothing we use while
+# keeping our own crates fully debuggable. Recompiling deps is unaffected —
+# this changes what is emitted, not what is built.
+[profile.dev]
+debug = true
+
+[profile.dev.package."*"]
+debug = false
+
 # CI profile — builds the relay for desktop e2e. Dependencies keep full
 # release optimization (warm from main's cache; they carry the runtime hot
 # path: tokio/sqlx/axum). Workspace crates build at opt-level 1 — enough for

crates/buzz-acp/src/acp.rs

@@ -20,6 +20,68 @@ use crate::usage::{TurnUsage, UsageTracker};
 /// Lines exceeding this limit are rejected to prevent OOM from rogue agents.
 const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB
 
+/// Build the process invocation for an ACP adapter.
+///
+/// npm installs expose adapters as `.cmd` shims on Windows. `CreateProcess`
+/// cannot execute those files directly (OS error 193), so route only those
+/// shims through the system command processor. Native executables and every
+/// non-Windows platform retain the direct-exec path.
+fn windows_batch_spawn(command: &str, args: &[String]) -> (String, Vec<String>) {
+    #[cfg(windows)]
+    {
+        let is_batch = std::path::Path::new(command)
+            .extension()
+            .map(|extension| {
+                matches!(
+                    extension.to_string_lossy().to_ascii_lowercase().as_str(),
+                    "cmd" | "bat"
+                )
+            })
+            .unwrap_or(false);
+        if is_batch {
+            // npm's PowerShell shim only forwards stdin when PowerShell itself
+            // detects pipeline input. A Rust pipe is not reported that way, so
+            // ACP initialize hangs. Execute the installed JavaScript entrypoint
+            // with Node directly, preserving the harness's stdio handles.
+            let shim_path = std::path::Path::new(command);
+            let package_name = shim_path.file_stem().and_then(|stem| stem.to_str());
+            let npm_root = shim_path.parent();
+            let script = npm_root.zip(package_name).map(|(root, package)| {
+                root.join("node_modules")
+                    .join("@agentclientprotocol")
+                    .join(package)
+                    .join("dist")
+                    .join("index.js")
+            });
+            if let Some(script) = script.filter(|path| path.is_file()) {
+                let sibling_node = npm_root
+                    .map(|root| root.join("node.exe"))
+                    .filter(|path| path.is_file());
+                let node = sibling_node
+                    .map(|path| path.display().to_string())
+                    .unwrap_or_else(|| "node.exe".to_string());
+                let mut node_args = vec![script.display().to_string()];
+                node_args.extend_from_slice(args);
+                return (node, node_args);
+            }
+
+            // Non-npm batch adapters have no sibling PowerShell shim. Keep the
+            // compatibility fallback for simple batch files; catalogued Buzz
+            // runtimes all take the safer branch above.
+            return (
+                std::env::var("COMSPEC").unwrap_or_else(|_| "cmd.exe".to_string()),
+                std::iter::once("/D".to_string())
+                    .chain(std::iter::once("/C".to_string()))
+                    .chain(std::iter::once(command.to_string()))
+                    .chain(args.iter().cloned())
+                    .collect(),
+            );
+        }
+    }
+
+    (command.to_string(), args.to_vec())
+}
+
 /// An MCP server configuration passed to `session/new`.
 ///
 /// Corresponds to the `McpServerStdio` variant in the ACP schema.
@@ -456,8 +518,9 @@ impl AcpClient {
     ) -> Result<Self, AcpError> {
         use std::process::Stdio;
 
-        let mut cmd = tokio::process::Command::new(command);
-        cmd.args(args)
+        let (spawn_command, spawn_args) = windows_batch_spawn(command, args);
+        let mut cmd = tokio::process::Command::new(spawn_command);
+        cmd.args(spawn_args)
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             // Inherit stderr so agent logs are visible in the harness terminal.
@@ -2256,6 +2319,39 @@ fn configure_no_window(cmd: &mut tokio::process::Command) {
 mod tests {
     use super::*;
 
+    #[test]
+    fn native_adapter_spawn_stays_direct() {
+        let args = vec!["acp".to_string()];
+        let (command, actual_args) = windows_batch_spawn("adapter.exe", &args);
+        assert_eq!(command, "adapter.exe");
+        assert_eq!(actual_args, args);
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn windows_npm_adapter_uses_node_entrypoint() {
+        let temp =
+            std::env::temp_dir().join(format!("buzz-acp-shim-test-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&temp).expect("tempdir");
+        let cmd_shim = temp.join("codex-acp.cmd");
+        let script = temp
+            .join("node_modules")
+            .join("@agentclientprotocol")
+            .join("codex-acp")
+            .join("dist")
+            .join("index.js");
+        std::fs::create_dir_all(script.parent().expect("script parent")).expect("package dirs");
+        std::fs::write(&cmd_shim, "@echo off\r\n").expect("cmd shim");
+        std::fs::write(&script, "").expect("Node entrypoint");
+        let args = vec!["acp".to_string(), "--flag=value with spaces".to_string()];
+        let (command, actual_args) =
+            windows_batch_spawn(cmd_shim.to_str().expect("utf8 path"), &args);
+        assert_eq!(command, "node.exe");
+        assert_eq!(actual_args[0], script.display().to_string());
+        assert_eq!(&actual_args[1..], args);
+        std::fs::remove_dir_all(temp).expect("remove tempdir");
+    }
+
     #[test]
     fn stop_reason_parses_all_known_values() {
         assert_eq!(StopReason::from_str("end_turn"), Some(StopReason::EndTurn));

crates/buzz-agent/src/agent.rs

@@ -13,8 +13,8 @@ use crate::mcp::McpRegistry;
 use crate::mcp::ResultBudget;
 
 use crate::types::{
-    AgentError, ContentBlock, HistoryItem, ProviderStop, StopReason, ToolCall, ToolResult,
-    ToolResultContent, TurnTotalState,
+    AgentError, ContentBlock, HistoryItem, ProviderStop, SessionUsageBaseline, StopReason,
+    ToolCall, ToolResult, ToolResultContent, TurnTotalState,
 };
 use crate::wire::{self, WireSender};
 
@@ -150,9 +150,40 @@ pub struct RunCtx<'a> {
     /// Reset to `Unseen` at turn start in `run()`. Callers must not derive a
     /// total by summing input+output — that is the UI display approximation only.
     pub turn_total_state: &'a mut TurnTotalState,
+    /// Session-cumulative counters as they stood when this turn began. Added to
+    /// the `turn_*` accumulators above to report a cumulative figure mid-turn;
+    /// the session's own copy is only advanced once, after the turn returns.
+    pub usage_baseline: SessionUsageBaseline,
 }
 
 impl RunCtx<'_> {
+    /// Send a session-cumulative `usage_update` reflecting everything observed
+    /// up to and including the most recent LLM response.
+    ///
+    /// The figure is the turn-start baseline plus this turn's running
+    /// accumulators, which is exactly what `session/prompt` will fold into the
+    /// session once the turn returns — so a mid-turn notification and the
+    /// end-of-turn one agree, and a turn that never returns has still reported
+    /// everything but its final in-flight request.
+    async fn emit_usage_update(&self) {
+        let base = self.usage_baseline;
+        let payload = wire::usage_update_payload(
+            base.input_tokens
+                .saturating_add(self.turn_input_tokens.unwrap_or(0)),
+            base.output_tokens
+                .saturating_add(self.turn_output_tokens.unwrap_or(0)),
+            base.cached_input_tokens
+                .saturating_add(self.turn_cached_input_tokens.unwrap_or(0)),
+            base.total_state.merge_session(*self.turn_total_state),
+            self.effective_model,
+        );
+        wire::send(
+            self.wire,
+            wire::goose_session_update(self.session_id, payload),
+        )
+        .await;
+    }
+
     pub async fn run(&mut self, prompt: Vec<ContentBlock>) -> Result<StopReason, AgentError> {
         let user_text = prompt_to_text(prompt)?;
         if user_text.len() > MAX_PROMPT_BYTES {
@@ -299,6 +330,23 @@ impl RunCtx<'_> {
             // this gate rather than representing absent categories as zero.
             if response.input_tokens.is_some() || response.output_tokens.is_some() {
                 *self.turn_total_state = self.turn_total_state.fold(response.total_tokens);
+                // Report what the turn has burned SO FAR, before running the
+                // next round. A turn is many provider round-trips over many
+                // minutes, and until this point the only report was the one
+                // `session/prompt` sends after the turn returns — so a turn
+                // that was cancelled, timed out, or whose process was killed
+                // reported nothing at all, and its tokens (already billed)
+                // existed only in this stack frame. Reporting per round bounds
+                // the loss to the single request in flight.
+                //
+                // Emitting more than one `usage_update` per turn is expected by
+                // the consumer: buzz-acp's UsageTracker advances its committed
+                // baseline only when the turn's metric is published, so every
+                // notification within a turn measures from the same frozen
+                // baseline and the last one seen is the turn's true total.
+                // goose behaves the same way, which is why the tracker was
+                // written to tolerate it.
+                self.emit_usage_update().await;
             }
 
             if !response.reasoning.is_empty() {

crates/buzz-agent/src/lib.rs

@@ -658,6 +658,7 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
         effective_model_override,
         run_id,
         mut steer_rx,
+        usage_baseline,
     ) = match acquire_session(&app, &p.session_id).await {
         Ok(v) => v,
         Err(reason) => {
@@ -709,6 +710,7 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
         turn_output_tokens: &mut turn_output_tokens,
         turn_cached_input_tokens: &mut turn_cached_input_tokens,
         turn_total_state: &mut turn_total_state,
+        usage_baseline,
     };
     let result = ctx.run(p.prompt).await;
     if let Some(s) = app.sessions.lock().await.get_mut(&sid) {
@@ -766,28 +768,16 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
         if let Some((accumulated_in, accumulated_out, accumulated_cached, accumulated_total)) =
             accumulated
         {
-            // Build the usage_update payload. `accumulatedTotalTokens` is only
-            // included when the cumulative is exactly known — never when Unseen
-            // (no total ever observed) or Unknown (at least one turn lacked a
-            // total). A goose consumer that doesn't recognise the field ignores it.
-            let mut update = serde_json::json!({
-                "sessionUpdate": "usage_update",
-                // used: total tokens as a context-usage proxy;
-                // contextLimit: 0 (buzz-agent has no context limit tracking).
-                "used": accumulated_in.saturating_add(accumulated_out),
-                "contextLimit": 0u64,
-                "accumulatedInputTokens": accumulated_in,
-                "accumulatedOutputTokens": accumulated_out,
-                // A subset of accumulatedInputTokens, not an addition to
-                // it. Extends goose's usage_update shape; a consumer that
-                // does not know the field ignores it and prices exactly as
-                // it did before.
-                "accumulatedCachedInputTokens": accumulated_cached,
-                "model": effective_model_str,
-            });
-            if let crate::types::TurnTotalState::Exact(total) = accumulated_total {
-                update["accumulatedTotalTokens"] = serde_json::json!(total);
-            }
+            // Same builder the run loop uses for its per-round reports, so the
+            // final notification is shape-identical to the ones that preceded
+            // it and a consumer taking the high-water mark lands on this one.
+            let update = wire::usage_update_payload(
+                accumulated_in,
+                accumulated_out,
+                accumulated_cached,
+                accumulated_total,
+                effective_model_str,
+            );
             wire::send(&wire_tx, goose_session_update(&sid, update)).await;
         }
     }
@@ -821,6 +811,7 @@ async fn acquire_session(
         Option<String>,
         String,
         mpsc::UnboundedReceiver<Vec<ContentBlock>>,
+        crate::types::SessionUsageBaseline,
     ),
     &'static str,
 > {
@@ -857,6 +848,17 @@ async fn acquire_session(
         effective_model,
         run_id,
         steer_rx,
+        // Snapshot rather than a handle: the run loop reports cumulative usage
+        // after every LLM round, and taking the sessions lock on each of those
+        // would serialise concurrent sessions behind one another's provider
+        // round-trips. Nothing else advances these counters while this turn
+        // holds `busy`, so the snapshot cannot go stale under it.
+        crate::types::SessionUsageBaseline {
+            input_tokens: s.accumulated_input_tokens,
+            output_tokens: s.accumulated_output_tokens,
+            cached_input_tokens: s.accumulated_cached_input_tokens,
+            total_state: s.accumulated_total_state,
+        },
     ))
 }
 

crates/buzz-agent/src/types.rs

@@ -308,6 +308,30 @@ impl TurnTotalState {
     }
 }
 
+/// The session-cumulative usage counters as of the START of a turn.
+///
+/// Copied out of the session under the lock when a turn begins and handed to
+/// `RunCtx` by value, so the run loop can emit a cumulative `usage_update`
+/// after every LLM round without reaching back into `App.sessions` (which it
+/// holds no handle to, and which is locked by the turn's own bookkeeping at
+/// both ends).
+///
+/// This exists so that usage is durable *during* a turn rather than only after
+/// it. The counters a turn accrues live in the prompt task's stack frame until
+/// the turn returns; a process killed mid-turn takes them with it and the
+/// tokens are billed by the provider but recorded nowhere. That is not
+/// hypothetical — it silently under-reported a long-horizon benchmark's cost by
+/// several-fold, because every phase of a `continue_until_timeout` run is
+/// terminated mid-turn by design.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct SessionUsageBaseline {
+    pub input_tokens: u64,
+    pub output_tokens: u64,
+    /// The cache-served subset of `input_tokens`, not an addition to it.
+    pub cached_input_tokens: u64,
+    pub total_state: TurnTotalState,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub enum StopReason {
     EndTurn,

crates/buzz-agent/src/wire.rs

@@ -148,6 +148,48 @@ pub fn goose_session_update(sid: &str, update: Value) -> Value {
     })
 }
 
+/// Build the `usage_update` payload for a `_goose/unstable/session/update`.
+///
+/// Shared by the two places that report usage — after each LLM round inside a
+/// turn, and once more when the turn completes — so the wire shape cannot drift
+/// between them. A consumer takes the high-water mark per session, so the
+/// mid-turn payloads are supersets of each other and the final one wins; a
+/// divergence in field names or units between the two call sites would instead
+/// show up as tokens silently vanishing, which is the failure this reporting
+/// exists to prevent.
+///
+/// All counts are SESSION-cumulative, matching goose, so buzz-acp's
+/// `UsageTracker` can compute per-turn deltas symmetrically for both agents.
+pub fn usage_update_payload(
+    accumulated_input_tokens: u64,
+    accumulated_output_tokens: u64,
+    accumulated_cached_input_tokens: u64,
+    accumulated_total: crate::types::TurnTotalState,
+    model: &str,
+) -> Value {
+    let mut update = json!({
+        "sessionUpdate": "usage_update",
+        // used: total tokens as a context-usage proxy;
+        // contextLimit: 0 (buzz-agent has no context limit tracking).
+        "used": accumulated_input_tokens.saturating_add(accumulated_output_tokens),
+        "contextLimit": 0u64,
+        "accumulatedInputTokens": accumulated_input_tokens,
+        "accumulatedOutputTokens": accumulated_output_tokens,
+        // A subset of accumulatedInputTokens, not an addition to it. Extends
+        // goose's usage_update shape; a consumer that does not know the field
+        // ignores it and prices exactly as it did before.
+        "accumulatedCachedInputTokens": accumulated_cached_input_tokens,
+        "model": model,
+    });
+    // Only when the cumulative is exactly known — never when Unseen (no total
+    // ever observed) or Unknown (at least one turn lacked a total). A goose
+    // consumer that doesn't recognise the field ignores it.
+    if let Some(total) = accumulated_total.exact_value() {
+        update["accumulatedTotalTokens"] = json!(total);
+    }
+    update
+}
+
 /// A `session/update` notification carrying a `update._meta.goose.<key>` field.
 /// Used to advertise `activeRunId` (so steer-capable clients can target the
 /// in-flight run) and `queuedSteer` (so they can correlate an accepted steer