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

fix(buzz-acp,buzz-agent): surface stall duration and fate (#2204)

Browse Source 
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
2026-07-21 01:10:28 -04:00
committed by GitHub
parent c5d00a3589
commit cd6b573f8d
3 changed files with 415 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files

557
crates/buzz-acp/src/lib.rs
View File

@@ -2779,6 +2779,15 @@ fn handle_prompt_result(
.retain(|_, meta| meta.agent_index != agent_index);
debug_assert_eq!(before, pool.task_map().len() + 1);
// The hard-timeout death_message (below) must describe the batch's
// *actual* fate, not just the `recently_active` eligibility flag — a
// recently-active batch that exhausts the retry budget in queue.requeue()
// is dead-lettered same as an immediate one, and both differ from a
// channel-removed drop or a heartbeat call with no batch at all. Each
// branch below records what actually happened; only the hard-timeout
// match arm in the death_message construction reads it.
let mut hard_timeout_fate_suffix: Option<&'static str> = None;
// Requeue BEFORE mark_complete: requeue() sets retry_after with a future
// deadline, and mark_complete() checks for it to decide whether to preserve
// retry_counts. If mark_complete runs first, retry_counts is cleared and
@@ -2825,6 +2834,7 @@ fn handle_prompt_result(
config.max_turn_duration_secs
);
spawn_failure_notice(rest_client, &batch, content);
hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)");
} else if matches!(
result.outcome,
PromptOutcome::Timeout(TimeoutKind::Hard {
@@ -2842,6 +2852,9 @@ fn handle_prompt_result(
config.max_turn_duration_secs
);
spawn_failure_notice(rest_client, &dead, content);
hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)");
} else {
hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)");
}
} else if let Some(dead) = queue.requeue(batch) {
let reason = match &result.outcome {
@@ -2864,6 +2877,7 @@ fn handle_prompt_result(
events = batch.events.len(),
"dropping failed batch for removed channel"
);
hard_timeout_fate_suffix = Some(" — batch dropped (channel removed)");
}
}
@@ -2888,10 +2902,6 @@ fn handle_prompt_result(
PromptOutcome::Cancelled => "cancelled",
PromptOutcome::CancelDrainTimeout(_) => "cancel_drain_timeout",
};
let hard_timeout_recently_active = match &result.outcome {
PromptOutcome::Timeout(TimeoutKind::Hard { recently_active }) => Some(*recently_active),
_ => None,
};
let agent_index = result.agent.index;
// Capture the spawn-time configured model and our PID before the agent is
// moved into match arms below. `desired_model` reflects the config/persona
@@ -2952,11 +2962,10 @@ fn handle_prompt_result(
let death_message: String = match outcome_label {
"exited" => "Agent process exited unexpectedly".to_string(),
"hard_timeout" => {
let suffix = if hard_timeout_recently_active == Some(true) {
" — requeued for retry (recently active)"
} else {
" — dead-lettered (no recent activity)"
};
// Neutral wording when no fate was recorded above: a
// heartbeat hard timeout carries no batch at all, so
// nothing was requeued or dead-lettered.
let suffix = hard_timeout_fate_suffix.unwrap_or(" (no batch to retry)");
format!(
"Agent turn exceeded the maximum duration ({}s){}",
config.max_turn_duration_secs, suffix
@@ -4853,6 +4862,186 @@ mod error_outcome_emission_tests {
);
}
/// The hard-timeout `death_message` must report what actually happened to
/// the batch, not just the `recently_active` eligibility flag: a
/// recently-active batch within its retry budget is requeued, so the
/// observer payload must say so.
#[tokio::test]
async fn hard_timeout_recently_active_requeue_success_reports_requeued_for_retry() {
let channel_id = Uuid::new_v4();
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,
turn_id: "test-turn-id".to_string(),
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 batch = FlushBatch {
channel_id,
events: vec![BatchEvent {
event: EventBuilder::new(Kind::Custom(9), "test")
.sign_with_keys(&Keys::generate())
.unwrap(),
prompt_tag: "test".into(),
received_at: std::time::Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
};
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".to_string(),
outcome: PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: true,
}),
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,
Some(observer.clone()),
None,
);
let events = observer.snapshot();
let turn_error = events
.iter()
.find(|e| e.kind == "turn_error")
.expect("exactly one turn_error event must be emitted");
assert_eq!(
turn_error.payload["error"].as_str().unwrap(),
format!(
"Agent turn exceeded the maximum duration ({}s) — requeued for retry (recently active)",
config.max_turn_duration_secs
),
);
assert_eq!(
queue.pending_channels(),
1,
"batch must be requeued, not dead-lettered, while within the retry budget"
);
}
/// Same recently-active hard timeout, but the channel has already
/// exhausted its retry budget ([`crate::queue::MAX_RETRIES`] prior
/// attempts) — `queue.requeue()` dead-letters instead of requeueing, and
/// the observer payload must report that fate, not the requeue wording
/// above.
#[tokio::test]
async fn hard_timeout_recently_active_budget_exhausted_reports_dead_lettered() {
let channel_id = Uuid::new_v4();
let mut queue = EventQueue::new(config::DedupMode::Queue);
// Simulate MAX_RETRIES prior failed attempts on this channel so the
// upcoming requeue() call in handle_prompt_result crosses the
// dead-letter threshold.
queue.set_retry_count_for_test(channel_id, crate::queue::MAX_RETRIES);
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,
turn_id: "test-turn-id".to_string(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
},
);
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 batch = FlushBatch {
channel_id,
events: vec![BatchEvent {
event: EventBuilder::new(Kind::Custom(9), "final-attempt")
.sign_with_keys(&Keys::generate())
.unwrap(),
prompt_tag: "test".into(),
received_at: std::time::Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
};
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".to_string(),
outcome: PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: true,
}),
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,
Some(observer.clone()),
None,
);
let events = observer.snapshot();
let turn_error = events
.iter()
.find(|e| e.kind == "turn_error")
.expect("exactly one turn_error event must be emitted");
assert_eq!(
turn_error.payload["error"].as_str().unwrap(),
format!(
"Agent turn exceeded the maximum duration ({}s) — dead-lettered (retry budget exhausted)",
config.max_turn_duration_secs
),
);
assert_eq!(
queue.queued_event_count(&channel_id),
0,
"batch with an exhausted retry budget must be dead-lettered, not requeued"
);
}
/// Cancel-drain-timeout batches are requeued as cancelled (merge into the
/// next flush, `CancelReason` preserved) — never dead-lettered like a real
/// hard-cap. The agent itself is NOT returned to the idle pool: it is
@@ -5384,353 +5573,3 @@ mod observer_payload_trim_tests {
assert!(leaf.contains("[elided"));
}
}
#[cfg(test)]
mod steer_renewal_tests {
//! Integration-level tests for F2 (steer-renewal deadline extension) and F3
//! (virtual-time regression: hard timeout with `recently_active: false` after
//! a steer renews the hard deadline past the idle deadline).
//!
//! These tests work through `EventQueue::extend_in_flight_deadline` (the
//! production code called by the `SteerAck::Success` handler at lib.rs:2334-
//! 2335) and through `handle_prompt_result` (which owns the fate decision for
//! every `PromptOutcome`).
use super::*;
use crate::acp::AcpClient;
use crate::observer::ObserverHandle;
use crate::pool::{
AgentPool, OwnedAgent, PromptOutcome, PromptResult, PromptSource, TimeoutKind,
};
use crate::queue::{BatchEvent, EventQueue, FlushBatch, QueuedEvent};
use nostr::{EventBuilder, Keys, Kind};
use std::collections::HashSet;
use std::time::Instant;
fn test_config() -> Config {
Config {
keys: nostr::Keys::generate(),
relay_url: "ws://localhost:3000".into(),
agent_command: "true".into(),
agent_args: vec![],
mcp_command: "test-mcp-server".into(),
idle_timeout_secs: config::DEFAULT_IDLE_TIMEOUT_SECS,
max_turn_duration_secs: config::DEFAULT_MAX_TURN_DURATION_SECS,
agents: 1,
heartbeat_interval_secs: 0,
turn_liveness_secs: 10,
heartbeat_prompt: None,
system_prompt: None,
team_instructions: None,
initial_message: None,
subscribe_mode: config::SubscribeMode::All,
dedup_mode: config::DedupMode::Queue,
multiple_event_handling: config::MultipleEventHandling::Queue,
ignore_self: true,
kinds_override: None,
channels_override: None,
no_mention_filter: false,
config_path: std::path::PathBuf::from("./buzz-acp.toml"),
context_message_limit: 12,
max_turns_per_session: 0,
presence_enabled: true,
typing_enabled: true,
memory_enabled: false,
model: None,
permission_mode: config::PermissionMode::BypassPermissions,
respond_to: config::RespondTo::Anyone,
respond_to_allowlist: HashSet::new(),
allowed_respond_to: vec![],
persona_env_vars: vec![],
has_generated_codex_config: false,
relay_observer: false,
agent_owner: None,
no_base_prompt: false,
base_prompt_content: None,
}
}
async fn dummy_agent(index: usize) -> OwnedAgent {
OwnedAgent {
index,
acp: AcpClient::spawn("cat", &[], &[], false)
.await
.expect("spawn cat as inert agent"),
state: Default::default(),
model_capabilities: None,
desired_model: None,
model_overridden: false,
agent_name: "unknown".into(),
goose_system_prompt_supported: None,
protocol_version: 1,
}
}
fn make_flush_batch(channel_id: Uuid) -> FlushBatch {
let keys = Keys::generate();
let event = EventBuilder::new(Kind::Custom(9), "test")
.sign_with_keys(&keys)
.unwrap();
FlushBatch {
channel_id,
events: vec![BatchEvent {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
}
}
// ── F2 case 2.1: successful steer extends the queue deadline ─────────────
/// A successful steer (`SteerAck::Success`) calls
/// `queue.extend_in_flight_deadline(channel_id, max_turn_secs)`. This test
/// verifies that call actually moves the deadline forward — simulating what
/// the ack handler does at lib.rs:2334-2335 — and that the channel remains
/// in-flight past the original (shorter) deadline.
///
/// We use `flush_next` to put the channel naturally in-flight (same path
/// as production code), then call `extend_in_flight_deadline` and confirm
/// that a subsequent `flush_next` on a second channel does NOT release ch
/// from in-flight (the extended deadline keeps it alive).
#[test]
fn steer_success_extends_queue_deadline() {
let keys = Keys::generate();
let mut q = EventQueue::new(config::DedupMode::Queue);
let ch = Uuid::new_v4();
// Put ch in-flight via normal flush path.
let event = EventBuilder::new(Kind::Custom(9), "original-work")
.sign_with_keys(&keys)
.unwrap();
q.push(QueuedEvent {
channel_id: ch,
event,
received_at: Instant::now(),
prompt_tag: "test".into(),
});
let _batch = q.flush_next().expect("first flush");
assert!(
q.is_channel_in_flight(ch),
"ch must be in-flight after flush"
);
// Simulate SteerAck::Success: extend the deadline by max_turn_secs.
// In production this is called at lib.rs:2335.
let max_turn_secs = 7200u64;
q.extend_in_flight_deadline(ch, max_turn_secs);
// Push an event for a second channel and flush — this triggers the
// expiry check inside flush_next. If extend_in_flight_deadline failed,
// ch's deadline might expire and it would be auto-released.
let ch2 = Uuid::new_v4();
let event2 = EventBuilder::new(Kind::Custom(9), "other")
.sign_with_keys(&keys)
.unwrap();
q.push(QueuedEvent {
channel_id: ch2,
event: event2,
received_at: Instant::now(),
prompt_tag: "test".into(),
});
let batch2 = q.flush_next().expect("ch2 should flush");
assert_eq!(batch2.channel_id, ch2, "ch2 flushed, not ch");
// ch must still be in-flight — the extended deadline protected it.
assert!(
q.is_channel_in_flight(ch),
"ch must remain in-flight after deadline extension (SteerAck::Success path)"
);
}
// ── F2 case 2.3: SteerAck::Err / non-success does NOT extend the deadline ─
/// A failed or neutral steer (`SteerAck::Err`, `SteerAck::PromptCompletedNeutral`)
/// must NOT call `extend_in_flight_deadline`. This test verifies that when
/// no extension is applied, the channel behaves according to its original
/// deadline — simulating the path where the condition at lib.rs:2334 is false.
#[test]
fn steer_error_does_not_extend_queue_deadline() {
let keys = Keys::generate();
let mut q = EventQueue::new(config::DedupMode::Queue);
let ch = Uuid::new_v4();
// Put ch in-flight.
let event = EventBuilder::new(Kind::Custom(9), "work")
.sign_with_keys(&keys)
.unwrap();
q.push(QueuedEvent {
channel_id: ch,
event,
received_at: Instant::now(),
prompt_tag: "test".into(),
});
let _batch = q.flush_next().expect("flush");
assert!(q.is_channel_in_flight(ch));
// SteerAck::Err path: the condition at lib.rs:2334 is false, so
// extend_in_flight_deadline is NOT called. The channel stays in-flight
// with its original deadline — which is the DEFAULT_IN_FLIGHT_DEADLINE_SECS
// (7300s) set by flush_next. Confirm the channel is still in-flight and
// has_flushable_work returns false (no pending events for ch).
assert!(
q.is_channel_in_flight(ch),
"channel must still be in-flight on SteerAck::Err (no extension applied)"
);
assert!(
!q.has_flushable_work(),
"no flushable work since ch is in-flight with its original deadline"
);
}
// ── F2 case 2.5: steer renewal is monotonic across repeated steers ───────
/// Calling the steer-renewal extension multiple times with the same
/// `max_turn_secs` must only move the deadline forward, never backward.
/// Uses the queue-public API only (flush_next to establish in-flight, then
/// repeated extend calls verified via observable behavior).
#[test]
fn steer_renewal_is_monotonic_across_repeated_steers() {
let keys = Keys::generate();
let mut q = EventQueue::new(config::DedupMode::Queue);
let ch = Uuid::new_v4();
// Put ch in-flight.
let event = EventBuilder::new(Kind::Custom(9), "work")
.sign_with_keys(&keys)
.unwrap();
q.push(QueuedEvent {
channel_id: ch,
event,
received_at: Instant::now(),
prompt_tag: "test".into(),
});
let _batch = q.flush_next().expect("flush");
let max_turn_secs = 7200u64;
// First extension.
q.extend_in_flight_deadline(ch, max_turn_secs);
// Second extension with same value — must not regress.
q.extend_in_flight_deadline(ch, max_turn_secs);
// Channel must still be in-flight (not expired, not completed).
assert!(
q.is_channel_in_flight(ch),
"channel must remain in-flight after repeated extend calls (monotonic)"
);
// Confirm still no flushable work — the repeated extensions must not
// accidentally complete the channel or release it.
assert!(
!q.has_flushable_work(),
"no flushable work after repeated extend — channel stays in-flight"
);
}
// ── F3: virtual-time regression ──────────────────────────────────────────
/// F3 — virtual-time regression.
///
/// When silence after a steer renews the hard deadline past the idle
/// deadline, and then exceeds BOTH `idle_timeout` AND
/// `RECENT_ACTIVITY_WINDOW` (60 s), the read loop produces
/// `PromptOutcome::Timeout(TimeoutKind::Hard { recently_active: false })`.
///
/// This test verifies that outcome:
/// 1. Dead-letters the batch (no requeue) — confirming `recently_active: false`
/// controls fate correctly.
/// 2. Does not panic or underflow — confirming the virtual-time path is robust.
///
/// The complementary assertion — that `recently_active: true` requeues —
/// pins the flag as the sole fate switch, not some other condition.
#[tokio::test]
async fn hard_timeout_recently_active_false_after_steer_renews_past_idle() {
let run = |outcome: PromptOutcome| async move {
let channel_id = Uuid::new_v4();
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,
turn_id: "test-turn-id".to_string(),
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 batch = make_flush_batch(channel_id);
let result = PromptResult {
agent,
source: PromptSource::Channel(channel_id),
turn_id: "test-turn-id".to_string(),
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,
Some(observer),
None,
);
(
queue.pending_channels(),
queue.queued_event_count(&channel_id),
)
};
// Scenario: steer renewed hard deadline past idle; then silence exceeded
// BOTH idle_timeout AND RECENT_ACTIVITY_WINDOW (60s) → recently_active: false.
// Expected fate: dead-letter (no requeue, no panic).
let (channels, events) = run(PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: false,
}))
.await;
assert_eq!(
channels, 0,
"hard timeout (recently_active: false) after steer renewal must dead-letter — no requeue"
);
assert_eq!(
events, 0,
"hard timeout (recently_active: false) must drop all events"
);
// Complementary: recently_active: true requeues (steer kept idle clock
// warm; only the hard deadline fired, not the combined silence check).
let (channels_ra, events_ra) = run(PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: true,
}))
.await;
assert_eq!(
channels_ra, 1,
"hard timeout (recently_active: true) must requeue the batch"
);
assert_eq!(
events_ra, 1,
"hard timeout (recently_active: true) must preserve the event"
);
}
}

12
crates/buzz-acp/src/queue.rs
View File

@@ -27,7 +27,7 @@ const MAX_PENDING_PER_CHANNEL: usize = 500;
const MAX_BATCH_EVENTS: usize = 50;
/// Maximum retry attempts before a batch is dead-lettered.
const MAX_RETRIES: u32 = 10;
pub(crate) const MAX_RETRIES: u32 = 10;
/// Base retry delay in seconds (doubled each attempt).
const BASE_RETRY_DELAY_SECS: u64 = 5;
@@ -601,6 +601,16 @@ impl EventQueue {
self.queues.get(channel_id).map_or(0, |q| q.len())
}
/// Force a channel's retry-attempt counter to `count`, simulating `count`
/// prior failed attempts without needing to drive fake flush/requeue
/// cycles through the queue (which would leave artifact events behind).
/// Test-only — lets integration tests outside this module exercise
/// `requeue()`'s dead-letter threshold directly.
#[cfg(test)]
pub fn set_retry_count_for_test(&mut self, channel_id: Uuid, count: u32) {
self.retry_counts.insert(channel_id, count);
}
/// Drop all queued (non-in-flight) events for a channel.
///
/// Used when the agent is removed from a channel — any pending events

240
crates/buzz-agent/src/llm.rs
View File

@@ -1027,6 +1027,28 @@ fn is_retryable_transport_error(e: &reqwest::Error) -> bool {
e.is_timeout() || e.is_connect() || e.is_request()
}
/// Build the terminal `AgentError::Llm` for a `post()` exit that has given up
/// retrying — persistent retryable status, transport failure, or a body-read
/// break. `detail` carries the specific cause (status/body, or the transport
/// error text); `elapsed` and `attempts` are the cumulative cost of every
/// attempt made on this call, including the retries that failed before this
/// one. When cumulative time crosses `STALL_NOTICE_THRESHOLD`, this also logs
/// a `tracing::warn!` so a slow-building outage is visible in logs even
/// before an operator reads the returned error text.
fn terminal_llm_error(elapsed: std::time::Duration, attempts: u32, detail: &str) -> AgentError {
if elapsed >= STALL_NOTICE_THRESHOLD {
tracing::warn!(
cumulative_stall = ?elapsed,
attempts,
"llm: cumulative stall {elapsed:?} across {attempts} attempts ({detail})"
);
}
AgentError::Llm(format!(
"{detail} (cumulative {elapsed:?}, {attempts} attempt{})",
if attempts == 1 { "" } else { "s" },
))
}
async fn post<F>(http: &Client, url: &str, body: &Value, apply: F) -> Result<Value, AgentError>
where
F: Fn(reqwest::RequestBuilder) -> reqwest::RequestBuilder,
@@ -1055,14 +1077,11 @@ where
backoff_with_jitter(attempt).await;
continue;
}
let elapsed = call_start.elapsed();
if elapsed >= STALL_NOTICE_THRESHOLD {
tracing::warn!(
cumulative_stall = ?elapsed,
"llm: cumulative stall {elapsed:?} across {attempt} retries (transport failure)"
);
}
return Err(AgentError::Llm(format!("transport: {e}")));
return Err(terminal_llm_error(
call_start.elapsed(),
attempt + 1,
&format!("transport: {e}"),
));
}
};
let status = resp.status();
@@ -1071,21 +1090,32 @@ where
// the two are indistinguishable at the HTTP-status layer. The caller's
// retry loop keys off `LlmAuth` and refreshes once; the per-call guard
// bounds a pure-authz 403 to one wasted refresh before it propagates.
// Not a stall path: auth failures are not surfaced through
// terminal_llm_error, they resolve on the next call after refresh.
if status == 401 || status == 403 {
return Err(AgentError::LlmAuth(read_error_body(resp).await));
}
if (status.is_server_error() || status == 429 || status.as_u16() == 499)
&& attempt + 1 < MAX_RETRIES
{
tracing::warn!(
attempt = attempt + 1,
max_attempts = MAX_RETRIES,
%status,
"llm: retryable status, retrying"
);
backoff_with_jitter(attempt).await;
continue;
if status.is_server_error() || status == 429 || status.as_u16() == 499 {
if attempt + 1 < MAX_RETRIES {
tracing::warn!(
attempt = attempt + 1,
max_attempts = MAX_RETRIES,
%status,
"llm: retryable status, retrying"
);
backoff_with_jitter(attempt).await;
continue;
}
let body = read_error_body(resp).await;
return Err(terminal_llm_error(
call_start.elapsed(),
attempt + 1,
&format!("exhausted retries: {status}: {body}"),
));
}
// Not a stall path: the model is misconfigured, not the transport or
// upstream capacity — no retry was attempted, so cumulative duration
// would be misleading.
if status == 404 {
return Err(AgentError::LlmModelNotFound(format!(
"{status}: {}",
@@ -1118,21 +1148,18 @@ where
buf.extend_from_slice(&chunk);
}
Ok(None) => break,
Err(e) => return Err(AgentError::Llm(format!("read: {e}"))),
Err(e) => {
return Err(terminal_llm_error(
call_start.elapsed(),
attempt + 1,
&format!("body read: {e}"),
));
}
}
}
return serde_json::from_slice(&buf).map_err(|e| AgentError::Llm(format!("json: {e}")));
}
let elapsed = call_start.elapsed();
if elapsed >= STALL_NOTICE_THRESHOLD {
tracing::warn!(
cumulative_stall = ?elapsed,
"llm: cumulative stall {elapsed:?} across {MAX_RETRIES} retries (exhausted)"
);
}
Err(AgentError::Llm(format!(
"exhausted retries (cumulative {elapsed:?})"
)))
unreachable!("loop always returns on its final iteration (attempt + 1 == MAX_RETRIES)");
}
/// Build the `TokenSource` for the configured provider.
@@ -1193,6 +1220,7 @@ mod tests {
use crate::config::{Config, HookServers, OpenAiApi, Provider};
use crate::types::{HistoryItem, ToolCall, ToolResult, ToolResultContent};
use std::time::Duration;
use tracing_subscriber::layer::SubscriberExt;
fn cfg(provider: Provider) -> Config {
Config {
@@ -2316,10 +2344,19 @@ mod tests {
let err = post(&client, &url, &serde_json::json!({}), |b| b)
.await
.unwrap_err();
assert!(
matches!(&err, AgentError::Llm(msg) if msg.contains("499")),
"expected AgentError::Llm mentioning 499, got: {err:?}"
);
match &err {
AgentError::Llm(msg) => {
assert!(
msg.contains("exhausted retries") && msg.contains("499"),
"expected 'exhausted retries' + '499' in error, got: {msg}"
);
assert!(
msg.contains("cumulative") && msg.contains("3 attempts"),
"expected cumulative duration + exact attempt count, got: {msg}"
);
}
other => panic!("expected AgentError::Llm, got: {other:?}"),
}
assert_eq!(
accepts.load(Ordering::SeqCst),
MAX_RETRIES,
@@ -2327,6 +2364,141 @@ mod tests {
);
}
/// `terminal_llm_error` below `STALL_NOTICE_THRESHOLD` carries the detail
/// and attempt count but no stall-specific text — this is the common case
/// (a handful of quick retries), not an outage.
#[test]
fn terminal_llm_error_below_threshold_carries_detail_no_stall_claim() {
let err = terminal_llm_error(Duration::from_secs(2), 3, "exhausted retries: 499: body");
match err {
AgentError::Llm(msg) => {
assert!(msg.contains("cumulative 2s"), "must carry duration: {msg}");
assert!(
msg.contains("3 attempts"),
"must carry attempt count: {msg}"
);
assert!(
msg.contains("exhausted retries") && msg.contains("499"),
"must preserve detail: {msg}"
);
}
other => panic!("expected Llm, got: {other:?}"),
}
}
/// Above `STALL_NOTICE_THRESHOLD`, the error still carries the same
/// detail/duration/count — the threshold only gates the extra
/// `tracing::warn!` (not separately assertable here), not the error text.
/// Singular "1 attempt" (not "1 attempts") confirms the pluralization.
#[test]
fn terminal_llm_error_above_threshold_uses_singular_attempt() {
let err = terminal_llm_error(Duration::from_secs(301), 1, "transport: connection reset");
match err {
AgentError::Llm(msg) => {
assert!(
msg.contains("cumulative 301s"),
"must carry duration: {msg}"
);
assert!(msg.contains("1 attempt"), "must carry count: {msg}");
assert!(
!msg.contains("1 attempts"),
"singular for count of 1: {msg}"
);
}
other => panic!("expected Llm, got: {other:?}"),
}
}
/// Captures `tracing::warn!` events emitted during a scoped subscriber
/// so a test can assert on the stall-notice threshold without a real
/// sleep or a global logger.
struct StallWarnCapture {
count: Arc<std::sync::atomic::AtomicUsize>,
}
struct StallWarnVisitor {
saw_cumulative_stall: bool,
saw_attempts: bool,
}
impl tracing::field::Visit for StallWarnVisitor {
fn record_debug(&mut self, field: &tracing::field::Field, _value: &dyn std::fmt::Debug) {
match field.name() {
"cumulative_stall" => self.saw_cumulative_stall = true,
"attempts" => self.saw_attempts = true,
_ => {}
}
}
fn record_u64(&mut self, field: &tracing::field::Field, _value: u64) {
if field.name() == "attempts" {
self.saw_attempts = true;
}
}
}
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for StallWarnCapture {
fn on_event(
&self,
event: &tracing::Event<'_>,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if *event.metadata().level() != tracing::Level::WARN {
return;
}
let mut visitor = StallWarnVisitor {
saw_cumulative_stall: false,
saw_attempts: false,
};
event.record(&mut visitor);
if visitor.saw_cumulative_stall && visitor.saw_attempts {
self.count.fetch_add(1, Ordering::SeqCst);
}
}
}
/// Runs `f` under a scoped subscriber that only counts `WARN` events
/// carrying both `cumulative_stall` and `attempts` fields — the shape
/// `terminal_llm_error`'s stall notice emits. Returns that count.
fn count_stall_warnings(f: impl FnOnce()) -> usize {
let count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let subscriber = tracing_subscriber::registry().with(StallWarnCapture {
count: count.clone(),
});
tracing::subscriber::with_default(subscriber, f);
count.load(Ordering::SeqCst)
}
/// Below `STALL_NOTICE_THRESHOLD`, `terminal_llm_error` must not emit the
/// stall warning at all — otherwise every ordinary retry-exhaustion would
/// falsely read as an outage. This is mutation-sensitive: deleting the
/// threshold branch, weakening `>=` to `>` at the boundary, or an
/// always-warn mutation are all caught by pairing this with the 300s case
/// below.
#[test]
fn terminal_llm_error_below_threshold_emits_no_stall_warning() {
let warnings = count_stall_warnings(|| {
let _ = terminal_llm_error(Duration::from_secs(299), 3, "exhausted retries: 499");
});
assert_eq!(warnings, 0, "no stall warning below STALL_NOTICE_THRESHOLD");
}
/// At `STALL_NOTICE_THRESHOLD`, `terminal_llm_error` must emit exactly
/// one stall warning carrying the duration and attempt count — the
/// never-warn mutation is caught here; combined with the 299s case above,
/// a deleted branch or an always-warn mutation are caught by whichever
/// assertion it violates.
#[test]
fn terminal_llm_error_at_threshold_emits_one_stall_warning() {
let warnings = count_stall_warnings(|| {
let _ = terminal_llm_error(Duration::from_secs(300), 5, "transport: connection reset");
});
assert_eq!(
warnings, 1,
"exactly one stall warning with duration+attempts fields at threshold"
);
}
// ---- usage / input-token extraction -------------------------------------
#[test]