diff --git a/crates/sprout-acp/README.md b/crates/sprout-acp/README.md
index 65f2f9f2f2..4d28c9bd7d 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,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.
 
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 71a0c71fd9..0046917851 100644
--- a/crates/sprout-acp/src/config.rs
+++ b/crates/sprout-acp/src/config.rs
@@ -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}"
+        );
+    }
 }
diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs
index 6d5b4eac05..672adf6ce3 100644
--- a/crates/sprout-acp/src/main.rs
+++ b/crates/sprout-acp/src/main.rs
@@ -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);
-
-    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();
-                            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()
-                    );
+    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(Box<PromptResult>),
+        Panic(tokio::task::JoinError),
+    }
 
-                    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 */ }
+    loop {
+        // 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))
+                }
+                // 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();
-                            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
+                }
+                _ = 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;
                 }
             }
-            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<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}");
-        }
-    }
-}
diff --git a/crates/sprout-acp/src/pool.rs b/crates/sprout-acp/src/pool.rs
new file mode 100644
index 0000000000..ca6aadf75b
--- /dev/null
+++ b/crates/sprout-acp/src/pool.rs
@@ -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}");
+        }
+    }
+}
diff --git a/crates/sprout-acp/src/queue.rs b/crates/sprout-acp/src/queue.rs
index 3ead39def4..50414a9df9 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<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));
+    }
+
+    /// 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 a prompt is currently in flight.
+    /// 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);
+    }
 }
diff --git a/justfile b/justfile
index e9b7aec628..62075c6758 100644
--- a/justfile
+++ b/justfile
@@ -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}}"