diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 24d59c2c93..da930dda45 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -506,7 +506,8 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)] pub relay_observer: bool, - /// Exit after this many seconds with no dispatched events and no turn in flight. + /// Exit after this many seconds since the last channel dispatch or completion, + /// with no turn in flight. Heartbeat completion does not renew this interval. /// 0 disables inactivity self-termination. #[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)] pub exit_after_inactivity: u64, @@ -516,8 +517,9 @@ pub struct CliArgs { pub lazy_pool: bool, /// Tear the woken pool back down to the lazy empty-slot state after this - /// many seconds with no dispatched turn in flight and an empty queue, - /// releasing worker subprocesses until the next accepted event re-wakes. + /// many seconds since the last channel dispatch or completion, with no turn + /// in flight and an empty queue. Heartbeat completion does not renew the interval. + /// Worker subprocesses are released until the next accepted event re-wakes. /// Requires `--lazy-pool`; ignored otherwise. 0 disables idle re-sleep. #[arg(long, env = "BUZZ_ACP_IDLE_POOL_SLEEP", default_value_t = 0)] pub idle_pool_sleep: u64, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 07dc4cce55..0249f5be4d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3998,6 +3998,7 @@ async fn run_harness( &mut respawn_tasks, observer.clone(), Some(&ctx.rest_client), + &mut last_activity, ) == LoopAction::Exit { break; @@ -5239,7 +5240,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 { + // Start a full idle interval when channel work settles, before releasing + // its in-flight guard. Heartbeats must not keep an otherwise idle pool alive. + match &result.source { + PromptSource::Channel(_) => *last_activity = tokio::time::Instant::now(), + PromptSource::Heartbeat => {} + } let before = pool.task_map().len(); let agent_index = result.agent.index; let successful_steer_deliveries = pool @@ -11286,6 +11294,114 @@ mod error_outcome_emission_tests { agent.state.set_scope_owner_generation(scope, generation); } + // Exercise result handling and both production idle decisions with a fake clock. + // No relay or provider is used; cat supplies only an owned process handle. + async fn assert_completion_idle_window(channel_turn: bool) { + let channel_id = Uuid::new_v4(); + let scope = scope::SessionScope::Conversation { channel_id }; + let mut agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + if channel_turn { + bind_agent_scope_owner(&mut pool, &mut agent, scope.clone()); + } + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: channel_turn.then_some(channel_id), + scope: channel_turn.then_some(scope.clone()), + turn_id: "idle-window".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 = !channel_turn; + 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 bound = Duration::from_secs(30); + let mut last_activity = tokio::time::Instant::now(); + tokio::time::advance(bound * 2).await; + let completed_at = tokio::time::Instant::now(); + let result = PromptResult { + agent, + source: if channel_turn { + PromptSource::Channel(scope) + } else { + PromptSource::Heartbeat + }, + turn_id: "idle-window".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &HashSet::new(), + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + &mut last_activity, + ); + // Match the main loop: consume finished prompt tasks before the idle check. + while pool.join_set.join_next().await.is_some() {} + assert!(pool.task_map().is_empty()); + assert!(!heartbeat_in_flight); + assert!(!queue.has_undispatched_work()); + let sleep_due = |now| { + idle_pool_sleep_due( + true, + last_activity, + now, + bound, + heartbeat_in_flight, + !pool.join_set.is_empty(), + queue.has_undispatched_work(), + any_respawn_in_flight(&crash_history), + ) + }; + assert_eq!(sleep_due(completed_at), !channel_turn); + assert_eq!( + inactivity_expired(last_activity, completed_at, bound, false), + !channel_turn + ); + if channel_turn { + assert_eq!(last_activity, completed_at); + assert!(!sleep_due(completed_at + bound - Duration::from_millis(1))); + } + assert!(sleep_due(completed_at + bound)); + assert!(inactivity_expired( + last_activity, + completed_at + bound, + bound, + false + )); + } + + #[tokio::test(start_paused = true)] + async fn completed_channel_turn_gets_a_full_bounded_idle_window() { + assert_completion_idle_window(true).await; + } + + #[tokio::test(start_paused = true)] + async fn completed_heartbeat_does_not_extend_the_idle_window() { + assert_completion_idle_window(false).await; + } + #[tokio::test] async fn successful_native_steer_is_transferred_to_live_session_delivery_state() { let channel_id = Uuid::new_v4(); @@ -11357,6 +11473,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"); @@ -11437,6 +11554,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"); @@ -11558,6 +11676,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"); @@ -11627,6 +11746,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let turn_errors: Vec<_> = observer @@ -11896,6 +12016,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(); @@ -11991,6 +12112,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); ( queue.pending_channels(), @@ -12100,6 +12222,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ); ( queue.pending_channels(), @@ -12195,6 +12318,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let events = observer.snapshot(); @@ -12292,6 +12416,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), None, + &mut tokio::time::Instant::now(), ); let events = observer.snapshot(); @@ -12413,6 +12538,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 @@ -12549,6 +12675,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. @@ -12680,6 +12807,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, None, None, + &mut tokio::time::Instant::now(), ), LoopAction::Continue )); @@ -12832,6 +12960,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. @@ -12940,6 +13069,7 @@ mod error_outcome_emission_tests { &mut respawn_tasks, Some(observer.clone()), Some(&rest), + &mut tokio::time::Instant::now(), ); // The batch must not be requeued: pending_channels returns 0. @@ -13211,6 +13341,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).