feat(acp): host-wide session pool with fleet slot cap, idle close and DM auto-publish

crates/buzz-acp/Cargo.toml

@@ -71,10 +71,11 @@ toml = "1.0"
 # Filter expressions
 evalexpr = { workspace = true }
 
-# Process-group kill (safe wrapper around killpg) — Unix-only; kill_process_group
-# has a #[cfg(not(unix))] fallback in acp.rs.
+# Process-group kill (safe wrapper around killpg) and flock(2) fleet slots —
+# Unix-only; kill_process_group has a #[cfg(not(unix))] fallback in acp.rs and
+# fleet.rs refuses to start without flock.
 [target.'cfg(unix)'.dependencies]
-nix = { version = "0.31", default-features = false, features = ["signal"] }
+nix = { version = "0.31", default-features = false, features = ["signal", "fs"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }

crates/buzz-acp/README.md

@@ -125,6 +125,11 @@ All configuration is via environment variables (or CLI flags — every env var h
 |------|---------|---------|-------------|
 | `--agents` | `BUZZ_ACP_AGENTS` | `1` | Number of agent subprocesses (1–32). |
 | `--lazy-pool` | `BUZZ_ACP_LAZY_POOL` | `false` | Connect, subscribe, and queue accepted work before starting ACP/LLM subprocesses. The first accepted event wakes one pool initialization task; failures retry with bounded exponential backoff while work remains. |
+| `--min-agents` | `BUZZ_ACP_MIN_AGENTS` | `--agents` | Agent subprocesses started eagerly. The rest of `--agents` are on-demand: spawned when every started agent is busy and channels are waiting, shut down again after `--session-idle-close` of holding no session. |
+| `--fleet-slots` | `BUZZ_ACP_FLEET_SLOTS` | `0` | Host-wide cap on live ACP sessions, shared by every harness using the same `--fleet-slot-dir`. One `flock`ed file per slot, so the kernel releases a dead harness's slots. Work that finds every slot held waits in the queue. `0` = no cap. |
+| `--fleet-slot-dir` | `BUZZ_ACP_FLEET_SLOT_DIR` | `$XDG_RUNTIME_DIR/buzz-acp/fleet` | Directory of fleet slot lock files (falls back to `<tmp>/buzz-acp-fleet`). |
+| `--session-idle-close` | `BUZZ_ACP_SESSION_IDLE_CLOSE` | `0` | Close a channel session — releasing its fleet slot — after this many seconds without a turn; the next event in that channel starts a fresh session. `0` = never. |
+| `--dm-autopublish` | `BUZZ_ACP_DM_AUTOPUBLISH` | `false` | In DM channels, when a turn ends without the agent having published a message, post the agent's trailing reply text as a top-level message so the human never gets silence. Skipped when the agent already posted in that turn; channels are unaffected. |
 | `--heartbeat-interval` | `BUZZ_ACP_HEARTBEAT_INTERVAL` | `0` | Seconds between heartbeat prompts. `0` = disabled. Must be `0` or ≥10 when enabled. |
 | `--heartbeat-prompt` | `BUZZ_ACP_HEARTBEAT_PROMPT` | (built-in) | Custom heartbeat prompt text. Conflicts with `--heartbeat-prompt-file`. |
 | `--heartbeat-prompt-file` | `BUZZ_ACP_HEARTBEAT_PROMPT_FILE` | — | Read heartbeat prompt from a file. Conflicts with `--heartbeat-prompt`. |
@@ -217,6 +222,8 @@ buzz-acp --agents 2 --heartbeat-interval 300 \
 
 ### Shared Identity
 
+**Session pool (many harnesses on one host):** each live ACP session is a worker process (for `claude-agent-acp`, a `claude` process plus its MCP servers), so memory scales with live sessions rather than harnesses. `--fleet-slots` caps live sessions host-wide the way a connection pool caps connections, `--session-idle-close` returns idle sessions to the pool, and `--min-agents` keeps per-harness subprocess count low until work actually arrives. A typical fleet of 25 harnesses on a 32 GB host runs `--agents 8 --min-agents 1 --fleet-slots 20 --session-idle-close 600`.
+
 All N agents authenticate as the **same Nostr bot identity** — users see one bot regardless of how many agents are running. The same channel is never processed by two agents simultaneously (the queue enforces this). Cross-channel message ordering is not guaranteed when N>1.
 
 ### Heartbeat Semantics

crates/buzz-acp/src/acp.rs

@@ -214,6 +214,12 @@ pub struct AcpClient {
     standard_usage: StandardUsageTracker,
     /// Known adapter identity for prompt-response usage mapping.
     standard_adapter: Option<StandardAdapterKind>,
+    /// Trailing assistant text of the in-flight turn — every
+    /// `agent_message_chunk` since the last `tool_call`. Consumed by
+    /// [`take_turn_reply`](Self::take_turn_reply) for DM auto-publish.
+    turn_reply_text: String,
+    /// Whether a `tool_call` in this turn carried `buzz messages send`.
+    turn_saw_buzz_send: bool,
 }
 
 /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
@@ -563,6 +569,8 @@ impl AcpClient {
             goose_usage: UsageTracker::default(),
             standard_usage: StandardUsageTracker::default(),
             standard_adapter,
+            turn_reply_text: String::new(),
+            turn_saw_buzz_send: false,
         })
     }
 
@@ -790,6 +798,8 @@ impl AcpClient {
         // misattributed to this turn.
         self.goose_usage.begin_turn(session_id);
         self.standard_usage.begin_turn(session_id);
+        self.turn_reply_text.clear();
+        self.turn_saw_buzz_send = false;
 
         self.last_prompt_id = Some(self.next_id);
         let id = self.next_id;
@@ -853,6 +863,19 @@ impl AcpClient {
         self.send_notification("session/cancel", params).await
     }
 
+    /// Send `session/close` and wait for the adapter to tear the session's
+    /// worker down. Bounded by [`CLOSE_TIMEOUT`](Self::CLOSE_TIMEOUT) rather
+    /// than the 60s request default because callers run this inline on an
+    /// idle agent from the main loop.
+    pub async fn session_close(&mut self, session_id: &str) -> Result<(), AcpError> {
+        let params = serde_json::json!({
+            "sessionId": session_id,
+        });
+        self.send_request_with_timeout("session/close", params, Self::CLOSE_TIMEOUT)
+            .await
+            .map(|_| ())
+    }
+
     /// Returns `true` if a `session/prompt` request is currently in flight.
     pub fn has_in_flight_prompt(&self) -> bool {
         self.last_prompt_id.is_some()
@@ -890,6 +913,14 @@ impl AcpClient {
         goose_usage.or(standard_usage)
     }
 
+    /// Consume the turn's trailing reply text and whether the agent ran
+    /// `buzz messages send`. Resets both. See `pool::maybe_autopublish_dm_reply`.
+    pub fn take_turn_reply(&mut self) -> (String, bool) {
+        let text = std::mem::take(&mut self.turn_reply_text);
+        let saw = std::mem::replace(&mut self.turn_saw_buzz_send, false);
+        (text.trim().to_string(), saw)
+    }
+
     /// Notify the usage tracker that buzz-acp just spawned a new session.
     ///
     /// Seeds a zero baseline so the first usage notification for `session_id`
@@ -1084,6 +1115,9 @@ impl AcpClient {
     /// Default timeout for non-prompt RPCs (initialize, session/new, etc.).
     const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 
+    /// Timeout for `session/close`, which only has to kill a worker process.
+    const CLOSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
     /// Send a JSON-RPC request and wait for the matching response.
     ///
     /// Assigns the next available id, writes the NDJSON line to stdin,
@@ -1097,6 +1131,17 @@ impl AcpClient {
         &mut self,
         method: &str,
         params: serde_json::Value,
+    ) -> Result<serde_json::Value, AcpError> {
+        self.send_request_with_timeout(method, params, Self::REQUEST_TIMEOUT)
+            .await
+    }
+
+    /// [`send_request`](Self::send_request) with an explicit per-phase timeout.
+    async fn send_request_with_timeout(
+        &mut self,
+        method: &str,
+        params: serde_json::Value,
+        timeout: std::time::Duration,
     ) -> Result<serde_json::Value, AcpError> {
         let id = self.next_id;
         self.next_id += 1;
@@ -1113,7 +1158,6 @@ impl AcpClient {
         // Wrap write + read in a single timeout so a hung agent can't block forever.
         // We cannot use an async block that borrows `self` mutably across two awaits
         // inside timeout(), so we sequence them with early-return on timeout.
-        let timeout = Self::REQUEST_TIMEOUT;
         match tokio::time::timeout(timeout, self.write_ndjson(&msg)).await {
             Ok(result) => result?,
             Err(_) => return Err(AcpError::Timeout(timeout)),
@@ -1756,6 +1800,7 @@ impl AcpClient {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
                     tracing::info!(target: "acp::stream", "{text}");
+                    self.turn_reply_text.push_str(text);
                 }
                 false
             }
@@ -1769,6 +1814,11 @@ impl AcpClient {
                     .and_then(|v| v.as_str())
                     .unwrap_or("unknown");
                 tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})");
+                // Only text AFTER the last tool call is the reply to the human.
+                self.turn_reply_text.clear();
+                if !self.turn_saw_buzz_send && update.to_string().contains("buzz messages send") {
+                    self.turn_saw_buzz_send = true;
+                }
                 true
             }
             "tool_call_update" => {
@@ -3125,6 +3175,48 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn session_close_sends_a_request_with_the_session_id() {
+        let capture = std::env::temp_dir().join(format!(
+            "buzz-acp-close-{}-{}",
+            std::process::id(),
+            uuid::Uuid::new_v4()
+        ));
+        let script = format!(
+            r#"IFS= read -r line; printf '%s\n' "$line" > '{}'; printf '{{"jsonrpc":"2.0","id":0,"result":{{}}}}\n'"#,
+            capture.display()
+        );
+        let mut client = spawn_script(&script).await;
+        client
+            .session_close("sess-42")
+            .await
+            .expect("empty result is a successful close");
+        let sent: serde_json::Value =
+            serde_json::from_str(&std::fs::read_to_string(&capture).expect("captured request"))
+                .expect("captured request is JSON");
+        assert_eq!(sent["method"], "session/close");
+        assert_eq!(sent["params"]["sessionId"], "sess-42");
+        assert!(
+            sent.get("id").is_some(),
+            "close is a request, not a notification"
+        );
+        let _ = std::fs::remove_file(capture);
+    }
+
+    #[tokio::test]
+    async fn session_close_surfaces_method_not_found_as_agent_error() {
+        let mut client = spawn_script(
+            r#"IFS= read -r line; printf '{"jsonrpc":"2.0","id":0,"error":{"code":-32601,"message":"Method not found"}}
+'"#,
+        )
+        .await;
+        let result = client.session_close("sess-1").await;
+        assert!(
+            matches!(result, Err(AcpError::AgentError { code: -32601, .. })),
+            "expected -32601 AgentError, got {result:?}"
+        );
+    }
+
     #[tokio::test]
     async fn idle_timeout_fires_on_silent_process() {
         let mut client = spawn_script("sleep 10").await;
@@ -3145,6 +3237,28 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn turn_reply_keeps_text_after_last_tool_call_and_flags_buzz_send() {
+        let mut client = spawn_script("cat").await;
+        let chunk = |t: &str| serde_json::json!({"params":{"update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":t}}}});
+        let tool = |cmd: &str| serde_json::json!({"params":{"update":{"sessionUpdate":"tool_call","title":"Bash","kind":"execute","rawInput":{"command":cmd}}}});
+        let _ = client.handle_session_update(&chunk("thinking out loud"));
+        let _ = client.handle_session_update(&tool("ls -la"));
+        let _ = client.handle_session_update(&chunk("final "));
+        let _ = client.handle_session_update(&chunk("answer\n"));
+        assert_eq!(
+            client.take_turn_reply(),
+            ("final answer".to_string(), false)
+        );
+        // take() resets both the text and the flag.
+        assert_eq!(client.take_turn_reply(), (String::new(), false));
+        let _ = client.handle_session_update(&tool(
+            "printf 'x' | buzz messages send --channel abc --content -",
+        ));
+        let _ = client.handle_session_update(&chunk("done"));
+        assert_eq!(client.take_turn_reply(), ("done".to_string(), true));
+    }
+
     #[tokio::test]
     async fn hard_timeout_fires_when_deadline_is_immediate() {
         let mut client = spawn_script("while true; do echo 'noise'; sleep 0.01; done").await;