External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(timeout): unified turn-timeout fix — cap inheritance, steer renewal, activity-aware requeue, LLM stall surfacing (#2175)

Browse Source
This commit is contained in:
Will Pfleger
2026-07-20 18:20:57 -04:00
committed by GitHub
parent 2eff2cb391
commit 0f86a608b9
11 changed files with 1046 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-acp/README.md
+1 -1
View File
@@ -103,7 +103,7 @@ All configuration is via environment variables (or CLI flags — every env var h
| `BUZZ_ACP_AGENT_ARGS` | no | `acp` | Agent arguments (comma-separated). |
| `BUZZ_ACP_MCP_COMMAND` | no | `""` (empty) | Path to an optional MCP server binary to provide to the agent subprocess. |
| `BUZZ_ACP_IDLE_TIMEOUT` | no | `620` | Idle timeout: max seconds of silence before cancelling a turn. Resets on any agent stdout activity. |
| `BUZZ_ACP_MAX_TURN_DURATION` | no | `3600` | Absolute wall-clock cap per turn (safety valve). |
| `BUZZ_ACP_MAX_TURN_DURATION` | no | `7200` | Absolute wall-clock cap per turn (safety valve). |
| `BUZZ_API_TOKEN` | no | — | API token (required if relay enforces token auth). |
**Note:** `BUZZ_ACP_AGENT_ARGS` splits on commas. For args with values, use: `-c,key="value"`.
crates/buzz-acp/src/acp.rs
+141 -35
View File
@@ -89,8 +89,8 @@ pub enum AcpError {
#[error("Idle timeout — no agent activity for {0:?}")]
IdleTimeout(std::time::Duration),
#[error("Hard turn timeout exceeded")]
HardTimeout,
#[error("Hard turn timeout exceeded (silence {silence:?})")]
HardTimeout { silence: std::time::Duration },
#[error("Agent did not stop within {0:?} after cancellation")]
CancelDrainTimeout(std::time::Duration),
@@ -704,7 +704,13 @@ impl AcpClient {
}
let result = self
.read_until_response_with_idle_timeout(session_id, id, idle_timeout, hard_deadline)
.read_until_response_with_idle_timeout(
session_id,
id,
idle_timeout,
hard_deadline,
max_duration,
)
.await;
// On timeout errors, leave current_hard_deadline set so cancel_with_cleanup
@@ -714,7 +720,7 @@ impl AcpClient {
self.last_prompt_id = None;
self.current_hard_deadline = None;
}
Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => {
Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout { .. }) => {
// Leave last_prompt_id and current_hard_deadline set —
// caller will invoke cancel_with_cleanup.
}
@@ -879,7 +885,7 @@ impl AcpClient {
.cancel_with_cleanup_until(session_id, hard_deadline)
.await
{
Err(AcpError::HardTimeout) => Err(AcpError::CancelDrainTimeout(grace)),
Err(AcpError::HardTimeout { .. }) => Err(AcpError::CancelDrainTimeout(grace)),
other => other,
}
}
@@ -919,12 +925,16 @@ impl AcpClient {
// The separate hard_deadline bounds agents that keep producing output
// but ignore cancellation.
let cleanup_idle = std::time::Duration::from_secs(30);
let remaining = hard_deadline
.checked_duration_since(tokio::time::Instant::now())
.unwrap_or_default();
let result = self
.read_until_response_with_idle_timeout(
session_id,
prompt_id,
cleanup_idle,
hard_deadline,
remaining,
)
.await?;
self.parse_stop_reason(&result)
@@ -1187,6 +1197,7 @@ impl AcpClient {
expected_id: u64,
idle_timeout: std::time::Duration,
hard_deadline: tokio::time::Instant,
max_duration: std::time::Duration,
) -> Result<serde_json::Value, AcpError> {
use tokio::time::Instant;
@@ -1205,7 +1216,10 @@ impl AcpClient {
let mut pending_steer: Option<(u64, tokio::sync::oneshot::Sender<crate::pool::SteerAck>)> =
None;
let mut idle_deadline = Instant::now() + idle_timeout;
let now = Instant::now();
let mut idle_deadline = now + idle_timeout;
let mut hard_deadline = hard_deadline;
let mut last_activity_at = now;
loop {
// Determine which deadline fires first BEFORE sleeping — this is
@@ -1236,8 +1250,9 @@ impl AcpClient {
tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity");
return Err(AcpError::IdleTimeout(idle_timeout));
} else {
tracing::warn!("hard turn timeout exceeded");
return Err(AcpError::HardTimeout);
let silence = Instant::now().saturating_duration_since(last_activity_at);
tracing::warn!("hard turn timeout exceeded (silence {silence:?})");
return Err(AcpError::HardTimeout { silence });
}
}
@@ -1331,8 +1346,9 @@ impl AcpClient {
tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity");
return Err(AcpError::IdleTimeout(idle_timeout));
} else {
tracing::warn!("hard turn timeout exceeded");
return Err(AcpError::HardTimeout);
let silence = Instant::now().saturating_duration_since(last_activity_at);
tracing::warn!("hard turn timeout exceeded (silence {silence:?})");
return Err(AcpError::HardTimeout { silence });
}
}
};
@@ -1393,9 +1409,9 @@ impl AcpClient {
};
self.observe("acp_read", msg.clone());
// Only reset the idle clock on lines that parse as valid JSON.
// Malformed lines (skipped above) don't count as real agent activity.
idle_deadline = Instant::now() + idle_timeout;
let activity_now = Instant::now();
idle_deadline = activity_now + idle_timeout;
last_activity_at = activity_now;
// Steer response routing must come BEFORE the prompt
// response check: a steer response is a regular
@@ -1421,6 +1437,15 @@ impl AcpClient {
crate::pool::SteerError::AgentError { code, message },
)
} else {
let renew_now = Instant::now();
let new_deadline = renew_now + max_duration;
if new_deadline > hard_deadline {
hard_deadline = new_deadline;
self.current_hard_deadline = Some(new_deadline);
tracing::info!(
"steer success: renewed hard deadline ({max_duration:?} from now)"
);
}
crate::pool::SteerAck::Success
};
let _ = ack_tx.send(ack);
@@ -1449,11 +1474,10 @@ impl AcpClient {
match method {
"session/update" => {
if self.handle_session_update(&msg) {
// Belt-and-suspenders — general reset already fired
// above, this is defense-in-depth in case the general
// reset is later narrowed.
let activity_now = Instant::now();
idle_deadline = activity_now + idle_timeout;
last_activity_at = activity_now;
tracing::debug!("idle clock reset: tool call started");
idle_deadline = Instant::now() + idle_timeout;
}
}
"_goose/unstable/session/update" => {
@@ -2550,7 +2574,9 @@ mod tests {
#[test]
fn hard_timeout_error_display() {
let err = AcpError::HardTimeout;
let err = AcpError::HardTimeout {
silence: std::time::Duration::from_secs(120),
};
let msg = err.to_string();
assert!(
msg.contains("Hard turn timeout"),
@@ -2567,13 +2593,15 @@ mod tests {
#[tokio::test]
async fn idle_timeout_fires_on_silent_process() {
let mut client = spawn_script("sleep 10").await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
let max_dur = std::time::Duration::from_secs(30);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let result = client
.read_until_response_with_idle_timeout(
"test",
999,
std::time::Duration::from_millis(100),
hard_deadline,
max_dur,
)
.await;
assert!(
@@ -2585,7 +2613,8 @@ mod tests {
#[tokio::test]
async fn hard_timeout_fires_when_deadline_is_immediate() {
let mut client = spawn_script("while true; do echo 'noise'; sleep 0.01; done").await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(1);
let max_dur = std::time::Duration::from_millis(1);
let hard_deadline = tokio::time::Instant::now() + max_dur;
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
let result = client
.read_until_response_with_idle_timeout(
@@ -2593,10 +2622,11 @@ mod tests {
999,
std::time::Duration::from_secs(60),
hard_deadline,
max_dur,
)
.await;
assert!(
matches!(result, Err(AcpError::HardTimeout)),
matches!(result, Err(AcpError::HardTimeout { .. })),
"expected HardTimeout, got {result:?}"
);
}
@@ -2630,7 +2660,8 @@ mod tests {
r#"for i in $(seq 1 10); do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_thought_chunk","content":{"text":"thinking"}}}}'; sleep 0.05; done; sleep 10"#,
)
.await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(10);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let start = std::time::Instant::now();
let result = client
.read_until_response_with_idle_timeout(
@@ -2638,6 +2669,7 @@ mod tests {
999,
std::time::Duration::from_millis(200),
hard_deadline,
max_dur,
)
.await;
let elapsed = start.elapsed();
@@ -2652,13 +2684,15 @@ mod tests {
let mut client =
spawn_script(r#"echo '{"jsonrpc":"2.0","id":42,"result":{"stopReason":"end_turn"}}'"#)
.await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
let max_dur = std::time::Duration::from_secs(5);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let result = client
.read_until_response_with_idle_timeout(
"test",
42,
std::time::Duration::from_secs(2),
hard_deadline,
max_dur,
)
.await;
assert!(result.is_ok());
@@ -2669,13 +2703,15 @@ mod tests {
async fn agent_exit_detected_as_eof() {
let mut client = spawn_script("exit 0").await;
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
let max_dur = std::time::Duration::from_secs(5);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let result = client
.read_until_response_with_idle_timeout(
"test",
999,
std::time::Duration::from_secs(2),
hard_deadline,
max_dur,
)
.await;
assert!(matches!(result, Err(AcpError::AgentExited)));
@@ -2697,13 +2733,15 @@ mod tests {
sleep 1
"#;
let mut client = spawn_script(script).await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
let max_dur = std::time::Duration::from_secs(5);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let result = client
.read_until_response_with_idle_timeout(
"test",
0,
std::time::Duration::from_secs(3),
hard_deadline,
max_dur,
)
.await;
assert!(result.is_ok(), "expected Ok response, got {result:?}");
@@ -2714,9 +2752,10 @@ mod tests {
async fn idle_fires_before_hard_when_idle_is_shorter() {
let mut client = spawn_script("sleep 10").await;
let idle = std::time::Duration::from_millis(100);
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(10);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let result = client
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline)
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline, max_dur)
.await;
assert!(
matches!(result, Err(AcpError::IdleTimeout(_))),
@@ -2763,11 +2802,11 @@ mod tests {
let idle = std::time::Duration::from_secs(60); // idle ≫ hard
let start = std::time::Instant::now();
let result = client
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline)
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline, hard)
.await;
let elapsed = start.elapsed();
assert!(
matches!(result, Err(AcpError::HardTimeout)),
matches!(result, Err(AcpError::HardTimeout { .. })),
"expected HardTimeout under gapless valid-JSON stream, got {result:?} (elapsed {elapsed:?})"
);
// Must fire close to the hard deadline, not late. Without the
@@ -2818,7 +2857,8 @@ mod tests {
r#"for i in $(seq 1 20); do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"keepalive"}}}'; sleep 0.05; done; sleep 10"#,
)
.await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(10);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let start = std::time::Instant::now();
let result = client
.read_until_response_with_idle_timeout(
@@ -2826,6 +2866,7 @@ mod tests {
999,
std::time::Duration::from_millis(100),
hard_deadline,
max_dur,
)
.await;
let elapsed = start.elapsed();
@@ -2852,7 +2893,8 @@ mod tests {
r#"echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"tool_call","title":"long_running","kind":"shell"}}}'; sleep 0.08; sleep 10"#,
)
.await;
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(10);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let start = std::time::Instant::now();
let result = client
.read_until_response_with_idle_timeout(
@@ -2860,6 +2902,7 @@ mod tests {
999,
std::time::Duration::from_millis(200),
hard_deadline,
max_dur,
)
.await;
let elapsed = start.elapsed();
@@ -3145,9 +3188,10 @@ mod tests {
// be matched (the script writes nothing); the read loop will
// exit via IdleTimeout shortly after the steer arm fires.
let idle = std::time::Duration::from_millis(500);
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
let max_dur = std::time::Duration::from_secs(5);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let read_result = client
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline)
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline, max_dur)
.await;
send_task.await.expect("send_task should complete");
@@ -3212,9 +3256,10 @@ mod tests {
// the script so the read loop exits via idle timeout after the
// steer response is routed to ack.
let idle = std::time::Duration::from_secs(2);
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(10);
let hard_deadline = tokio::time::Instant::now() + max_dur;
let read_result = client
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline)
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline, max_dur)
.await;
send_task.await.expect("send_task should complete");
@@ -3241,6 +3286,67 @@ mod tests {
}
}
/// Steer-success renewal keeps the turn alive past the original hard
/// deadline. This is the red-on-old/green-on-new test for the core bug
/// fix (acp.rs:1440-1444): without renewal, the read loop returns
/// `HardTimeout` before the prompt response arrives.
///
/// Timeline:
/// t≈0: read loop starts, `hard_deadline = now + 1s`
/// t≈0.5s: script emits steer response (id=0) → Success renewal
/// moves `hard_deadline` to `now + 3s` (≈3.5s from start)
/// t≈1.5s: script emits prompt response (id=999) → `Ok`
///
/// Old code: `HardTimeout` at t≈1s (before prompt response).
/// New code: deadline renewed at t≈0.5s → prompt response at t≈1.5s → `Ok`.
#[tokio::test]
async fn steer_success_renews_hard_deadline_and_survives_past_original() {
let script = "sleep 0.5; \
echo '{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{\"stopReason\":\"end_turn\"}}'; \
sleep 1; \
echo '{\"jsonrpc\":\"2.0\",\"id\":999,\"result\":{\"done\":true}}'";
let mut client = spawn_script(script).await;
let update = session_info_update_msg(Some(serde_json::json!("run-99")));
let _ = client.handle_session_update(&update);
let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::<crate::pool::SteerRequest>(1);
client.install_steer_rx(steer_rx);
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<crate::pool::SteerAck>();
let send_task = tokio::spawn(async move {
steer_tx
.send(crate::pool::SteerRequest {
prompt_blocks: vec!["steer body".into()],
ack_tx,
})
.await
.expect("steer_tx send should succeed");
});
let idle = std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(3);
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
let result = client
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline, max_dur)
.await;
send_task.await.expect("send_task should complete");
assert!(
result.is_ok(),
"expected Ok (prompt response after renewed deadline), got {result:?}"
);
assert_eq!(result.unwrap()["done"], serde_json::json!(true));
let ack = ack_rx
.await
.expect("ack oneshot must have received a SteerAck");
match ack {
crate::pool::SteerAck::Success => {}
other => panic!("expected SteerAck::Success, got {other:?}"),
}
}
// ── Goose usage notification integration ──────────────────────────────
/// Build a `_goose/unstable/session/update` JSON-RPC notification.
crates/buzz-acp/src/lib.rs
+510 -25
View File
@@ -2331,6 +2331,9 @@ async fn tokio_main() -> Result<()> {
signal_fallback,
"non-cancelling steer ack received"
);
if matches!(ack, Ok(pool::SteerAck::Success)) {
queue.extend_in_flight_deadline(channel_id, config.max_turn_duration_secs);
}
if drop_withheld {
queue.remove_event(channel_id, &event_id);
}
@@ -2808,15 +2811,16 @@ fn handle_prompt_result(
// accounting, same as a clean cancel.
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.
} else if matches!(
result.outcome,
PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: false
})
) {
tracing::error!(
channel_id = %batch.channel_id,
events = batch.events.len(),
"dead-lettering batch after hard-cap timeout — discarding {} events",
"dead-lettering batch after hard-cap timeout (no recent activity) — discarding {} events",
batch.events.len(),
);
let content = format!(
@@ -2824,16 +2828,28 @@ fn handle_prompt_result(
config.max_turn_duration_secs
);
spawn_failure_notice(rest_client, &batch, content);
} else if matches!(
result.outcome,
PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: true
})
) {
tracing::warn!(
channel_id = %batch.channel_id,
events = batch.events.len(),
"hard-cap timeout with recent activity — requeueing for retry"
);
if let Some(dead) = queue.requeue(batch) {
let content = format!(
"⚠️ I couldn't process the last request after multiple retries (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, &dead, 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.
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) => {
PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => {
"the turn exceeded the maximum duration".to_string()
}
PromptOutcome::AgentExited => "the agent process exited".to_string(),
@@ -2870,11 +2886,15 @@ fn handle_prompt_result(
PromptOutcome::Ok(_) => "ok",
PromptOutcome::Error(_) => "error",
PromptOutcome::Timeout(TimeoutKind::Idle) => "idle_timeout",
PromptOutcome::Timeout(TimeoutKind::Hard) => "hard_timeout",
PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => "hard_timeout",
PromptOutcome::AgentExited => "exited",
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
@@ -2934,10 +2954,17 @@ fn handle_prompt_result(
);
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
),
"hard_timeout" => {
let suffix = if hard_timeout_recently_active == Some(true) {
" — requeued for retry (recently active)"
} else {
" — dead-lettered (no recent activity)"
};
format!(
"Agent turn exceeded the maximum duration ({}s){}",
config.max_turn_duration_secs, suffix
)
}
_ => "Agent session timed out due to inactivity".to_string(),
};
emit_turn_error(&death_message, None);
@@ -4541,7 +4568,10 @@ mod error_outcome_emission_tests {
#[tokio::test]
async fn hard_timeout_emits_exactly_one_feed_event() {
assert_eq!(
turn_errors_emitted_for(PromptOutcome::Timeout(TimeoutKind::Hard)).await,
turn_errors_emitted_for(PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: false
}))
.await,
1
);
}
@@ -4615,7 +4645,13 @@ mod error_outcome_emission_tests {
);
};
check_label(PromptOutcome::Timeout(TimeoutKind::Idle), "idle_timeout").await;
check_label(PromptOutcome::Timeout(TimeoutKind::Hard), "hard_timeout").await;
check_label(
PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: false,
}),
"hard_timeout",
)
.await;
check_label(
PromptOutcome::CancelDrainTimeout(std::time::Duration::from_secs(5)),
"cancel_drain_timeout",
@@ -4697,15 +4733,23 @@ mod error_outcome_emission_tests {
)
};
// Hard timeout: batch must NOT be requeued (dead-lettered immediately).
// Hard timeout (not recently active): dead-lettered immediately.
let hard_batch = make_batch();
let (hard_channels, hard_events) =
run(PromptOutcome::Timeout(TimeoutKind::Hard), hard_batch).await;
let (hard_channels, hard_events) = run(
PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: false,
}),
hard_batch,
)
.await;
assert_eq!(
hard_channels, 0,
"hard-cap timeout must not requeue the batch"
"hard-cap timeout (not recently active) must not requeue the batch"
);
assert_eq!(
hard_events, 0,
"hard-cap timeout (not recently active) must drop all events"
);
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();
@@ -4721,6 +4765,97 @@ mod error_outcome_emission_tests {
);
}
#[tokio::test]
async fn hard_timeout_recently_active_requeues_batch() {
let channel_id = Uuid::new_v4();
let make_batch = || {
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: std::time::Instant::now(),
}],
cancelled_events: vec![],
cancel_reason: None,
}
};
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();
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 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,
None,
None,
);
(
queue.pending_channels(),
queue.queued_event_count(&channel_id),
)
};
let batch = make_batch();
let (channels, events) = run(
PromptOutcome::Timeout(TimeoutKind::Hard {
recently_active: true,
}),
batch,
)
.await;
assert_eq!(
channels, 1,
"hard-cap timeout with recent activity must requeue the batch"
);
assert_eq!(
events, 1,
"hard-cap timeout with recent activity must preserve the event"
);
}
/// 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
@@ -5252,3 +5387,353 @@ 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"
);
}
}
crates/buzz-acp/src/pool.rs
+20 -10
View File
@@ -40,6 +40,10 @@ use crate::queue::{
};
use crate::relay::{ChannelInfo, RestClient};
/// Window within which agent activity before a hard-cap death qualifies
/// the turn as "recently active" (eligible for requeue instead of dead-letter).
const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60);
// FlushBatch and BatchEvent derive Clone (added in queue.rs) so we can store
// a recoverable copy in TaskMeta for panic recovery in Queue mode.
@@ -392,7 +396,9 @@ pub enum TimeoutKind {
/// No ACP wire activity for `idle_timeout` seconds.
Idle,
/// Turn ran for `max_turn_duration` seconds of wall-clock time.
Hard,
/// `recently_active` is true when the agent produced output within
/// `RECENT_ACTIVITY_WINDOW` of the hard-cap firing.
Hard { recently_active: bool },
}
/// Outcome of a prompt task.
@@ -1617,10 +1623,11 @@ pub async fn run_prompt_task(
);
return;
}
Err(AcpError::HardTimeout) => {
Err(AcpError::HardTimeout { silence }) => {
let recently_active = silence < RECENT_ACTIVITY_WINDOW;
tracing::error!(
target: "pool::session",
"hard timeout ({}s cap) during initial_message for channel {cid} — agent process is unrecoverable",
"hard timeout ({}s cap, silence {silence:?}, recently_active={recently_active}) during initial_message for channel {cid} — agent process is unrecoverable",
ctx.max_turn_duration.as_secs()
);
agent.state.invalidate_all();
@@ -1629,7 +1636,7 @@ pub async fn run_prompt_task(
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Hard),
PromptOutcome::Timeout(TimeoutKind::Hard { recently_active }),
requeue_batch_if_queue(&ctx, batch),
);
return;
@@ -2094,10 +2101,11 @@ pub async fn run_prompt_task(
}
}
}
Err(AcpError::HardTimeout) => {
Err(AcpError::HardTimeout { silence }) => {
let recently_active = silence < RECENT_ACTIVITY_WINDOW;
tracing::error!(
target: "pool::prompt",
"hard timeout ({}s cap) — agent process is unrecoverable, invalidating all sessions",
"hard timeout ({}s cap, silence {silence:?}, recently_active={recently_active}) — agent process is unrecoverable, invalidating all sessions",
ctx.max_turn_duration.as_secs()
);
agent.state.invalidate_all();
@@ -2116,7 +2124,7 @@ pub async fn run_prompt_task(
&turn_id,
agent,
source,
PromptOutcome::Timeout(TimeoutKind::Hard),
PromptOutcome::Timeout(TimeoutKind::Hard { recently_active }),
requeue_batch_if_queue(&ctx, batch),
);
}
@@ -2989,7 +2997,7 @@ fn classify_control_cancel_failure(
// should be unreachable in practice. If it ever fires anyway, still
// report the truthful non-hard outcome rather than the real hard-cap
// (which would dead-letter the batch and claim the configured cap).
AcpError::HardTimeout => (
AcpError::HardTimeout { .. } => (
PromptOutcome::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
false,
),
@@ -4455,7 +4463,7 @@ mod tests {
let label = match outcome {
PromptOutcome::AgentExited => "AgentExited",
PromptOutcome::Timeout(TimeoutKind::Idle) => "Timeout(Idle)",
PromptOutcome::Timeout(TimeoutKind::Hard) => "Timeout(Hard)",
PromptOutcome::Timeout(TimeoutKind::Hard { .. }) => "Timeout(Hard)",
PromptOutcome::CancelDrainTimeout(_) => "CancelDrainTimeout",
PromptOutcome::Error(_) => "Error",
PromptOutcome::Cancelled => "Cancelled",
@@ -4533,7 +4541,9 @@ mod tests {
},
Case {
name: "unexpected HardTimeout cannot become Timeout(Hard)",
error: || AcpError::HardTimeout,
error: || AcpError::HardTimeout {
silence: Duration::from_secs(300),
},
signal: ControlSignal::Steer,
expected_outcome: "CancelDrainTimeout",
batch_preserved: true,
crates/buzz-acp/src/queue.rs
+224
View File
@@ -200,6 +200,27 @@ impl EventQueue {
self
}
/// Monotonically extend an existing in-flight deadline for `channel_id`.
///
/// Called when a successful steer grants a fresh turn budget. The new
/// deadline is `max(current, now + max_turn_secs + buffer)` — it never
/// moves backward. If the channel is not in-flight (already completed
/// via `mark_complete`), this is a no-op: a late ack never resurrects
/// a deadline.
pub fn extend_in_flight_deadline(&mut self, channel_id: Uuid, max_turn_secs: u64) {
if let Some(current) = self.in_flight_deadlines.get_mut(&channel_id) {
let extended = Instant::now()
+ Duration::from_secs(max_turn_secs + IN_FLIGHT_DEADLINE_BUFFER_SECS);
if extended > *current {
tracing::info!(
%channel_id,
"extending in-flight deadline by {max_turn_secs}s + {IN_FLIGHT_DEADLINE_BUFFER_SECS}s buffer"
);
*current = extended;
}
}
}
/// Push an event into the queue for its channel.
///
/// In [`DedupMode::Drop`], events for any currently in-flight channel are
@@ -4522,4 +4543,207 @@ mod tests {
"in_flight_deadline must be strictly greater than max_turn_duration"
);
}
#[test]
fn extend_in_flight_deadline_advances_existing_entry() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
let old_deadline = Instant::now() + Duration::from_secs(100);
q.in_flight_channels.insert(ch);
q.in_flight_deadlines.insert(ch, old_deadline);
q.extend_in_flight_deadline(ch, 7200);
let new = *q.in_flight_deadlines.get(&ch).unwrap();
assert!(
new > old_deadline,
"extended deadline must be past the original"
);
}
#[test]
fn extend_in_flight_deadline_is_monotonic() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
let far_future = Instant::now() + Duration::from_secs(999_999);
q.in_flight_channels.insert(ch);
q.in_flight_deadlines.insert(ch, far_future);
q.extend_in_flight_deadline(ch, 7200);
let after = *q.in_flight_deadlines.get(&ch).unwrap();
assert_eq!(after, far_future, "deadline must never move backward");
}
#[test]
fn extend_in_flight_deadline_noop_after_mark_complete() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
q.in_flight_channels.insert(ch);
q.in_flight_deadlines
.insert(ch, Instant::now() + Duration::from_secs(100));
q.in_flight_batch_sizes.insert(ch, 1);
q.mark_complete(ch);
assert!(!q.in_flight_deadlines.contains_key(&ch));
q.extend_in_flight_deadline(ch, 7200);
assert!(
!q.in_flight_deadlines.contains_key(&ch),
"extend after mark_complete must not resurrect a deadline"
);
}
#[test]
fn compact_expired_state_preserves_extended_in_flight_deadline() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
let extended = Instant::now() + Duration::from_secs(9999);
q.in_flight_channels.insert(ch);
q.in_flight_deadlines.insert(ch, extended);
q.compact_expired_state();
assert!(
q.in_flight_deadlines.contains_key(&ch),
"compaction must not touch in-flight deadlines"
);
assert_eq!(
*q.in_flight_deadlines.get(&ch).unwrap(),
extended,
"compaction must leave extended deadline intact"
);
}
// ── F2 case 2.2: extend_in_flight_deadline prevents flush_next expiry ────
/// A channel in-flight with a past deadline is auto-expired by
/// `flush_next` and the "BUG: in-flight channel expired" path fires.
/// Verify this baseline — we need the expiry to actually happen for the
/// next test's "extended deadline prevents it" assertion to be meaningful.
#[test]
fn expired_in_flight_deadline_is_auto_released_by_flush_next() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
// Insert the channel as in-flight with a deadline already in the past
// (Instant::now() — by the time flush_next runs, now >= deadline).
q.in_flight_channels.insert(ch);
q.in_flight_deadlines.insert(ch, Instant::now());
q.in_flight_batch_sizes.insert(ch, 1);
// Also push an event so flush_next has something to do after expiry.
q.push(make_queued(ch, "after-expiry"));
// flush_next auto-expires the stuck entry and then dispatches the
// pending event for the now-freed channel.
let batch = q
.flush_next()
.expect("channel should be dispatchable after auto-expiry");
assert_eq!(batch.channel_id, ch);
assert_eq!(batch.events[0].event.content, "after-expiry");
}
/// F2 case 2.2 — `flush_next` path.
///
/// A channel whose in-flight deadline was extended far into the future
/// must NOT be auto-expired by `flush_next`. No "BUG: in-flight channel
/// expired" logic fires; the channel stays in-flight; the pending event
/// for it is not dispatched a second time.
#[test]
fn extended_deadline_prevents_flush_next_expiry() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
// Put the channel in-flight with an extended deadline far in the future.
q.in_flight_channels.insert(ch);
q.in_flight_deadlines
.insert(ch, Instant::now() + Duration::from_secs(9999));
q.in_flight_batch_sizes.insert(ch, 1);
// Push an event for another channel so flush_next has work to do.
let ch2 = Uuid::new_v4();
q.push(make_queued(ch2, "other-channel"));
let batch = q.flush_next().expect("other channel should flush");
assert_eq!(
batch.channel_id, ch2,
"only ch2 should be flushed; ch is still in-flight with extended deadline"
);
// ch must still be in-flight — the extended deadline did not expire.
assert!(
q.in_flight_channels.contains(&ch),
"ch must remain in-flight after flush_next with an extended deadline"
);
assert!(
q.in_flight_deadlines.contains_key(&ch),
"in-flight deadline for ch must not be removed by flush_next"
);
}
/// F2 case 2.2 — `has_flushable_work` path.
///
/// Same guarantee as above but exercising `has_flushable_work` instead of
/// `flush_next`. A channel with an extended deadline far in the future
/// must not be auto-expired, and the method must return the correct answer
/// based on the other pending channel only.
#[test]
fn extended_deadline_prevents_has_flushable_work_expiry() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
// In-flight channel with extended (far-future) deadline.
q.in_flight_channels.insert(ch);
q.in_flight_deadlines
.insert(ch, Instant::now() + Duration::from_secs(9999));
q.in_flight_batch_sizes.insert(ch, 1);
// No other channels — nothing flushable.
assert!(
!q.has_flushable_work(),
"has_flushable_work must return false when the only channel is in-flight with extended deadline"
);
assert!(
q.in_flight_channels.contains(&ch),
"ch must remain in-flight after has_flushable_work with extended deadline"
);
// Add a pending event for a different channel.
let ch2 = Uuid::new_v4();
q.push(make_queued(ch2, "pending"));
assert!(
q.has_flushable_work(),
"has_flushable_work must return true for the pending ch2 event"
);
// ch still in-flight and not expired.
assert!(
q.in_flight_channels.contains(&ch),
"ch must still be in-flight after has_flushable_work finds ch2 work"
);
}
// ── F2 case 2.5: steer renewal is monotonic across repeated steers ───────
/// Calling `extend_in_flight_deadline` twice with the same `max_turn_secs`
/// must only move the deadline forward; the second call must produce a
/// deadline >= the first (monotonic guarantee, tested at queue-unit level).
#[test]
fn repeated_extend_in_flight_deadline_is_monotonic() {
let mut q = EventQueue::new(DedupMode::Queue);
let ch = Uuid::new_v4();
q.in_flight_channels.insert(ch);
q.in_flight_deadlines
.insert(ch, Instant::now() + Duration::from_secs(100));
q.extend_in_flight_deadline(ch, 7200);
let after_first = *q.in_flight_deadlines.get(&ch).unwrap();
q.extend_in_flight_deadline(ch, 7200);
let after_second = *q.in_flight_deadlines.get(&ch).unwrap();
assert!(
after_second >= after_first,
"second extend must not move deadline backward (monotonic)"
);
}
}
crates/buzz-agent/README.md
+2 -2
View File
@@ -147,7 +147,7 @@ Everything is environment variables. No flags, no config files. (We are a subpro
| `BUZZ_AGENT_MAX_OUTPUT_TOKENS` | `32768` | Per LLM call. Headroom for large tool-call inputs (e.g. file writes via heredoc); Sonnet 4 / Opus 4 cap at 64K. |
| `BUZZ_AGENT_MAX_CONTEXT_TOKENS` | `200000` | Provider context window used by the handoff gate. |
| `BUZZ_AGENT_MAX_HANDOFFS` | `10` | Max context handoffs per session before falling back to truncation. |
| `BUZZ_AGENT_LLM_TIMEOUT_SECS` | `240` | |
| `BUZZ_AGENT_LLM_TIMEOUT_SECS` | `240` | Max seconds with no response bytes before abandoning an LLM call (per-read inactivity, not wall-clock). |
| `BUZZ_AGENT_TOOL_TIMEOUT_SECS` | `660` | Per-tool call timeout in seconds |
| `BUZZ_AGENT_MAX_PARALLEL_TOOLS` | `8` | Max concurrent tool calls per turn (1 = sequential) |
| `BUZZ_AGENT_MAX_SESSIONS` | unlimited | Max concurrent ACP sessions. Sessions are cheap; default has no cap. |
@@ -244,7 +244,7 @@ The trust boundary is **the operator who launched the agent**. The harness, MCP
| Tool schema bytes | 4 KiB | `MAX_SCHEMA_BYTES` (oversize → replaced with `{}`) |
| Tool calls per turn | 64 | `MAX_TOOL_CALLS_PER_TURN` |
| Loop rounds | 0 (unlimited) | `BUZZ_AGENT_MAX_ROUNDS` |
| LLM call timeout | 240 s | `BUZZ_AGENT_LLM_TIMEOUT_SECS` |
| LLM read inactivity timeout | 240 s | `BUZZ_AGENT_LLM_TIMEOUT_SECS` |
| Tool call timeout | 660 s | `BUZZ_AGENT_TOOL_TIMEOUT_SECS` |
## What This Is NOT
crates/buzz-agent/src/llm.rs
+141 -3
View File
@@ -21,6 +21,7 @@ const DATABRICKS_OAUTH_SCOPES: &[&str] = &["all-apis", "offline_access"];
const MAX_LLM_RESPONSE_BYTES: usize = 16 * 1024 * 1024;
const MAX_LLM_ERROR_BODY_BYTES: usize = 4 * 1024;
const STALL_NOTICE_THRESHOLD: std::time::Duration = std::time::Duration::from_secs(300);
/// Parser for an OpenAI-family JSON response. Per-endpoint pair lives
/// alongside its `_body` serializer.
@@ -52,7 +53,7 @@ impl Llm {
pub fn new(cfg: &Config) -> Result<Self, AgentError> {
let http = Client::builder()
.connect_timeout(std::time::Duration::from_secs(10))
.timeout(cfg.llm_timeout)
.read_timeout(cfg.llm_timeout)
.build()
.map_err(|e| AgentError::Llm(format!("http: {e}")))?;
let auth = build_token_source(cfg)?;
@@ -1032,6 +1033,7 @@ where
{
let body_bytes =
serde_json::to_vec(body).map_err(|e| AgentError::Llm(format!("serialize: {e}")))?;
let call_start = std::time::Instant::now();
for attempt in 0..MAX_RETRIES {
let resp = match apply(
http.post(url)
@@ -1053,6 +1055,13 @@ 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}")));
}
};
@@ -1065,7 +1074,9 @@ where
if status == 401 || status == 403 {
return Err(AgentError::LlmAuth(read_error_body(resp).await));
}
if (status.is_server_error() || status == 429) && attempt + 1 < MAX_RETRIES {
if (status.is_server_error() || status == 429 || status.as_u16() == 499)
&& attempt + 1 < MAX_RETRIES
{
tracing::warn!(
attempt = attempt + 1,
max_attempts = MAX_RETRIES,
@@ -1112,7 +1123,16 @@ where
}
return serde_json::from_slice(&buf).map_err(|e| AgentError::Llm(format!("json: {e}")));
}
Err(AgentError::Llm("exhausted retries".into()))
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:?})"
)))
}
/// Build the `TokenSource` for the configured provider.
@@ -2189,6 +2209,124 @@ mod tests {
);
}
/// A 499 response is retried and the call succeeds on the second attempt.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_retries_499_and_succeeds() {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1/x", listener.local_addr().unwrap());
let accepts = Arc::new(AtomicU32::new(0));
let accepts_srv = accepts.clone();
tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
let n = accepts_srv.fetch_add(1, Ordering::SeqCst);
// Read the full request headers before responding.
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(k) => buf.extend_from_slice(&tmp[..k]),
}
}
if n == 0 {
// First attempt: respond with 499.
let resp = "HTTP/1.1 499 Client Closed Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
continue;
}
// Subsequent attempts: 200 OK with a tiny JSON body.
let body = "{\"ok\":true}";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
}
});
let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let out = post(&client, &url, &serde_json::json!({}), |b| b)
.await
.expect("post should succeed after 499 retry");
assert_eq!(out, serde_json::json!({ "ok": true }));
assert!(
accepts.load(Ordering::SeqCst) >= 2,
"server must see at least 2 attempts (got {})",
accepts.load(Ordering::SeqCst)
);
}
/// When all MAX_RETRIES attempts return 499 the error includes "exhausted retries".
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_exhausts_retries_on_persistent_499() {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1/x", listener.local_addr().unwrap());
let accepts = Arc::new(AtomicU32::new(0));
let accepts_srv = accepts.clone();
tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
accepts_srv.fetch_add(1, Ordering::SeqCst);
// Read the full request headers before responding.
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(k) => buf.extend_from_slice(&tmp[..k]),
}
}
// Always 499.
let resp = "HTTP/1.1 499 Client Closed Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
}
});
let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
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:?}"
);
assert_eq!(
accepts.load(Ordering::SeqCst),
MAX_RETRIES,
"server must see exactly MAX_RETRIES attempts — 499 must be retried"
);
}
// ---- usage / input-token extraction -------------------------------------
#[test]
desktop/src-tauri/src/managed_agents/runtime.rs
+3 -4
View File
@@ -1706,10 +1706,9 @@ pub fn spawn_agent_child(
command.env("BUZZ_ACP_IDLE_TIMEOUT", idle.to_string());
}
let max_dur = record
.max_turn_duration_seconds
.unwrap_or(super::types::DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS);
command.env("BUZZ_ACP_MAX_TURN_DURATION", max_dur.to_string());
if let Some(max_dur) = record.max_turn_duration_seconds {
command.env("BUZZ_ACP_MAX_TURN_DURATION", max_dur.to_string());
}
command.env("BUZZ_ACP_AGENTS", record.parallelism.to_string());
command.env("BUZZ_ACP_MULTIPLE_EVENT_HANDLING", "steer");
command.env("BUZZ_ACP_DEDUP", "queue");
desktop/src-tauri/src/managed_agents/spawn_hash.rs
+1 -5
View File
@@ -128,11 +128,7 @@ pub(crate) fn spawn_config_hash(
.hash(&mut hasher);
}
record.idle_timeout_seconds.hash(&mut hasher);
// Spawn writes BUZZ_ACP_MAX_TURN_DURATION with a default.
record
.max_turn_duration_seconds
.unwrap_or(super::types::DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS)
.hash(&mut hasher);
record.max_turn_duration_seconds.hash(&mut hasher);
record.parallelism.hash(&mut hasher);
hasher.finish()
desktop/src-tauri/src/managed_agents/spawn_hash/tests.rs
+3 -6
View File
@@ -257,14 +257,11 @@ fn allowlist_content_edit_still_changes_hash() {
}
#[test]
fn explicit_default_max_turn_duration_does_not_change_hash() {
// Spawn writes BUZZ_ACP_MAX_TURN_DURATION with the default filled in, so
// None → Some(default) is the same spawned value and must not badge.
fn explicit_max_turn_duration_changes_hash_from_none() {
let rec = record();
let mut edited = record();
edited.max_turn_duration_seconds =
Some(crate::managed_agents::types::DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS);
assert_eq!(
edited.max_turn_duration_seconds = Some(7200);
assert_ne!(
spawn_config_hash(&rec, &[], &[], "wss://ws.example", &Default::default()),
spawn_config_hash(&edited, &[], &[], "wss://ws.example", &Default::default())
);
desktop/src-tauri/src/managed_agents/types.rs
-2
View File
@@ -726,8 +726,6 @@ pub struct UpdateTeamRequest {
pub const DEFAULT_ACP_COMMAND: &str = "buzz-acp";
/// ~5 min (320s) — matches the CLI harness default (BUZZ_ACP_IDLE_TIMEOUT).
pub const DEFAULT_AGENT_TURN_TIMEOUT_SECONDS: u64 = 320;
/// 1 hour — absolute wall-clock safety cap per turn.
pub const DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS: u64 = 3600;
pub const DEFAULT_AGENT_PARALLELISM: u32 = 24;
fn default_agent_parallelism() -> u32 {