From 6a416a6e31cd57a58fab9dac746a3f138583d815 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 08:13:57 -0400 Subject: [PATCH 1/7] feat(sprout-acp): add pool.rs, multi-channel queue, heartbeat config Steps 1-3 of parallel agents + heartbeat implementation: - queue.rs: HashSet in_flight_channels, retry_after throttle, requeue_preserve_timestamps, has_flushable_work, Clone derives - pool.rs: AgentPool, OwnedAgent, PromptResult, run_prompt_task - config.rs: --agents, --heartbeat-interval, --heartbeat-prompt flags All 116 tests pass. main.rs loop rewrite (Step 4) follows. --- crates/sprout-acp/src/config.rs | 110 ++++++- crates/sprout-acp/src/main.rs | 12 +- crates/sprout-acp/src/pool.rs | 527 ++++++++++++++++++++++++++++++++ crates/sprout-acp/src/queue.rs | 410 ++++++++++++++++++++++--- 4 files changed, 1003 insertions(+), 56 deletions(-) create mode 100644 crates/sprout-acp/src/pool.rs diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 96ebfe4546..5ef1890d24 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -94,6 +94,25 @@ pub struct CliArgs { )] pub system_prompt_file: Option, + /// Number of parallel agent subprocesses. + #[arg(long, env = "SPROUT_ACP_AGENTS", default_value_t = 1, + value_parser = clap::value_parser!(u32).range(1..=32))] + pub agents: u32, + + /// Seconds between heartbeat prompts. 0 = disabled. + #[arg(long, env = "SPROUT_ACP_HEARTBEAT_INTERVAL", default_value_t = 0)] + pub heartbeat_interval: u64, + + /// Heartbeat prompt text. Conflicts with --heartbeat-prompt-file. + #[arg(long, env = "SPROUT_ACP_HEARTBEAT_PROMPT", + conflicts_with = "heartbeat_prompt_file")] + pub heartbeat_prompt: Option, + + /// Read heartbeat prompt from file. + #[arg(long, env = "SPROUT_ACP_HEARTBEAT_PROMPT_FILE", + conflicts_with = "heartbeat_prompt")] + pub heartbeat_prompt_file: Option, + #[arg(long, env = "SPROUT_ACP_INITIAL_MESSAGE")] pub initial_message: Option, @@ -146,6 +165,9 @@ pub struct Config { pub agent_args: Vec, pub mcp_command: String, pub turn_timeout_secs: u64, + pub agents: u32, + pub heartbeat_interval_secs: u64, + pub heartbeat_prompt: Option, pub system_prompt: Option, pub initial_message: Option, pub subscribe_mode: SubscribeMode, @@ -187,6 +209,20 @@ impl Config { None }; + if args.heartbeat_interval > 0 && args.heartbeat_interval < 10 { + return Err(ConfigError::ConfigFile( + "heartbeat interval must be 0 (disabled) or ≥10 seconds".into(), + )); + } + + let heartbeat_prompt = if let Some(text) = args.heartbeat_prompt { + Some(text) + } else if let Some(ref path) = args.heartbeat_prompt_file { + Some(std::fs::read_to_string(path)?) + } else { + None + }; + if matches!(args.subscribe, SubscribeMode::Config) { if args.kinds.is_some() { tracing::warn!("--kinds is ignored in config mode"); @@ -207,6 +243,9 @@ impl Config { agent_args: args.agent_args, mcp_command: args.mcp_command, turn_timeout_secs: args.turn_timeout, + agents: args.agents, + heartbeat_interval_secs: args.heartbeat_interval, + heartbeat_prompt, system_prompt, initial_message: args.initial_message, subscribe_mode: args.subscribe, @@ -222,13 +261,15 @@ impl Config { /// Human-readable summary (no secrets). pub fn summary(&self) -> String { format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} timeout={}s subscribe={:?} dedup={:?} ignore_self={}", + "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} timeout={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} ignore_self={}", self.relay_url, self.keys.public_key().to_hex(), self.agent_command, self.agent_args.join(" "), self.mcp_command, self.turn_timeout_secs, + self.agents, + self.heartbeat_interval_secs, self.subscribe_mode, self.dedup_mode, self.ignore_self, @@ -424,6 +465,9 @@ mod tests { agent_args: vec!["acp".into()], mcp_command: "sprout-mcp-server".into(), turn_timeout_secs: 300, + agents: 1, + heartbeat_interval_secs: 0, + heartbeat_prompt: None, system_prompt: None, initial_message: None, subscribe_mode: mode, @@ -842,4 +886,68 @@ channels = "ALL" assert!(err.to_string().contains("filter too long")); std::fs::remove_dir_all(&dir).ok(); } + + // ── heartbeat validation ───────────────────────────────────────────────── + + fn validate_heartbeat_interval(secs: u64) -> Result<(), ConfigError> { + if secs > 0 && secs < 10 { + return Err(ConfigError::ConfigFile( + "heartbeat interval must be 0 (disabled) or ≥10 seconds".into(), + )); + } + Ok(()) + } + + #[test] + fn test_heartbeat_interval_zero_ok() { + assert!(validate_heartbeat_interval(0).is_ok()); + } + + #[test] + fn test_heartbeat_interval_ten_ok() { + assert!(validate_heartbeat_interval(10).is_ok()); + } + + #[test] + fn test_heartbeat_interval_large_ok() { + assert!(validate_heartbeat_interval(300).is_ok()); + } + + #[test] + fn test_heartbeat_interval_five_rejected() { + let err = validate_heartbeat_interval(5).unwrap_err(); + assert!(err.to_string().contains("heartbeat interval must be 0")); + } + + #[test] + fn test_heartbeat_interval_one_rejected() { + let err = validate_heartbeat_interval(1).unwrap_err(); + assert!(err.to_string().contains("heartbeat interval must be 0")); + } + + #[test] + fn test_heartbeat_interval_nine_rejected() { + let err = validate_heartbeat_interval(9).unwrap_err(); + assert!(err.to_string().contains("heartbeat interval must be 0")); + } + + // ── summary includes agents and heartbeat ──────────────────────────────── + + #[test] + fn test_summary_includes_agents_and_heartbeat() { + let config = test_config(SubscribeMode::Mentions); + let s = config.summary(); + assert!(s.contains("agents=1"), "summary should include agents=1, got: {s}"); + assert!(s.contains("heartbeat=0s"), "summary should include heartbeat=0s, got: {s}"); + } + + #[test] + fn test_summary_reflects_custom_agents_and_heartbeat() { + let mut config = test_config(SubscribeMode::Mentions); + config.agents = 4; + config.heartbeat_interval_secs = 30; + let s = config.summary(); + assert!(s.contains("agents=4"), "summary should include agents=4, got: {s}"); + assert!(s.contains("heartbeat=30s"), "summary should include heartbeat=30s, got: {s}"); + } } diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 6d5b4eac05..dd1b3e1624 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -199,7 +199,7 @@ async fn main() -> Result<()> { DedupMode::Queue => queue.requeue(batch), DedupMode::Drop => { /* discard */ } } - queue.mark_complete(); + queue.mark_complete(channel_id); acp = respawn_agent(&config).await?; break; } @@ -211,7 +211,7 @@ async fn main() -> Result<()> { DedupMode::Queue => queue.requeue(batch), DedupMode::Drop => { /* discard */ } } - queue.mark_complete(); + queue.mark_complete(channel_id); break; } }; @@ -228,7 +228,7 @@ async fn main() -> Result<()> { match prompt_result { Ok(Ok(stop_reason)) => { log_stop_reason(channel_id, &stop_reason); - queue.mark_complete(); + queue.mark_complete(channel_id); } Ok(Err(AcpError::AgentExited)) => { tracing::error!("agent process exited — respawning"); @@ -237,7 +237,7 @@ async fn main() -> Result<()> { DedupMode::Queue => queue.requeue(batch), DedupMode::Drop => { /* discard */ } } - queue.mark_complete(); + queue.mark_complete(channel_id); acp = respawn_agent(&config).await?; break; } @@ -247,7 +247,7 @@ async fn main() -> Result<()> { DedupMode::Queue => queue.requeue(batch), DedupMode::Drop => { /* discard */ } } - queue.mark_complete(); + queue.mark_complete(channel_id); sessions.remove(&channel_id); break; } @@ -272,7 +272,7 @@ async fn main() -> Result<()> { sessions.remove(&channel_id); } } - queue.mark_complete(); + queue.mark_complete(channel_id); break; } } diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs new file mode 100644 index 0000000000..a2514d7d8f --- /dev/null +++ b/crates/sprout-acp/src/pool.rs @@ -0,0 +1,527 @@ +//! Agent pool — owns N AcpClient instances and dispatches prompt tasks. +//! +//! # Mental model +//! +//! ```text +//! AgentPool +//! ├── agents: Vec> ← idle agents sit here +//! ├── join_set: JoinSet<()> ← in-flight tasks +//! ├── task_map: HashMap ← panic recovery metadata +//! └── result_tx/rx: mpsc channel ← tasks return agents here +//! +//! Dispatch: +//! try_claim() → OwnedAgent (removed from slot) +//! spawn run_prompt_task(agent, ...) into join_set +//! task sends PromptResult { agent, outcome } via result_tx +//! next_result() → PromptResult +//! return_agent(agent) → puts agent back in slot +//! ``` +//! +//! `AcpClient` is NOT Clone — ownership moves out on claim and back on return. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::mpsc; +use tokio::task::JoinSet; +use tokio::time::timeout; +use uuid::Uuid; + +use crate::acp::{AcpClient, AcpError, McpServer, StopReason}; +use crate::config::DedupMode; +use crate::queue::FlushBatch; + +// ── FlushBatch Clone note ───────────────────────────────────────────────────── +// FlushBatch and BatchEvent derive Clone (added in queue.rs) so we can store +// a recoverable copy in TaskMeta for panic recovery in Queue mode. + +// ── Types ───────────────────────────────────────────────────────────────────── + +/// Metadata stored per in-flight task for panic recovery. +struct TaskMeta { + agent_index: usize, + channel_id: Option, + /// Clone of batch for Queue mode panic recovery. + recoverable_batch: Option, +} + +/// An agent with its session state, owned by the pool or a running task. +pub struct OwnedAgent { + pub index: usize, + pub acp: AcpClient, + /// channel_id → session_id + pub sessions: HashMap, + pub heartbeat_session: Option, +} + +/// Pool of agents with take-and-return ownership semantics. +/// +/// Agents are either idle (sitting in `agents[i]`) or checked out +/// (running inside a spawned task). The `task_map` tracks in-flight +/// tasks for panic recovery. +pub struct AgentPool { + agents: Vec>, + result_tx: mpsc::UnboundedSender, + result_rx: mpsc::UnboundedReceiver, + pub join_set: JoinSet<()>, + task_map: HashMap, +} + +/// Result returned by a completed prompt task. +pub struct PromptResult { + pub agent: OwnedAgent, + pub source: PromptSource, + pub outcome: PromptOutcome, + /// Present on failure in Queue mode, for requeue. + pub batch: Option, +} + +/// Whether the prompt came from a channel event or a heartbeat. +pub enum PromptSource { + Channel(Uuid), + Heartbeat, +} + +/// Outcome of a prompt task. +pub enum PromptOutcome { + Ok(StopReason), + Error(AcpError), + AgentExited, + Timeout, +} + +/// Immutable config subset shared (via `Arc`) by all spawned prompt tasks. +/// +/// Built once from `Config` at startup. Avoids cloning the full config +/// into every task. +pub struct PromptContext { + pub mcp_servers: Vec, + pub initial_message: Option, + pub turn_timeout: Duration, + pub dedup_mode: DedupMode, + pub system_prompt: Option, + pub heartbeat_prompt: Option, + pub cwd: String, +} + +// ── AgentPool impl ──────────────────────────────────────────────────────────── + +impl AgentPool { + /// Create a new pool from a list of initialized agents. + /// + /// Agents are placed into indexed slots. The unbounded channel is created + /// here; tasks send results back through `result_tx`. + pub fn new(agents: Vec) -> Self { + let (result_tx, result_rx) = mpsc::unbounded_channel(); + let slots = agents.into_iter().map(Some).collect(); + Self { + agents: slots, + result_tx, + result_rx, + join_set: JoinSet::new(), + task_map: HashMap::new(), + } + } + + /// Try to claim an idle agent for the given channel (or heartbeat if `None`). + /// + /// Pass 1: prefer an agent that already has a session for `channel_id`. + /// Pass 2: any idle agent. + /// + /// Returns `None` if all agents are checked out. + pub fn try_claim(&mut self, channel_id: Option) -> Option { + // Pass 1: prefer agent with existing session for this channel. + if let Some(cid) = channel_id { + let idx = self + .agents + .iter() + .position(|slot| { + slot.as_ref() + .map(|a| a.sessions.contains_key(&cid)) + .unwrap_or(false) + }); + if let Some(i) = idx { + return self.agents[i].take(); + } + } + + // Pass 2: first idle agent. + let idx = self.agents.iter().position(|slot| slot.is_some()); + idx.map(|i| self.agents[i].take().unwrap()) + } + + /// Return an agent to its slot after a task completes. + pub fn return_agent(&mut self, agent: OwnedAgent) { + self.agents[agent.index] = Some(agent); + } + + /// Whether any agent is currently idle (sitting in its slot). + pub fn any_idle(&self) -> bool { + self.agents.iter().any(|slot| slot.is_some()) + } + + /// Count of agents that are alive: idle OR checked out (have a task_map entry). + /// + /// Used to detect when all agents have exited so the caller can respawn. + pub fn live_count(&self) -> usize { + let idle = self.agents.iter().filter(|s| s.is_some()).count(); + let checked_out = self.task_map.len(); + idle + checked_out + } + + /// Wait for the next completed prompt result. + /// + /// Panics if the channel closes (impossible while pool is alive). + pub async fn next_result(&mut self) -> PromptResult { + self.result_rx + .recv() + .await + .expect("result channel closed — pool invariant violated") + } + + // ── Accessors ───────────────────────────────────────────────────────── + + pub fn task_map(&self) -> &HashMap { + &self.task_map + } + + pub fn task_map_mut(&mut self) -> &mut HashMap { + &mut self.task_map + } + + pub fn result_tx(&self) -> mpsc::UnboundedSender { + self.result_tx.clone() + } + + pub fn agents_mut(&mut self) -> &mut Vec> { + &mut self.agents + } +} + +// ── run_prompt_task ─────────────────────────────────────────────────────────── + +/// Core async function spawned for each prompt. +/// +/// Lifecycle: +/// 1. Resolve or create a session (channel or heartbeat). +/// 2. Send `initial_message` on new channel sessions (if configured). +/// 3. Send the actual prompt with turn timeout. +/// 4. Handle all error paths, always returning the agent via `result_tx`. +/// +/// The agent is ALWAYS returned — even on panic the `JoinSet` detects the +/// abort and the caller uses `task_map` to recover the agent index. +pub async fn run_prompt_task( + mut agent: OwnedAgent, + batch: Option, + prompt_text: String, + ctx: Arc, + result_tx: mpsc::UnboundedSender, +) { + // ── Determine source and resolve/create session ─────────────────────── + + // Is this a channel prompt or a heartbeat? + let source = match &batch { + Some(b) => PromptSource::Channel(b.channel_id), + None => PromptSource::Heartbeat, + }; + + let (session_id, is_new_session) = match &source { + PromptSource::Channel(cid) => { + if let Some(sid) = agent.sessions.get(cid) { + (sid.clone(), false) + } else { + // Create new session. + match agent.acp.session_new(&ctx.cwd, ctx.mcp_servers.clone()).await { + Ok(sid) => { + tracing::info!( + target: "pool::session", + "created session {sid} for channel {cid}" + ); + agent.sessions.insert(*cid, sid.clone()); + (sid, true) + } + Err(AcpError::AgentExited) => { + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + Err(e) => { + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Error(e), + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + } + } + } + PromptSource::Heartbeat => { + if let Some(sid) = &agent.heartbeat_session { + (sid.clone(), false) + } else { + match agent.acp.session_new(&ctx.cwd, ctx.mcp_servers.clone()).await { + Ok(sid) => { + tracing::info!( + target: "pool::session", + "created heartbeat session {sid} for agent {}", + agent.index + ); + agent.heartbeat_session = Some(sid.clone()); + (sid, true) + } + Err(AcpError::AgentExited) => { + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: None, + }); + return; + } + Err(e) => { + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Error(e), + batch: None, + }); + return; + } + } + } + } + }; + + // ── Send initial_message on new channel sessions ────────────────────── + + if is_new_session { + if let (PromptSource::Channel(cid), Some(ref initial_msg)) = + (&source, &ctx.initial_message) + { + tracing::info!( + target: "pool::session", + "sending initial_message to session {session_id} for channel {cid}" + ); + let init_result = + timeout(ctx.turn_timeout, agent.acp.session_prompt(&session_id, initial_msg)) + .await; + + match init_result { + Ok(Ok(stop_reason)) => { + tracing::info!( + target: "pool::session", + "initial_message complete for channel {cid}: {stop_reason:?}" + ); + } + Ok(Err(AcpError::AgentExited)) => { + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + Ok(Err(e)) => { + tracing::error!( + target: "pool::session", + "initial_message failed for channel {cid}: {e} — invalidating session" + ); + agent.sessions.remove(cid); + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Error(e), + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + Err(_elapsed) => { + tracing::warn!( + target: "pool::session", + "initial_message timed out for channel {cid} — cancelling" + ); + match agent.acp.cancel_with_cleanup(&session_id).await { + Ok(_) => { + agent.sessions.remove(cid); + } + Err(AcpError::AgentExited) => { + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + Err(e) => { + tracing::error!( + target: "pool::session", + "cancel_with_cleanup failed during initial_message timeout: {e}" + ); + agent.sessions.remove(cid); + } + } + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Timeout, + batch: requeue_batch_if_queue(&ctx, batch), + }); + return; + } + } + } + } + + // ── Send the actual prompt ──────────────────────────────────────────── + + let prompt_result = + timeout(ctx.turn_timeout, agent.acp.session_prompt(&session_id, &prompt_text)).await; + + match prompt_result { + Ok(Ok(stop_reason)) => { + log_stop_reason(&source, &stop_reason); + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Ok(stop_reason), + batch: None, + }); + } + Ok(Err(AcpError::AgentExited)) => { + tracing::error!(target: "pool::prompt", "agent {} exited during prompt", agent.index); + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: requeue_batch_if_queue(&ctx, batch), + }); + } + Ok(Err(e)) => { + tracing::error!(target: "pool::prompt", "session_prompt error: {e}"); + // Invalidate only the affected session. + match &source { + PromptSource::Channel(cid) => { + agent.sessions.remove(cid); + } + PromptSource::Heartbeat => { + agent.heartbeat_session = None; + } + } + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Error(e), + batch: requeue_batch_if_queue(&ctx, batch), + }); + } + Err(_elapsed) => { + tracing::warn!( + target: "pool::prompt", + "turn timeout ({}s) — cancelling session {session_id}", + ctx.turn_timeout.as_secs() + ); + match agent.acp.cancel_with_cleanup(&session_id).await { + Ok(stop_reason) => { + log_stop_reason(&source, &stop_reason); + // Session is still valid after a clean cancel. + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Timeout, + batch: requeue_batch_if_queue(&ctx, batch), + }); + } + Err(AcpError::AgentExited) => { + tracing::error!( + target: "pool::prompt", + "agent {} exited during cancel_with_cleanup", + agent.index + ); + agent.sessions.clear(); + agent.heartbeat_session = None; + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::AgentExited, + batch: requeue_batch_if_queue(&ctx, batch), + }); + } + Err(e) => { + tracing::error!( + target: "pool::prompt", + "cancel_with_cleanup error: {e} — invalidating session" + ); + match &source { + PromptSource::Channel(cid) => { + agent.sessions.remove(cid); + } + PromptSource::Heartbeat => { + agent.heartbeat_session = None; + } + } + let _ = result_tx.send(PromptResult { + agent, + source, + outcome: PromptOutcome::Timeout, + batch: requeue_batch_if_queue(&ctx, batch), + }); + } + } + } + } +} + +// ── Internal helpers ────────────────────────────────────────────────────────── + +/// Return the batch for requeue only in Queue mode; drop it in Drop mode. +#[inline] +fn requeue_batch_if_queue(ctx: &PromptContext, batch: Option) -> Option { + match ctx.dedup_mode { + DedupMode::Queue => batch, + DedupMode::Drop => None, + } +} + +/// Log a stop reason at the appropriate tracing level. +fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { + let label = match source { + PromptSource::Channel(cid) => format!("channel {cid}"), + PromptSource::Heartbeat => "heartbeat".to_string(), + }; + match stop_reason { + StopReason::EndTurn => { + tracing::info!(target: "pool::prompt", "turn complete for {label}: end_turn"); + } + StopReason::Cancelled => { + tracing::warn!(target: "pool::prompt", "turn cancelled for {label}"); + } + StopReason::MaxTokens => { + tracing::warn!(target: "pool::prompt", "turn hit max_tokens for {label}"); + } + StopReason::MaxTurnRequests => { + tracing::warn!(target: "pool::prompt", "turn hit max_turn_requests for {label}"); + } + StopReason::Refusal => { + tracing::warn!(target: "pool::prompt", "turn refused for {label}"); + } + } +} diff --git a/crates/sprout-acp/src/queue.rs b/crates/sprout-acp/src/queue.rs index 732dc800c5..29429f9fa9 100644 --- a/crates/sprout-acp/src/queue.rs +++ b/crates/sprout-acp/src/queue.rs @@ -1,9 +1,10 @@ //! Event queue state machine for sprout-acp. //! -//! Manages per-channel event queues with a global one-in-flight constraint. +//! Manages per-channel event queues with per-channel in-flight tracking. //! When the harness is ready to prompt the agent, it flushes the channel with //! the oldest pending event, draining ALL events for that channel into a single -//! batch. Only one `session/prompt` is in flight at a time across all channels. +//! batch. Multiple channels can be in-flight simultaneously; each channel is +//! independent. //! //! ## Dedup modes //! @@ -13,8 +14,8 @@ //! - **Queue** — all events accumulate; batched on the next flush cycle. use nostr::{Event, ToBech32}; -use std::collections::{HashMap, VecDeque}; -use std::time::Instant; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::time::{Duration, Instant}; use uuid::Uuid; use crate::config::DedupMode; @@ -32,14 +33,15 @@ pub struct QueuedEvent { } /// A single event inside a [`FlushBatch`]. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct BatchEvent { pub event: Event, pub prompt_tag: String, + pub received_at: Instant, } /// A batch of events to prompt the agent with. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct FlushBatch { pub channel_id: Uuid, pub events: Vec, @@ -47,37 +49,42 @@ pub struct FlushBatch { // ── EventQueue ──────────────────────────────────────────────────────────────── -/// Per-channel event queue with global one-in-flight enforcement. +/// Per-channel event queue with per-channel in-flight enforcement. /// /// # State Machine /// /// ```text /// State: /// queues: Map> -/// in_flight_channel: Option +/// in_flight_channels: HashSet +/// retry_after: Map /// dedup_mode: DedupMode /// /// Transitions: /// push(event): -/// if dedup_mode == Drop AND in_flight_channel == Some(event.channel_id): +/// if dedup_mode == Drop AND in_flight_channels.contains(event.channel_id): /// debug log + discard /// else: /// queues[event.channel_id].push_back(event) /// /// flush_next() → Option: -/// if in_flight_channel.is_some(): return None -/// if all queues empty: return None -/// channel = pick channel with oldest head event (min received_at) +/// candidates = channels where queue non-empty +/// AND NOT in in_flight_channels +/// AND (no retry_after OR retry_after[c] <= now) +/// if candidates empty: return None +/// channel = pick candidate with oldest head event (min received_at) /// events = drain queues[channel] -/// in_flight_channel = Some(channel) +/// in_flight_channels.insert(channel) /// return Some(FlushBatch { channel, events }) /// -/// mark_complete(): -/// in_flight_channel = None +/// mark_complete(channel_id): +/// in_flight_channels.remove(channel_id) +/// (retry_after entries expire naturally via Instant check) /// ``` pub struct EventQueue { queues: HashMap>, - in_flight_channel: Option, + in_flight_channels: HashSet, + retry_after: HashMap, dedup_mode: DedupMode, } @@ -86,18 +93,19 @@ impl EventQueue { pub fn new(dedup_mode: DedupMode) -> Self { Self { queues: HashMap::new(), - in_flight_channel: None, + in_flight_channels: HashSet::new(), + retry_after: HashMap::new(), dedup_mode, } } /// Push an event into the queue for its channel. /// - /// In [`DedupMode::Drop`], events for the currently in-flight channel are + /// In [`DedupMode::Drop`], events for any currently in-flight channel are /// silently discarded (debug-logged). pub fn push(&mut self, event: QueuedEvent) { if matches!(self.dedup_mode, DedupMode::Drop) - && self.in_flight_channel == Some(event.channel_id) + && self.in_flight_channels.contains(&event.channel_id) { tracing::debug!( channel_id = %event.channel_id, @@ -113,20 +121,23 @@ impl EventQueue { /// Try to flush the next batch. /// - /// Returns `None` if a prompt is already in flight or if all queues are - /// empty. Otherwise picks the channel with the oldest pending event (FIFO - /// fairness across channels), drains ALL events for that channel into a - /// single batch, sets `in_flight_channel`, and returns the batch. + /// Returns `None` if all non-in-flight, non-throttled queues are empty. + /// Otherwise picks the channel with the oldest pending event (FIFO fairness + /// across channels), drains ALL events for that channel into a single batch, + /// inserts into `in_flight_channels`, and returns the batch. pub fn flush_next(&mut self) -> Option { - if self.in_flight_channel.is_some() { - return None; - } + let now = Instant::now(); - // Find the channel whose head event has the oldest received_at. + // Find the channel whose head event has the oldest received_at, + // excluding in-flight channels and throttled channels. let channel_id = self .queues .iter() - .filter(|(_, q)| !q.is_empty()) + .filter(|(id, q)| { + !q.is_empty() + && !self.in_flight_channels.contains(id) + && self.retry_after.get(id).map_or(true, |&t| t <= now) + }) .min_by_key(|(_, q)| q.front().unwrap().received_at) .map(|(id, _)| *id)?; @@ -137,17 +148,21 @@ impl EventQueue { .map(|qe| BatchEvent { event: qe.event, prompt_tag: qe.prompt_tag, + received_at: qe.received_at, }) .collect(); - self.in_flight_channel = Some(channel_id); + self.in_flight_channels.insert(channel_id); Some(FlushBatch { channel_id, events }) } - /// Mark the current prompt as complete. Clears `in_flight_channel`. - pub fn mark_complete(&mut self) { - self.in_flight_channel = None; + /// Mark the prompt for `channel_id` as complete. + /// + /// Removes the channel from `in_flight_channels`. Does NOT clear + /// `retry_after` — those entries expire naturally via Instant check. + pub fn mark_complete(&mut self, channel_id: Uuid) { + self.in_flight_channels.remove(&channel_id); } /// Re-queue a batch of events that failed to process. @@ -156,9 +171,12 @@ impl EventQueue { /// are processed first on the next flush cycle. This prevents event loss /// when session creation or `session/prompt` fails transiently. /// - /// Note: `received_at` is reset to `Instant::now()` for re-queued events. - /// This means a re-queued channel competes fairly with other channels rather - /// than always winning due to stale timestamps. + /// `received_at` is reset to `Instant::now()` for re-queued events. + /// A 5-second `retry_after` throttle is set so the channel is not + /// immediately re-flushed. + /// + /// Note: does NOT remove from `in_flight_channels` — caller must call + /// `mark_complete` separately. pub fn requeue(&mut self, batch: FlushBatch) { let queue = self.queues.entry(batch.channel_id).or_default(); // Push to front in reverse order so original order is preserved. @@ -170,12 +188,46 @@ impl EventQueue { received_at: Instant::now(), }); } + self.retry_after + .insert(batch.channel_id, Instant::now() + Duration::from_secs(5)); } - /// Whether a prompt is currently in flight. + /// Re-queue a batch preserving original `received_at` timestamps. + /// + /// Used when a batch was flushed but no agent was available — we want to + /// retry without penalizing the channel's position in the fairness queue + /// and without imposing a retry throttle. + /// + /// Does NOT set `retry_after`. Does NOT remove from `in_flight_channels` — + /// caller must call `mark_complete` separately. + pub fn requeue_preserve_timestamps(&mut self, batch: FlushBatch) { + let queue = self.queues.entry(batch.channel_id).or_default(); + // Push to front in reverse order so original order is preserved. + for be in batch.events.into_iter().rev() { + queue.push_front(QueuedEvent { + channel_id: batch.channel_id, + event: be.event, + prompt_tag: be.prompt_tag, + received_at: be.received_at, + }); + } + } + + /// Returns `true` if any channel has pending events that are not in-flight + /// and not throttled by `retry_after`. + pub fn has_flushable_work(&self) -> bool { + let now = Instant::now(); + self.queues.iter().any(|(id, q)| { + !q.is_empty() + && !self.in_flight_channels.contains(id) + && self.retry_after.get(id).map_or(true, |&t| t <= now) + }) + } + + /// Whether any prompt is currently in flight. #[allow(dead_code)] pub fn is_in_flight(&self) -> bool { - self.in_flight_channel.is_some() + !self.in_flight_channels.is_empty() } /// Total number of pending events across all channels. @@ -342,10 +394,10 @@ mod tests { assert_eq!(q.pending_channels(), 0); } - // ── Test 2: in_flight blocks flush ─────────────────────────────────────── + // ── Test 2: same channel cannot be flushed twice ───────────────────────── #[test] - fn test_in_flight_blocks_flush() { + fn test_in_flight_blocks_same_channel() { let mut q = EventQueue::new(DedupMode::Queue); let ch = Uuid::new_v4(); @@ -356,7 +408,8 @@ mod tests { // Push another event while in-flight. q.push(make_queued(ch, "second")); - // flush_next must return None while in-flight. + // flush_next for the same channel must return None (it's in-flight). + // No other channels exist, so result is None. assert!(q.flush_next().is_none()); } @@ -370,12 +423,12 @@ mod tests { q.push(make_queued(ch, "first")); let _batch = q.flush_next().expect("first flush should succeed"); - // Push while in-flight; flush blocked. + // Push while in-flight; flush blocked (same channel in-flight). q.push(make_queued(ch, "second")); assert!(q.flush_next().is_none()); // Complete the in-flight prompt. - q.mark_complete(); + q.mark_complete(ch); assert!(!q.is_in_flight()); // Now flush should succeed. @@ -449,7 +502,7 @@ mod tests { assert_eq!(q.pending_count(), 1); assert_eq!(q.pending_channels(), 1); - q.mark_complete(); + q.mark_complete(ch_a); // Second flush picks B. let batch_b = q.flush_next().expect("second flush"); @@ -490,7 +543,7 @@ mod tests { assert_eq!(q.pending_count(), 1); assert_eq!(q.pending_channels(), 1); - q.mark_complete(); + q.mark_complete(ch_a); // Flush B (1 event drained). let _ = q.flush_next(); @@ -514,6 +567,7 @@ mod tests { events: vec![BatchEvent { event, prompt_tag: "@mention".into(), + received_at: Instant::now(), }], }; @@ -542,7 +596,10 @@ mod tests { // Simulate failure — requeue the batch. queue.requeue(batch); - queue.mark_complete(); + queue.mark_complete(ch); + + // retry_after is set, so manually clear it for this test. + queue.retry_after.remove(&ch); // Should be able to flush again and get the same events in order. let batch2 = queue.flush_next().unwrap(); @@ -567,9 +624,9 @@ mod tests { // Requeue ch_a (simulating failure) and complete. queue.requeue(batch_a); - queue.mark_complete(); + queue.mark_complete(ch_a); - // After requeue, ch_a's received_at is reset to now, so ch_b (older) goes first. + // After requeue, ch_a has retry_after set (5s), so ch_b goes first. let next_batch = queue.flush_next().unwrap(); assert_eq!(next_batch.channel_id, ch_b); } @@ -589,14 +646,17 @@ mod tests { BatchEvent { event: e1, prompt_tag: "tag-a".into(), + received_at: Instant::now(), }, BatchEvent { event: e2, prompt_tag: "tag-b".into(), + received_at: Instant::now(), }, BatchEvent { event: e3, prompt_tag: "tag-c".into(), + received_at: Instant::now(), }, ], }; @@ -630,6 +690,7 @@ mod tests { events: vec![BatchEvent { event, prompt_tag: "test".into(), + received_at: Instant::now(), }], }; @@ -652,7 +713,7 @@ mod tests { q.push(make_queued(ch, "dropped")); assert_eq!(q.pending_count(), 0, "event should be dropped"); - q.mark_complete(); + q.mark_complete(ch); // Nothing to flush. assert!(q.flush_next().is_none()); } @@ -673,8 +734,259 @@ mod tests { q.push(make_queued(ch_b, "B-event")); assert_eq!(q.pending_count(), 1); - q.mark_complete(); + q.mark_complete(ch_a); let batch_b = q.flush_next().expect("flush B"); assert_eq!(batch_b.channel_id, ch_b); } + + // ── Test 14: multiple channels can be in-flight simultaneously ──────────── + + #[test] + fn test_multiple_channels_in_flight_simultaneously() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + + q.push(make_queued_at(ch_a, "A-event", Duration::from_secs(2))); + q.push(make_queued_at(ch_b, "B-event", Duration::from_secs(1))); + + // Flush A — now A is in-flight. + let batch_a = q.flush_next().expect("flush A"); + assert_eq!(batch_a.channel_id, ch_a); + assert!(q.is_in_flight()); + + // Flush B — B should also be flushable (different channel). + let batch_b = q.flush_next().expect("flush B while A in-flight"); + assert_eq!(batch_b.channel_id, ch_b); + + // Both in-flight. + assert_eq!(q.in_flight_channels.len(), 2); + + // Complete A only. + q.mark_complete(ch_a); + assert!(q.is_in_flight()); // B still in-flight. + + // Complete B. + q.mark_complete(ch_b); + assert!(!q.is_in_flight()); + } + + // ── Test 15: same channel cannot be flushed twice ───────────────────────── + + #[test] + fn test_same_channel_not_flushed_twice() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let ch2 = Uuid::new_v4(); + + q.push(make_queued(ch, "first")); + let _batch = q.flush_next().expect("first flush"); + + // Push more events for same channel while in-flight. + q.push(make_queued(ch, "second")); + // Also push for another channel. + q.push(make_queued(ch2, "other")); + + // flush_next should pick ch2, not ch (ch is in-flight). + let batch2 = q.flush_next().expect("should flush ch2"); + assert_eq!(batch2.channel_id, ch2); + + // ch still in-flight — no more candidates. + assert!(q.flush_next().is_none()); + } + + // ── Test 16: drop mode drops events for any in-flight channel ───────────── + + #[test] + fn test_drop_mode_drops_for_any_in_flight_channel() { + let mut q = EventQueue::new(DedupMode::Drop); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + + q.push(make_queued_at(ch_a, "A-event", Duration::from_secs(2))); + q.push(make_queued_at(ch_b, "B-event", Duration::from_secs(1))); + + // Flush both — both in-flight. + let _batch_a = q.flush_next().expect("flush A"); + let _batch_b = q.flush_next().expect("flush B"); + + // Drop mode: pushing to either in-flight channel is dropped. + q.push(make_queued(ch_a, "A-dropped")); + q.push(make_queued(ch_b, "B-dropped")); + assert_eq!(q.pending_count(), 0); + + q.mark_complete(ch_a); + q.mark_complete(ch_b); + } + + // ── Test 17: flush_next picks oldest non-in-flight, non-throttled channel ─ + + #[test] + fn test_flush_next_picks_oldest_non_throttled() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + let ch_c = Uuid::new_v4(); + + // A is oldest, B is middle, C is newest. + q.push(make_queued_at(ch_a, "A", Duration::from_secs(10))); + q.push(make_queued_at(ch_b, "B", Duration::from_secs(5))); + q.push(make_queued_at(ch_c, "C", Duration::from_secs(1))); + + // Flush A (oldest). + let batch = q.flush_next().expect("flush A"); + assert_eq!(batch.channel_id, ch_a); + + // A is in-flight; next oldest non-in-flight is B. + let batch2 = q.flush_next().expect("flush B"); + assert_eq!(batch2.channel_id, ch_b); + + // A and B in-flight; only C left. + let batch3 = q.flush_next().expect("flush C"); + assert_eq!(batch3.channel_id, ch_c); + + // All in-flight. + assert!(q.flush_next().is_none()); + + q.mark_complete(ch_a); + q.mark_complete(ch_b); + q.mark_complete(ch_c); + } + + // ── Test 18: mark_complete(channel_id) clears only that channel ─────────── + + #[test] + fn test_mark_complete_clears_only_specified_channel() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + + q.push(make_queued_at(ch_a, "A", Duration::from_secs(2))); + q.push(make_queued_at(ch_b, "B", Duration::from_secs(1))); + + let _batch_a = q.flush_next().expect("flush A"); + let _batch_b = q.flush_next().expect("flush B"); + + assert_eq!(q.in_flight_channels.len(), 2); + + // Complete only A. + q.mark_complete(ch_a); + assert_eq!(q.in_flight_channels.len(), 1); + assert!(q.in_flight_channels.contains(&ch_b)); + assert!(!q.in_flight_channels.contains(&ch_a)); + + // B still in-flight. + assert!(q.is_in_flight()); + + q.mark_complete(ch_b); + assert!(!q.is_in_flight()); + } + + // ── Test 19: requeue_preserve_timestamps preserves received_at ─────────── + + #[test] + fn test_requeue_preserve_timestamps() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let old_time = Instant::now() - Duration::from_secs(10); + + q.push(QueuedEvent { + channel_id: ch, + event: make_event("old-msg"), + received_at: old_time, + prompt_tag: "test".into(), + }); + + let batch = q.flush_next().expect("flush"); + let original_received_at = batch.events[0].received_at; + + // requeue_preserve_timestamps should keep the original timestamp. + q.requeue_preserve_timestamps(batch); + q.mark_complete(ch); + + // No retry_after set — should be immediately flushable. + let batch2 = q.flush_next().expect("flush after requeue_preserve"); + assert_eq!(batch2.events[0].received_at, original_received_at); + } + + // ── Test 20: requeue_preserve_timestamps does not set retry_after ───────── + + #[test] + fn test_requeue_preserve_timestamps_no_retry_after() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + q.push(make_queued(ch, "msg")); + let batch = q.flush_next().expect("flush"); + + q.requeue_preserve_timestamps(batch); + q.mark_complete(ch); + + // No retry_after — channel should be immediately flushable. + assert!(q.retry_after.get(&ch).is_none()); + assert!(q.flush_next().is_some()); + } + + // ── Test 21: has_flushable_work returns correct results ─────────────────── + + #[test] + fn test_has_flushable_work() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + // Empty queue — no flushable work. + assert!(!q.has_flushable_work()); + + q.push(make_queued(ch, "msg")); + assert!(q.has_flushable_work()); + + // Flush — now in-flight, no flushable work. + let _batch = q.flush_next().expect("flush"); + assert!(!q.has_flushable_work()); + + // Complete — no pending events, no flushable work. + q.mark_complete(ch); + assert!(!q.has_flushable_work()); + + // Requeue with retry_after — throttled, no flushable work. + q.push(make_queued(ch, "msg2")); + let batch2 = q.flush_next().expect("flush2"); + q.requeue(batch2); + q.mark_complete(ch); + assert!(!q.has_flushable_work(), "throttled channel should not be flushable"); + + // Manually expire the retry_after to simulate time passing. + q.retry_after.insert(ch, Instant::now() - Duration::from_secs(1)); + assert!(q.has_flushable_work(), "expired throttle should be flushable"); + } + + // ── Test 22: retry throttle blocks re-flush for 5 seconds ───────────────── + + #[test] + fn test_retry_throttle_blocks_requeue_channel() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let ch2 = Uuid::new_v4(); + + q.push(make_queued(ch, "msg")); + let batch = q.flush_next().expect("flush"); + + // Requeue sets retry_after. + q.requeue(batch); + q.mark_complete(ch); + + // Channel is throttled — flush_next should return None (no other channels). + assert!(q.flush_next().is_none()); + + // Add a different channel — it should be flushable. + q.push(make_queued(ch2, "other")); + let batch2 = q.flush_next().expect("ch2 should be flushable"); + assert_eq!(batch2.channel_id, ch2); + + // After retry_after expires, ch should be flushable again. + q.retry_after.insert(ch, Instant::now() - Duration::from_secs(1)); + q.mark_complete(ch2); + let batch3 = q.flush_next().expect("ch should be flushable after throttle expires"); + assert_eq!(batch3.channel_id, ch); + } } From 2b974634c9ab8e3d7a884a54111f7f3d3ace7aa1 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 08:20:05 -0400 Subject: [PATCH 2/7] feat(sprout-acp): rewrite main loop for parallel agents + heartbeat Step 4: Replace 2-branch select with 5-branch biased select. - N-agent startup with AgentPool - dispatch_pending: flush queued work to idle agents - handle_prompt_result: reclaim agent, requeue on failure, respawn on exit - recover_panicked_agent: JoinError::id() -> task_map O(1) lookup - drain_ready_join_results: now_or_never() prevents panic starvation - dispatch_heartbeat: at-most-one-globally guard - Unified SIGINT+SIGTERM shutdown via watch channel - Grace period drain + JoinSet::shutdown() abort - rx_and_join_set() split-borrow helper for select! All 116 tests pass. --- crates/sprout-acp/src/main.rs | 676 ++++++++++++++++++++-------------- crates/sprout-acp/src/pool.rs | 27 +- 2 files changed, 416 insertions(+), 287 deletions(-) diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index dd1b3e1624..bbc62260c0 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -3,21 +3,23 @@ mod acp; mod config; mod filter; +mod pool; mod queue; mod relay; use std::collections::HashMap; +use std::sync::Arc; use std::time::Duration; use anyhow::Result; +use futures_util::FutureExt; use nostr::ToBech32; -use tokio::time::timeout; +use tokio::sync::watch; use tracing_subscriber::EnvFilter; -use uuid::Uuid; - -use acp::{AcpClient, AcpError, EnvVar, McpServer, StopReason}; +use acp::{AcpClient, EnvVar, McpServer}; use config::{Config, DedupMode, SubscribeMode}; use filter::SubscriptionRule; +use pool::{AgentPool, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource}; use queue::{EventQueue, QueuedEvent}; use relay::HarnessRelay; use sprout_core::kind::{ @@ -36,8 +38,19 @@ async fn main() -> Result<()> { let config = Config::from_cli().map_err(|e| anyhow::anyhow!("configuration error: {e}"))?; tracing::info!("sprout-acp starting: {}", config.summary()); - // ── Step 1: Spawn ACP agent subprocess and initialize ───────────────────── - let mut acp = spawn_and_init(&config).await?; + // ── Step 1: Spawn N ACP agent subprocesses and initialize ───────────────── + let mut agents = Vec::with_capacity(config.agents as usize); + for i in 0..config.agents as usize { + let acp = spawn_and_init(&config).await?; + agents.push(OwnedAgent { + index: i, + acp, + sessions: HashMap::new(), + heartbeat_session: None, + }); + } + tracing::info!("agent_pool_ready agents={}", agents.len()); + let mut pool = AgentPool::new(agents); // ── Step 2: Connect to Sprout relay ────────────────────────────────────── let pubkey_hex = config.keys.public_key().to_hex(); @@ -60,7 +73,6 @@ async fn main() -> Result<()> { tracing::info!("discovered {} channel(s)", channels.len()); - // Build subscription rules from the configured mode. let rules: Vec = match config.subscribe_mode { SubscribeMode::Mentions => { vec![SubscriptionRule { @@ -113,207 +125,410 @@ async fn main() -> Result<()> { } } - // ── Step 5: Main orchestration loop ────────────────────────────────────── - let mut sessions: HashMap = HashMap::new(); + // ── Step 5: Build shared prompt context ────────────────────────────────── let dedup_mode = config.dedup_mode; let mut queue = EventQueue::new(dedup_mode); - let mcp_servers = build_mcp_servers(&config); - let turn_timeout = Duration::from_secs(config.turn_timeout_secs); - - loop { - // Wait for the next relay event or shutdown signal. - let sprout_event = tokio::select! { - event = relay.next_event() => event, - _ = tokio::signal::ctrl_c() => { - tracing::info!("shutting down (SIGINT/SIGTERM)"); - break; - } - }; - - match sprout_event { - Some(sprout_event) => { - // Self-event filter: drop events authored by this agent. - if config.ignore_self && sprout_event.event.pubkey.to_hex() == pubkey_hex { - tracing::debug!( - channel_id = %sprout_event.channel_id, - "dropping self-authored event" - ); - continue; - } - - // Rule match: find first matching rule. - let matched = filter::match_event( - &sprout_event.event, - sprout_event.channel_id, - &rules, - &pubkey_hex, - ) - .await; - - let prompt_tag = match matched { - Some(m) => m.prompt_tag, - None => { - tracing::debug!( - channel_id = %sprout_event.channel_id, - kind = sprout_event.event.kind.as_u16(), - "event matched no rule — dropping" - ); - continue; - } - }; - - // Push event into the queue. - queue.push(QueuedEvent { - channel_id: sprout_event.channel_id, - event: sprout_event.event, - received_at: std::time::Instant::now(), - prompt_tag, - }); - - // Try to flush and process batches. - loop { - let batch = match queue.flush_next() { - Some(b) => b, - None => break, - }; - - let channel_id = batch.channel_id; - // Format prompt before potentially requeuing (borrows batch by ref). - let prompt_text = queue::format_prompt(&batch, config.system_prompt.as_deref()); - - // Get or create session for this channel. - let session_id = match get_or_create_session( - &mut sessions, - channel_id, - &mut acp, - &mcp_servers, - &config, - ) - .await - { - Ok(id) => id, - Err(AcpError::AgentExited) => { - tracing::error!("agent exited during session setup — respawning"); - sessions.clear(); - match dedup_mode { - DedupMode::Queue => queue.requeue(batch), - DedupMode::Drop => { /* discard */ } - } - queue.mark_complete(channel_id); - acp = respawn_agent(&config).await?; - break; - } - Err(e) => { - tracing::error!( - "failed to create session for channel {channel_id}: {e}" - ); - match dedup_mode { - DedupMode::Queue => queue.requeue(batch), - DedupMode::Drop => { /* discard */ } - } - queue.mark_complete(channel_id); - break; - } - }; - tracing::info!( - "prompting agent for channel {channel_id} (session {session_id}, {} event(s))", - batch.events.len() - ); + let ctx = Arc::new(PromptContext { + mcp_servers: build_mcp_servers(&config), + initial_message: config.initial_message.clone(), + turn_timeout: Duration::from_secs(config.turn_timeout_secs), + dedup_mode: config.dedup_mode, + system_prompt: config.system_prompt.clone(), + heartbeat_prompt: config.heartbeat_prompt.clone(), + cwd: std::env::current_dir() + .unwrap_or_else(|_| std::path::PathBuf::from("/")) + .to_string_lossy() + .to_string(), + }); + + // ── Step 6: Heartbeat timer ─────────────────────────────────────────────── + let mut heartbeat = if config.heartbeat_interval_secs > 0 { + let interval = Duration::from_secs(config.heartbeat_interval_secs); + Some(tokio::time::interval_at( + tokio::time::Instant::now() + interval, + interval, + )) + } else { + None + }; + let mut heartbeat_in_flight = false; + + // ── Step 7: Shutdown signal ─────────────────────────────────────────────── + let (shutdown_tx, mut shutdown_rx) = watch::channel(()); + + let tx = shutdown_tx.clone(); + tokio::spawn(async move { + tokio::signal::ctrl_c().await.ok(); + let _ = tx.send(()); + }); + + #[cfg(unix)] + { + let tx = shutdown_tx.clone(); + tokio::spawn(async move { + use tokio::signal::unix::{signal, SignalKind}; + let mut sigterm = signal(SignalKind::terminate()).expect("SIGTERM handler"); + sigterm.recv().await; + let _ = tx.send(()); + }); + } - // Send prompt with turn timeout. - let prompt_result = - timeout(turn_timeout, acp.session_prompt(&session_id, &prompt_text)).await; + // ── Step 8: Main orchestration loop ────────────────────────────────────── + // + // 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 + // borrow, yielding a typed enum so the outer code can dispatch cleanly. + enum PoolEvent { + Result(PromptResult), + Panic(tokio::task::JoinError), + } - match prompt_result { - Ok(Ok(stop_reason)) => { - log_stop_reason(channel_id, &stop_reason); - queue.mark_complete(channel_id); - } - Ok(Err(AcpError::AgentExited)) => { - tracing::error!("agent process exited — respawning"); - sessions.clear(); - match dedup_mode { - DedupMode::Queue => queue.requeue(batch), - DedupMode::Drop => { /* discard */ } - } - queue.mark_complete(channel_id); - acp = respawn_agent(&config).await?; - break; - } - Ok(Err(e)) => { - tracing::error!("session_prompt error for channel {channel_id}: {e}"); - match dedup_mode { - DedupMode::Queue => queue.requeue(batch), - DedupMode::Drop => { /* discard */ } + loop { + // Borrow result_rx and join_set simultaneously via split-borrow helper. + let pool_event: Option = { + let (result_rx, join_set) = pool.rx_and_join_set(); + tokio::select! { + biased; + r = result_rx.recv() => Some(PoolEvent::Result(r.expect("result channel closed"))), + j = join_set.join_next() => match j { + Some(Err(e)) => Some(PoolEvent::Panic(e)), + _ => None, + }, + // Remaining branches don't touch pool — evaluated when pool is idle. + sprout_event = relay.next_event() => { + let _ = result_rx; // end split borrow before relay handling + match sprout_event { + Some(sprout_event) => { + if config.ignore_self && sprout_event.event.pubkey.to_hex() == pubkey_hex { + tracing::debug!(channel_id = %sprout_event.channel_id, "dropping self-authored event"); + continue; } - queue.mark_complete(channel_id); - sessions.remove(&channel_id); - break; - } - Err(_elapsed) => { - tracing::warn!( - "turn timeout ({}s) for channel {channel_id} — cancelling", - config.turn_timeout_secs - ); - match acp.cancel_with_cleanup(&session_id).await { - Ok(stop_reason) => { - log_stop_reason(channel_id, &stop_reason); - } - Err(AcpError::AgentExited) => { - tracing::error!("agent exited during cancel — respawning"); - sessions.clear(); - acp = respawn_agent(&config).await?; - } - Err(e) => { - tracing::error!( - "cancel_with_cleanup error for channel {channel_id}: {e} — invalidating session" - ); - sessions.remove(&channel_id); + let matched = filter::match_event(&sprout_event.event, sprout_event.channel_id, &rules, &pubkey_hex).await; + let prompt_tag = match matched { + Some(m) => m.prompt_tag, + None => { + tracing::debug!(channel_id = %sprout_event.channel_id, kind = sprout_event.event.kind.as_u16(), "event matched no rule — dropping"); + continue; } + }; + queue.push(QueuedEvent { + channel_id: sprout_event.channel_id, + event: sprout_event.event, + received_at: std::time::Instant::now(), + prompt_tag, + }); + dispatch_pending(&mut pool, &mut queue, &ctx); + } + None => { + tracing::warn!("relay event stream ended — requesting reconnect"); + if let Err(e) = relay.reconnect().await { + tracing::error!("relay background task is gone: {e} — exiting"); + tokio::time::sleep(Duration::from_secs(1)).await; + break; } - queue.mark_complete(channel_id); - break; } } + None + } + _ = async { + match heartbeat.as_mut() { + Some(hb) => hb.tick().await, + None => std::future::pending().await, + } + } => { + let _ = result_rx; + if queue.has_flushable_work() { + dispatch_pending(&mut pool, &mut queue, &ctx); + } else if pool.any_idle() { + dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); + } + None + } + _ = shutdown_rx.changed() => { + tracing::info!("shutting down"); + break; } } - None => { - // Relay event stream ended — request background reconnect. - // The background task handles the actual reconnection and - // resubscription asynchronously; we just resume waiting for events. - tracing::warn!("relay event stream ended — requesting reconnect"); - if let Err(e) = relay.reconnect().await { - // Background task is dead (cmd_tx closed). Sleep to prevent - // a CPU-burning spin loop, then exit — no recovery possible. - tracing::error!("relay background task is gone: {e} — exiting"); - tokio::time::sleep(Duration::from_secs(1)).await; + }; + + match pool_event { + Some(PoolEvent::Result(result)) => { + if handle_prompt_result(&mut pool, &mut queue, &config, result, &mut heartbeat_in_flight).await == LoopAction::Exit { + break; + } + if drain_ready_join_results(&mut pool, &mut queue, &config, &mut heartbeat_in_flight).await == LoopAction::Exit { + break; + } + dispatch_pending(&mut pool, &mut queue, &ctx); + } + Some(PoolEvent::Panic(join_error)) => { + tracing::error!("agent task panicked: {join_error}"); + recover_panicked_agent(&mut pool, &mut queue, &config, join_error, &mut heartbeat_in_flight).await; + if pool.live_count() == 0 { + tracing::error!("all agents dead — exiting"); break; } + dispatch_pending(&mut pool, &mut queue, &ctx); } + None => {} // relay/heartbeat/shutdown branches handled inline above } } + // ── Shutdown sequence ───────────────────────────────────────────────────── + tracing::info!("shutdown: waiting for in-flight prompts"); + let grace = Duration::from_secs(config.turn_timeout_secs + 5); + let shutdown_result = tokio::time::timeout(grace, async { + while let Some(result) = pool.join_set.join_next().await { + if let Err(e) = result { + tracing::warn!("task finished with error during shutdown: {e}"); + } + } + }) + .await; + if shutdown_result.is_err() { + tracing::warn!("grace period expired, aborting remaining tasks"); + pool.join_set.shutdown().await; + } + drop(pool); tracing::info!("sprout-acp stopped"); Ok(()) } -// ── Helper: respawn agent after exit ───────────────────────────────────────── +// ── Loop control ────────────────────────────────────────────────────────────── + +#[derive(PartialEq)] +enum LoopAction { + Continue, + Exit, +} + +// ── dispatch_pending ────────────────────────────────────────────────────────── + +/// Flush queued work to available agents. +fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc) { + loop { + let batch = match queue.flush_next() { + Some(b) => b, + None => break, + }; + let channel_id = batch.channel_id; + let agent = match pool.try_claim(Some(channel_id)) { + Some(a) => a, + None => { + queue.requeue_preserve_timestamps(batch); + queue.mark_complete(channel_id); + break; + } + }; + + let prompt_text = queue::format_prompt(&batch, ctx.system_prompt.as_deref()); + let recoverable_batch = match ctx.dedup_mode { + DedupMode::Queue => Some(batch.clone()), + DedupMode::Drop => None, + }; + + let result_tx = pool.result_tx(); + let ctx_clone = Arc::clone(ctx); + let agent_index = agent.index; + + let abort_handle = pool.join_set.spawn(async move { + pool::run_prompt_task(agent, Some(batch), prompt_text, ctx_clone, result_tx).await; + }); + + pool.task_map_mut().insert( + abort_handle.id(), + pool::TaskMeta { + agent_index, + channel_id: Some(channel_id), + recoverable_batch, + }, + ); + } +} + +// ── handle_prompt_result ────────────────────────────────────────────────────── + +async fn handle_prompt_result( + pool: &mut AgentPool, + queue: &mut EventQueue, + config: &Config, + result: PromptResult, + heartbeat_in_flight: &mut bool, +) -> LoopAction { + let before = pool.task_map().len(); + let agent_index = result.agent.index; + pool.task_map_mut().retain(|_, meta| meta.agent_index != agent_index); + debug_assert_eq!(before, pool.task_map().len() + 1); + + match &result.source { + PromptSource::Channel(ch) => queue.mark_complete(*ch), + PromptSource::Heartbeat => *heartbeat_in_flight = false, + } + + if let Some(batch) = result.batch { + queue.requeue(batch); + } + + match result.outcome { + PromptOutcome::AgentExited => { + let index = result.agent.index; + match respawn_agent_into(result.agent, config).await { + Ok(agent) => pool.return_agent(agent), + Err(e) => { + tracing::error!("failed to respawn agent {index}: {e}"); + if pool.live_count() == 0 { + tracing::error!("all agents dead — exiting"); + return LoopAction::Exit; + } + } + } + } + _ => { + pool.return_agent(result.agent); + } + } + LoopAction::Continue +} + +// ── recover_panicked_agent ──────────────────────────────────────────────────── + +async fn recover_panicked_agent( + pool: &mut AgentPool, + queue: &mut EventQueue, + config: &Config, + join_error: tokio::task::JoinError, + heartbeat_in_flight: &mut bool, +) { + let task_id = join_error.id(); + let Some(meta) = pool.task_map_mut().remove(&task_id) else { + tracing::error!("panic for unknown task {task_id:?} — bug"); + return; + }; + let i = meta.agent_index; + + if let Some(ch) = meta.channel_id { + queue.mark_complete(ch); + tracing::warn!("cleared wedged in-flight channel {ch} from panicked agent {i}"); + } else { + *heartbeat_in_flight = false; + tracing::warn!("cleared wedged heartbeat_in_flight from panicked agent {i}"); + } + + if let Some(batch) = meta.recoverable_batch { + queue.requeue(batch); + tracing::warn!("requeued batch for panicked agent {i}"); + } -async fn respawn_agent(config: &Config) -> Result { match spawn_and_init(config).await { - Ok(new_acp) => { - tracing::info!("agent respawned successfully"); - Ok(new_acp) + Ok(acp) => { + pool.agents_mut()[i] = Some(OwnedAgent { + index: i, + acp, + sessions: HashMap::new(), + heartbeat_session: None, + }); + tracing::info!("respawned agent {i} after panic"); } Err(e) => { - tracing::error!("failed to respawn agent: {e}"); - Err(e) + tracing::error!("failed to respawn agent {i} after panic: {e}"); } } } -// ── Helper: spawn agent and initialize ─────────────────────────────────────── +// ── drain_ready_join_results ────────────────────────────────────────────────── + +async fn drain_ready_join_results( + pool: &mut AgentPool, + queue: &mut EventQueue, + config: &Config, + heartbeat_in_flight: &mut bool, +) -> LoopAction { + while let Some(Some(join_result)) = pool.join_set.join_next().now_or_never() { + if let Err(join_error) = join_result { + tracing::error!("agent task panicked: {join_error}"); + recover_panicked_agent(pool, queue, config, join_error, heartbeat_in_flight).await; + if pool.live_count() == 0 { + return LoopAction::Exit; + } + } + } + LoopAction::Continue +} + +// ── dispatch_heartbeat ──────────────────────────────────────────────────────── + +fn dispatch_heartbeat( + pool: &mut AgentPool, + ctx: &Arc, + heartbeat_in_flight: &mut bool, +) { + if *heartbeat_in_flight { + return; + } + let agent = match pool.try_claim(None) { + Some(a) => a, + None => return, + }; + + let prompt_text = ctx + .heartbeat_prompt + .clone() + .unwrap_or_else(default_heartbeat_prompt); + let result_tx = pool.result_tx(); + let ctx_clone = Arc::clone(ctx); + let agent_index = agent.index; + + let abort_handle = pool.join_set.spawn(async move { + pool::run_prompt_task(agent, None, prompt_text, ctx_clone, result_tx).await; + }); + + pool.task_map_mut().insert( + abort_handle.id(), + pool::TaskMeta { + agent_index, + channel_id: None, + recoverable_batch: None, + }, + ); + *heartbeat_in_flight = true; +} + +// ── default_heartbeat_prompt ────────────────────────────────────────────────── + +fn default_heartbeat_prompt() -> String { + let now = chrono::Utc::now().to_rfc3339(); + format!( + "[System: Heartbeat]\nTime: {now}\n\n\ + You have been awakened for a routine heartbeat. You have NO incoming messages or\n\ + active channel context for this turn.\n\n\ + Your tasks:\n\ + 1. Call `get_feed_actions()` to check for pending workflow approvals or\n\ + high-priority requests addressed to you.\n\ + 2. Call `get_feed_mentions()` to check for unanswered @mentions.\n\ + 3. If you find actionable items, address them using the appropriate tools\n\ + (e.g., `approve_workflow_step`, `send_message`, `send_reply`).\n\ + 4. If there are no pending actions or mentions, end your turn immediately.\n\n\ + Do not call `list_channels()` or `search()` unless you have a specific reason.\n\ + Do not invent work — only act on items surfaced by the feed tools." + ) +} + +// ── respawn_agent_into ──────────────────────────────────────────────────────── + +async fn respawn_agent_into(old_agent: OwnedAgent, config: &Config) -> Result { + let index = old_agent.index; + drop(old_agent); // kill the old process via AcpClient Drop + let acp = spawn_and_init(config).await?; + Ok(OwnedAgent { + index, + acp, + sessions: HashMap::new(), + heartbeat_session: None, + }) +} + +// ── spawn_and_init ──────────────────────────────────────────────────────────── async fn spawn_and_init(config: &Config) -> Result { let mut acp = AcpClient::spawn(&config.agent_command, &config.agent_args) @@ -329,88 +544,7 @@ async fn spawn_and_init(config: &Config) -> Result { Ok(acp) } -// ── Helper: get or create session for a channel ─────────────────────────────── - -/// Get or create an ACP session for the given channel. -/// -/// If a session already exists, returns it immediately. Otherwise creates a new -/// session and — if `config.initial_message` is set — sends it as the first -/// prompt before returning. The initial-message turn counts against -/// `in_flight_channel`, so in `drop` mode events for this channel are dropped -/// while it runs. -async fn get_or_create_session( - sessions: &mut HashMap, - channel_id: Uuid, - acp: &mut AcpClient, - mcp_servers: &[McpServer], - config: &Config, -) -> Result { - if let Some(session_id) = sessions.get(&channel_id) { - return Ok(session_id.clone()); - } - - let cwd = std::env::current_dir() - .unwrap_or_else(|_| std::path::PathBuf::from("/")) - .to_string_lossy() - .to_string(); - - let session_id = acp.session_new(&cwd, mcp_servers.to_vec()).await?; - tracing::info!("created session {session_id} for channel {channel_id}"); - sessions.insert(channel_id, session_id.clone()); - - // Send initial message if configured. - if let Some(ref initial_message) = config.initial_message { - tracing::info!("sending initial message to session {session_id} for channel {channel_id}"); - let turn_timeout = Duration::from_secs(config.turn_timeout_secs); - let result = timeout( - turn_timeout, - acp.session_prompt(&session_id, initial_message), - ) - .await; - match result { - Ok(Ok(stop_reason)) => { - tracing::info!( - "initial message complete for channel {channel_id}: {stop_reason:?}" - ); - } - Ok(Err(e)) => { - tracing::error!( - "initial message failed for channel {channel_id}: {e} — invalidating session" - ); - sessions.remove(&channel_id); - return Err(e); - } - Err(_elapsed) => { - tracing::warn!( - "initial message timed out for channel {channel_id} — cancelling and invalidating session" - ); - // Cancel the in-flight prompt to keep the NDJSON stream in sync, - // matching the cleanup path used by the main event loop. - match acp.cancel_with_cleanup(&session_id).await { - Ok(_) => { - // Agent is still alive — just this session timed out. - sessions.remove(&channel_id); - return Err(AcpError::Timeout); - } - Err(AcpError::AgentExited) => { - // Agent actually died during cancel. - sessions.remove(&channel_id); - return Err(AcpError::AgentExited); - } - Err(cancel_err) => { - tracing::error!("cancel_with_cleanup failed during initial message timeout: {cancel_err}"); - sessions.remove(&channel_id); - return Err(AcpError::Timeout); - } - } - } - } - } - - Ok(session_id) -} - -// ── Helper: build MCP server config from Config ─────────────────────────────── +// ── build_mcp_servers ───────────────────────────────────────────────────────── fn build_mcp_servers(config: &Config) -> Vec { vec![McpServer { @@ -442,25 +576,3 @@ fn build_mcp_servers(config: &Config) -> Vec { }, }] } - -// ── Helper: log stop reason at appropriate level ────────────────────────────── - -fn log_stop_reason(channel_id: Uuid, stop_reason: &StopReason) { - match stop_reason { - StopReason::EndTurn => { - tracing::info!("turn complete for channel {channel_id}: end_turn"); - } - StopReason::Cancelled => { - tracing::warn!("turn cancelled for channel {channel_id}"); - } - StopReason::MaxTokens => { - tracing::warn!("turn hit max_tokens for channel {channel_id}"); - } - StopReason::MaxTurnRequests => { - tracing::warn!("turn hit max_turn_requests for channel {channel_id}"); - } - StopReason::Refusal => { - tracing::warn!("turn refused for channel {channel_id}"); - } - } -} diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index a2514d7d8f..8e60b289cb 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -39,11 +39,11 @@ use crate::queue::FlushBatch; // ── Types ───────────────────────────────────────────────────────────────────── /// Metadata stored per in-flight task for panic recovery. -struct TaskMeta { - agent_index: usize, - channel_id: Option, +pub struct TaskMeta { + pub agent_index: usize, + pub channel_id: Option, /// Clone of batch for Queue mode panic recovery. - recoverable_batch: Option, + pub recoverable_batch: Option, } /// An agent with its session state, owned by the pool or a running task. @@ -153,7 +153,8 @@ impl AgentPool { /// Return an agent to its slot after a task completes. pub fn return_agent(&mut self, agent: OwnedAgent) { - self.agents[agent.index] = Some(agent); + let index = agent.index; + self.agents[index] = Some(agent); } /// Whether any agent is currently idle (sitting in its slot). @@ -194,6 +195,22 @@ impl AgentPool { self.result_tx.clone() } + pub fn result_rx_mut(&mut self) -> &mut mpsc::UnboundedReceiver { + &mut self.result_rx + } + + /// Split-borrow: returns mutable refs to `result_rx` and `join_set` + /// simultaneously. This lets callers poll both in a single `select!` + /// without a double-borrow error on `&mut AgentPool`. + pub fn rx_and_join_set( + &mut self, + ) -> ( + &mut mpsc::UnboundedReceiver, + &mut JoinSet<()>, + ) { + (&mut self.result_rx, &mut self.join_set) + } + pub fn agents_mut(&mut self) -> &mut Vec> { &mut self.agents } From 3377fcd439e33330e1d1c92cf5d7141b7236af0b Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 08:23:38 -0400 Subject: [PATCH 3/7] feat(sprout-acp): operational logging + README docs Steps 5-6: Add structured tracing events for pool health monitoring and update README with parallel agents + heartbeat documentation. - 7 new structured log events (agent_claimed, agent_returned, heartbeat_fired, heartbeat_skipped_*, dispatch_pending, pool_exhausted) - has_session_for() helper in pool.rs for affinity_hit detection - README: new flags table, config examples, shared identity note, heartbeat semantics, choosing N guidance - Fix dead_code warnings in pool.rs --- crates/sprout-acp/README.md | 55 ++++++++++++++++++++++++++++++++++- crates/sprout-acp/src/main.rs | 21 +++++++++++++ crates/sprout-acp/src/pool.rs | 13 +++++++++ 3 files changed, 88 insertions(+), 1 deletion(-) diff --git a/crates/sprout-acp/README.md b/crates/sprout-acp/README.md index 02478d2ed3..ea34a0a301 100644 --- a/crates/sprout-acp/README.md +++ b/crates/sprout-acp/README.md @@ -92,7 +92,9 @@ sprout-acp ## Configuration -All configuration is via environment variables. +All configuration is via environment variables (or CLI flags — every env var has a matching flag). + +### Core | Variable | Required | Default | Description | |----------|----------|---------|-------------| @@ -108,6 +110,57 @@ All configuration is via environment variables. **Legacy env vars:** `SPROUT_ACP_PRIVATE_KEY` and `SPROUT_ACP_API_TOKEN` are still accepted as fallbacks. +### Parallel Agents & Heartbeat + +| Flag | Env Var | Default | Description | +|------|---------|---------|-------------| +| `--agents` | `SPROUT_ACP_AGENTS` | `1` | Number of agent subprocesses (1–32). | +| `--heartbeat-interval` | `SPROUT_ACP_HEARTBEAT_INTERVAL` | `0` | Seconds between heartbeat prompts. `0` = disabled. Must be `0` or ≥10 when enabled. | +| `--heartbeat-prompt` | `SPROUT_ACP_HEARTBEAT_PROMPT` | (built-in) | Custom heartbeat prompt text. Conflicts with `--heartbeat-prompt-file`. | +| `--heartbeat-prompt-file` | `SPROUT_ACP_HEARTBEAT_PROMPT_FILE` | — | Read heartbeat prompt from a file. Conflicts with `--heartbeat-prompt`. | + +### Configuration Examples + +**Single agent, no heartbeat (default — backward compatible):** +```bash +sprout-acp +``` + +**Four agents, no heartbeat (high-throughput event processing):** +```bash +sprout-acp --agents 4 +``` + +**Two agents with 5-minute heartbeat:** +```bash +sprout-acp --agents 2 --heartbeat-interval 300 +``` + +**Custom heartbeat prompt:** +```bash +sprout-acp --agents 2 --heartbeat-interval 300 \ + --heartbeat-prompt "Check get_feed_actions() for pending approvals, then get_feed_mentions() for unanswered mentions. If nothing actionable, end your turn immediately." +``` + +### Shared Identity + +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 + +When `--heartbeat-interval` is set, the harness fires a prompt on an idle agent at the configured interval. Heartbeat rules: + +- **Lower priority than queued events** — if events are pending, they are dispatched first. +- **Skipped when all agents are busy** — no queuing; the tick is simply dropped. +- **At most one heartbeat in flight globally** — the next tick is suppressed until the current one completes. +- **Default prompt** (when `--heartbeat-prompt` is not set) calls `get_feed_actions()` and `get_feed_mentions()` to surface pending work. + +Heartbeat is designed for idle periods. Under sustained event load it will rarely fire — that's expected. + +### Choosing N + +Start with **N=2** for most deployments. Increase if queue depth grows under load. Each agent spawns its own MCP server subprocess, so resource usage scales approximately as N × (agent memory + MCP server memory). Maximum is 32. + ## How It Works 1. **Startup** — Spawns the agent subprocess, sends ACP `initialize`, connects to the relay with NIP-42 auth. diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index bbc62260c0..b2844d4eb8 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -239,9 +239,12 @@ async fn main() -> Result<()> { } => { let _ = result_rx; if queue.has_flushable_work() { + tracing::debug!("heartbeat_skipped_events"); dispatch_pending(&mut pool, &mut queue, &ctx); } else if pool.any_idle() { dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); + } else { + tracing::debug!("heartbeat_skipped_busy"); } None } @@ -307,20 +310,25 @@ enum LoopAction { /// Flush queued work to available agents. fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc) { + let mut dispatched: usize = 0; loop { let batch = match queue.flush_next() { Some(b) => b, None => break, }; let channel_id = batch.channel_id; + let affinity_hit = pool.has_session_for(channel_id); let agent = match pool.try_claim(Some(channel_id)) { Some(a) => a, None => { + let pending = queue.pending_channels(); + tracing::warn!(pending_channels = pending, "pool_exhausted"); queue.requeue_preserve_timestamps(batch); queue.mark_complete(channel_id); break; } }; + tracing::debug!(agent = agent.index, channel = %channel_id, affinity_hit, "agent_claimed"); let prompt_text = queue::format_prompt(&batch, ctx.system_prompt.as_deref()); let recoverable_batch = match ctx.dedup_mode { @@ -344,7 +352,9 @@ fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc "ok", + PromptOutcome::Error(_) => "error", + PromptOutcome::Timeout => "timeout", + PromptOutcome::AgentExited => "exited", + }; + let agent_index = result.agent.index; + match result.outcome { PromptOutcome::AgentExited => { + tracing::debug!(agent = agent_index, outcome = outcome_label, "agent_returned"); let index = result.agent.index; match respawn_agent_into(result.agent, config).await { Ok(agent) => pool.return_agent(agent), @@ -385,6 +404,7 @@ async fn handle_prompt_result( } } _ => { + tracing::debug!(agent = agent_index, outcome = outcome_label, "agent_returned"); pool.return_agent(result.agent); } } @@ -492,6 +512,7 @@ fn dispatch_heartbeat( }, ); *heartbeat_in_flight = true; + tracing::info!(agent = agent_index, "heartbeat_fired"); } // ── default_heartbeat_prompt ────────────────────────────────────────────────── diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index 8e60b289cb..e6eb254afe 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -84,6 +84,7 @@ pub enum PromptSource { } /// Outcome of a prompt task. +#[allow(dead_code)] pub enum PromptOutcome { Ok(StopReason), Error(AcpError), @@ -162,6 +163,16 @@ impl AgentPool { self.agents.iter().any(|slot| slot.is_some()) } + /// Whether any idle agent already has a session for `channel_id`. + /// Used to compute `affinity_hit` before calling `try_claim`. + pub fn has_session_for(&self, channel_id: Uuid) -> bool { + self.agents.iter().any(|slot| { + slot.as_ref() + .map(|a| a.sessions.contains_key(&channel_id)) + .unwrap_or(false) + }) + } + /// Count of agents that are alive: idle OR checked out (have a task_map entry). /// /// Used to detect when all agents have exited so the caller can respawn. @@ -174,6 +185,7 @@ impl AgentPool { /// Wait for the next completed prompt result. /// /// Panics if the channel closes (impossible while pool is alive). + #[allow(dead_code)] pub async fn next_result(&mut self) -> PromptResult { self.result_rx .recv() @@ -195,6 +207,7 @@ impl AgentPool { self.result_tx.clone() } + #[allow(dead_code)] pub fn result_rx_mut(&mut self) -> &mut mpsc::UnboundedReceiver { &mut self.result_rx } From 2bacb96f1e93955d9d5ec8f97e39bd5cac71afd7 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 11:02:55 -0400 Subject: [PATCH 4/7] fix(sprout-acp): prevent 100% CPU spin when JoinSet is empty join_set.join_next() returns None immediately when the JoinSet has no tasks. Without a guard, the biased select loop spins at 100% CPU in the idle state (no in-flight prompts). Add is_empty() precondition so the branch is disabled when there are no tasks to join. --- crates/sprout-acp/src/main.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index b2844d4eb8..0cfe8225f1 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -191,10 +191,12 @@ async fn main() -> Result<()> { tokio::select! { biased; r = result_rx.recv() => Some(PoolEvent::Result(r.expect("result channel closed"))), - j = join_set.join_next() => match j { - Some(Err(e)) => Some(PoolEvent::Panic(e)), - _ => None, - }, + // Guard: join_next() returns None immediately when JoinSet is + // empty, which would cause a tight spin. Only poll when there + // are in-flight tasks. + Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => { + Some(PoolEvent::Panic(e)) + } // Remaining branches don't touch pool — evaluated when pool is idle. sprout_event = relay.next_event() => { let _ = result_rx; // end split borrow before relay handling From fe5612b6bc3a1307f0cbfba8d879dedf89df8a39 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 11:10:04 -0400 Subject: [PATCH 5/7] feat: add 'just goose' and 'just goose-bg' agent harness recipes Convenience recipes to launch a goose agent connected to a Sprout relay. Accepts relay URL, agent count, system prompt, private key, and API token. goose-bg runs in a detached screen session. --- justfile | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/justfile b/justfile index e9b7aec628..c3cedb5536 100644 --- a/justfile +++ b/justfile @@ -170,3 +170,42 @@ clean: # Check the Rust workspace compiles without producing binaries check-compile: cargo check --workspace --all-targets + +# ─── Agent Harness ──────────────────────────────────────────────────────────── + +# Run a goose agent connected to a Sprout relay (foreground) +goose relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": + #!/usr/bin/env bash + set -euo pipefail + cargo build --release -p sprout-acp -p sprout-mcp + env_args=( + SPROUT_RELAY_URL="{{relay}}" + SPROUT_PRIVATE_KEY="{{key}}" + SPROUT_ACP_AGENT_COMMAND=goose + SPROUT_ACP_AGENT_ARGS=acp + SPROUT_ACP_MCP_COMMAND=./target/release/sprout-mcp-server + SPROUT_ACP_AGENTS="{{agents}}" + GOOSE_MODE=auto + ) + [[ -n "{{token}}" ]] && env_args+=(SPROUT_ACP_API_TOKEN="{{token}}") + [[ -n "{{prompt}}" ]] && env_args+=(SPROUT_ACP_SYSTEM_PROMPT="{{prompt}}") + exec env "${env_args[@]}" ./target/release/sprout-acp + +# Run a goose agent in the background (screen session named 'goose-agent') +goose-bg relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": + #!/usr/bin/env bash + set -euo pipefail + cargo build --release -p sprout-acp -p sprout-mcp + env_args=( + SPROUT_RELAY_URL="{{relay}}" + SPROUT_PRIVATE_KEY="{{key}}" + SPROUT_ACP_AGENT_COMMAND=goose + SPROUT_ACP_AGENT_ARGS=acp + SPROUT_ACP_MCP_COMMAND=./target/release/sprout-mcp-server + SPROUT_ACP_AGENTS="{{agents}}" + GOOSE_MODE=auto + ) + [[ -n "{{token}}" ]] && env_args+=(SPROUT_ACP_API_TOKEN="{{token}}") + [[ -n "{{prompt}}" ]] && env_args+=(SPROUT_ACP_SYSTEM_PROMPT="{{prompt}}") + screen -dmS goose-agent bash -c "$(printf '%q ' env "${env_args[@]}") ./target/release/sprout-acp" + echo "Agent running in screen session 'goose-agent'. Attach with: screen -r goose-agent" From 2a3cd85fe94e68eea19d8e49c2e90b397d37bb47 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 11:27:36 -0400 Subject: [PATCH 6/7] fix: address crossfire R1 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README: update stale 'How It Works' for multi-agent semantics - pool.rs: debug_assert slot empty in return_agent - pool.rs: remove dead next_result/result_rx_mut methods - main.rs: pool_exhausted log warn→debug (normal under load) - justfile: unique screen session name, add heartbeat param --- crates/sprout-acp/README.md | 6 +++--- crates/sprout-acp/src/main.rs | 2 +- crates/sprout-acp/src/pool.rs | 23 ++++------------------- justfile | 16 +++++++++++----- 4 files changed, 19 insertions(+), 28 deletions(-) diff --git a/crates/sprout-acp/README.md b/crates/sprout-acp/README.md index 884f34ede1..4d28c9bd7d 100644 --- a/crates/sprout-acp/README.md +++ b/crates/sprout-acp/README.md @@ -163,14 +163,14 @@ Start with **N=2** for most deployments. Increase if queue depth grows under loa ## How It Works -1. **Startup** — Spawns the agent subprocess, sends ACP `initialize`, connects to the relay with NIP-42 auth. +1. **Startup** — Spawns N agent subprocesses (default 1), sends ACP `initialize` to each, connects to the relay with NIP-42 auth. 2. **Channel discovery** — Queries the relay REST API for accessible channels, subscribes to each. 3. **Event loop** — Listens for @mention events (kind 9 with the agent's pubkey in a `#p` tag). Events queue per channel. -4. **Prompting** — When events are pending and no prompt is in flight, drains all queued events for the oldest channel into a single batched prompt via ACP `session/prompt`. +4. **Prompting** — When events are pending and no prompt is in flight for that channel, drains all queued events for the oldest channel into a single batched prompt via ACP `session/prompt`. 5. **Agent response** — The agent processes the prompt and uses Sprout MCP tools (`send_message`, `get_channel_history`, etc.) to interact with Sprout. 6. **Recovery** — If the agent crashes, the harness respawns it. If the relay disconnects, the harness reconnects with a `since` filter to avoid missing events. -Only one prompt is in flight at a time (globally, not per-session). This matches the concurrency model of current ACP agents. +Each channel has at most one prompt in flight. Multiple channels can be processed concurrently when agents > 1. > **Note:** On startup, the harness replays all unprocessed @mentions since the last run. Expect a burst of activity if there are stale events in the channel. diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 0cfe8225f1..cfe2cce89b 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -324,7 +324,7 @@ fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc a, None => { let pending = queue.pending_channels(); - tracing::warn!(pending_channels = pending, "pool_exhausted"); + tracing::debug!(pending_channels = pending, "pool_exhausted"); queue.requeue_preserve_timestamps(batch); queue.mark_complete(channel_id); break; diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index e6eb254afe..711256fc26 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -13,7 +13,7 @@ //! try_claim() → OwnedAgent (removed from slot) //! spawn run_prompt_task(agent, ...) into join_set //! task sends PromptResult { agent, outcome } via result_tx -//! next_result() → PromptResult +//! rx_and_join_set() → poll result_rx for PromptResult //! return_agent(agent) → puts agent back in slot //! ``` //! @@ -154,8 +154,9 @@ impl AgentPool { /// Return an agent to its slot after a task completes. pub fn return_agent(&mut self, agent: OwnedAgent) { - let index = agent.index; - self.agents[index] = Some(agent); + let idx = agent.index; + debug_assert!(self.agents[idx].is_none(), "return_agent: slot {idx} already occupied"); + self.agents[idx] = Some(agent); } /// Whether any agent is currently idle (sitting in its slot). @@ -182,17 +183,6 @@ impl AgentPool { idle + checked_out } - /// Wait for the next completed prompt result. - /// - /// Panics if the channel closes (impossible while pool is alive). - #[allow(dead_code)] - pub async fn next_result(&mut self) -> PromptResult { - self.result_rx - .recv() - .await - .expect("result channel closed — pool invariant violated") - } - // ── Accessors ───────────────────────────────────────────────────────── pub fn task_map(&self) -> &HashMap { @@ -207,11 +197,6 @@ impl AgentPool { self.result_tx.clone() } - #[allow(dead_code)] - pub fn result_rx_mut(&mut self) -> &mut mpsc::UnboundedReceiver { - &mut self.result_rx - } - /// Split-borrow: returns mutable refs to `result_rx` and `join_set` /// simultaneously. This lets callers poll both in a single `select!` /// without a double-borrow error on `&mut AgentPool`. diff --git a/justfile b/justfile index c3cedb5536..62075c6758 100644 --- a/justfile +++ b/justfile @@ -174,7 +174,7 @@ check-compile: # ─── Agent Harness ──────────────────────────────────────────────────────────── # Run a goose agent connected to a Sprout relay (foreground) -goose relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": +goose relay="ws://localhost:3000" agents="1" heartbeat="0" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": #!/usr/bin/env bash set -euo pipefail cargo build --release -p sprout-acp -p sprout-mcp @@ -189,10 +189,13 @@ goose relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_KEY" ) [[ -n "{{token}}" ]] && env_args+=(SPROUT_ACP_API_TOKEN="{{token}}") [[ -n "{{prompt}}" ]] && env_args+=(SPROUT_ACP_SYSTEM_PROMPT="{{prompt}}") + if [[ "{{heartbeat}}" != "0" ]]; then + env_args+=(SPROUT_ACP_HEARTBEAT_INTERVAL={{heartbeat}}) + fi exec env "${env_args[@]}" ./target/release/sprout-acp -# Run a goose agent in the background (screen session named 'goose-agent') -goose-bg relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": +# Run a goose agent in the background (screen session named 'goose-agent-N') +goose-bg relay="ws://localhost:3000" agents="1" heartbeat="0" prompt="" key="$SPROUT_PRIVATE_KEY" token="$SPROUT_ACP_API_TOKEN": #!/usr/bin/env bash set -euo pipefail cargo build --release -p sprout-acp -p sprout-mcp @@ -207,5 +210,8 @@ goose-bg relay="ws://localhost:3000" agents="1" prompt="" key="$SPROUT_PRIVATE_K ) [[ -n "{{token}}" ]] && env_args+=(SPROUT_ACP_API_TOKEN="{{token}}") [[ -n "{{prompt}}" ]] && env_args+=(SPROUT_ACP_SYSTEM_PROMPT="{{prompt}}") - screen -dmS goose-agent bash -c "$(printf '%q ' env "${env_args[@]}") ./target/release/sprout-acp" - echo "Agent running in screen session 'goose-agent'. Attach with: screen -r goose-agent" + if [[ "{{heartbeat}}" != "0" ]]; then + env_args+=(SPROUT_ACP_HEARTBEAT_INTERVAL={{heartbeat}}) + fi + screen -dmS goose-agent-{{agents}} bash -c "$(printf '%q ' env "${env_args[@]}") ./target/release/sprout-acp" + echo "Agent running in screen session 'goose-agent-{{agents}}'. Attach with: screen -r goose-agent-{{agents}}" From 7e977de4a824c0331907f36c023a9ec5373c2d09 Mon Sep 17 00:00:00 2001 From: Tyler Longwell Date: Sat, 14 Mar 2026 18:14:39 -0400 Subject: [PATCH 7/7] fix: rustfmt + clippy -D warnings (map_or, Box enum variant, contains_key, dead_code) --- crates/sprout-acp/src/acp.rs | 1 + crates/sprout-acp/src/config.rs | 34 +++++++++++++---- crates/sprout-acp/src/main.rs | 65 ++++++++++++++++++++++++++------- crates/sprout-acp/src/pool.rs | 53 ++++++++++++++++----------- crates/sprout-acp/src/queue.rs | 26 +++++++++---- 5 files changed, 127 insertions(+), 52 deletions(-) diff --git a/crates/sprout-acp/src/acp.rs b/crates/sprout-acp/src/acp.rs index 0da61d3cc5..774b7abd4f 100644 --- a/crates/sprout-acp/src/acp.rs +++ b/crates/sprout-acp/src/acp.rs @@ -76,6 +76,7 @@ pub enum AcpError { #[error("Agent process exited unexpectedly")] AgentExited, + #[allow(dead_code)] #[error("Turn timed out")] Timeout, diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index d8e6392c0c..0046917851 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -104,13 +104,19 @@ pub struct CliArgs { pub heartbeat_interval: u64, /// Heartbeat prompt text. Conflicts with --heartbeat-prompt-file. - #[arg(long, env = "SPROUT_ACP_HEARTBEAT_PROMPT", - conflicts_with = "heartbeat_prompt_file")] + #[arg( + long, + env = "SPROUT_ACP_HEARTBEAT_PROMPT", + conflicts_with = "heartbeat_prompt_file" + )] pub heartbeat_prompt: Option, /// Read heartbeat prompt from file. - #[arg(long, env = "SPROUT_ACP_HEARTBEAT_PROMPT_FILE", - conflicts_with = "heartbeat_prompt")] + #[arg( + long, + env = "SPROUT_ACP_HEARTBEAT_PROMPT_FILE", + conflicts_with = "heartbeat_prompt" + )] pub heartbeat_prompt_file: Option, #[arg(long, env = "SPROUT_ACP_INITIAL_MESSAGE")] @@ -927,8 +933,14 @@ channels = "ALL" fn test_summary_includes_agents_and_heartbeat() { let config = test_config(SubscribeMode::Mentions); let s = config.summary(); - assert!(s.contains("agents=1"), "summary should include agents=1, got: {s}"); - assert!(s.contains("heartbeat=0s"), "summary should include heartbeat=0s, got: {s}"); + assert!( + s.contains("agents=1"), + "summary should include agents=1, got: {s}" + ); + assert!( + s.contains("heartbeat=0s"), + "summary should include heartbeat=0s, got: {s}" + ); } #[test] @@ -937,7 +949,13 @@ channels = "ALL" config.agents = 4; config.heartbeat_interval_secs = 30; let s = config.summary(); - assert!(s.contains("agents=4"), "summary should include agents=4, got: {s}"); - assert!(s.contains("heartbeat=30s"), "summary should include heartbeat=30s, got: {s}"); + assert!( + s.contains("agents=4"), + "summary should include agents=4, got: {s}" + ); + assert!( + s.contains("heartbeat=30s"), + "summary should include heartbeat=30s, got: {s}" + ); } } diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index cfe2cce89b..672adf6ce3 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -11,20 +11,20 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use anyhow::Result; -use futures_util::FutureExt; -use nostr::ToBech32; -use tokio::sync::watch; -use tracing_subscriber::EnvFilter; use acp::{AcpClient, EnvVar, McpServer}; +use anyhow::Result; use config::{Config, DedupMode, SubscribeMode}; use filter::SubscriptionRule; +use futures_util::FutureExt; +use nostr::ToBech32; use pool::{AgentPool, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource}; use queue::{EventQueue, QueuedEvent}; use relay::HarnessRelay; use sprout_core::kind::{ KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, }; +use tokio::sync::watch; +use tracing_subscriber::EnvFilter; #[tokio::main] async fn main() -> Result<()> { @@ -180,7 +180,7 @@ async fn main() -> Result<()> { // fields (result_rx vs join_set). We use `rx_and_join_set()` to split the // borrow, yielding a typed enum so the outer code can dispatch cleanly. enum PoolEvent { - Result(PromptResult), + Result(Box), Panic(tokio::task::JoinError), } @@ -190,7 +190,7 @@ async fn main() -> Result<()> { let (result_rx, join_set) = pool.rx_and_join_set(); tokio::select! { biased; - r = result_rx.recv() => Some(PoolEvent::Result(r.expect("result channel closed"))), + r = result_rx.recv() => Some(PoolEvent::Result(Box::new(r.expect("result channel closed")))), // Guard: join_next() returns None immediately when JoinSet is // empty, which would cause a tight spin. Only poll when there // are in-flight tasks. @@ -259,17 +259,41 @@ async fn main() -> Result<()> { match pool_event { Some(PoolEvent::Result(result)) => { - if handle_prompt_result(&mut pool, &mut queue, &config, result, &mut heartbeat_in_flight).await == LoopAction::Exit { + if handle_prompt_result( + &mut pool, + &mut queue, + &config, + *result, + &mut heartbeat_in_flight, + ) + .await + == LoopAction::Exit + { break; } - if drain_ready_join_results(&mut pool, &mut queue, &config, &mut heartbeat_in_flight).await == LoopAction::Exit { + if drain_ready_join_results( + &mut pool, + &mut queue, + &config, + &mut heartbeat_in_flight, + ) + .await + == LoopAction::Exit + { break; } dispatch_pending(&mut pool, &mut queue, &ctx); } Some(PoolEvent::Panic(join_error)) => { tracing::error!("agent task panicked: {join_error}"); - recover_panicked_agent(&mut pool, &mut queue, &config, join_error, &mut heartbeat_in_flight).await; + recover_panicked_agent( + &mut pool, + &mut queue, + &config, + join_error, + &mut heartbeat_in_flight, + ) + .await; if pool.live_count() == 0 { tracing::error!("all agents dead — exiting"); break; @@ -356,7 +380,11 @@ fn dispatch_pending(pool: &mut AgentPool, queue: &mut EventQueue, ctx: &Arc LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; - pool.task_map_mut().retain(|_, meta| meta.agent_index != agent_index); + pool.task_map_mut() + .retain(|_, meta| meta.agent_index != agent_index); debug_assert_eq!(before, pool.task_map().len() + 1); match &result.source { @@ -392,7 +421,11 @@ async fn handle_prompt_result( match result.outcome { PromptOutcome::AgentExited => { - tracing::debug!(agent = agent_index, outcome = outcome_label, "agent_returned"); + tracing::debug!( + agent = agent_index, + outcome = outcome_label, + "agent_returned" + ); let index = result.agent.index; match respawn_agent_into(result.agent, config).await { Ok(agent) => pool.return_agent(agent), @@ -406,7 +439,11 @@ async fn handle_prompt_result( } } _ => { - tracing::debug!(agent = agent_index, outcome = outcome_label, "agent_returned"); + tracing::debug!( + agent = agent_index, + outcome = outcome_label, + "agent_returned" + ); pool.return_agent(result.agent); } } diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs index 711256fc26..ca6aadf75b 100644 --- a/crates/sprout-acp/src/pool.rs +++ b/crates/sprout-acp/src/pool.rs @@ -134,14 +134,11 @@ impl AgentPool { pub fn try_claim(&mut self, channel_id: Option) -> Option { // Pass 1: prefer agent with existing session for this channel. if let Some(cid) = channel_id { - let idx = self - .agents - .iter() - .position(|slot| { - slot.as_ref() - .map(|a| a.sessions.contains_key(&cid)) - .unwrap_or(false) - }); + let idx = self.agents.iter().position(|slot| { + slot.as_ref() + .map(|a| a.sessions.contains_key(&cid)) + .unwrap_or(false) + }); if let Some(i) = idx { return self.agents[i].take(); } @@ -155,7 +152,10 @@ impl AgentPool { /// Return an agent to its slot after a task completes. pub fn return_agent(&mut self, agent: OwnedAgent) { let idx = agent.index; - debug_assert!(self.agents[idx].is_none(), "return_agent: slot {idx} already occupied"); + debug_assert!( + self.agents[idx].is_none(), + "return_agent: slot {idx} already occupied" + ); self.agents[idx] = Some(agent); } @@ -202,10 +202,7 @@ impl AgentPool { /// without a double-borrow error on `&mut AgentPool`. pub fn rx_and_join_set( &mut self, - ) -> ( - &mut mpsc::UnboundedReceiver, - &mut JoinSet<()>, - ) { + ) -> (&mut mpsc::UnboundedReceiver, &mut JoinSet<()>) { (&mut self.result_rx, &mut self.join_set) } @@ -247,7 +244,11 @@ pub async fn run_prompt_task( (sid.clone(), false) } else { // Create new session. - match agent.acp.session_new(&ctx.cwd, ctx.mcp_servers.clone()).await { + match agent + .acp + .session_new(&ctx.cwd, ctx.mcp_servers.clone()) + .await + { Ok(sid) => { tracing::info!( target: "pool::session", @@ -283,7 +284,11 @@ pub async fn run_prompt_task( if let Some(sid) = &agent.heartbeat_session { (sid.clone(), false) } else { - match agent.acp.session_new(&ctx.cwd, ctx.mcp_servers.clone()).await { + match agent + .acp + .session_new(&ctx.cwd, ctx.mcp_servers.clone()) + .await + { Ok(sid) => { tracing::info!( target: "pool::session", @@ -321,16 +326,17 @@ pub async fn run_prompt_task( // ── Send initial_message on new channel sessions ────────────────────── if is_new_session { - if let (PromptSource::Channel(cid), Some(ref initial_msg)) = - (&source, &ctx.initial_message) + if let (PromptSource::Channel(cid), Some(ref initial_msg)) = (&source, &ctx.initial_message) { tracing::info!( target: "pool::session", "sending initial_message to session {session_id} for channel {cid}" ); - let init_result = - timeout(ctx.turn_timeout, agent.acp.session_prompt(&session_id, initial_msg)) - .await; + let init_result = timeout( + ctx.turn_timeout, + agent.acp.session_prompt(&session_id, initial_msg), + ) + .await; match init_result { Ok(Ok(stop_reason)) => { @@ -406,8 +412,11 @@ pub async fn run_prompt_task( // ── Send the actual prompt ──────────────────────────────────────────── - let prompt_result = - timeout(ctx.turn_timeout, agent.acp.session_prompt(&session_id, &prompt_text)).await; + let prompt_result = timeout( + ctx.turn_timeout, + agent.acp.session_prompt(&session_id, &prompt_text), + ) + .await; match prompt_result { Ok(Ok(stop_reason)) => { diff --git a/crates/sprout-acp/src/queue.rs b/crates/sprout-acp/src/queue.rs index 1cf6e71ad3..50414a9df9 100644 --- a/crates/sprout-acp/src/queue.rs +++ b/crates/sprout-acp/src/queue.rs @@ -136,7 +136,7 @@ impl EventQueue { .filter(|(id, q)| { !q.is_empty() && !self.in_flight_channels.contains(id) - && self.retry_after.get(id).map_or(true, |&t| t <= now) + && self.retry_after.get(id).is_none_or(|&t| t <= now) }) .min_by_key(|(_, q)| q.front().unwrap().received_at) .map(|(id, _)| *id)?; @@ -220,7 +220,7 @@ impl EventQueue { self.queues.iter().any(|(id, q)| { !q.is_empty() && !self.in_flight_channels.contains(id) - && self.retry_after.get(id).map_or(true, |&t| t <= now) + && self.retry_after.get(id).is_none_or(|&t| t <= now) }) } @@ -923,7 +923,7 @@ mod tests { q.mark_complete(ch); // No retry_after — channel should be immediately flushable. - assert!(q.retry_after.get(&ch).is_none()); + assert!(!q.retry_after.contains_key(&ch)); assert!(q.flush_next().is_some()); } @@ -953,11 +953,18 @@ mod tests { let batch2 = q.flush_next().expect("flush2"); q.requeue(batch2); q.mark_complete(ch); - assert!(!q.has_flushable_work(), "throttled channel should not be flushable"); + assert!( + !q.has_flushable_work(), + "throttled channel should not be flushable" + ); // Manually expire the retry_after to simulate time passing. - q.retry_after.insert(ch, Instant::now() - Duration::from_secs(1)); - assert!(q.has_flushable_work(), "expired throttle should be flushable"); + q.retry_after + .insert(ch, Instant::now() - Duration::from_secs(1)); + assert!( + q.has_flushable_work(), + "expired throttle should be flushable" + ); } // ── Test 22: retry throttle blocks re-flush for 5 seconds ───────────────── @@ -984,9 +991,12 @@ mod tests { assert_eq!(batch2.channel_id, ch2); // After retry_after expires, ch should be flushable again. - q.retry_after.insert(ch, Instant::now() - Duration::from_secs(1)); + q.retry_after + .insert(ch, Instant::now() - Duration::from_secs(1)); q.mark_complete(ch2); - let batch3 = q.flush_next().expect("ch should be flushable after throttle expires"); + let batch3 = q + .flush_next() + .expect("ch should be flushable after throttle expires"); assert_eq!(batch3.channel_id, ch); } }