feat(acp): make event queue max-retries configurable via BUZZ_ACP_QUEUE_MAX_RETRIES

crates/buzz-acp/src/config.rs

@@ -373,6 +373,12 @@ pub struct CliArgs {
           value_parser = clap::value_parser!(u32))]
     pub max_turns_per_session: u32,
 
+    /// Maximum retry attempts before a stuck event batch is dead-lettered.
+    #[arg(long, env = "BUZZ_ACP_QUEUE_MAX_RETRIES",
+          default_value_t = crate::queue::MAX_RETRIES,
+          value_parser = clap::value_parser!(u32).range(1..))]
+    pub queue_max_retries: u32,
+
     /// Disable automatic presence (online/offline) status.
     #[arg(long, env = "BUZZ_ACP_NO_PRESENCE")]
     pub no_presence: bool,
@@ -531,6 +537,9 @@ pub struct Config {
     pub context_message_limit: u32,
     /// Maximum turns per session before proactive rotation. 0 = disabled.
     pub max_turns_per_session: u32,
+    /// Maximum retry attempts before a stuck event batch is dead-lettered.
+    /// Defaults to [`crate::queue::MAX_RETRIES`] (10).
+    pub queue_max_retries: u32,
     pub presence_enabled: bool,
     pub typing_enabled: bool,
     /// Whether NIP-AE agent core memory injection is enabled. When false,
@@ -1101,6 +1110,7 @@ impl Config {
             config_path: args.config,
             context_message_limit: args.context_message_limit,
             max_turns_per_session: args.max_turns_per_session,
+            queue_max_retries: args.queue_max_retries,
             presence_enabled: !args.no_presence,
             typing_enabled: !args.no_typing,
             memory_enabled: args.memory && !args.no_memory,
@@ -1143,7 +1153,7 @@ impl Config {
             format!(" allowed_respond_to=[{}]", modes.join(","))
         };
         format!(
-            "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}",
+            "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} queue_max_retries={} presence={} typing={} memory={} model={} permission_mode={} {}{}",
             self.relay_url,
             self.keys.public_key().to_hex(),
             self.agent_command,
@@ -1159,6 +1169,7 @@ impl Config {
             self.ignore_self,
             self.context_message_limit,
             self.max_turns_per_session,
+            self.queue_max_retries,
             self.presence_enabled,
             self.typing_enabled,
             self.memory_enabled,
@@ -1476,6 +1487,7 @@ mod tests {
             config_path: PathBuf::from("./buzz-acp.toml"),
             context_message_limit: 12,
             max_turns_per_session: 0,
+            queue_max_retries: crate::queue::MAX_RETRIES,
             presence_enabled: true,
             typing_enabled: true,
             memory_enabled: true,

crates/buzz-acp/src/lib.rs

@@ -1984,8 +1984,9 @@ async fn tokio_main() -> Result<()> {
 
     let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default();
     let dedup_mode = config.dedup_mode;
-    let mut queue =
-        EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs);
+    let mut queue = EventQueue::new(dedup_mode)
+        .with_in_flight_deadline(config.max_turn_duration_secs)
+        .with_max_retries(config.queue_max_retries);
 
     // Online means the harness can receive work, not merely that its socket is
     // connected. Publishing after channel subscriptions gives desktop callers
@@ -6520,6 +6521,7 @@ mod build_mcp_servers_tests {
             config_path: std::path::PathBuf::from("./buzz-acp.toml"),
             context_message_limit: 12,
             max_turns_per_session: 0,
+            queue_max_retries: crate::queue::MAX_RETRIES,
             presence_enabled: true,
             typing_enabled: true,
             memory_enabled: false,
@@ -6743,6 +6745,7 @@ mod error_outcome_emission_tests {
             config_path: std::path::PathBuf::from("./buzz-acp.toml"),
             context_message_limit: 12,
             max_turns_per_session: 0,
+            queue_max_retries: crate::queue::MAX_RETRIES,
             presence_enabled: true,
             typing_enabled: true,
             memory_enabled: false,

crates/buzz-acp/src/queue.rs

@@ -26,7 +26,9 @@ const MAX_PENDING_PER_CHANNEL: usize = 500;
 /// Maximum events drained into a single batch.
 const MAX_BATCH_EVENTS: usize = 50;
 
-/// Maximum retry attempts before a batch is dead-lettered.
+/// Default maximum retry attempts before a batch is dead-lettered.
+/// Configurable via `--queue-max-retries` / `BUZZ_ACP_QUEUE_MAX_RETRIES`
+/// ([`Config::queue_max_retries`](crate::config::Config::queue_max_retries)).
 pub(crate) const MAX_RETRIES: u32 = 10;
 
 /// Base retry delay in seconds (doubled each attempt).
@@ -168,6 +170,10 @@ pub struct EventQueue {
     /// Must be strictly greater than `max_turn_duration` so a turn running to
     /// the hard cap returns via `mark_complete` before the backstop fires.
     in_flight_deadline: Duration,
+    /// Maximum retry attempts before a batch is dead-lettered. Defaults to
+    /// [`MAX_RETRIES`]; overridden via
+    /// [`with_max_retries`](Self::with_max_retries).
+    max_retries: u32,
 }
 
 impl EventQueue {
@@ -189,6 +195,7 @@ impl EventQueue {
             cancel_reasons: HashMap::new(),
             withheld_native_steer: HashMap::new(),
             in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS),
+            max_retries: MAX_RETRIES,
         }
     }
 
@@ -200,6 +207,13 @@ impl EventQueue {
         self
     }
 
+    /// Override the maximum retry attempts before a batch is dead-lettered.
+    /// Defaults to [`MAX_RETRIES`] (10) when not called.
+    pub fn with_max_retries(mut self, max_retries: u32) -> Self {
+        self.max_retries = max_retries;
+        self
+    }
+
     /// Monotonically extend an existing in-flight deadline for `channel_id`.
     ///
     /// Called when a successful steer grants a fresh turn budget. The new
@@ -434,13 +448,13 @@ impl EventQueue {
             *count
         };
 
-        if attempt > MAX_RETRIES {
+        if attempt > self.max_retries {
             tracing::error!(
                 channel_id = %channel_id,
                 attempt,
                 events = batch.events.len(),
                 "dead-lettering batch after {} retries — discarding {} events",
-                MAX_RETRIES,
+                self.max_retries,
                 batch.events.len(),
             );
             self.retry_counts.remove(&channel_id);
@@ -466,7 +480,7 @@ impl EventQueue {
         tracing::warn!(
             channel_id = %channel_id,
             attempt,
-            max = MAX_RETRIES,
+            max = self.max_retries,
             delay_secs = delay.as_secs_f64(),
             events = batch.events.len(),
             "requeueing failed batch with backoff"
@@ -3121,6 +3135,34 @@ mod tests {
         assert!(!q.retry_after.contains_key(&ch));
     }
 
+    #[test]
+    fn test_requeue_dead_letters_after_custom_max_retries() {
+        let custom_max_retries = 2;
+        let mut q = EventQueue::new(DedupMode::Queue).with_max_retries(custom_max_retries);
+        let ch = Uuid::new_v4();
+
+        q.push(make_queued(ch, "poison"));
+        for attempt in 1..=custom_max_retries {
+            q.retry_after
+                .insert(ch, Instant::now() - Duration::from_secs(1));
+            let batch = q.flush_next().expect("flush");
+            assert!(
+                q.requeue(batch).is_none(),
+                "attempt {attempt} should requeue, not dead-letter"
+            );
+            q.mark_complete(ch);
+        }
+
+        // The (custom_max_retries + 1)'th failure dead-letters: batch is returned.
+        q.retry_after
+            .insert(ch, Instant::now() - Duration::from_secs(1));
+        let batch = q.flush_next().expect("flush");
+        let dead = q
+            .requeue(batch)
+            .expect("should dead-letter at custom threshold");
+        assert_eq!(dead.channel_id, ch);
+    }
+
     #[test]
     fn test_retry_throttle_blocks_requeue_channel() {
         let mut q = EventQueue::new(DedupMode::Queue);