From 57278133e6e5dc0114d42354ecfcffd6d1a5a227 Mon Sep 17 00:00:00 2001 From: gruming Date: Thu, 3 Sep 2026 14:50:59 +0900 Subject: [PATCH 1/2] feat(acp): host-wide session pool with fleet slot cap, idle close and DM auto-publish Memory on a multi-harness host scales with live ACP sessions (each is a `claude` process plus its MCP servers), not with harness count. This adds a connection-pool style cap and idle reclamation so 25 harnesses fit on one host: - `--fleet-slots N` / `--fleet-slot-dir`: host-wide cap on live sessions, shared by every harness via one flock(2)ed file per slot, so the kernel releases a dead harness's slots. Work that finds every slot held waits in the channel queue instead of failing. - `--session-idle-close S`: close a channel session (releasing its slot) after S seconds without a turn; the next event starts a fresh session. - `--min-agents M`: start only M subprocesses eagerly; the rest of `--agents` spawn on demand when every started agent is busy and shut down again after the idle bound. - `--dm-autopublish` (default on): in DM channels, when a turn ends without the agent having published, post its trailing reply text so the human never gets silence. Skipped when the agent already posted in that turn; channel turns are unaffected. All flags default off (autopublish on) so existing deployments are unchanged. README documents the flags and a 25-harness / 32 GB example. Co-Authored-By: Claude Fable 5.1 Signed-off-by: gruming --- crates/buzz-acp/Cargo.toml | 7 +- crates/buzz-acp/README.md | 7 + crates/buzz-acp/src/acp.rs | 116 +++++- crates/buzz-acp/src/config.rs | 115 +++++- crates/buzz-acp/src/fleet.rs | 241 ++++++++++++ crates/buzz-acp/src/lib.rs | 685 ++++++++++++++++++++++++++++++++-- crates/buzz-acp/src/pool.rs | 496 +++++++++++++++++++++++- crates/buzz-acp/src/queue.rs | 27 ++ 8 files changed, 1648 insertions(+), 46 deletions(-) create mode 100644 crates/buzz-acp/src/fleet.rs diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d047849806f..ac445f50bf7 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/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"] } diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 41d9a214bdd..1e3bdebab66 100644 --- a/crates/buzz-acp/README.md +++ b/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 `/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` | `true` | 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 diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 0d87bca028c..df1950eed07 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/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, + /// 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 { + 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 { 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; diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 5b7e27131ec..cdc291c3ea8 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -525,6 +525,39 @@ pub struct CliArgs { /// ignored (the watermark stays at startup time). #[arg(long, env = "BUZZ_ACP_REPLAY_FLOOR")] pub replay_floor: Option, + + /// Agent subprocesses started eagerly. The rest of `--agents` are + /// on-demand: spawned when every started agent is busy and channels are + /// waiting, and shut down again after `--session-idle-close` of holding + /// no session. Defaults to `--agents` (everything eager). + #[arg(long, env = "BUZZ_ACP_MIN_AGENTS", + value_parser = clap::value_parser!(u32).range(1..=32))] + pub min_agents: Option, + + /// Host-wide cap on live ACP sessions, shared by every harness using the + /// same `--fleet-slot-dir`. One slot per live session, held as a file + /// lock so the kernel releases it if the harness dies. Work that finds + /// every slot held waits in the queue. 0 = no cap. + #[arg(long, env = "BUZZ_ACP_FLEET_SLOTS", default_value_t = 0)] + pub fleet_slots: u32, + + /// Directory of fleet slot lock files. Defaults to + /// `$XDG_RUNTIME_DIR/buzz-acp/fleet`, else `/buzz-acp-fleet`. + #[arg(long, env = "BUZZ_ACP_FLEET_SLOT_DIR")] + pub fleet_slot_dir: Option, + + /// Close a channel session — releasing its fleet slot — after this many + /// seconds without a turn; on-demand agents holding no session are shut + /// down on the same bound. The next event in that channel starts a fresh + /// session. 0 = never. + #[arg(long, env = "BUZZ_ACP_SESSION_IDLE_CLOSE", default_value_t = 0)] + pub session_idle_close: u64, + + /// 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 channel + /// message so the human never gets silence. Channels are unaffected. + #[arg(long, env = "BUZZ_ACP_DM_AUTOPUBLISH", default_value_t = true, action = clap::ArgAction::Set)] + pub dm_autopublish: bool, } /// Merged NIP-01 subscription filter for a single channel. @@ -620,6 +653,16 @@ pub struct Config { /// triggering message. Clamped where consumed — see /// `startup_watermark_with_floor`. pub replay_floor_unix: Option, + /// Agent slots spawned eagerly (`<= agents`); the rest are on-demand. + pub min_agents: u32, + /// Host-wide live-session cap shared through `fleet_slot_dir`. 0 = off. + pub fleet_slots: u32, + /// Directory of the fleet slot lock files. + pub fleet_slot_dir: PathBuf, + /// Seconds a session may sit without a turn before it is closed. 0 = never. + pub session_idle_close_secs: u64, + /// Auto-post trailing reply text in DMs when the agent published nothing. + pub dm_autopublish: bool, /// Agent owner pubkey (hex). Used for `--respond-to=owner-only` gate. /// Replaces the old REST-based owner lookup. pub agent_owner: Option, @@ -1201,6 +1244,11 @@ impl Config { lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, replay_floor_unix: args.replay_floor, + min_agents: resolve_min_agents(args.min_agents, args.agents), + fleet_slots: args.fleet_slots, + fleet_slot_dir: args.fleet_slot_dir.unwrap_or_else(default_fleet_slot_dir), + session_idle_close_secs: args.session_idle_close, + dm_autopublish: args.dm_autopublish, agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()), no_base_prompt: args.no_base_prompt, base_prompt_content, @@ -1225,7 +1273,7 @@ impl Config { format!(" allowed_respond_to=[{}]", modes.join(",")) }; format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} session_policy={} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} session_policy={} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{} min_agents={} fleet_slots={} fleet_slot_dir={} session_idle_close={}s dm_autopublish={}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, @@ -1249,10 +1297,29 @@ impl Config { self.permission_mode, respond_to_detail, allowed_respond_to_detail, + self.min_agents, + self.fleet_slots, + self.fleet_slot_dir.display(), + self.session_idle_close_secs, + self.dm_autopublish, ) } } +/// `--min-agents` defaults to `--agents` and can never exceed it. +pub(crate) fn resolve_min_agents(min_agents: Option, agents: u32) -> u32 { + min_agents.unwrap_or(agents).clamp(1, agents.max(1)) +} + +/// `$XDG_RUNTIME_DIR/buzz-acp/fleet` when the runtime dir is set (per-user +/// tmpfs shared by every harness the user runs), else the system temp dir. +pub(crate) fn default_fleet_slot_dir() -> PathBuf { + match std::env::var_os("XDG_RUNTIME_DIR") { + Some(dir) if !dir.is_empty() => PathBuf::from(dir).join("buzz-acp").join("fleet"), + _ => std::env::temp_dir().join("buzz-acp-fleet"), + } +} + #[derive(Debug, serde::Deserialize)] struct TomlConfig { #[serde(default)] @@ -1577,6 +1644,11 @@ mod tests { lazy_pool: false, idle_pool_sleep_secs: 0, replay_floor_unix: None, + min_agents: 1, + fleet_slots: 0, + fleet_slot_dir: PathBuf::new(), + session_idle_close_secs: 0, + dm_autopublish: true, agent_owner: None, no_base_prompt: false, base_prompt_content: None, @@ -2313,6 +2385,47 @@ channels = "ALL" assert_eq!(configured.idle_pool_sleep, 300); } + #[test] + fn session_pool_flags_default_off_and_accept_cli_values() { + let key = "0".repeat(64); + let default = CliArgs::parse_from(["buzz-acp", "--private-key", &key]); + assert_eq!(default.min_agents, None); + assert_eq!(default.fleet_slots, 0); + assert_eq!(default.fleet_slot_dir, None); + assert_eq!(default.session_idle_close, 0); + + let configured = CliArgs::parse_from([ + "buzz-acp", + "--private-key", + &key, + "--agents", + "8", + "--min-agents", + "2", + "--fleet-slots", + "20", + "--fleet-slot-dir", + "/tmp/fleet", + "--session-idle-close", + "600", + ]); + assert_eq!(configured.min_agents, Some(2)); + assert_eq!(configured.fleet_slots, 20); + assert_eq!( + configured.fleet_slot_dir.as_deref(), + Some(std::path::Path::new("/tmp/fleet")) + ); + assert_eq!(configured.session_idle_close, 600); + } + + #[test] + fn min_agents_defaults_to_agents_and_is_capped_by_it() { + assert_eq!(resolve_min_agents(None, 4), 4); + assert_eq!(resolve_min_agents(Some(2), 8), 2); + assert_eq!(resolve_min_agents(Some(9), 8), 8); + assert_eq!(resolve_min_agents(Some(0), 3), 1); + } + #[test] fn lazy_pool_cli_flag_enables_deferred_startup() { let key = "0".repeat(64); diff --git a/crates/buzz-acp/src/fleet.rs b/crates/buzz-acp/src/fleet.rs new file mode 100644 index 00000000000..998f22bb3d4 --- /dev/null +++ b/crates/buzz-acp/src/fleet.rs @@ -0,0 +1,241 @@ +//! Fleet-wide session slots shared by every harness process on this host. +//! +//! One slot is one live ACP session — the adapter's per-channel worker (for +//! `claude-agent-acp`, a `claude` process plus its MCP servers), which is what +//! actually costs memory. Slots are `flock`ed lock files in a shared +//! directory: a harness holds a slot for as long as it keeps the lock file +//! open, and the kernel releases the lock the moment that process exits for +//! any reason, so a crashed harness can never strand a slot. +//! +//! This is the `maximumPoolSize` half of a connection pool. The per-harness +//! `--agents` bound and the `--session-idle-close` return path live in +//! `pool.rs` and the main loop. +//! +//! Unix-only (`flock(2)`); elsewhere `FleetPool::new` refuses to start. + +use std::path::{Path, PathBuf}; + +use tokio::time::Instant; + +/// One acquired fleet slot. Dropping it releases the slot. +pub struct FleetSlot { + index: u32, + #[cfg(unix)] + _lock: nix::fcntl::Flock, +} + +impl FleetSlot { + pub fn index(&self) -> u32 { + self.index + } +} + +/// Handle on the shared slot directory for one harness. +pub struct FleetPool { + dir: PathBuf, + slots: u32, + /// Set on the first failed acquire, cleared on the next success, so the + /// exhausted/acquired transitions are logged once each instead of on + /// every dispatch pass. + waiting_since: Option, +} + +impl FleetPool { + pub fn new(dir: impl Into, slots: u32) -> std::io::Result { + let dir = dir.into(); + if !cfg!(unix) { + return Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "fleet slots need flock(2); unavailable on this platform", + )); + } + std::fs::create_dir_all(&dir)?; + Ok(Self { + dir, + slots, + waiting_since: None, + }) + } + + pub fn slots(&self) -> u32 { + self.slots + } + + pub fn dir(&self) -> &Path { + &self.dir + } + + /// True while the last acquire attempt found every slot held. + pub fn is_waiting(&self) -> bool { + self.waiting_since.is_some() + } + + /// Claim the lowest free slot without blocking. `None` when every slot is + /// held (by this or another harness); the caller parks its work and tries + /// again on its next dispatch pass. + /// + /// A slot released while another thread is between `fork` and `exec` can + /// look held for a few microseconds (the child briefly holds a duplicate + /// of the descriptor). Callers already retry on their next pass, so this + /// is never worse than a short wait. + pub fn try_acquire(&mut self) -> Option { + for index in 0..self.slots { + let path = self.dir.join(format!("slot-{index:02}.lock")); + match try_lock_file(&path) { + Ok(Some(lock)) => { + if let Some(since) = self.waiting_since.take() { + tracing::info!( + target: "fleet", + slot = index, + waited_secs = since.elapsed().as_secs(), + "fleet slot acquired after wait" + ); + } + return Some(FleetSlot { + index, + #[cfg(unix)] + _lock: lock, + }); + } + Ok(None) => continue, + Err(error) => { + tracing::warn!( + target: "fleet", + path = %path.display(), + "fleet slot unusable: {error}" + ); + continue; + } + } + } + if self.waiting_since.is_none() { + self.waiting_since = Some(Instant::now()); + tracing::info!( + target: "fleet", + slots = self.slots, + dir = %self.dir.display(), + "fleet_exhausted — parking work until a slot is released" + ); + } + None + } +} + +/// `Ok(Some)` = locked, `Ok(None)` = held by someone else, `Err` = cannot +/// open or lock the file at all. +#[cfg(unix)] +fn try_lock_file(path: &Path) -> std::io::Result>> { + use nix::errno::Errno; + use nix::fcntl::{Flock, FlockArg}; + + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path)?; + match Flock::lock(file, FlockArg::LockExclusiveNonblock) { + Ok(lock) => Ok(Some(lock)), + Err((_, Errno::EWOULDBLOCK)) => Ok(None), + Err((_, errno)) => Err(std::io::Error::from(errno)), + } +} + +#[cfg(not(unix))] +fn try_lock_file(_path: &Path) -> std::io::Result> { + Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "flock(2) unavailable", + )) +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + + fn scratch_dir(name: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "buzz-acp-fleet-{name}-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )) + } + + /// Retry briefly: a sibling test mid-`spawn` can hold a duplicate of a + /// just-dropped lock descriptor until its `exec` (see `try_acquire`). + async fn acquire_soon(pool: &mut FleetPool) -> Option { + for _ in 0..50 { + if let Some(slot) = pool.try_acquire() { + return Some(slot); + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + None + } + + #[tokio::test] + async fn acquires_up_to_the_cap_and_reuses_released_slots() { + let dir = scratch_dir("cap"); + let mut pool = FleetPool::new(&dir, 2).expect("create slot dir"); + assert!(dir.is_dir()); + + let first = pool.try_acquire().expect("slot 0"); + let second = pool.try_acquire().expect("slot 1"); + assert_eq!((first.index(), second.index()), (0, 1)); + assert!(pool.try_acquire().is_none()); + assert!(pool.is_waiting()); + + drop(first); + let reused = acquire_soon(&mut pool).await.expect("slot 0 again"); + assert_eq!(reused.index(), 0); + assert!(!pool.is_waiting()); + drop(second); + drop(reused); + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn slots_are_shared_between_pools_on_the_same_dir() { + // Two pools on one directory stand in for two harness processes: the + // lock lives on the file, not in this pool's memory. + let dir = scratch_dir("shared"); + let mut one = FleetPool::new(&dir, 1).expect("create slot dir"); + let mut two = FleetPool::new(&dir, 1).expect("reuse slot dir"); + + let held = one.try_acquire().expect("first pool takes the slot"); + assert!(two.try_acquire().is_none()); + drop(held); + assert!(acquire_soon(&mut two).await.is_some()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn a_dead_holder_releases_its_slot_to_the_kernel() { + // A child process takes the lock and is killed; the slot must come + // back without any bookkeeping on our side. + let dir = scratch_dir("dead"); + let mut pool = FleetPool::new(&dir, 1).expect("create slot dir"); + let lock_path = dir.join("slot-00.lock"); + let mut child = tokio::process::Command::new("bash") + .arg("-c") + .arg(format!( + "exec 9>'{}'; flock 9; echo locked; sleep 30 9>&-", + lock_path.display() + )) + .stdout(std::process::Stdio::piped()) + .spawn() + .expect("spawn locking child"); + let mut stdout = child.stdout.take().expect("child stdout"); + let mut buf = [0u8; 8]; + let _ = tokio::io::AsyncReadExt::read(&mut stdout, &mut buf).await; + + assert!(pool.try_acquire().is_none(), "child holds the only slot"); + child.kill().await.expect("kill child"); + let _ = child.wait().await; + assert!( + acquire_soon(&mut pool).await.is_some(), + "slot released on child exit" + ); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index d72d4cf482a..a7c3113c31a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4,6 +4,7 @@ mod acp; mod config; mod engram_fetch; mod filter; +mod fleet; mod observer; mod pool; mod pool_lifecycle; @@ -37,6 +38,7 @@ use config::{ MultipleEventHandling, RespondTo, SubscribeMode, }; use filter::SubscriptionRule; +use fleet::FleetPool; use futures_util::FutureExt; use nostr::{PublicKey, ToBech32}; use pool::{ @@ -2789,6 +2791,7 @@ async fn tokio_main() -> Result<()> { rest_client: relay.rest_client(), channel_info: pool::ChannelInfoResolver::new(channel_info_map, relay.rest_client()), context_message_limit: config.context_message_limit, + dm_autopublish: config.dm_autopublish, max_turns_per_session: config.max_turns_per_session, permission_mode: config.permission_mode, agent_keys: config.keys.clone(), @@ -2967,6 +2970,40 @@ async fn tokio_main() -> Result<()> { }) .collect(); + // Host-wide live-session cap (`--fleet-slots`). Acquired by the dispatcher + // right before a turn that will create a session; released when that + // session is closed, or by the kernel when this process dies. + let mut fleet: Option = if config.fleet_slots > 0 { + let pool = + FleetPool::new(&config.fleet_slot_dir, config.fleet_slots).with_context(|| { + format!( + "cannot prepare fleet slot dir {}", + config.fleet_slot_dir.display() + ) + })?; + tracing::info!( + slots = pool.slots(), + dir = %pool.dir().display(), + "fleet session cap enabled" + ); + Some(pool) + } else { + None + }; + + // Session-level idle housekeeping (see `reap_idle_sessions`). Always + // ticks so sessions invalidated on a live adapter get their + // `session/close`; the idle bound itself is opt-in. + let session_idle_close_bound = Duration::from_secs(config.session_idle_close_secs); + let mut session_reaper = { + let interval = if session_idle_close_bound.is_zero() { + Duration::from_secs(30) + } else { + session_idle_close_bound.min(Duration::from_secs(30)) + }; + tokio::time::interval_at(tokio::time::Instant::now() + interval, interval) + }; + // // Branches 1 & 2 both need to borrow `pool`, but they access different // fields (result_rx vs join_set). We use `rx_and_join_set()` to split the @@ -3015,33 +3052,55 @@ async fn tokio_main() -> Result<()> { } } + // On-demand agents: when every started agent is busy and channels are + // waiting, fill surplus slots now rather than on the next 30s + // maintenance tick. Baseline slots keep their maintenance cadence. + if pool_ready && config.min_agents < config.agents { + let wanted = on_demand_fill( + &pool, + &queue, + &crash_history, + fleet.as_ref().is_some_and(FleetPool::is_waiting), + ); + if wanted > 0 { + refill_slots( + &pool, + &mut crash_history, + &config, + &respawn_tx, + &mut respawn_tasks, + observer.clone(), + false, + wanted, + ); + } + } + if pool_ready && last_maintenance.elapsed() >= maintenance_interval { last_maintenance = std::time::Instant::now(); queue.compact_expired_state(); // Slot refill: spawn background tasks for empty slots whose // circuit breaker allows it. spawn_and_init runs off the main - // loop so it never blocks event processing. - for (idx, slot) in crash_history.iter_mut().enumerate() { - if pool.slot_alive(idx) || slot.respawn_in_flight { - continue; - } - if !slot.can_refill() { - continue; - } - slot.respawn_in_flight = true; - tracing::info!(agent = idx, "slot refill: spawning background respawn"); - let cmd = config.agent_command.clone(); - let args = config.agent_args.clone(); - let env = config.persona_env_vars.clone(); - let has_codex = config.has_generated_codex_config; - let observer = observer.clone(); - let guard = RespawnGuard::new(idx, respawn_tx.clone()); - respawn_tasks.spawn(async move { - let result = spawn_and_init(&cmd, &args, &env, has_codex, idx, observer).await; - guard.send(result); - }); - } + // loop so it never blocks event processing. Slots at or above + // `min_agents` are on-demand and refill only while queued + // channels outnumber idle agents. + let wanted = on_demand_fill( + &pool, + &queue, + &crash_history, + fleet.as_ref().is_some_and(FleetPool::is_waiting), + ); + refill_slots( + &pool, + &mut crash_history, + &config, + &respawn_tx, + &mut respawn_tasks, + observer.clone(), + true, + wanted, + ); // Flush requeued batches whose retry_after has expired. Without // this, a batch requeued during crash recovery can sit idle @@ -3050,7 +3109,7 @@ async fn tokio_main() -> Result<()> { // arrive when the channel is silent. if queue.has_flushable_work() { for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3102,7 +3161,7 @@ async fn tokio_main() -> Result<()> { // next relay event arrives — which can be minutes on quiet channels. if respawn_collected { for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3541,7 +3600,7 @@ async fn tokio_main() -> Result<()> { ); if pool_ready { for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3579,6 +3638,26 @@ async fn tokio_main() -> Result<()> { } None } + _ = session_reaper.tick() => { + let _ = result_rx; // end split borrow before touching pool + if pool_ready { + let summary = reap_idle_sessions( + &mut pool, + tokio::time::Instant::now(), + session_idle_close_bound, + config.min_agents as usize, + ) + .await; + if summary.sessions_closed > 0 || summary.agents_released > 0 { + tracing::info!( + sessions_closed = summary.sessions_closed, + agents_released = summary.agents_released, + "session_reaper" + ); + } + } + None + } _ = async { match idle_pool_sleep_reaper.as_mut() { Some(timer) => timer.tick().await, @@ -3641,12 +3720,12 @@ async fn tokio_main() -> Result<()> { } else if queue.has_flushable_work() { tracing::debug!("heartbeat_skipped_events"); for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } } else if pool.any_idle() { - dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); + dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight, &mut fleet); } else { tracing::debug!("heartbeat_skipped_busy"); } @@ -3743,7 +3822,7 @@ async fn tokio_main() -> Result<()> { break; } for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3768,7 +3847,7 @@ async fn tokio_main() -> Result<()> { break; } for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3922,7 +4001,7 @@ async fn tokio_main() -> Result<()> { // queue drains. We still try here in case the in-flight // task has already returned. for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) + dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet) { typing_channels.insert(scope, thread_tags); } @@ -3949,9 +4028,13 @@ async fn tokio_main() -> Result<()> { "ready", None, ); - for (scope, thread_tags) in - dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity) - { + for (scope, thread_tags) in dispatch_pending( + &mut pool, + &mut queue, + &ctx, + &mut last_activity, + &mut fleet, + ) { typing_channels.insert(scope, thread_tags); } } @@ -4340,6 +4423,7 @@ fn dispatch_pending( queue: &mut EventQueue, ctx: &Arc, last_activity: &mut tokio::time::Instant, + fleet: &mut Option, ) -> Vec<(scope::SessionScope, ThreadTags)> { // Keyed by the exact session scope, not the channel: two threads dispatching // concurrently in one channel get distinct typing entries so completing one @@ -4351,6 +4435,11 @@ fn dispatch_pending( // releasing requeues them so the next dispatch (when the owner returns) // reuses that exact session instead of forking a duplicate. let mut held: Vec = Vec::new(); + // Batches parked because the fleet has no free session slot. They stay + // in-flight for the length of this pass so `flush_next` moves on to scopes + // that already hold a session, then go back to the queue front untouched + // (no retry accounting — nothing failed). + let mut fleet_parked: Vec = Vec::new(); loop { let batch = match queue.flush_next() { Some(b) => b, @@ -4379,6 +4468,19 @@ fn dispatch_pending( // thread's provider session so a temporarily busy worker cannot cause // another to open a duplicate session for the same thread. let affinity_hit = pool.has_session_for(&scope); + // A turn without a session on any idle agent will create one, which + // needs a fleet slot. Only reach for the fleet when an agent is + // actually free, so pool exhaustion never churns a slot. + let fleet_slot = match fleet.as_mut() { + Some(fleet) if !affinity_hit && pool.any_idle() => match fleet.try_acquire() { + Some(slot) => Some(slot), + None => { + fleet_parked.push(batch); + continue; + } + }, + _ => None, + }; let mut agent = match pool.try_claim(Some(&scope)) { Some(a) => a, None => { @@ -4389,7 +4491,15 @@ fn dispatch_pending( break; } }; - tracing::debug!(agent = agent.index, channel = %channel_id, scope = %scope.telemetry_label(), affinity_hit, "agent_claimed"); + tracing::debug!( + agent = agent.index, + channel = %channel_id, + scope = %scope.telemetry_label(), + affinity_hit, + fleet_slot = fleet_slot.as_ref().map(|slot| slot.index()), + "agent_claimed" + ); + agent.state.pending_slot = fleet_slot; let recoverable_batch = match ctx.dedup_mode { DedupMode::Queue => Some(batch.clone()), @@ -4460,8 +4570,13 @@ fn dispatch_pending( queue.requeue_preserve_timestamps(batch); queue.mark_complete(scope); } + let parked = fleet_parked.len(); + for batch in fleet_parked { + park_batch(queue, batch); + } tracing::debug!( dispatched = dispatched_channels.len(), + fleet_parked = parked, queue_depth = queue.pending_channels(), "dispatch_pending" ); @@ -4545,6 +4660,13 @@ fn handle_prompt_result( pool.task_map_mut() .retain(|_, meta| meta.agent_index != agent_index); debug_assert_eq!(before, pool.task_map().len() + 1); + // Stamp the session's idle clock at return time. The reaper only ever + // sees idle agents, so this is exact; a session the task already + // invalidated is simply not stamped. + result + .agent + .state + .touch(&result.source, tokio::time::Instant::now()); if let PromptSource::Channel(scope) = &result.source { // The task may have invalidated this session before returning. Never // resurrect delivery state for a dead session; its replacement must @@ -5074,14 +5196,35 @@ fn dispatch_heartbeat( pool: &mut AgentPool, ctx: &Arc, heartbeat_in_flight: &mut bool, + fleet: &mut Option, ) { if *heartbeat_in_flight { return; } - let agent = match pool.try_claim(None) { + let mut agent = match pool.try_claim(None) { Some(a) => a, None => return, }; + // A heartbeat session counts against the fleet like any other. Heartbeats + // are droppable by design, so an exhausted fleet just skips this tick. + if agent.state.heartbeat_session.is_none() { + if let Some(fleet) = fleet.as_mut() { + match fleet.try_acquire() { + Some(slot) => agent.state.pending_slot = Some(slot), + None => { + tracing::debug!(agent = agent.index, "heartbeat skipped — fleet exhausted"); + // Not a turn: keep the surplus-idle clock where it was. + let idle_since = agent.state.idle_since; + let index = agent.index; + pool.return_agent(agent); + if let Some(agent) = pool.agents_mut()[index].as_mut() { + agent.state.idle_since = idle_since; + } + return; + } + } + } + } let prompt_text = ctx .heartbeat_prompt @@ -5280,6 +5423,188 @@ fn normalized_agent_name(init_result: &serde_json::Value) -> String { .to_ascii_lowercase() } +/// Put a flushed batch back exactly as it was: live events to the queue front +/// with their timestamps, cancelled events back to the cancelled store (which +/// `requeue_preserve_timestamps` alone would drop), and the channel released +/// from in-flight with no retry accounting. +fn park_batch(queue: &mut EventQueue, mut batch: FlushBatch) { + let channel_id = batch.channel_id; + let scope = batch.scope.clone(); + let cancelled_events = std::mem::take(&mut batch.cancelled_events); + let cancel_reason = batch.cancel_reason; + queue.requeue_preserve_timestamps(batch); + if !cancelled_events.is_empty() { + queue.requeue_as_cancelled( + FlushBatch { + channel_id, + scope: scope.clone(), + events: Vec::new(), + cancelled_events, + cancel_reason, + }, + cancel_reason.unwrap_or(CancelReason::Steer), + ); + } + queue.mark_complete(&scope); +} + +/// How many on-demand slots to fill right now: waiting channels beyond what +/// in-flight (re)spawns will absorb, and only while agents are the bottleneck. +fn on_demand_fill( + pool: &AgentPool, + queue: &EventQueue, + crash_history: &[SlotCircuit], + fleet_waiting: bool, +) -> usize { + // An idle agent means the queue is not agent-bound; an exhausted fleet + // means no new session could start on a fresh agent anyway. + if pool.any_idle() || fleet_waiting { + return 0; + } + let arriving = crash_history + .iter() + .filter(|slot| slot.respawn_in_flight) + .count(); + queue.undispatched_scopes().saturating_sub(arriving) +} + +/// Spawn background (re)spawns for empty slots. Baseline slots +/// (`< min_agents`) refill whenever `include_baseline` and their circuit +/// allows; on-demand slots refill only while `on_demand` remains. +#[allow(clippy::too_many_arguments)] +fn refill_slots( + pool: &AgentPool, + crash_history: &mut [SlotCircuit], + config: &Config, + respawn_tx: &mpsc::Sender, + respawn_tasks: &mut tokio::task::JoinSet<()>, + observer: Option, + include_baseline: bool, + mut on_demand: usize, +) { + let min_agents = config.min_agents as usize; + for (idx, slot) in crash_history.iter_mut().enumerate() { + if pool.slot_alive(idx) || slot.respawn_in_flight { + continue; + } + let baseline = idx < min_agents; + if baseline { + if !include_baseline { + continue; + } + } else if on_demand == 0 { + continue; + } + if !slot.can_refill() { + continue; + } + if !baseline { + on_demand -= 1; + } + slot.respawn_in_flight = true; + tracing::info!( + agent = idx, + on_demand = !baseline, + "slot refill: spawning background respawn" + ); + let cmd = config.agent_command.clone(); + let args = config.agent_args.clone(); + let env = config.persona_env_vars.clone(); + let has_codex = config.has_generated_codex_config; + let observer = observer.clone(); + let guard = RespawnGuard::new(idx, respawn_tx.clone()); + respawn_tasks.spawn(async move { + let result = spawn_and_init(&cmd, &args, &env, has_codex, idx, observer).await; + guard.send(result); + }); + } +} + +#[derive(Debug, Default, PartialEq, Eq)] +struct ReapSummary { + sessions_closed: usize, + agents_released: usize, +} + +/// Idle housekeeping over the agents sitting in the pool (never checked-out +/// ones): flush queued `session/close`es, close channel/heartbeat sessions +/// that have not run a turn for `bound`, and shut down on-demand agents +/// (slot index `>= min_agents`) left holding no session for `bound`. +/// `bound == 0` keeps only the flush. +async fn reap_idle_sessions( + pool: &mut AgentPool, + now: tokio::time::Instant, + bound: Duration, + min_agents: usize, +) -> ReapSummary { + let mut summary = ReapSummary::default(); + for slot in pool.agents_mut().iter_mut() { + let Some(agent) = slot.as_mut() else { + continue; + }; + if !bound.is_zero() { + let due: Vec = agent + .state + .sessions + .keys() + .filter(|scope| { + agent + .state + .last_used + .get(scope) + .is_some_and(|last| now.duration_since(*last) >= bound) + }) + .cloned() + .collect(); + for scope in &due { + tracing::info!( + target: "pool::session", + agent = agent.index, + channel = %scope.channel_id(), + scope = %scope.telemetry_label(), + idle_bound_secs = bound.as_secs(), + "session idle — closing" + ); + agent.state.invalidate_scope(scope); + } + summary.sessions_closed += due.len(); + if agent + .state + .heartbeat_last_used + .is_some_and(|last| now.duration_since(last) >= bound) + { + tracing::info!( + target: "pool::session", + agent = agent.index, + idle_bound_secs = bound.as_secs(), + "heartbeat session idle — closing" + ); + agent.state.invalidate(&PromptSource::Heartbeat); + summary.sessions_closed += 1; + } + } + agent.flush_pending_close().await; + + let surplus_idle = !bound.is_zero() + && agent.index >= min_agents + && agent.state.sessions.is_empty() + && agent.state.heartbeat_session.is_none() + && agent.state.pending_close.is_empty() + && agent + .state + .idle_since + .is_some_and(|since| now.duration_since(since) >= bound); + if surplus_idle { + if let Some(mut agent) = slot.take() { + tracing::info!(agent = agent.index, "idle on-demand agent released"); + agent.acp.shutdown().await; + summary.agents_released += 1; + } + } + } + summary +} + async fn shutdown_agent_slots(slots: &mut [Option]) { for slot in slots { if let Some(mut agent) = slot.take() { @@ -5302,6 +5627,7 @@ async fn shutdown_agent_pool(pool: &mut AgentPool) { struct PoolStartup { agents: u32, + min_agents: u32, command: String, args: Vec, extra_env: Vec<(String, String)>, @@ -5315,6 +5641,7 @@ impl PoolStartup { fn from_config(config: &Config, observer: Option) -> Self { Self { agents: config.agents, + min_agents: config.min_agents, command: config.agent_command.clone(), args: config.agent_args.clone(), extra_env: config.persona_env_vars.clone(), @@ -5334,6 +5661,12 @@ async fn initialize_agent_pool( // Attempt each spawn under a 60-second timeout; a partial pool is valid. let mut agent_slots: Vec> = Vec::with_capacity(startup.agents as usize); for i in 0..startup.agents as usize { + if i >= startup.min_agents as usize { + // On-demand slot: filled by `refill_slots` when waiting channels + // outgrow the started agents, emptied again once idle. + agent_slots.push(None); + continue; + } let spawn_result = AcpClient::spawn( &startup.command, &startup.args, @@ -5421,14 +5754,18 @@ async fn initialize_agent_pool( startup.agents )); } - if live_count < startup.agents as usize { + if live_count < startup.min_agents as usize { tracing::warn!( "started {}/{} agents — continuing with reduced pool", live_count, - startup.agents + startup.min_agents ); } - tracing::info!("agent_pool_ready agents={}", live_count); + tracing::info!( + "agent_pool_ready agents={} max_agents={}", + live_count, + startup.agents + ); Ok(AgentPool::from_slots(agent_slots)) } @@ -8934,6 +9271,11 @@ mod build_mcp_servers_tests { lazy_pool: false, idle_pool_sleep_secs: 0, replay_floor_unix: None, + min_agents: 1, + fleet_slots: 0, + fleet_slot_dir: std::path::PathBuf::new(), + session_idle_close_secs: 0, + dm_autopublish: true, agent_owner: None, no_base_prompt: false, base_prompt_content: None, @@ -9160,6 +9502,11 @@ mod error_outcome_emission_tests { lazy_pool: false, idle_pool_sleep_secs: 0, replay_floor_unix: None, + min_agents: 1, + fleet_slots: 0, + fleet_slot_dir: std::path::PathBuf::new(), + session_idle_close_secs: 0, + dm_autopublish: true, agent_owner: None, no_base_prompt: false, base_prompt_content: None, @@ -9206,6 +9553,270 @@ mod error_outcome_emission_tests { } } + mod session_pool_tests { + use super::*; + use crate::{on_demand_fill, reap_idle_sessions, refill_slots, ReapSummary}; + + fn capture_path(name: &str) -> std::path::PathBuf { + std::env::temp_dir().join(format!( + "buzz-acp-reaper-{name}-{}-{}", + std::process::id(), + Uuid::new_v4() + )) + } + + /// Fake adapter: appends every request line to `capture` and answers + /// each one with an empty result, so `session/close` succeeds. + async fn recording_agent(index: usize, capture: &std::path::Path) -> OwnedAgent { + let script = format!( + r#"n=0; while IFS= read -r line; do printf '%s\n' "$line" >> '{}'; printf '{{"jsonrpc":"2.0","id":%d,"result":{{}}}}\n' "$n"; n=$((n+1)); done"#, + capture.display() + ); + let mut agent = dummy_agent(index).await; + agent.acp = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .expect("spawn recording adapter"); + agent + } + + fn closed_session_ids(capture: &std::path::Path) -> Vec { + std::fs::read_to_string(capture) + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter(|req| req["method"] == "session/close") + .filter_map(|req| req["params"]["sessionId"].as_str().map(str::to_string)) + .collect() + } + + #[tokio::test] + async fn idle_sessions_are_closed_and_recent_ones_kept() { + let capture = capture_path("idle"); + let mut agent = recording_agent(0, &capture).await; + let base = tokio::time::Instant::now(); + let stale = scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let fresh = scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + agent + .state + .sessions + .insert(stale.clone(), "sess-stale".into()); + agent.state.last_used.insert(stale.clone(), base); + agent + .state + .sessions + .insert(fresh.clone(), "sess-fresh".into()); + agent + .state + .last_used + .insert(fresh.clone(), base + Duration::from_secs(500)); + let mut pool = AgentPool::from_slots(vec![Some(agent)]); + + let summary = reap_idle_sessions( + &mut pool, + base + Duration::from_secs(600), + Duration::from_secs(600), + 1, + ) + .await; + + assert_eq!( + summary, + ReapSummary { + sessions_closed: 1, + agents_released: 0 + } + ); + let agent = pool.agents_mut()[0].as_ref().expect("agent kept"); + assert!(!agent.state.sessions.contains_key(&stale)); + assert!(agent.state.sessions.contains_key(&fresh)); + assert!(agent.state.pending_close.is_empty(), "close was flushed"); + assert_eq!(closed_session_ids(&capture), vec!["sess-stale".to_string()]); + let _ = std::fs::remove_file(capture); + } + + #[tokio::test] + async fn invalidated_sessions_get_session_close_even_without_an_idle_bound() { + let capture = capture_path("rotate"); + let mut agent = recording_agent(0, &capture).await; + let cid = Uuid::new_v4(); + let scope = scope::SessionScope::Conversation { channel_id: cid }; + agent + .state + .sessions + .insert(scope.clone(), "sess-rotated".into()); + assert_eq!(agent.state.invalidate_channel(&cid), 1); + let mut pool = AgentPool::from_slots(vec![Some(agent)]); + + let summary = + reap_idle_sessions(&mut pool, tokio::time::Instant::now(), Duration::ZERO, 1).await; + + assert_eq!(summary, ReapSummary::default()); + let agent = pool.agents_mut()[0] + .as_ref() + .expect("bound 0 never releases"); + assert!(agent.state.pending_close.is_empty()); + assert_eq!( + closed_session_ids(&capture), + vec!["sess-rotated".to_string()] + ); + let _ = std::fs::remove_file(capture); + } + + #[tokio::test] + async fn surplus_idle_agent_is_released_but_baseline_and_busy_ones_are_kept() { + let base = tokio::time::Instant::now(); + let mut baseline = dummy_agent(0).await; + baseline.state.idle_since = Some(base); + let mut surplus_idle = dummy_agent(1).await; + surplus_idle.state.idle_since = Some(base); + let mut surplus_with_session = dummy_agent(2).await; + surplus_with_session.state.idle_since = Some(base); + let live = scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + surplus_with_session + .state + .sessions + .insert(live.clone(), "sess-live".into()); + surplus_with_session + .state + .last_used + .insert(live.clone(), base + Duration::from_secs(300)); + let mut pool = AgentPool::from_slots(vec![ + Some(baseline), + Some(surplus_idle), + Some(surplus_with_session), + ]); + + let summary = reap_idle_sessions( + &mut pool, + base + Duration::from_secs(600), + Duration::from_secs(600), + 1, + ) + .await; + + assert_eq!( + summary, + ReapSummary { + sessions_closed: 0, + agents_released: 1 + } + ); + let slots = pool.agents_mut(); + assert!(slots[0].is_some(), "baseline slot is never released"); + assert!(slots[1].is_none(), "idle on-demand agent released"); + assert!( + slots[2].is_some(), + "on-demand agent with a live session kept" + ); + } + + #[tokio::test] + async fn on_demand_fill_counts_waiting_channels_beyond_arriving_agents() { + let mut queue = EventQueue::new(config::DedupMode::Queue); + for _ in 0..3 { + let channel_id = Uuid::new_v4(); + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "x") + .sign_with_keys(&keys) + .unwrap(); + queue.push(QueuedEvent { + channel_id, + scope: scope::SessionScope::Conversation { channel_id }, + event, + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + } + let circuit = |in_flight: bool| SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: in_flight, + }; + + let empty = AgentPool::from_slots(vec![None, None, None]); + assert_eq!( + on_demand_fill( + &empty, + &queue, + &[circuit(false), circuit(true), circuit(false)], + false + ), + 2, + "three waiting channels minus one agent already on its way" + ); + assert_eq!( + on_demand_fill(&empty, &queue, &[], true), + 0, + "an exhausted fleet means a fresh agent could not start a session anyway" + ); + + let with_idle = AgentPool::from_slots(vec![Some(dummy_agent(0).await), None]); + assert_eq!( + on_demand_fill(&with_idle, &queue, &[circuit(false), circuit(false)], false), + 0, + "an idle agent means the queue is not agent-bound" + ); + } + + #[tokio::test] + async fn refill_spawns_on_demand_slots_only_while_demand_remains() { + let mut config = test_config(); + config.agents = 3; + config.min_agents = 1; + let pool = AgentPool::from_slots(vec![None, None, None]); + let mut crash_history: Vec = (0..3) + .map(|_| SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }) + .collect(); + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + + refill_slots( + &pool, + &mut crash_history, + &config, + &respawn_tx, + &mut respawn_tasks, + None, + false, + 1, + ); + let in_flight: Vec = crash_history.iter().map(|s| s.respawn_in_flight).collect(); + assert_eq!( + in_flight, + vec![false, true, false], + "one unit of demand fills exactly one on-demand slot; baseline untouched" + ); + + refill_slots( + &pool, + &mut crash_history, + &config, + &respawn_tx, + &mut respawn_tasks, + None, + true, + 0, + ); + let in_flight: Vec = crash_history.iter().map(|s| s.respawn_in_flight).collect(); + assert_eq!( + in_flight, + vec![true, true, false], + "maintenance refills the baseline slot, no demand → slot 2 stays empty" + ); + respawn_tasks.abort_all(); + } + } + #[tokio::test] async fn successful_native_steer_is_transferred_to_live_session_delivery_state() { let channel_id = Uuid::new_v4(); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index c73bd56031f..6cf00af54f5 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -35,6 +35,7 @@ use crate::acp::{ ModelSwitchMethod, StopReason, SystemPromptTransport, }; use crate::config::{compose_scoped_session_title, DedupMode, PermissionMode}; +use crate::fleet::FleetSlot; use crate::observer; use crate::prompt_project::{pick_authoritative_project_home, PromptProjectInfo}; use crate::queue::{ @@ -141,6 +142,27 @@ pub struct SessionState { /// Per-scope successful-delivery state. Created with the ACP session and /// cleared atomically with every invalidation path. pub deliveries: HashMap, + /// session scope → fleet slot held for the live session (`--fleet-slots`). + /// Absent when the fleet cap is off. + pub session_slots: HashMap, + /// Fleet slot held for the heartbeat session. + pub heartbeat_slot: Option, + /// Fleet slot the dispatcher acquired for a session this turn is about + /// to create. Moved into `session_slots` / `heartbeat_slot` on creation; + /// released by `AgentPool::return_agent` if the turn never created one. + pub pending_slot: Option, + /// Sessions invalidated while their adapter stayed alive, still awaiting + /// `session/close`. The fleet slot rides along so it is released only + /// once the adapter has actually torn the session's worker down. + pub pending_close: Vec<(String, Option)>, + /// session scope → when the session last finished a turn (idle clock for + /// `--session-idle-close`). + pub last_used: HashMap, + /// Idle clock for the heartbeat session. + pub heartbeat_last_used: Option, + /// When this agent was last returned to the pool. Drives the release of + /// surplus on-demand adapters that hold no session. + pub idle_since: Option, } impl SessionState { @@ -151,26 +173,51 @@ impl SessionState { self.invalidate_scope(scope); } PromptSource::Heartbeat => { - self.heartbeat_session = None; + if let Some(session_id) = self.heartbeat_session.take() { + let slot = self.heartbeat_slot.take(); + self.pending_close.push((session_id, slot)); + } self.heartbeat_turn_count = 0; self.heartbeat_standing_context_sent = false; + self.heartbeat_last_used = None; } } } /// Invalidate a single session scope's session and turn counter. /// Returns `true` if the scope had an active session. + /// + /// The session is queued for `session/close` (see + /// `OwnedAgent::flush_pending_close`) so its worker — and fleet slot — are + /// reclaimed while the adapter lives on. pub fn invalidate_scope(&mut self, scope: &SessionScope) -> bool { + match self.forget_scope(scope) { + Some((session_id, slot)) => { + self.pending_close.push((session_id, slot)); + true + } + None => false, + } + } + + /// Drop every record of `scope`'s session without scheduling a + /// `session/close`. Returns the session id and its fleet slot so the + /// caller can close the session itself and release the slot afterwards. + pub fn forget_scope(&mut self, scope: &SessionScope) -> Option<(String, Option)> { self.turn_counts.remove(scope); self.core_sections.remove(scope); self.canvas_sections.remove(scope); self.deliveries.remove(scope); - self.sessions.remove(scope).is_some() + self.last_used.remove(scope); + let session_id = self.sessions.remove(scope)?; + Some((session_id, self.session_slots.remove(scope))) } /// Invalidate every session scope belonging to `channel_id` (channel-wide /// cleanup, e.g. when the agent is removed from a channel). Returns the - /// number of scopes that had an active session. + /// number of scopes that had an active session. Each closed session is + /// queued for `session/close` via `invalidate_scope`, so its worker and + /// fleet slot are reclaimed once `flush_pending_close` runs. pub fn invalidate_channel(&mut self, channel_id: &Uuid) -> usize { let scopes: Vec = self .sessions @@ -193,6 +240,19 @@ impl SessionState { count } + /// Record that `source`'s live session just finished a turn. + pub fn touch(&mut self, source: &PromptSource, now: tokio::time::Instant) { + match source { + PromptSource::Channel(scope) if self.sessions.contains_key(scope) => { + self.last_used.insert(scope.clone(), now); + } + PromptSource::Heartbeat if self.heartbeat_session.is_some() => { + self.heartbeat_last_used = Some(now); + } + _ => {} + } + } + /// Invalidate all sessions and turn counters (e.g. after agent exit). pub fn invalidate_all(&mut self) { self.sessions.clear(); @@ -203,6 +263,14 @@ impl SessionState { self.core_sections.clear(); self.canvas_sections.clear(); self.deliveries.clear(); + // The adapter is gone, so every worker is already dead: release the + // slots outright and forget the closes that can no longer be sent. + self.session_slots.clear(); + self.heartbeat_slot = None; + self.pending_slot = None; + self.pending_close.clear(); + self.last_used.clear(); + self.heartbeat_last_used = None; } pub(crate) fn mark_scope_delivery_success( @@ -309,7 +377,65 @@ fn session_new_system_prompt<'a>( } } +/// What `session/close` did for one session. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CloseOutcome { + Closed, + /// Adapter has no `session/close`; bookkeeping dropped, worker lives + /// until the adapter itself is torn down (pre-existing behavior). + Unsupported, + Failed, + AgentExited, +} + +/// Send `session/close` for `session_id`, mapping the result to a +/// [`CloseOutcome`]. Never fails the caller — a session we cannot close is +/// still one we stop using. +pub(crate) async fn close_session_quietly(acp: &mut AcpClient, session_id: &str) -> CloseOutcome { + match acp.session_close(session_id).await { + Ok(()) => { + tracing::info!(target: "pool::session", "closed session {session_id}"); + CloseOutcome::Closed + } + Err(AcpError::AgentExited) => { + tracing::warn!( + target: "pool::session", + "agent exited while closing session {session_id}" + ); + CloseOutcome::AgentExited + } + Err(AcpError::AgentError { code: -32601, .. }) => { + tracing::debug!( + target: "pool::session", + "adapter has no session/close — dropping session {session_id} bookkeeping only" + ); + CloseOutcome::Unsupported + } + Err(error) => { + tracing::warn!( + target: "pool::session", + "session/close failed for {session_id}: {error}" + ); + CloseOutcome::Failed + } + } +} + impl OwnedAgent { + /// Send `session/close` for every session this agent invalidated while + /// its adapter stayed alive, releasing each fleet slot afterwards. Must + /// only run on an idle agent (nothing else may be reading its stdout). + pub(crate) async fn flush_pending_close(&mut self) { + while let Some((session_id, slot)) = self.state.pending_close.pop() { + let outcome = close_session_quietly(&mut self.acp, &session_id).await; + drop(slot); + if outcome == CloseOutcome::AgentExited { + self.state.invalidate_all(); + return; + } + } + } + pub(crate) fn has_system_prompt_support(&self) -> bool { has_system_prompt_support( self.protocol_version, @@ -783,6 +909,8 @@ pub struct PromptContext { pub channel_info: ChannelInfoResolver, /// Max messages to include in thread/DM context. 0 = disabled. pub context_message_limit: u32, + /// Post trailing reply text in DMs when the agent published nothing. + pub dm_autopublish: bool, /// Max turns per session before proactive rotation. 0 = disabled. pub max_turns_per_session: u32, /// Permission mode to apply after session creation. `Default` = skip. @@ -880,8 +1008,12 @@ impl AgentPool { } /// Return an agent to its slot after a task completes. - pub fn return_agent(&mut self, agent: OwnedAgent) { + pub fn return_agent(&mut self, mut agent: OwnedAgent) { let idx = agent.index; + // A slot handed out for a session the turn never created goes back to + // the fleet here; a created session moved it into `session_slots`. + agent.state.pending_slot = None; + agent.state.idle_since = Some(tokio::time::Instant::now()); if self.agents[idx].is_some() { // This is a bug: two tasks returned the same agent index. Log it // loudly so it shows up in production logs, then overwrite — the @@ -2305,6 +2437,13 @@ pub async fn run_prompt_task( .state .deliveries .insert(scope.clone(), ChannelDeliveryState::default()); + if let Some(slot) = agent.state.pending_slot.take() { + agent.state.session_slots.insert(scope.clone(), slot); + } + agent + .state + .last_used + .insert(scope.clone(), tokio::time::Instant::now()); // Seed a zero usage baseline: buzz-acp spawned this session // so prior usage is zero by definition — first turn is reliable. agent.acp.notify_session_spawned(&sid); @@ -2367,6 +2506,8 @@ pub async fn run_prompt_task( agent.index ); agent.state.heartbeat_session = Some(sid.clone()); + agent.state.heartbeat_slot = agent.state.pending_slot.take(); + agent.state.heartbeat_last_used = Some(tokio::time::Instant::now()); // Seed a zero usage baseline: buzz-acp spawned this session. agent.acp.notify_session_spawned(&sid); (sid, true) @@ -2774,6 +2915,22 @@ pub async fn run_prompt_task( "turn starting for {}", prompt_label(&source) ); + // DM auto-publish inputs: a confirmed DM only (unknown type ⇒ no post), + // and the turn's start so the self-post lookup can bound its window. + // Self-authored batches (e.g. `[wake-at]` self-mentions) never auto-publish: + // the human asked nothing, so trailing text is bookkeeping, not a reply. + let self_pk = ctx.agent_keys.public_key(); + let batch_has_foreign_author = batch + .as_ref() + .is_some_and(|b| b.events.iter().any(|e| e.event.pubkey != self_pk)); + let turn_is_dm = batch_has_foreign_author + && resolved_channel_info + .as_ref() + .is_some_and(|ci| ci.channel_type == "dm"); + let turn_started_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); // When control_rx is Some (channel tasks), wrap the prompt in select! so // the main loop can cancel, interrupt, or rotate it. Heartbeats @@ -2919,6 +3076,16 @@ pub async fn run_prompt_task( ); } log_stop_reason(&source, &StopReason::EndTurn); + if let PromptSource::Channel(scope) = &source { + maybe_autopublish_dm_reply( + &ctx, + &mut agent.acp, + scope.channel_id(), + turn_is_dm, + turn_started_at, + ) + .await; + } if let PromptSource::Channel(scope) = &source { let standing_sent = !agent.has_system_prompt_support(); record_scope_delivery_success( @@ -2961,6 +3128,16 @@ pub async fn run_prompt_task( match prompt_result { Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + if let PromptSource::Channel(scope) = &source { + maybe_autopublish_dm_reply( + &ctx, + &mut agent.acp, + scope.channel_id(), + turn_is_dm, + turn_started_at, + ) + .await; + } if let PromptSource::Channel(scope) = &source { let standing_sent = !agent.has_system_prompt_support(); @@ -5018,6 +5195,82 @@ pub(crate) async fn post_failure_notice( } } +/// DM turns must end with a message to the human. When the agent produced +/// trailing reply text but never ran `buzz messages send` — and the relay +/// shows nothing authored by us in the channel since the turn started — post +/// that text as a top-level channel message. Channels are never touched. +pub(crate) async fn maybe_autopublish_dm_reply( + ctx: &PromptContext, + acp: &mut crate::acp::AcpClient, + channel_id: Uuid, + is_dm: bool, + turn_started_at: u64, +) { + let (text, saw_send) = acp.take_turn_reply(); + if !dm_autopublish_wanted(ctx.dm_autopublish, is_dm, saw_send, &text) { + return; + } + if self_posted_since(&ctx.rest_client, channel_id, turn_started_at).await { + tracing::debug!( + target: "pool::dm", + channel = %channel_id, + "dm autopublish skipped: agent already posted this turn" + ); + return; + } + tracing::info!( + target: "pool::dm", + channel = %channel_id, + chars = text.chars().count(), + "dm autopublish: turn ended without a publish — posting trailing reply text" + ); + post_failure_notice(&ctx.rest_client, channel_id, &ThreadTags::default(), &text).await; +} + +/// Pure decision: only in DMs, only when enabled, only when the agent did not +/// already run `buzz messages send`, and only when there is text to post. +pub(crate) fn dm_autopublish_wanted( + enabled: bool, + is_dm: bool, + saw_send: bool, + text: &str, +) -> bool { + enabled && is_dm && !saw_send && !text.trim().is_empty() +} + +/// Whether the relay holds a message authored by us in `channel_id` at or +/// after `since` (unix seconds, 2s slack). Fails closed (`true`) on error so +/// a lookup outage never double-posts. +async fn self_posted_since(rest: &RestClient, channel_id: Uuid, since: u64) -> bool { + use nostr::{Alphabet, SingleLetterTag}; + let ch = channel_id.to_string(); + let filter = nostr::Filter::new() + .kinds([ + nostr::Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE as u16), + nostr::Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + ]) + .author(rest.keys.public_key()) + .custom_tags(SingleLetterTag::lowercase(Alphabet::H), [ch.as_str()]) + .since(nostr::Timestamp::from(since.saturating_sub(2))) + .limit(1); + match timeout( + CONTEXT_FETCH_TIMEOUT, + rest.query(std::slice::from_ref(&filter)), + ) + .await + { + Ok(Ok(json)) => json.as_array().is_some_and(|a| !a.is_empty()), + Ok(Err(e)) => { + tracing::warn!(channel = %channel_id, "dm autopublish: self-post lookup failed: {e} — skipping"); + true + } + Err(_) => { + tracing::warn!(channel = %channel_id, "dm autopublish: self-post lookup timed out — skipping"); + true + } + } +} + /// Best-effort: remove a reaction via a signed kind:5 (NIP-09) deletion event. /// /// Queries kind:7 reactions by our pubkey targeting the event, finds the matching @@ -7595,6 +7848,231 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" } } + // ── fleet slots & session idle bookkeeping ─────────────────────────────── + + fn fleet_dir(name: &str) -> std::path::PathBuf { + std::env::temp_dir().join(format!( + "buzz-acp-pool-fleet-{name}-{}-{}", + std::process::id(), + Uuid::new_v4() + )) + } + + async fn inert_agent(index: usize) -> OwnedAgent { + OwnedAgent { + index, + acp: AcpClient::spawn("cat", &[], &[], false) + .await + .expect("spawn cat as inert agent"), + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + desired_model_request_id: None, + desired_model_pending_ack: false, + startup_effort: None, + agent_name: "unknown".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + } + } + + fn queued_event(channel_id: Uuid) -> crate::queue::QueuedEvent { + crate::queue::QueuedEvent { + channel_id, + scope: crate::scope::SessionScope::Conversation { channel_id }, + event: EventBuilder::new(Kind::Custom(9), "test") + .sign_with_keys(&Keys::generate()) + .unwrap(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + } + } + + #[tokio::test] + async fn invalidating_a_channel_holds_its_slot_until_the_close_is_flushed() { + let dir = fleet_dir("invalidate"); + let mut fleet = crate::fleet::FleetPool::new(&dir, 1).expect("fleet dir"); + let mut state = SessionState::default(); + let cid = Uuid::new_v4(); + let scope = crate::scope::SessionScope::Conversation { channel_id: cid }; + state.sessions.insert(scope.clone(), "sess-1".into()); + state + .session_slots + .insert(scope.clone(), fleet.try_acquire().expect("slot")); + state + .last_used + .insert(scope.clone(), tokio::time::Instant::now()); + + assert_eq!(state.invalidate_channel(&cid), 1); + assert!(!state.sessions.contains_key(&scope)); + assert!(!state.last_used.contains_key(&scope)); + assert_eq!(state.pending_close.len(), 1); + assert_eq!(state.pending_close[0].0, "sess-1"); + assert!( + fleet.try_acquire().is_none(), + "slot must ride along with the pending close, not be released early" + ); + + state.pending_close.clear(); + assert!( + fleet.try_acquire().is_some(), + "slot released with the close entry" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn invalidate_all_releases_every_slot_and_forgets_pending_closes() { + let dir = fleet_dir("all"); + let mut fleet = crate::fleet::FleetPool::new(&dir, 3).expect("fleet dir"); + let mut state = SessionState::default(); + let scope = crate::scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + state.sessions.insert(scope.clone(), "sess-1".into()); + state + .session_slots + .insert(scope.clone(), fleet.try_acquire().unwrap()); + state.heartbeat_session = Some("hb".into()); + state.heartbeat_slot = fleet.try_acquire(); + state.pending_slot = fleet.try_acquire(); + assert!(fleet.try_acquire().is_none()); + + state.invalidate_all(); + assert!( + state.pending_close.is_empty(), + "a dead adapter has nothing to close" + ); + assert!(state.session_slots.is_empty() && state.heartbeat_slot.is_none()); + assert!(state.pending_slot.is_none()); + let reclaimed = (0..3).filter_map(|_| fleet.try_acquire()).count(); + assert_eq!(reclaimed, 3); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn touch_stamps_only_live_sessions() { + let mut state = SessionState::default(); + let live = crate::scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let dead = crate::scope::SessionScope::Conversation { + channel_id: Uuid::new_v4(), + }; + let now = tokio::time::Instant::now(); + state.sessions.insert(live.clone(), "sess-live".into()); + state.touch(&PromptSource::Channel(live.clone()), now); + state.touch(&PromptSource::Channel(dead.clone()), now); + state.touch(&PromptSource::Heartbeat, now); + assert_eq!(state.last_used.get(&live), Some(&now)); + assert!(!state.last_used.contains_key(&dead)); + assert!(state.heartbeat_last_used.is_none()); + state.heartbeat_session = Some("hb".into()); + state.touch(&PromptSource::Heartbeat, now); + assert_eq!(state.heartbeat_last_used, Some(now)); + } + + #[tokio::test] + async fn return_agent_releases_an_unused_pending_slot() { + let dir = fleet_dir("return"); + let mut fleet = crate::fleet::FleetPool::new(&dir, 1).expect("fleet dir"); + let mut agent = inert_agent(0).await; + agent.state.pending_slot = fleet.try_acquire(); + assert!(fleet.try_acquire().is_none()); + let mut pool = AgentPool::from_slots(vec![None]); + pool.return_agent(agent); + // Retry briefly: a sibling test mid-spawn can hold a duplicate of the + // just-dropped lock descriptor until its exec. + let mut reclaimed = fleet.try_acquire(); + for _ in 0..50 { + if reclaimed.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + reclaimed = fleet.try_acquire(); + } + assert!(reclaimed.is_some(), "returned agent gave the slot back"); + assert!(pool.agents_mut()[0] + .as_ref() + .unwrap() + .state + .idle_since + .is_some()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn dispatch_parks_channels_needing_a_session_when_the_fleet_is_full() { + let dir = fleet_dir("dispatch"); + let mut fleet = Some(crate::fleet::FleetPool::new(&dir, 1).expect("fleet dir")); + let ctx = Arc::new(make_prompt_context_no_owner()); + let mut pool = + AgentPool::from_slots(vec![Some(inert_agent(0).await), Some(inert_agent(1).await)]); + let mut queue = crate::queue::EventQueue::new(DedupMode::Queue); + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + assert!(queue.push(queued_event(first))); + assert!(queue.push(queued_event(second))); + let mut last_activity = tokio::time::Instant::now(); + + let dispatched = + crate::dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet); + assert_eq!(dispatched.len(), 1, "one slot, one new session"); + let started = dispatched[0].0.clone(); + let parked = if started.channel_id() == first { + second + } else { + first + }; + assert!(queue.is_scope_in_flight(&started)); + assert!( + !queue.is_scope_in_flight(parked), + "parked channel is released back to the queue, not left in flight" + ); + assert_eq!( + queue.queued_event_count(parked), + 1, + "parked event is not lost" + ); + assert!( + pool.any_idle(), + "no slot means no agent is claimed for the parked channel" + ); + assert!(fleet.as_ref().unwrap().is_waiting()); + + // A channel that already holds a session on the idle agent needs no + // slot and dispatches even while the fleet is full. + let affinity = Uuid::new_v4(); + { + let idle = pool + .agents_mut() + .iter_mut() + .flatten() + .next() + .expect("an idle agent"); + idle.state.sessions.insert( + crate::scope::SessionScope::Conversation { + channel_id: affinity, + }, + "sess-affinity".into(), + ); + } + assert!(queue.push(queued_event(affinity))); + let dispatched = + crate::dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, &mut fleet); + assert_eq!(dispatched.len(), 1); + assert_eq!(dispatched[0].0.channel_id(), affinity); + assert_eq!( + queue.queued_event_count(parked), + 1, + "still parked, still intact" + ); + + pool.join_set.abort_all(); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn test_requeue_cancelled_batch_maps_control_signal_to_cancel_reason() { let cases = [ @@ -8641,6 +9119,15 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" ); } + #[test] + fn dm_autopublish_wanted_only_in_dm_without_send() { + assert!(dm_autopublish_wanted(true, true, false, "hi")); + assert!(!dm_autopublish_wanted(false, true, false, "hi")); + assert!(!dm_autopublish_wanted(true, false, false, "hi")); + assert!(!dm_autopublish_wanted(true, true, true, "hi")); + assert!(!dm_autopublish_wanted(true, true, false, " \n")); + } + pub(super) fn make_prompt_context_no_owner() -> PromptContext { let agent_keys = nostr::Keys::generate(); make_prompt_context_impl(&agent_keys, None) @@ -8687,6 +9174,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" }, ), context_message_limit: 0, + dm_autopublish: false, max_turns_per_session: 0, permission_mode: PermissionMode::Default, agent_keys: agent_keys.clone(), diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index b2fbde6242f..41678b2fe00 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -774,6 +774,33 @@ impl EventQueue { has_queued || has_cancelled || has_withheld } + /// Number of distinct pending session scopes with undispatched work — the + /// per-scope count behind [`has_undispatched_work`](Self::has_undispatched_work). + /// + /// On-demand agent sizing keys on this: each pending scope needs its own + /// provider session, so this is the number of sessions' worth of waiting + /// work. Under `channel` policy every scope is a `Conversation`, so it + /// equals the count of channels with undispatched work; under `thread` + /// policy it counts distinct waiting thread partitions. + pub fn undispatched_scopes(&self) -> usize { + let mut scopes: HashSet<&SessionScope> = HashSet::new(); + scopes.extend( + self.queues + .iter() + .filter(|(_, q)| !q.is_empty()) + .map(|(scope, _)| scope), + ); + scopes.extend(self.cancelled_batches.keys()); + scopes.extend( + self.withheld_native_steer + .iter() + .filter(|(_, v)| !v.is_empty()) + .map(|(scope, _)| scope), + ); + scopes.retain(|scope| !self.in_flight_scopes.contains(scope)); + scopes.len() + } + /// Number of pending partitions (session scopes) with queued events. /// /// Under `channel` policy this equals the number of channels with pending From 800e694b268500df00b6089e96c066315b28a4b3 Mon Sep 17 00:00:00 2001 From: gruming Date: Thu, 3 Sep 2026 18:38:55 +0900 Subject: [PATCH 2/2] feat(acp): make --dm-autopublish opt-in (default off) Align the DM auto-publish flag with the pool flags: every new flag in this change now defaults off, so a deployment that passes none of them behaves exactly like upstream. Set BUZZ_ACP_DM_AUTOPUBLISH=true (or --dm-autopublish true) to enable it. README default updated and the CLI default test now covers the flag. Co-Authored-By: Claude Fable 5.1 Signed-off-by: gruming --- crates/buzz-acp/README.md | 2 +- crates/buzz-acp/src/config.rs | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 1e3bdebab66..9c3bdd032ac 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -129,7 +129,7 @@ All configuration is via environment variables (or CLI flags — every env var h | `--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 `/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` | `true` | 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. | +| `--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`. | diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index cdc291c3ea8..08c9101d853 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -556,7 +556,7 @@ pub struct CliArgs { /// 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 channel /// message so the human never gets silence. Channels are unaffected. - #[arg(long, env = "BUZZ_ACP_DM_AUTOPUBLISH", default_value_t = true, action = clap::ArgAction::Set)] + #[arg(long, env = "BUZZ_ACP_DM_AUTOPUBLISH", default_value_t = false, action = clap::ArgAction::Set)] pub dm_autopublish: bool, } @@ -2393,6 +2393,7 @@ channels = "ALL" assert_eq!(default.fleet_slots, 0); assert_eq!(default.fleet_slot_dir, None); assert_eq!(default.session_idle_close, 0); + assert!(!default.dm_autopublish); let configured = CliArgs::parse_from([ "buzz-acp", @@ -2408,6 +2409,8 @@ channels = "ALL" "/tmp/fleet", "--session-idle-close", "600", + "--dm-autopublish", + "true", ]); assert_eq!(configured.min_agents, Some(2)); assert_eq!(configured.fleet_slots, 20); @@ -2416,6 +2419,7 @@ channels = "ALL" Some(std::path::Path::new("/tmp/fleet")) ); assert_eq!(configured.session_idle_close, 600); + assert!(configured.dm_autopublish); } #[test]