From d443b724a10ed91aa3801a380882d1dfdbe49842 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 13 Jul 2026 22:34:36 -0400 Subject: [PATCH 1/4] fix(buzz-acp): distinguish idle vs hard-cap timeout, dead-letter hard kills immediately MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hard-cap timeouts (wall-clock limit) were mislabeled as idle-timeout inactivity and then retried up to 10× — each cycle re-running the same >1h task from a fresh session before dying again, producing ~10h of silent churn. Three coupled fixes: - Plumb `TimeoutKind { Idle, Hard }` through `PromptOutcome::Timeout` so every downstream site can tell which clock fired. Split the combined cancel-drain `IdleTimeout | HardTimeout` arm to carry the correct kind through. - Dead-letter hard-cap kills immediately in `handle_prompt_result` without calling `queue.requeue()`. Hard-cap failure is deterministic; a fresh session will reproduce the same death. Idle timeout keeps its current retry behavior (may be a transient provider stall). Observer `outcome_label` now emits `"idle_timeout"` or `"hard_timeout"` instead of `"timeout"` for observability. - Raise `max_turn_duration` default from 3600s to 7200s so genuinely long turns (heavy implementation work) have headroom. Idle stays at 900s, preserving the `idle < max_turn` invariant and the real hang detector. Adds unit tests verifying hard timeout is NOT requeued and idle IS, and that `outcome_label` differs between the two kinds. --- crates/buzz-acp/src/config.rs | 10 +- crates/buzz-acp/src/lib.rs | 249 +++++++++++++++++++++++++++++----- crates/buzz-acp/src/pool.rs | 52 +++++-- 3 files changed, 263 insertions(+), 48 deletions(-) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index b94468ea0..78eb8810d 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -220,7 +220,7 @@ pub struct CliArgs { pub idle_timeout: Option, /// Absolute wall-clock cap per turn (safety valve). - #[arg(long, env = "BUZZ_ACP_MAX_TURN_DURATION", default_value = "3600")] + #[arg(long, env = "BUZZ_ACP_MAX_TURN_DURATION", default_value = "7200")] pub max_turn_duration: u64, /// Deprecated: alias for --idle-timeout. If both set, --idle-timeout wins. @@ -1326,7 +1326,7 @@ mod tests { agent_args: vec!["acp".into()], mcp_command: "".into(), idle_timeout_secs: DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 3600, + max_turn_duration_secs: 7200, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -2231,7 +2231,7 @@ channels = "ALL" "summary should include {expected_idle}: {summary}" ); assert!( - summary.contains("max_turn=3600s"), + summary.contains("max_turn=7200s"), "summary should include max_turn: {summary}" ); } @@ -2438,10 +2438,10 @@ channels = "ALL" // And the valid case: let idle_valid = 900u64; - let max_turn_valid = 3600u64; + let max_turn_valid = 7200u64; assert!( idle_valid < max_turn_valid, - "default idle (900) must be less than default max_turn (3600)" + "default idle (900) must be less than default max_turn (7200)" ); } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 1835b7810..a55d3de06 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -34,9 +34,9 @@ use futures_util::FutureExt; use nostr::{PublicKey, ToBech32}; use pool::{ AgentPool, ControlSignal, IdleSwitchResult, OwnedAgent, PromptContext, PromptOutcome, - PromptResult, PromptSource, SessionState, + PromptResult, PromptSource, SessionState, TimeoutKind, }; -use queue::{CancelReason, EventQueue, QueuedEvent, ThreadTags}; +use queue::{CancelReason, EventQueue, FlushBatch, QueuedEvent, ThreadTags}; use relay::{HarnessRelay, RelayEventPublisher}; use tokio::sync::{mpsc, watch}; use tracing_subscriber::EnvFilter; @@ -2671,6 +2671,29 @@ fn dispatch_pending( dispatched_channels } +/// Spawn a task that posts a user-visible failure notice to the relay. +/// +/// Shared by the hard-cap immediate dead-letter path and the retries-exhausted +/// dead-letter path so neither duplicates the tokio::spawn block. +fn spawn_failure_notice( + rest_client: Option<&relay::RestClient>, + batch: &FlushBatch, + content: String, +) { + if let Some(rest) = rest_client { + let thread_tags = batch + .events + .last() + .map(|be| queue::parse_thread_tags(&be.event)) + .unwrap_or_default(); + let rest = rest.clone(); + let channel_id = batch.channel_id; + tokio::spawn(async move { + pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await; + }); + } +} + #[allow(clippy::too_many_arguments)] fn handle_prompt_result( pool: &mut AgentPool, @@ -2711,31 +2734,33 @@ fn handle_prompt_result( // system default — rather than telling the agent to supersede. let reason = batch.cancel_reason.unwrap_or(CancelReason::Steer); queue.requeue_as_cancelled(batch, reason); + } else if matches!(result.outcome, PromptOutcome::Timeout(TimeoutKind::Hard)) { + // Hard-cap timeout is deterministic: re-running the same task + // from a fresh session will reproduce the same death. Dead-letter + // immediately without requeueing so the channel isn't subjected to + // up to 10 × 1-hour retry cycles. + let content = format!( + "⚠️ I couldn't process the last request (the turn exceeded the maximum duration ({}s)). Please re-send if it's still needed.", + config.max_turn_duration_secs + ); + spawn_failure_notice(rest_client, &batch, content); } else if let Some(dead) = queue.requeue(batch) { // Dead-lettered: retries exhausted and the events are gone. // Post a visible notice so the channel isn't left waiting on // a turn that will never happen. - if let Some(rest) = rest_client { - let thread_tags = dead - .events - .last() - .map(|be| queue::parse_thread_tags(&be.event)) - .unwrap_or_default(); - let reason = match &result.outcome { - PromptOutcome::Timeout => "the turn timed out".to_string(), - PromptOutcome::AgentExited => "the agent process exited".to_string(), - PromptOutcome::Error(e) => format!("{e}"), - _ => "repeated failures".to_string(), - }; - let content = format!( - "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed." - ); - let rest = rest.clone(); - let channel_id = dead.channel_id; - tokio::spawn(async move { - pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await; - }); - } + let reason = match &result.outcome { + PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), + PromptOutcome::Timeout(TimeoutKind::Hard) => { + unreachable!("hard timeout handled above") + } + PromptOutcome::AgentExited => "the agent process exited".to_string(), + PromptOutcome::Error(e) => format!("{e}"), + _ => "repeated failures".to_string(), + }; + let content = format!( + "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed." + ); + spawn_failure_notice(rest_client, &dead, content); } } else { tracing::debug!( @@ -2761,7 +2786,8 @@ fn handle_prompt_result( let outcome_label = match &result.outcome { PromptOutcome::Ok(_) => "ok", PromptOutcome::Error(_) => "error", - PromptOutcome::Timeout => "timeout", + PromptOutcome::Timeout(TimeoutKind::Idle) => "idle_timeout", + PromptOutcome::Timeout(TimeoutKind::Hard) => "hard_timeout", PromptOutcome::AgentExited => "exited", PromptOutcome::Cancelled => "cancelled", }; @@ -2813,7 +2839,7 @@ fn handle_prompt_result( pool.return_agent(result.agent); } // Fatal outcomes: the agent subprocess is dead or poisoned — respawn it. - PromptOutcome::AgentExited | PromptOutcome::Timeout => { + PromptOutcome::AgentExited | PromptOutcome::Timeout(_) => { tracing::warn!( agent = agent_index, outcome = outcome_label, @@ -2821,11 +2847,15 @@ fn handle_prompt_result( pid = harness_pid, "agent_returned — respawning" ); - let death_message = match outcome_label { - "exited" => "Agent process exited unexpectedly", - _ => "Agent session timed out due to inactivity", + let death_message: String = match outcome_label { + "exited" => "Agent process exited unexpectedly".to_string(), + "hard_timeout" => format!( + "Agent turn exceeded the maximum duration ({}s)", + config.max_turn_duration_secs + ), + _ => "Agent session timed out due to inactivity".to_string(), }; - emit_turn_error(death_message, None); + emit_turn_error(&death_message, None); let index = result.agent.index; let slot_history = &mut crash_history[index]; @@ -3835,7 +3865,7 @@ mod build_mcp_servers_tests { agent_args: vec!["acp".into()], mcp_command: "test-mcp-server".into(), idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 3600, + max_turn_duration_secs: 7200, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -3981,7 +4011,10 @@ mod error_outcome_emission_tests { use super::*; use crate::acp::{AcpClient, AcpError}; use crate::observer::ObserverHandle; - use crate::pool::{AgentPool, OwnedAgent, PromptOutcome, PromptResult, PromptSource}; + use crate::pool::{ + AgentPool, OwnedAgent, PromptOutcome, PromptResult, PromptSource, TimeoutKind, + }; + use crate::queue::FlushBatch; use std::collections::HashSet; fn test_config() -> Config { @@ -3995,7 +4028,7 @@ mod error_outcome_emission_tests { agent_args: vec![], mcp_command: "test-mcp-server".into(), idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 3600, + max_turn_duration_secs: 7200, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -4117,8 +4150,156 @@ mod error_outcome_emission_tests { } #[tokio::test] - async fn timeout_emits_exactly_one_feed_event() { - assert_eq!(turn_errors_emitted_for(PromptOutcome::Timeout).await, 1); + async fn idle_timeout_emits_exactly_one_feed_event() { + assert_eq!( + turn_errors_emitted_for(PromptOutcome::Timeout(TimeoutKind::Idle)).await, + 1 + ); + } + + #[tokio::test] + async fn hard_timeout_emits_exactly_one_feed_event() { + assert_eq!( + turn_errors_emitted_for(PromptOutcome::Timeout(TimeoutKind::Hard)).await, + 1 + ); + } + + /// idle_timeout outcome_label is "idle_timeout"; hard_timeout is "hard_timeout". + #[tokio::test] + async fn timeout_outcome_labels_differ() { + let check_label = |outcome: PromptOutcome, expected_label: &'static str| async move { + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: None, + recoverable_batch: None, + control_tx: None, + steer_tx: None, + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = false; + let removed_channels = HashSet::new(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let observer = ObserverHandle::in_process(); + let result = PromptResult { + agent, + source: PromptSource::Channel(Uuid::new_v4()), + outcome, + batch: None, + }; + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + Some(observer.clone()), + None, + ); + let events = observer.snapshot(); + let turn_error = events.iter().find(|e| e.kind == "turn_error").unwrap(); + assert_eq!( + turn_error.payload["outcome"].as_str().unwrap(), + expected_label + ); + }; + check_label(PromptOutcome::Timeout(TimeoutKind::Idle), "idle_timeout").await; + check_label(PromptOutcome::Timeout(TimeoutKind::Hard), "hard_timeout").await; + } + + /// hard-cap timeout dead-letters immediately (no requeue); idle timeout is requeued. + #[tokio::test] + async fn hard_timeout_not_requeued_idle_timeout_is_requeued() { + let make_batch = || FlushBatch { + channel_id: Uuid::new_v4(), + events: vec![], + cancelled_events: vec![], + cancel_reason: None, + }; + + let run = |outcome: PromptOutcome, batch: FlushBatch| async move { + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: None, + recoverable_batch: None, + control_tx: None, + steer_tx: None, + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = false; + let removed_channels = HashSet::new(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let result = PromptResult { + agent, + source: PromptSource::Channel(batch.channel_id), + outcome, + batch: Some(batch), + }; + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + // Return how many channels have pending items in the queue. + // requeue() inserts a queue entry (even for empty event lists) and + // sets a retry_after delay, so flush_next() won't fire immediately. + // pending_channels() reflects the queue map directly. + queue.pending_channels() + }; + + // Hard timeout: batch must NOT be requeued (dead-lettered immediately). + let hard_batch = make_batch(); + assert_eq!( + run(PromptOutcome::Timeout(TimeoutKind::Hard), hard_batch).await, + 0, + "hard-cap timeout must not requeue the batch" + ); + + // Idle timeout: batch IS requeued (first attempt, not yet dead-lettered). + let idle_batch = make_batch(); + assert_eq!( + run(PromptOutcome::Timeout(TimeoutKind::Idle), idle_batch).await, + 1, + "idle timeout must requeue the batch for retry" + ); } #[tokio::test] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 96302f117..f6dae27ae 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -343,13 +343,22 @@ pub enum SteerAck { PromptCompletedNeutral, } +/// Whether a turn was cut by the idle clock or the hard wall-clock cap. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TimeoutKind { + /// No ACP wire activity for `idle_timeout` seconds. + Idle, + /// Turn ran for `max_turn_duration` seconds of wall-clock time. + Hard, +} + /// Outcome of a prompt task. #[allow(dead_code)] pub enum PromptOutcome { Ok(StopReason), Error(AcpError), AgentExited, - Timeout, + Timeout(TimeoutKind), /// Intentional cancel via `!cancel` command or interrupt mode. /// Agent is healthy — no respawn, no retry penalty. Cancelled, @@ -1448,7 +1457,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), ); return; @@ -1464,7 +1473,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Hard), requeue_batch_if_queue(&ctx, batch), ); return; @@ -1698,8 +1707,33 @@ pub async fn run_prompt_task( ); return; } - Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => { - // Cancel drain timed out — agent state uncertain. + Err(AcpError::IdleTimeout(_)) => { + // Cancel drain idle timed out — agent state uncertain. + agent.state.invalidate(&source); + let retry_batch = + requeue_cancelled_batch(&ctx, control_signal, batch); + + let usage = agent.acp.take_turn_usage(); + publish_agent_turn_metric( + &ctx, + usage, + observer_channel_id, + &session_id, + &turn_id, + Some(buzz_core::agent_turn_metric::StopReason::Error), + ) + .await; + send_prompt_result( + &result_tx, + agent, + source, + PromptOutcome::Timeout(TimeoutKind::Idle), + retry_batch, + ); + return; + } + Err(AcpError::HardTimeout) => { + // Cancel drain hard timed out — agent state uncertain. agent.state.invalidate(&source); let retry_batch = requeue_cancelled_batch(&ctx, control_signal, batch); @@ -1718,7 +1752,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Hard), retry_batch, ); return; @@ -1912,7 +1946,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), ); } @@ -1961,7 +1995,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Idle), requeue_batch_if_queue(&ctx, batch), ); } @@ -1988,7 +2022,7 @@ pub async fn run_prompt_task( &result_tx, agent, source, - PromptOutcome::Timeout, + PromptOutcome::Timeout(TimeoutKind::Hard), requeue_batch_if_queue(&ctx, batch), ); } From d01b02908914a139eca13a3b7efae885edcacc7a Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 13 Jul 2026 23:05:27 -0400 Subject: [PATCH 2/4] fix(acp): derive in-flight deadline from max_turn_duration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The queue's IN_FLIGHT_DEADLINE_SECS (3700) was hardcoded to the old 3600s max_turn_duration + 100s buffer. With the cap raised to 7200s, a legitimate 2h turn hits the 3700s backstop at ~62 min and gets force-released while still running — enabling duplicate same-channel processing. Replace the static constant with a configurable `in_flight_deadline` field on EventQueue, derived from max_turn_duration_secs + 100s buffer at the prod construction site. Tests keep the default (7300s) via EventQueue::new(); prod calls .with_in_flight_deadline(). Also: harden the hard-timeout requeue test with a real event batch instead of an empty FlushBatch, and add invariant tests asserting in_flight_deadline > max_turn_duration. --- crates/buzz-acp/src/config.rs | 21 ++++----- crates/buzz-acp/src/lib.rs | 30 +++++++++---- crates/buzz-acp/src/queue.rs | 82 +++++++++++++++++++++++++++-------- 3 files changed, 96 insertions(+), 37 deletions(-) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 78eb8810d..c2ca548aa 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -26,6 +26,10 @@ use crate::filter::SubscriptionRule; /// Override via `--idle-timeout` / `BUZZ_ACP_IDLE_TIMEOUT`. pub(crate) const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 900; +/// Default absolute wall-clock cap per agent turn (2 hours). +/// Override via `--max-turn-duration` / `BUZZ_ACP_MAX_TURN_DURATION`. +pub(crate) const DEFAULT_MAX_TURN_DURATION_SECS: u64 = 7200; + #[derive(Debug, Error)] pub enum ConfigError { #[error("failed to parse nostr keys: {0}")] @@ -220,7 +224,7 @@ pub struct CliArgs { pub idle_timeout: Option, /// Absolute wall-clock cap per turn (safety valve). - #[arg(long, env = "BUZZ_ACP_MAX_TURN_DURATION", default_value = "7200")] + #[arg(long, env = "BUZZ_ACP_MAX_TURN_DURATION", default_value_t = DEFAULT_MAX_TURN_DURATION_SECS)] pub max_turn_duration: u64, /// Deprecated: alias for --idle-timeout. If both set, --idle-timeout wins. @@ -1326,7 +1330,7 @@ mod tests { agent_args: vec!["acp".into()], mcp_command: "".into(), idle_timeout_secs: DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 7200, + max_turn_duration_secs: DEFAULT_MAX_TURN_DURATION_SECS, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -2231,7 +2235,7 @@ channels = "ALL" "summary should include {expected_idle}: {summary}" ); assert!( - summary.contains("max_turn=7200s"), + summary.contains(&format!("max_turn={DEFAULT_MAX_TURN_DURATION_SECS}s")), "summary should include max_turn: {summary}" ); } @@ -2436,13 +2440,10 @@ channels = "ALL" "test precondition: idle must be >= max_turn to trigger guard" ); - // And the valid case: - let idle_valid = 900u64; - let max_turn_valid = 7200u64; - assert!( - idle_valid < max_turn_valid, - "default idle (900) must be less than default max_turn (7200)" - ); + // And the valid case (const assertion so clippy doesn't flag it): + const { + assert!(DEFAULT_IDLE_TIMEOUT_SECS < DEFAULT_MAX_TURN_DURATION_SECS); + } } // --- BUZZ_ACP_ALLOWED_RESPOND_TO gate --- diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index a55d3de06..2865ee2cb 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1387,7 +1387,8 @@ async fn tokio_main() -> Result<()> { } let dedup_mode = config.dedup_mode; - let mut queue = EventQueue::new(dedup_mode); + let mut queue = + EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); let base_prompt_content = config.base_prompt_content.take(); let ctx = Arc::new(PromptContext { @@ -3865,7 +3866,7 @@ mod build_mcp_servers_tests { agent_args: vec!["acp".into()], mcp_command: "test-mcp-server".into(), idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 7200, + max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -4014,7 +4015,8 @@ mod error_outcome_emission_tests { use crate::pool::{ AgentPool, OwnedAgent, PromptOutcome, PromptResult, PromptSource, TimeoutKind, }; - use crate::queue::FlushBatch; + use crate::queue::{BatchEvent, FlushBatch}; + use nostr::{EventBuilder, Keys, Kind}; use std::collections::HashSet; fn test_config() -> Config { @@ -4028,7 +4030,7 @@ mod error_outcome_emission_tests { agent_args: vec![], mcp_command: "test-mcp-server".into(), idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS, - max_turn_duration_secs: 7200, + max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS, agents: 1, heartbeat_interval_secs: 0, turn_liveness_secs: 10, @@ -4227,11 +4229,21 @@ mod error_outcome_emission_tests { /// hard-cap timeout dead-letters immediately (no requeue); idle timeout is requeued. #[tokio::test] async fn hard_timeout_not_requeued_idle_timeout_is_requeued() { - let make_batch = || FlushBatch { - channel_id: Uuid::new_v4(), - events: vec![], - cancelled_events: vec![], - cancel_reason: None, + let make_batch = || { + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "test") + .sign_with_keys(&keys) + .unwrap(); + FlushBatch { + channel_id: Uuid::new_v4(), + events: vec![BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + } }; let run = |outcome: PromptOutcome, batch: FlushBatch| async move { diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 2a427feea..37f7a1d63 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -35,8 +35,11 @@ const BASE_RETRY_DELAY_SECS: u64 = 5; /// Cap on retry delay in seconds. const MAX_RETRY_DELAY_SECS: u64 = 300; -/// In-flight deadline: max_turn (3600s) + 100s buffer. -const IN_FLIGHT_DEADLINE_SECS: u64 = 3700; +/// Buffer added to `max_turn_duration` to derive the in-flight deadline. +const IN_FLIGHT_DEADLINE_BUFFER_SECS: u64 = 100; + +/// Default in-flight deadline: default max_turn (7200s) + 100s buffer. +const DEFAULT_IN_FLIGHT_DEADLINE_SECS: u64 = 7300; /// An event waiting in the queue. #[derive(Debug, Clone)] @@ -94,7 +97,7 @@ pub struct FlushBatch { /// State: /// queues: Map> (capped at MAX_PENDING_PER_CHANNEL) /// in_flight_channels: HashSet -/// in_flight_deadlines: Map (auto-expire after IN_FLIGHT_DEADLINE_SECS) +/// in_flight_deadlines: Map (auto-expire after in_flight_deadline) /// retry_after: Map /// retry_counts: Map (dead-letter after MAX_RETRIES) /// dedup_mode: DedupMode @@ -117,7 +120,7 @@ pub struct FlushBatch { /// channel = pick candidate with oldest head event (min received_at) /// events = drain up to MAX_BATCH_EVENTS from queues[channel] /// in_flight_channels.insert(channel) -/// in_flight_deadlines.insert(channel, now + IN_FLIGHT_DEADLINE_SECS) +/// in_flight_deadlines.insert(channel, now + in_flight_deadline) /// return Some(FlushBatch { channel, events }) /// /// mark_complete(channel_id): @@ -157,14 +160,22 @@ pub struct EventQueue { /// path. Populated by [`mark_native_steer_pending`]; drained back to the /// queue front by [`release_native_steer`] (preserving original /// `received_at` fairness, same discipline as `requeue_preserve_timestamps` - /// at line 453). Bulk recovery on `IN_FLIGHT_DEADLINE_SECS` expiry is - /// performed by `flush_next` / `has_flushable_work` (recover, not - /// log-and-drop — the events were never delivered to the agent). + /// at line 453). Bulk recovery on in-flight deadline expiry is performed + /// by `flush_next` / `has_flushable_work` (recover, not log-and-drop — + /// the events were never delivered to the agent). withheld_native_steer: HashMap>, + /// Duration after which an in-flight channel is auto-expired as orphaned. + /// 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, } impl EventQueue { /// Create a new empty event queue with the given dedup mode. + /// + /// Uses [`DEFAULT_IN_FLIGHT_DEADLINE_SECS`] for the in-flight backstop. + /// Call [`with_in_flight_deadline`](Self::with_in_flight_deadline) to + /// derive the deadline from the configured `max_turn_duration`. pub fn new(dedup_mode: DedupMode) -> Self { Self { queues: HashMap::new(), @@ -177,9 +188,18 @@ impl EventQueue { cancelled_batches: HashMap::new(), cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), + in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), } } + /// Set the in-flight backstop deadline from the configured max turn + /// duration, preserving the 100s buffer for cancel-drain grace + respawn. + pub fn with_in_flight_deadline(mut self, max_turn_duration_secs: u64) -> Self { + self.in_flight_deadline = + Duration::from_secs(max_turn_duration_secs + IN_FLIGHT_DEADLINE_BUFFER_SECS); + self + } + /// Push an event into the queue for its channel. /// /// In [`DedupMode::Drop`], events for any currently in-flight channel are @@ -231,7 +251,7 @@ impl EventQueue { tracing::error!( channel_id = %id, lost_events, - deadline_secs = IN_FLIGHT_DEADLINE_SECS, + deadline_secs = self.in_flight_deadline.as_secs(), "BUG: in-flight channel expired without mark_complete — \ auto-releasing; {lost_events} dispatched event(s) orphaned" ); @@ -277,7 +297,7 @@ impl EventQueue { let cancel_reason = self.cancel_reasons.remove(&id); self.in_flight_channels.insert(id); self.in_flight_deadlines - .insert(id, now + Duration::from_secs(IN_FLIGHT_DEADLINE_SECS)); + .insert(id, now + self.in_flight_deadline); self.in_flight_batch_sizes.insert(id, cancelled.len()); return Some(FlushBatch { channel_id: id, @@ -309,10 +329,8 @@ impl EventQueue { } self.in_flight_channels.insert(channel_id); - self.in_flight_deadlines.insert( - channel_id, - now + Duration::from_secs(IN_FLIGHT_DEADLINE_SECS), - ); + self.in_flight_deadlines + .insert(channel_id, now + self.in_flight_deadline); self.in_flight_batch_sizes.insert(channel_id, events.len()); // Merge any cancelled events stored by requeue_as_cancelled(). @@ -524,7 +542,7 @@ impl EventQueue { tracing::error!( channel_id = %id, lost_events, - deadline_secs = IN_FLIGHT_DEADLINE_SECS, + deadline_secs = self.in_flight_deadline.as_secs(), "BUG: in-flight channel expired without mark_complete — \ auto-releasing; {lost_events} dispatched event(s) orphaned" ); @@ -693,7 +711,7 @@ impl EventQueue { /// Bulk-release every withheld event for `channel_id` back to the queue /// front, preserving relative FIFO order. /// - /// Called from the `IN_FLIGHT_DEADLINE_SECS` expiry blocks in + /// Called from the `in_flight_deadline` expiry blocks in /// `flush_next` and `has_flushable_work` — if a steer ack never arrives /// (read loop hung, watcher never posted), the withheld events would /// otherwise be permanently orphaned. Recover, do not log-and-drop: the @@ -4150,7 +4168,7 @@ mod tests { // `flush_next` / `has_flushable_work` / contiguous drain. `Success` ack // drops it via `remove_event`; `Err` / `PromptCompletedNeutral` ack // restores it to the queue front via `release_native_steer`. The - // `IN_FLIGHT_DEADLINE_SECS` expiry bulk-recovers withheld events so they + // `in_flight_deadline` expiry bulk-recovers withheld events so they // are never permanently orphaned. /// A channel whose only queued event has been withheld for a goose-native @@ -4230,7 +4248,7 @@ mod tests { } /// If the steer ack never arrives — read loop hung, watcher never posted — - /// the `IN_FLIGHT_DEADLINE_SECS` auto-expiry block must bulk-recover the + /// the `in_flight_deadline` auto-expiry block must bulk-recover the /// withheld events back to the queue front so normal dispatch can deliver /// them. Recover, not log-and-drop: the events were never seen by the /// agent. @@ -4252,7 +4270,7 @@ mod tests { // Force the in-flight deadline to be in the past, simulating the // steer ack never arriving and the read loop hanging long enough - // for `IN_FLIGHT_DEADLINE_SECS` to elapse. Same expiry-simulation + // for `in_flight_deadline` to elapse. Same expiry-simulation // trick used by `test_retry_throttle_blocks_requeue_channel`. q.in_flight_deadlines .insert(ch, Instant::now() - Duration::from_secs(1)); @@ -4408,4 +4426,32 @@ mod tests { "no canvas section expected when agent_canvas is None; got: {prompt}" ); } + + #[test] + fn default_in_flight_deadline_exceeds_default_max_turn_duration() { + let q = EventQueue::new(DedupMode::Queue); + let default_max_turn = Duration::from_secs(crate::config::DEFAULT_MAX_TURN_DURATION_SECS); + assert!( + q.in_flight_deadline > default_max_turn, + "in_flight_deadline ({:?}) must be strictly greater than \ + default max_turn_duration ({:?})", + q.in_flight_deadline, + default_max_turn, + ); + } + + #[test] + fn with_in_flight_deadline_derives_from_max_turn_duration() { + let max_turn = 9000u64; + let q = EventQueue::new(DedupMode::Queue).with_in_flight_deadline(max_turn); + let expected = Duration::from_secs(max_turn + IN_FLIGHT_DEADLINE_BUFFER_SECS); + assert_eq!( + q.in_flight_deadline, expected, + "in_flight_deadline should be max_turn_duration + buffer" + ); + assert!( + q.in_flight_deadline > Duration::from_secs(max_turn), + "in_flight_deadline must be strictly greater than max_turn_duration" + ); + } } From 8f04f68df24fd2df9a9bbb0c13e51695f61036ad Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Mon, 13 Jul 2026 23:50:26 -0400 Subject: [PATCH 3/4] fix(acp): validate max_turn_duration ceiling and harden requeue test Reject max_turn_duration above 604800s (7 days) at the config boundary so the unchecked u64 addition in with_in_flight_deadline cannot overflow. Make the requeue test assert queued event counts (not just channel keys) to verify event preservation on idle timeout and event drop on hard timeout. --- crates/buzz-acp/src/config.rs | 63 +++++++++++++++++++++++++++++++++++ crates/buzz-acp/src/lib.rs | 28 ++++++++++------ crates/buzz-acp/src/queue.rs | 6 ++++ 3 files changed, 87 insertions(+), 10 deletions(-) diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index c2ca548aa..b9984adc7 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -30,6 +30,11 @@ pub(crate) const DEFAULT_IDLE_TIMEOUT_SECS: u64 = 900; /// Override via `--max-turn-duration` / `BUZZ_ACP_MAX_TURN_DURATION`. pub(crate) const DEFAULT_MAX_TURN_DURATION_SECS: u64 = 7200; +/// Upper bound for `max_turn_duration` (7 days). Any higher is operationally +/// meaningless and risks arithmetic overflow when deriving the in-flight +/// deadline (`max_turn_duration + IN_FLIGHT_DEADLINE_BUFFER_SECS`). +pub(crate) const MAX_TURN_DURATION_CEILING_SECS: u64 = 604_800; + #[derive(Debug, Error)] pub enum ConfigError { #[error("failed to parse nostr keys: {0}")] @@ -835,6 +840,11 @@ impl Config { if raw == 0 { tracing::warn!("max turn duration of 0 is invalid — using 60s minimum"); 60 + } else if raw > MAX_TURN_DURATION_CEILING_SECS { + return Err(ConfigError::ConfigFile(format!( + "max_turn_duration ({}s) exceeds ceiling ({}s / 7 days)", + raw, MAX_TURN_DURATION_CEILING_SECS + ))); } else { raw } @@ -2628,4 +2638,57 @@ channels = "ALL" "from_args should accept any mode when allowed list is unset: {result:?}" ); } + + // --- max_turn_duration ceiling gate --- + + #[test] + fn max_turn_duration_at_ceiling_is_accepted() { + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--max-turn-duration", + &MAX_TURN_DURATION_CEILING_SECS.to_string(), + ]) + .expect("clap should parse args"); + let result = Config::from_args(args); + + assert!( + result.is_ok(), + "from_args should accept max_turn_duration at the ceiling: {result:?}" + ); + } + + #[test] + fn max_turn_duration_above_ceiling_is_rejected() { + let over = MAX_TURN_DURATION_CEILING_SECS + 1; + let args = CliArgs::try_parse_from([ + "buzz-acp", + "--private-key", + TEST_PRIVATE_KEY, + "--max-turn-duration", + &over.to_string(), + ]) + .expect("clap should parse args"); + let result = Config::from_args(args); + + assert!( + result.is_err(), + "from_args should reject max_turn_duration above the ceiling" + ); + let msg = result.unwrap_err().to_string(); + assert!( + msg.contains("exceeds ceiling"), + "error should mention 'exceeds ceiling': {msg}" + ); + } + + #[test] + fn max_turn_duration_ceiling_cannot_overflow_in_flight_deadline() { + // The in-flight deadline is max_turn + 100s buffer (IN_FLIGHT_DEADLINE_BUFFER_SECS). + // Verify that even at the ceiling, this addition cannot overflow u64. + const { + assert!(MAX_TURN_DURATION_CEILING_SECS < u64::MAX - 100); + } + } } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 2865ee2cb..53060e2db 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4246,7 +4246,9 @@ mod error_outcome_emission_tests { } }; + // Returns (pending_channels, queued_event_count_for_channel). let run = |outcome: PromptOutcome, batch: FlushBatch| async move { + let channel_id = batch.channel_id; let agent = dummy_agent(0).await; let mut pool = AgentPool::from_slots(vec![None]); let task_id = pool.join_set.spawn(async {}).id(); @@ -4273,7 +4275,7 @@ mod error_outcome_emission_tests { let mut respawn_tasks = tokio::task::JoinSet::new(); let result = PromptResult { agent, - source: PromptSource::Channel(batch.channel_id), + source: PromptSource::Channel(channel_id), outcome, batch: Some(batch), }; @@ -4290,28 +4292,34 @@ mod error_outcome_emission_tests { None, None, ); - // Return how many channels have pending items in the queue. - // requeue() inserts a queue entry (even for empty event lists) and - // sets a retry_after delay, so flush_next() won't fire immediately. - // pending_channels() reflects the queue map directly. - queue.pending_channels() + ( + queue.pending_channels(), + queue.queued_event_count(&channel_id), + ) }; // Hard timeout: batch must NOT be requeued (dead-lettered immediately). let hard_batch = make_batch(); + let (hard_channels, hard_events) = + run(PromptOutcome::Timeout(TimeoutKind::Hard), hard_batch).await; assert_eq!( - run(PromptOutcome::Timeout(TimeoutKind::Hard), hard_batch).await, - 0, + hard_channels, 0, "hard-cap timeout must not requeue the batch" ); + assert_eq!(hard_events, 0, "hard-cap timeout must drop all events"); // Idle timeout: batch IS requeued (first attempt, not yet dead-lettered). let idle_batch = make_batch(); + let (idle_channels, idle_events) = + run(PromptOutcome::Timeout(TimeoutKind::Idle), idle_batch).await; assert_eq!( - run(PromptOutcome::Timeout(TimeoutKind::Idle), idle_batch).await, - 1, + idle_channels, 1, "idle timeout must requeue the batch for retry" ); + assert_eq!( + idle_events, 1, + "idle timeout must preserve the event for retry" + ); } #[tokio::test] diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 37f7a1d63..c1b4c56f7 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -569,6 +569,12 @@ impl EventQueue { self.queues.len() } + /// Number of queued events for a specific channel. Test-only. + #[cfg(test)] + pub fn queued_event_count(&self, channel_id: &Uuid) -> usize { + self.queues.get(channel_id).map_or(0, |q| q.len()) + } + /// Drop all queued (non-in-flight) events for a channel. /// /// Used when the agent is removed from a channel — any pending events From e54dccf26ecf65f33e8ff782901d137116d27cbd Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Tue, 14 Jul 2026 09:33:35 -0400 Subject: [PATCH 4/4] chore(buzz-acp): replace unreachable! with soft fallback and add dead-letter log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hard-timeout match arm in handle_prompt_result used unreachable!(), which would panic the main loop if the else-if chain were ever reordered. Replace with a graceful fallback returning the honest reason string. Add a tracing::error! log line in the hard-cap dead-letter arm to match the retries-exhausted dead-letter path — ops greps for dead-letters now catch both paths symmetrically. --- crates/buzz-acp/src/lib.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 53060e2db..ea8764b8a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2740,6 +2740,12 @@ fn handle_prompt_result( // from a fresh session will reproduce the same death. Dead-letter // immediately without requeueing so the channel isn't subjected to // up to 10 × 1-hour retry cycles. + tracing::error!( + channel_id = %batch.channel_id, + events = batch.events.len(), + "dead-lettering batch after hard-cap timeout — discarding {} events", + batch.events.len(), + ); let content = format!( "⚠️ I couldn't process the last request (the turn exceeded the maximum duration ({}s)). Please re-send if it's still needed.", config.max_turn_duration_secs @@ -2751,8 +2757,11 @@ fn handle_prompt_result( // a turn that will never happen. let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), + // Unreachable today: Timeout(Hard) is consumed by the immediate + // dead-letter arm above before requeue() runs. Fail soft rather + // than panicking the main loop if that chain is ever reordered. PromptOutcome::Timeout(TimeoutKind::Hard) => { - unreachable!("hard timeout handled above") + "the turn exceeded the maximum duration".to_string() } PromptOutcome::AgentExited => "the agent process exited".to_string(), PromptOutcome::Error(e) => format!("{e}"),