jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(sprout-acp): parallel agent pool, heartbeat timer, and just goose recipes (#64)

Browse Source
This commit is contained in:
tlongwell-block
2026-03-14 18:19:42 -04:00
committed by GitHub
parent 124047ef88
commit 3c68325c67
7 changed files with 1609 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
crates/sprout-acp/README.md
View File

@@ -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,16 +110,67 @@ 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.
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.

1
crates/sprout-acp/src/acp.rs
View File

@@ -76,6 +76,7 @@ pub enum AcpError {
#[error("Agent process exited unexpectedly")]
AgentExited,
#[allow(dead_code)]
#[error("Turn timed out")]
Timeout,

128
crates/sprout-acp/src/config.rs
View File

@@ -94,6 +94,31 @@ pub struct CliArgs {
)]
pub system_prompt_file: Option<PathBuf>,
/// 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<String>,
/// Read heartbeat prompt from file.
#[arg(
long,
env = "SPROUT_ACP_HEARTBEAT_PROMPT_FILE",
conflicts_with = "heartbeat_prompt"
)]
pub heartbeat_prompt_file: Option<PathBuf>,
#[arg(long, env = "SPROUT_ACP_INITIAL_MESSAGE")]
pub initial_message: Option<String>,
@@ -146,6 +171,9 @@ pub struct Config {
pub agent_args: Vec<String>,
pub mcp_command: String,
pub turn_timeout_secs: u64,
pub agents: u32,
pub heartbeat_interval_secs: u64,
pub heartbeat_prompt: Option<String>,
pub system_prompt: Option<String>,
pub initial_message: Option<String>,
pub subscribe_mode: SubscribeMode,
@@ -187,6 +215,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 +249,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 +267,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 +471,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,
@@ -832,4 +882,80 @@ 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}"
);
}
}

742
crates/sprout-acp/src/main.rs
View File

@@ -3,26 +3,28 @@
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 acp::{AcpClient, EnvVar, McpServer};
use anyhow::Result;
use nostr::ToBech32;
use tokio::time::timeout;
use tracing_subscriber::EnvFilter;
use uuid::Uuid;
use acp::{AcpClient, AcpError, EnvVar, McpServer, StopReason};
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<()> {
@@ -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<SubscriptionRule> = match config.subscribe_mode {
SubscribeMode::Mentions => {
vec![SubscriptionRule {
@@ -113,207 +125,470 @@ async fn main() -> Result<()> {
}
}
// ── Step 5: Main orchestration loop ──────────────────────────────────────
let mut sessions: HashMap<Uuid, String> = 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);
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(());
});
}
// ── 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(Box<PromptResult>),
Panic(tokio::task::JoinError),
}
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;
// Borrow result_rx and join_set simultaneously via split-borrow helper.
let pool_event: Option<PoolEvent> = {
let (result_rx, join_set) = pool.rx_and_join_set();
tokio::select! {
biased;
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.
Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => {
Some(PoolEvent::Panic(e))
}
// 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 */ }
// 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();
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();
break;
}
};
tracing::info!(
"prompting agent for channel {channel_id} (session {session_id}, {} event(s))",
batch.events.len()
);
// Send prompt with turn timeout.
let prompt_result =
timeout(turn_timeout, acp.session_prompt(&session_id, &prompt_text)).await;
match prompt_result {
Ok(Ok(stop_reason)) => {
log_stop_reason(channel_id, &stop_reason);
queue.mark_complete();
}
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();
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 */ }
}
queue.mark_complete();
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();
break;
}
}
None
}
}
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;
_ = async {
match heartbeat.as_mut() {
Some(hb) => hb.tick().await,
None => std::future::pending().await,
}
} => {
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
}
_ = shutdown_rx.changed() => {
tracing::info!("shutting down");
break;
}
}
};
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<PromptContext>) {
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::debug!(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 {
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,
},
);
dispatched += 1;
}
tracing::debug!(
dispatched,
queue_depth = queue.pending_channels(),
"dispatch_pending"
);
}
// ── 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);
}
let outcome_label = match &result.outcome {
PromptOutcome::Ok(_) => "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),
Err(e) => {
tracing::error!("failed to respawn agent {index}: {e}");
if pool.live_count() == 0 {
tracing::error!("all agents dead — exiting");
return LoopAction::Exit;
}
}
}
}
_ => {
tracing::debug!(
agent = agent_index,
outcome = outcome_label,
"agent_returned"
);
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<AcpClient> {
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<PromptContext>,
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;
tracing::info!(agent = agent_index, "heartbeat_fired");
}
// ── 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<OwnedAgent> {
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<AcpClient> {
let mut acp = AcpClient::spawn(&config.agent_command, &config.agent_args)
@@ -329,88 +604,7 @@ async fn spawn_and_init(config: &Config) -> Result<AcpClient> {
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<Uuid, String>,
channel_id: Uuid,
acp: &mut AcpClient,
mcp_servers: &[McpServer],
config: &Config,
) -> Result<String, AcpError> {
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<McpServer> {
vec![McpServer {
@@ -442,25 +636,3 @@ fn build_mcp_servers(config: &Config) -> Vec<McpServer> {
},
}]
}
// ── 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}");
}
}
}

551
crates/sprout-acp/src/pool.rs Normal file
View File

@@ -0,0 +1,551 @@
//! Agent pool — owns N AcpClient instances and dispatches prompt tasks.
//!
//! # Mental model
//!
//! ```text
//! AgentPool
//! ├── agents: Vec<Option<OwnedAgent>> ← idle agents sit here
//! ├── join_set: JoinSet<()> ← in-flight tasks
//! ├── task_map: HashMap<Id, TaskMeta> ← 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
//! rx_and_join_set() → poll result_rx for 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.
pub struct TaskMeta {
pub agent_index: usize,
pub channel_id: Option<Uuid>,
/// Clone of batch for Queue mode panic recovery.
pub recoverable_batch: Option<FlushBatch>,
}
/// 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<Uuid, String>,
pub heartbeat_session: Option<String>,
}
/// 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<Option<OwnedAgent>>,
result_tx: mpsc::UnboundedSender<PromptResult>,
result_rx: mpsc::UnboundedReceiver<PromptResult>,
pub join_set: JoinSet<()>,
task_map: HashMap<tokio::task::Id, TaskMeta>,
}
/// 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<FlushBatch>,
}
/// Whether the prompt came from a channel event or a heartbeat.
pub enum PromptSource {
Channel(Uuid),
Heartbeat,
}
/// Outcome of a prompt task.
#[allow(dead_code)]
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<McpServer>,
pub initial_message: Option<String>,
pub turn_timeout: Duration,
pub dedup_mode: DedupMode,
pub system_prompt: Option<String>,
pub heartbeat_prompt: Option<String>,
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<OwnedAgent>) -> 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<Uuid>) -> Option<OwnedAgent> {
// 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) {
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).
pub fn any_idle(&self) -> bool {
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.
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
}
// ── Accessors ─────────────────────────────────────────────────────────
pub fn task_map(&self) -> &HashMap<tokio::task::Id, TaskMeta> {
&self.task_map
}
pub fn task_map_mut(&mut self) -> &mut HashMap<tokio::task::Id, TaskMeta> {
&mut self.task_map
}
pub fn result_tx(&self) -> mpsc::UnboundedSender<PromptResult> {
self.result_tx.clone()
}
/// 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<PromptResult>, &mut JoinSet<()>) {
(&mut self.result_rx, &mut self.join_set)
}
pub fn agents_mut(&mut self) -> &mut Vec<Option<OwnedAgent>> {
&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<FlushBatch>,
prompt_text: String,
ctx: Arc<PromptContext>,
result_tx: mpsc::UnboundedSender<PromptResult>,
) {
// ── 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<FlushBatch>) -> Option<FlushBatch> {
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}");
}
}
}

420
crates/sprout-acp/src/queue.rs
View File

@@ -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<BatchEvent>,
@@ -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<channel_id, VecDeque<QueuedEvent>>
/// in_flight_channel: Option<Uuid>
/// in_flight_channels: HashSet<Uuid>
/// retry_after: Map<channel_id, Instant>
/// 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<FlushBatch>:
/// 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<Uuid, VecDeque<QueuedEvent>>,
in_flight_channel: Option<Uuid>,
in_flight_channels: HashSet<Uuid>,
retry_after: HashMap<Uuid, Instant>,
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<FlushBatch> {
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).is_none_or(|&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).is_none_or(|&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,269 @@ 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.contains_key(&ch));
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);
}
}

45
justfile
View File

@@ -170,3 +170,48 @@ 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" 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
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}}")
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-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
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}}")
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}}"