From 08913d3612a4fae8bb568c39f50f941193caf10a Mon Sep 17 00:00:00 2001 From: Stalker23b Date: Tue, 1 Sep 2026 21:31:39 +0800 Subject: [PATCH 1/2] fix(buzz-acp): refresh idle clocks on turn completion and steer ack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The idle-pool sleep and exit-after-inactivity clocks were only refreshed at dispatch, so any turn longer than the idle bound was torn down on the first reaper tick after it completed — killing the CLI process together with its live background tasks and pending wake-ups (observed in production: a 28-minute turn ended and the pool was reaped 13 seconds later, orphaning two background tasks whose completion the agent was waiting on). Turn completion and steer acks now count as activity. Heartbeat results are excluded: heartbeat dispatch is gated on an idle slot with no flushable work, so counting them would keep a heartbeat-enabled pool awake forever. Signed-off-by: Stalker23b Co-authored-by: Пчелинский Co-authored-by: Claude Fable 5 --- crates/buzz-acp/src/lib.rs | 43 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 146214197a8..517f9a25976 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1675,6 +1675,20 @@ fn idle_pool_sleep_due( && inactivity_expired(last_activity, now, bound, turn_in_flight) } +/// Whether a completed prompt result counts as activity for the idle clocks +/// (`idle_pool_sleep_secs` and `exit_after_inactivity_secs`). +/// +/// Turn completion must refresh `last_activity`: the clock is otherwise only +/// refreshed at dispatch, so a turn longer than the idle bound would be torn +/// down on the first reaper tick after it completes — killing the CLI process +/// together with any background tasks and autonomous wake-ups it still owns +/// (the waiter-loss incident of 2026-09-01). Heartbeat results are excluded: +/// heartbeats fire only into an idle pool, so counting them as activity would +/// keep a heartbeat-enabled pool awake forever. +fn result_refreshes_idle_clock(source: &PromptSource) -> bool { + matches!(source, PromptSource::Channel(_)) +} + #[cfg(test)] mod inactivity_tests { use super::*; @@ -1719,6 +1733,28 @@ mod inactivity_tests { } } +#[cfg(test)] +mod idle_clock_refresh_tests { + use super::*; + + #[test] + fn channel_turn_completion_refreshes_idle_clock() { + // A completed channel turn is activity: without this, any turn longer + // than the idle bound is torn down on the first reaper tick after it + // completes, orphaning its background tasks (waiter-loss incident). + assert!(result_refreshes_idle_clock(&PromptSource::Channel( + Uuid::new_v4() + ))); + } + + #[test] + fn heartbeat_completion_does_not_refresh_idle_clock() { + // Heartbeats fire only into an idle pool — counting them as activity + // would keep a heartbeat-enabled pool awake forever. + assert!(!result_refreshes_idle_clock(&PromptSource::Heartbeat)); + } +} + #[cfg(test)] mod idle_pool_sleep_tests { use super::*; @@ -3148,6 +3184,9 @@ async fn tokio_main() -> Result<()> { if let PromptSource::Channel(ch) = &result.source { typing_channels.remove(ch); } + if result_refreshes_idle_clock(&result.source) { + last_activity = tokio::time::Instant::now(); + } if handle_prompt_result( &mut pool, &mut queue, @@ -3290,6 +3329,10 @@ async fn tokio_main() -> Result<()> { // pending_steer on every return path. If it does, // treat as PromptCompletedNeutral to avoid leaking // the withheld event in `withheld_native_steer`. + // A steer ack is channel activity by definition (steers only + // exist for channel turns) — refresh the idle clocks so a + // mid-turn steer keeps the pool's idle window honest. + last_activity = tokio::time::Instant::now(); let (release_withheld, drop_withheld, signal_fallback) = match &ack { Ok(pool::SteerAck::Success { .. }) => (false, true, false), // -32601 = method_not_found: agent does not implement the From 91957a5493f44b2e5ef20ee00d5b62c503eba78e Mon Sep 17 00:00:00 2001 From: Stalker23b Date: Tue, 1 Sep 2026 21:31:40 +0800 Subject: [PATCH 2/2] fix(buzz-acp): move idle-clock refresh into handle_prompt_result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings: - The refresh now lives inside handle_prompt_result, so the unit tests covering that function also guard the wiring — previously deleting the call-site line left the suite green. - result_refreshes_idle_clock uses an exhaustive match: a new PromptSource variant must decide explicitly whether its completion keeps the pool alive. - Corrected the heartbeat comment: dispatch is gated on an idle slot with no flushable work (not "only into an idle pool" — in a multi-agent pool a heartbeat can fire while another agent's turn is live). - Two wiring tests: a channel result refreshes the idle clock through handle_prompt_result; a heartbeat result leaves it stale. Signed-off-by: Stalker23b Co-authored-by: Пчелинский Co-authored-by: Claude Fable 5 --- crates/buzz-acp/src/lib.rs | 160 +++++++++++++++++++++++++++++++++++-- 1 file changed, 154 insertions(+), 6 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 517f9a25976..4e22e31bf02 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1683,10 +1683,16 @@ fn idle_pool_sleep_due( /// down on the first reaper tick after it completes — killing the CLI process /// together with any background tasks and autonomous wake-ups it still owns /// (the waiter-loss incident of 2026-09-01). Heartbeat results are excluded: -/// heartbeats fire only into an idle pool, so counting them as activity would -/// keep a heartbeat-enabled pool awake forever. +/// heartbeat dispatch is gated on an idle slot with no flushable work, so a +/// pool kept alive only by its own heartbeats would never be reaped if their +/// completions counted as activity. fn result_refreshes_idle_clock(source: &PromptSource) -> bool { - matches!(source, PromptSource::Channel(_)) + // Exhaustive on purpose: a new PromptSource variant must decide + // explicitly whether its completion keeps the pool alive. + match source { + PromptSource::Channel(_) => true, + PromptSource::Heartbeat => false, + } } #[cfg(test)] @@ -3184,9 +3190,6 @@ async fn tokio_main() -> Result<()> { if let PromptSource::Channel(ch) = &result.source { typing_channels.remove(ch); } - if result_refreshes_idle_clock(&result.source) { - last_activity = tokio::time::Instant::now(); - } if handle_prompt_result( &mut pool, &mut queue, @@ -3199,6 +3202,7 @@ async fn tokio_main() -> Result<()> { &mut respawn_tasks, observer.clone(), Some(&ctx.rest_client), + &mut last_activity, ) == LoopAction::Exit { break; @@ -3921,7 +3925,14 @@ fn handle_prompt_result( respawn_tasks: &mut tokio::task::JoinSet<()>, observer: Option, rest_client: Option<&relay::RestClient>, + last_activity: &mut tokio::time::Instant, ) -> LoopAction { + // Refresh the idle clocks before anything else (and in particular before + // mark_complete drops the in-flight guard): a turn longer than the idle + // bound must not be reaped on the first tick after it completes. + if result_refreshes_idle_clock(&result.source) { + *last_activity = tokio::time::Instant::now(); + } let before = pool.task_map().len(); let agent_index = result.agent.index; let successful_steer_deliveries = pool @@ -7173,6 +7184,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7181,6 +7193,130 @@ mod error_outcome_emission_tests { .contains(steer_event_id)); } + #[tokio::test] + async fn channel_result_refreshes_idle_clock_wiring() { + // Wiring guard for the idle-clock refresh: the refresh lives inside + // handle_prompt_result, so this test fails if that call is dropped — + // not just if the result_refreshes_idle_clock mapper regresses. + 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: Some(channel_id), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + + 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".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + + let stale = tokio::time::Instant::now() - std::time::Duration::from_secs(3600); + let mut last_activity = stale; + 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, + &mut last_activity, + ); + + assert!(last_activity > stale); + } + + #[tokio::test] + async fn heartbeat_result_leaves_idle_clock_stale() { + // Heartbeat completions must NOT count as activity: heartbeat + // dispatch is gated on an idle slot, so a pool kept alive only by + // its own heartbeats would never be reaped. + 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: "hb-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = true; + 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::Heartbeat, + turn_id: "hb-turn-id".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + + let stale = tokio::time::Instant::now() - std::time::Duration::from_secs(3600); + let mut last_activity = stale; + 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, + &mut last_activity, + ); + + assert_eq!(last_activity, stale); + assert!(!heartbeat_in_flight); + } + #[tokio::test] async fn in_flight_stale_native_steer_ack_cannot_update_replacement_session() { let channel_id = Uuid::new_v4(); @@ -7245,6 +7381,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7359,6 +7496,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); @@ -7422,6 +7560,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let turn_errors: Vec<_> = observer @@ -7589,6 +7728,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let events = observer.snapshot(); let turn_error = events.iter().find(|e| e.kind == "turn_error").unwrap(); @@ -7680,6 +7820,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); ( queue.pending_channels(), @@ -7786,6 +7927,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); ( queue.pending_channels(), @@ -7878,6 +8020,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let events = observer.snapshot(); @@ -7972,6 +8115,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let events = observer.snapshot(); @@ -8088,6 +8232,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); // Batch preserved as a cancelled merge, not dead-lettered — same @@ -8221,6 +8366,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); // No batch to merge — the queue has nothing pending for any channel. @@ -8404,6 +8550,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); // The batch must not be requeued: pending_channels returns 0. @@ -8490,6 +8637,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); // Non-auth application error: batch IS requeued (first attempt, retry budget > 0).