From 229f7b638ca2a5dd1a602062c021a3665cc77dd3 Mon Sep 17 00:00:00 2001 From: daniel Date: Sat, 5 Sep 2026 10:44:16 +0100 Subject: [PATCH 1/5] fix(acp): centralize session activity lifecycle --- src/protocols/acp.rs | 233 +++++++++++++++++--- src/protocols/acp/activity.rs | 88 ++++++++ src/protocols/acp/v2.rs | 388 +++++++++++++++++++++++++--------- src/tui/app.rs | 277 ++++++++++++++---------- src/tui/mod.rs | 53 +++-- src/tui/ui.rs | 27 ++- 6 files changed, 786 insertions(+), 280 deletions(-) create mode 100644 src/protocols/acp/activity.rs diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index f525899..c5949c4 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -46,6 +46,7 @@ use tokio::{ time::timeout, }; +mod activity; mod skill_catalog; pub mod v2; @@ -951,10 +952,86 @@ struct SessionBindingGuard { session_id: agentkit_acp::SessionId, } +/// Legacy wire projection of the shared activity reducer. Foreground activity +/// is settled by the standard prompt response; only unsolicited activity needs +/// the extension notification. Both paths observe the same per-session reducer. +#[derive(Clone)] +struct LegacyActivity { + state: Arc>, + session_id: agentkit_acp::SessionId, + notifications: mpsc::UnboundedSender, +} + +#[derive(Default)] +struct LegacyActivityProjection { + activity: activity::SessionActivity, + unsolicited: bool, + next_id: u64, + notification_id: Option, +} + +impl LegacyActivity { + fn new( + session_id: agentkit_acp::SessionId, + notifications: mpsc::UnboundedSender, + ) -> Self { + Self { + state: Arc::new(Mutex::new(LegacyActivityProjection::default())), + session_id, + notifications, + } + } + + fn begin_unsolicited(&self) { + self.state + .lock() + .unwrap_or_else(|e| e.into_inner()) + .unsolicited = true; + } + + fn settle(&self, error: Option) { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + let transition = state.activity.settle(); + state.unsolicited = false; + if transition.is_some() + && let Some(turn_id) = state.notification_id.take() + { + let _ = self.notifications.send(TurnStateNotification { + session_id: self.session_id.clone(), + turn_id, + active: false, + error, + }); + } + } + + fn observe(&self, event: &AgentEvent) { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if state.activity.observe(event) == Some(true) && state.unsolicited { + state.next_id = state.next_id.wrapping_add(1); + let turn_id = state.next_id; + state.notification_id = Some(turn_id); + let _ = self.notifications.send(TurnStateNotification { + session_id: self.session_id.clone(), + turn_id, + active: true, + error: None, + }); + } + } +} + +impl LoopObserver for LegacyActivity { + fn handle_event(&self, event: ObservedEvent) { + self.observe(&event.event); + } +} + #[derive(Clone)] struct ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, + activity: LegacyActivity, session_id: agentkit_acp::SessionId, } @@ -963,8 +1040,10 @@ impl ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, session_id: agentkit_acp::SessionId, + activity: LegacyActivity, ) -> Self { Self { + activity, inner, client, session_id, @@ -974,6 +1053,7 @@ impl ResponseInterruptionNoticeObserver { impl LoopObserver for ResponseInterruptionNoticeObserver { fn handle_event(&self, event: ObservedEvent) { + self.activity.observe(&event.event); if matches!(&event.event, AgentEvent::ResponseAttemptSuperseded) { let notification = SessionNotification::new( self.session_id.clone(), @@ -1258,6 +1338,7 @@ impl Server { let (client, messages) = AcpClientHandle::channel(); tokio::spawn(drain_client_messages(messages, connection.clone())); let (turn_states, turn_state_messages) = mpsc::unbounded_channel(); + let activity = LegacyActivity::new(session_id.clone(), turn_states); tokio::spawn(drain_turn_states(turn_state_messages, connection.clone())); let mut metadata = MetadataMap::new(); @@ -1281,6 +1362,7 @@ impl Server { self.integration.as_ref().clone(), client, session_id.clone(), + activity.clone(), ); let context = AcpDriverContext { cwd, @@ -1333,7 +1415,7 @@ impl Server { adapter: driver.adapter, catalog, commands: rx, - turn_states, + activity, mcp_events, }; let token = self.registry.next_token(); @@ -1574,7 +1656,7 @@ struct SessionActor { adapter: SelectableAdapter, catalog: Vec, commands: mpsc::Receiver, - turn_states: mpsc::UnboundedSender, + activity: LegacyActivity, mcp_events: crate::tools::mcp::McpSubscription, } @@ -1592,11 +1674,10 @@ async fn session_actor(actor: SessionActor) { adapter, catalog, mut commands, - turn_states, + activity, mut mcp_events, } = actor; let mut binding = Some(binding); - let mut next_autonomous_turn_id = 0_u64; loop { tokio::select! { // A queued cancel or close wins over a simultaneously-ready task @@ -1615,6 +1696,7 @@ async fn session_actor(actor: SessionActor) { &background_jobs, structured_completion, ).await; + activity.settle(None); let _ = reply.send(result); } // The server already interrupted the shared controller; this @@ -1665,8 +1747,7 @@ async fn session_actor(actor: SessionActor) { &session_id, &integration, &mut driver, - &turn_states, - &mut next_autonomous_turn_id, + &activity, ).await, Err(error) => Err(error), }; @@ -1685,8 +1766,7 @@ async fn session_actor(actor: SessionActor) { &session_id, &integration, &mut driver, - &turn_states, - &mut next_autonomous_turn_id, + &activity, ).await; if let Err(error) = result { eprintln!("autonomous ACP continuation failed for {session_id}: {error}"); @@ -2054,25 +2134,11 @@ async fn drive_unsolicited( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, driver: &mut LoopDriver, - turn_states: &mpsc::UnboundedSender, - next_turn_id: &mut u64, + activity: &LegacyActivity, ) -> Result<(), AcpRuntimeError> { - *next_turn_id = next_turn_id.wrapping_add(1); - let turn_id = *next_turn_id; - let _ = turn_states.send(TurnStateNotification { - session_id: session_id.clone(), - turn_id, - active: true, - error: None, - }); + activity.begin_unsolicited(); let result = drive_autonomous(session_id, integration, driver).await; - let error = result.as_ref().err().map(ToString::to_string); - let _ = turn_states.send(TurnStateNotification { - session_id: session_id.clone(), - turn_id, - active: false, - error, - }); + activity.settle(result.as_ref().err().map(ToString::to_string)); result } @@ -3209,6 +3275,7 @@ pub(super) mod tests { integration.clone(), client, session_id.clone(), + LegacyActivity::new(session_id.clone(), mpsc::unbounded_channel().0), ); let emit = |event| { observer.handle_event(ObservedEvent { @@ -4039,6 +4106,115 @@ pub(super) mod tests { drain.abort(); } + #[tokio::test] + async fn legacy_activity_ignores_empty_wakes_and_coalesces_real_continuations() { + let session_id = agentkit_acp::SessionId::new("legacy-activity"); + let loop_id = AgentkitSessionId::new("legacy-activity-loop"); + let integration = AcpIntegration::builder() + .name("legacy-activity-test") + .approval_resolver(AutoDenyResolver) + .build() + .unwrap(); + let (client, mut messages) = AcpClientHandle::channel(); + integration + .bind_session(AcpSessionBinding::new( + session_id.clone(), + loop_id.clone(), + client.clone(), + )) + .unwrap(); + let drain = tokio::spawn(async move { + while let Some(message) = messages.recv().await { + if let AcpClientMessage::Flush { response } = message { + let _ = response.send(()); + } + } + }); + let (notifications, mut states) = mpsc::unbounded_channel(); + let activity = LegacyActivity::new(session_id.clone(), notifications); + // Start the script at its plain content response, with no tool prerequisite. + let turns = Arc::new(AtomicUsize::new(2)); + let mut driver = Agent::builder() + .model(ScriptAdapter { + turns: turns.clone(), + user_items_seen: Arc::new(AtomicUsize::new(0)), + notification_items_seen: Arc::new(AtomicUsize::new(0)), + }) + .observer(ResponseInterruptionNoticeObserver::new( + integration.clone(), + client, + session_id.clone(), + activity.clone(), + )) + .build() + .unwrap() + .start(SessionConfig::new(loop_id).without_cache()) + .await + .unwrap(); + + for _ in 0..2 { + drive_unsolicited(&session_id, &integration, &mut driver, &activity) + .await + .unwrap(); + } + assert_eq!(turns.load(Ordering::SeqCst), 2); + assert!(states.try_recv().is_err()); + + driver + .submit_input(vec![Item::notification("background result")]) + .unwrap(); + drive_unsolicited(&session_id, &integration, &mut driver, &activity) + .await + .unwrap(); + let started = states.try_recv().unwrap(); + let ended = states.try_recv().unwrap(); + assert!(started.active && !ended.active); + assert_eq!(started.turn_id, 1); + assert_eq!(started.turn_id, ended.turn_id); + assert!(ended.error.is_none()); + drive_unsolicited(&session_id, &integration, &mut driver, &activity) + .await + .unwrap(); + assert_eq!(turns.load(Ordering::SeqCst), 3); + assert!(states.try_recv().is_err()); + + // Foreground uses the very same reducer, but retains standard v1 response + // semantics without extension notifications (or consuming extension IDs). + driver + .submit_input(vec![Item::text(ItemKind::User, "foreground")]) + .unwrap(); + let response = drive_until_pause(&session_id, &integration, &mut driver, true, None) + .await + .unwrap() + .unwrap(); + activity.settle(None); + assert_eq!(response.stop_reason, StopReason::EndTurn); + assert!(states.try_recv().is_err()); + + // Multiple logical turns drained within one autonomous interval do not + // publish an intermediate terminal state or allocate another turn ID. + activity.begin_unsolicited(); + for text in ["first continuation", "second continuation"] { + driver.submit_input(vec![Item::notification(text)]).unwrap(); + drive_autonomous(&session_id, &integration, &mut driver) + .await + .unwrap(); + } + let started = states.try_recv().unwrap(); + assert!(started.active); + assert_eq!(started.turn_id, 2); + assert!(states.try_recv().is_err()); + activity.settle(Some("terminal error".into())); + let ended = states.try_recv().unwrap(); + assert!(!ended.active); + assert_eq!(ended.turn_id, started.turn_id); + assert_eq!(ended.error.as_deref(), Some("terminal error")); + activity.settle(None); + assert!(states.try_recv().is_err()); + assert_eq!(turns.load(Ordering::SeqCst), 6); + drain.abort(); + } + #[tokio::test] async fn foreground_compose_detaches_out_of_band_and_completes_autonomously() { let turns = Arc::new(AtomicUsize::new(0)); @@ -4097,6 +4273,8 @@ pub(super) mod tests { entered: Arc::clone(&entered), release: Arc::clone(&release), }); + let (turn_states_tx, mut turn_states_rx) = mpsc::unbounded_channel(); + let activity = LegacyActivity::new(acp_session_id.clone(), turn_states_tx); let driver = Agent::builder() .model(ScriptAdapter { turns: Arc::clone(&turns), @@ -4106,6 +4284,7 @@ pub(super) mod tests { .add_tool_source(tools) .task_manager(task_manager) .observer(integration.as_ref().clone()) + .observer(activity.clone()) .cancellation(cancellation.handle()) .build() .unwrap() @@ -4113,7 +4292,7 @@ pub(super) mod tests { .await .unwrap(); let (commands_tx, commands_rx) = mpsc::channel(8); - let (turn_states_tx, mut turn_states_rx) = mpsc::unbounded_channel(); + let test_mcp = crate::tools::mcp::empty(); let mcp_events = test_mcp.subscribe(acp_session_id.to_string()); let root = tempfile::tempdir().unwrap(); @@ -4133,7 +4312,7 @@ pub(super) mod tests { .unwrap(), catalog: Vec::new(), commands: commands_rx, - turn_states: turn_states_tx, + activity, mcp_events, })); diff --git a/src/protocols/acp/activity.rs b/src/protocols/acp/activity.rs new file mode 100644 index 0000000..0a08e14 --- /dev/null +++ b/src/protocols/acp/activity.rs @@ -0,0 +1,88 @@ +use agentkit_loop::AgentEvent; + +/// Authoritative activity state for an attached session. A drive/wake-up is +/// not itself activity: only a logical turn started by the loop enters Running. +/// A finished logical turn remains Settling until the actor has drained steering +/// and structured background work. Those continuations belong to the same client-visible +/// activity interval, rather than producing an Idle/Running flicker. +#[derive(Default)] +pub(super) enum SessionActivity { + #[default] + Idle, + Running, + Settling, +} + +impl SessionActivity { + pub(super) fn observe(&mut self, event: &AgentEvent) -> Option { + match event { + AgentEvent::TurnStarted { .. } => { + let was_idle = matches!(self, Self::Idle); + *self = Self::Running; + was_idle.then_some(true) + } + AgentEvent::TurnFinished(_) if !matches!(self, Self::Idle) => { + *self = Self::Settling; + None + } + _ => None, + } + } + + pub(super) fn settle(&mut self) -> Option { + if matches!(self, Self::Idle) { + return None; + } + *self = Self::Idle; + Some(false) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use agentkit_core::{FinishReason, MetadataMap, SessionId}; + + #[test] + fn session_activity_coalesces_continuations_and_settles_once() { + for reason in [ + FinishReason::Completed, + FinishReason::Cancelled, + FinishReason::MaxTokens, + FinishReason::Blocked, + FinishReason::Error, + ] { + let mut state = SessionActivity::default(); + assert!(state.settle().is_none()); + let started = AgentEvent::TurnStarted { + session_id: SessionId::new("state-loop"), + turn_id: agentkit_core::TurnId::new("first"), + }; + assert!(matches!(state.observe(&started), Some(true))); + assert!(state.observe(&started).is_none()); + let finished = AgentEvent::TurnFinished(agentkit_loop::TurnResult { + turn_id: agentkit_core::TurnId::new("first"), + finish_reason: reason, + items: Vec::new(), + usage: None, + metadata: MetadataMap::new(), + }); + assert!(state.observe(&finished).is_none()); + assert!(matches!(state, SessionActivity::Settling)); + // A queued steer or structured synthesis continues before actor settlement. + assert!( + state + .observe(&AgentEvent::TurnStarted { + session_id: SessionId::new("state-loop"), + turn_id: agentkit_core::TurnId::new("continuation"), + }) + .is_none() + ); + assert!(matches!(state.settle(), Some(false))); + assert!(state.settle().is_none()); + assert!(state.observe(&finished).is_none()); + assert!(matches!(state, SessionActivity::Idle)); + assert!(matches!(state.observe(&started), Some(true))); + } + } +} diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 5548a46..13f35f5 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -32,6 +32,8 @@ use crate::{ runtime::{AcpDriverContext, BackgroundJobs, Runtime}, }; +use super::activity::SessionActivity; + use super::{ AuthenticationRequiredData, CancelBackgroundRequest, CancelBackgroundResponse, DetachComposeRequest, DetachComposeResponse, FileSearchRequest, FileSearchResponse, @@ -137,6 +139,9 @@ fn loop_error_stop_reason( } } +// Admission excludes competing requests while the actor prepares or drains work. +// It is not observable activity: an admitted autonomous drive may find no work. +// SessionActivity alone owns the ACP lifecycle projected to clients. fn claim_prompt(busy: &AtomicBool) -> Result<(), AcpRuntimeError> { busy.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) .map(|_| ()) @@ -209,6 +214,7 @@ impl ReplacementGeneration { struct ResponseReplacementSink { inner: S, current: Arc>, + activity: Arc>, } impl ResponseReplacementSink { @@ -216,6 +222,7 @@ impl ResponseReplacementSink { Self { inner, current: Arc::new(Mutex::new(CurrentReplacementMessages::default())), + activity: Arc::new(Mutex::new(SessionActivity::default())), } } @@ -322,6 +329,35 @@ impl ResponseReplacementSink { } } +impl ResponseReplacementSink { + fn transition_activity( + &self, + session_id: &wire::SessionId, + transition: impl FnOnce(&mut SessionActivity) -> Option, + ) -> Result<(), AcpRuntimeError> { + let mut activity = self + .activity + .lock() + .unwrap_or_else(|error| error.into_inner()); + if let Some(state) = transition(&mut activity) { + send_state(&self.inner, session_id, state)?; + } + Ok(()) + } + + fn settle_activity( + &self, + session_id: &wire::SessionId, + reason: wire::StopReason, + ) -> Result<(), AcpRuntimeError> { + self.transition_activity(session_id, |activity| { + activity + .settle() + .map(|_| wire::StateUpdate::Idle(wire::IdleStateUpdate::new().stop_reason(reason))) + }) + } +} + #[async_trait] impl AcpSessionUpdateSink for ResponseReplacementSink { fn update( @@ -396,6 +432,13 @@ where S: AcpSessionUpdateSink + Clone, { fn handle_event(&self, event: ObservedEvent) { + if let Err(error) = self.sink.transition_activity(&self.session_id, |activity| { + activity + .observe(&event.event) + .map(|_| wire::StateUpdate::Running(wire::RunningStateUpdate::new())) + }) { + tracing::debug!(%error, "failed to queue ACP v2 activity update"); + } if let AgentEvent::UsageUpdated(usage) = &event.event { let Some(update) = usage_update(usage) else { return; @@ -1210,7 +1253,7 @@ async fn prepare_prompt( skill_catalog: &mut skill_catalog::SkillCatalogMonitor, driver: &mut LoopDriver, command: PromptCommand, - sink: &impl AcpSessionUpdateSink, + sink: &ResponseReplacementSink, tasks: &TaskManagerHandle, background_jobs: &BackgroundJobs, structured_completion: bool, @@ -1448,16 +1491,11 @@ async fn run_active_turn( integration: &AcpIntegration, handle: &AcpSessionHandle, driver: &mut LoopDriver, - sink: &impl AcpSessionUpdateSink, + sink: &ResponseReplacementSink, cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, ) -> Result<(), AcpRuntimeError> { - send_state( - sink, - session_id, - wire::StateUpdate::Running(wire::RunningStateUpdate::new()), - )?; - let stop_reason = match drive_prompt( + let mut result = match drive_prompt( session_id, driver, handle, @@ -1466,36 +1504,42 @@ async fn run_active_turn( ) .await { - Ok(stop_reason) => stop_reason, Err(_) if handle .cancellation_handle() .is_cancelled_since(cancellation_generation) => { - wire::StopReason::Cancelled - } - Err(error) => { - if let Some((tasks, background_jobs)) = structured { - super::cancel_background_jobs(tasks, background_jobs).await; - let _ = super::settle_background_jobs(tasks, background_jobs).await; - } - terminalize_running_error(session_id, integration, handle, sink, &error).await?; - return Err(error); + Ok(wire::StopReason::Cancelled) } + result => result, }; - if stop_reason == wire::StopReason::Cancelled + if (result.is_err() || matches!(&result, Ok(wire::StopReason::Cancelled))) && let Some((tasks, background_jobs)) = structured { super::cancel_background_jobs(tasks, background_jobs).await; - let _ = super::settle_background_jobs(tasks, background_jobs).await?; + if let Err(error) = super::settle_background_jobs(tasks, background_jobs).await + && result.is_ok() + { + result = Err(error); + } } + handle.stop_injection_turn(); let _ = integration.flush_session_updates(session_id).await; integration.finish_prompt(session_id); - send_state( - sink, - session_id, - wire::StateUpdate::Idle(wire::IdleStateUpdate::new().stop_reason(stop_reason)), - ) + + // Every exit goes through the same state settlement, including cleanup errors. + // Diagnostic delivery failure must not prevent the Idle transition. + let diagnostic = match &result { + Err(error) => sink.update(error_diagnostic_notification(session_id, error)), + Ok(_) => Ok(()), + }; + let stop_reason = result + .as_ref() + .cloned() + .unwrap_or_else(|_| error_stop_reason()); + let settled = sink.settle_activity(session_id, stop_reason); + diagnostic.and(settled)?; + result.map(|_| ()) } async fn drive_autonomous( @@ -1504,7 +1548,7 @@ async fn drive_autonomous( handle: &AcpSessionHandle, busy: &AtomicBool, driver: &mut LoopDriver, - sink: &impl AcpSessionUpdateSink, + sink: &ResponseReplacementSink, ) -> Result<(), AcpRuntimeError> { if claim_prompt(busy).is_err() { return Ok(()); @@ -1529,26 +1573,6 @@ async fn drive_autonomous( result } -async fn terminalize_running_error( - session_id: &wire::SessionId, - integration: &AcpIntegration, - handle: &AcpSessionHandle, - sink: &impl AcpSessionUpdateSink, - error: &AcpRuntimeError, -) -> Result<(), AcpRuntimeError> { - handle.stop_injection_turn(); - let _ = integration.flush_session_updates(session_id).await; - integration.finish_prompt(session_id); - - let diagnostic_result = sink.update(error_diagnostic_notification(session_id, error)); - let idle_result = send_state( - sink, - session_id, - wire::StateUpdate::Idle(wire::IdleStateUpdate::new().stop_reason(error_stop_reason())), - ); - diagnostic_result.and(idle_result) -} - fn error_diagnostic_notification( session_id: &wire::SessionId, error: &AcpRuntimeError, @@ -2369,7 +2393,12 @@ mod tests { chunk: "answer".into(), })); - let updates = recording.updates.lock().unwrap(); + let recorded = recording.updates.lock().unwrap(); + assert!(matches!( + recorded[0].update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let updates = &recorded[1..]; assert_eq!(updates.len(), 6); let stale_message_id = match &updates[0].update { wire::SessionUpdate::AgentMessageChunk(chunk) => chunk.message_id.clone(), @@ -2400,7 +2429,7 @@ mod tests { wire::SessionUpdate::AgentMessageChunk(chunk) if chunk.message_id == replacement_message_id )); - drop(updates); + drop(recorded); emit(AgentEvent::ResponseAttemptSuperseded); emit(AgentEvent::ContentDelta(agentkit_core::Delta::BeginPart { @@ -2411,7 +2440,12 @@ mod tests { part_id: agentkit_core::PartId::new("message-3"), chunk: "third answer".into(), })); - let updates = recording.updates.lock().unwrap(); + let recorded = recording.updates.lock().unwrap(); + assert!(matches!( + recorded[0].update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let updates = &recorded[1..]; assert_eq!(updates.len(), 9); assert!(matches!( &updates[6].update, @@ -2431,14 +2465,14 @@ mod tests { if chunk.message_id != replacement_message_id && chunk.message_id != stale_message_id )); - drop(updates); + drop(recorded); emit(AgentEvent::TurnStarted { session_id: loop_session_id.clone(), turn_id: agentkit_core::TurnId::new("turn-2"), }); emit(AgentEvent::ResponseAttemptSuperseded); - assert_eq!(recording.updates.lock().unwrap().len(), 9); + assert_eq!(recording.updates.lock().unwrap().len(), 10); } #[test] @@ -2490,7 +2524,12 @@ mod tests { emit_part("thought-1", agentkit_core::PartKind::Reasoning, "thinking"); emit(finish("turn-1", FinishReason::Cancelled)); - let updates = recording.updates.lock().unwrap(); + let recorded = recording.updates.lock().unwrap(); + assert!(matches!( + recorded[0].update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let updates = &recorded[1..]; assert_eq!(updates.len(), 2); assert!(matches!( &updates[0].update, @@ -2502,7 +2541,7 @@ mod tests { wire::SessionUpdate::AgentThoughtChunk(chunk) if chunk.content == wire::ContentBlock::Text(wire::TextContent::new("thinking")) )); - drop(updates); + drop(recorded); emit(AgentEvent::TurnStarted { session_id: loop_session_id.clone(), @@ -2510,7 +2549,7 @@ mod tests { }); emit_part("message-2", agentkit_core::PartKind::Text, "completed"); emit(finish("turn-2", FinishReason::Completed)); - assert_eq!(recording.updates.lock().unwrap().len(), 3); + assert_eq!(recording.updates.lock().unwrap().len(), 4); emit(AgentEvent::TurnStarted { session_id: loop_session_id.clone(), @@ -2524,7 +2563,12 @@ mod tests { ); emit(finish("turn-3", FinishReason::Cancelled)); - let updates = recording.updates.lock().unwrap(); + let recorded = recording.updates.lock().unwrap(); + assert!(matches!( + recorded[0].update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let updates = &recorded[1..]; assert_eq!(updates.len(), 4); assert!(matches!( &updates[3].update, @@ -2800,7 +2844,12 @@ mod tests { }; assert_eq!(result.finish_reason, FinishReason::Cancelled); - let updates = recording.updates.lock().unwrap(); + let recorded = recording.updates.lock().unwrap(); + assert!(matches!( + recorded[0].update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let updates = &recorded[1..]; assert_eq!(updates.len(), 1); assert!(matches!( &updates[0].update, @@ -2862,19 +2911,38 @@ mod tests { #[tokio::test] async fn foreground_provider_error_after_running_terminalizes_once() { let integration = AcpIntegration::default(); - let sink = RecordingSink::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("foreground-provider-error"); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), - SessionId::new("foreground-provider-error-loop"), + SessionId::new("foreground_provider_error_after_running_terminalizes_once-loop"), sink.clone(), )) .unwrap(); handle.prepare_injection_turn(); let cancellation_generation = handle.cancellation_handle().generation(); - let (mut driver, turns) = - test_driver(TestOutcome::ProviderError, "foreground-provider-error-loop").await; + let turns = Arc::new(AtomicU64::new(0)); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::ProviderError, + turns: turns.clone(), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start( + SessionConfig::new(SessionId::new( + "foreground_provider_error_after_running_terminalizes_once-loop", + )) + .without_cache(), + ) + .await + .unwrap(); let (reply, response) = oneshot::channel(); let command = PromptCommand { request: wire::PromptRequest::new( @@ -2911,8 +2979,8 @@ mod tests { assert!(matches!(result, Err(AcpRuntimeError::Loop(_)))); assert_eq!(turns.load(Ordering::Relaxed), 1); - assert_eq!(sink.flushes.load(Ordering::Relaxed), 1); - let updates = sink.updates.lock().unwrap(); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + let updates = recording.updates.lock().unwrap(); assert_eq!(updates.len(), 4); assert!(matches!( updates[1].update, @@ -2964,12 +3032,26 @@ mod tests { entered: Arc::clone(&entered), release: Arc::clone(&release), }); + let integration = AcpIntegration::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); + let session_id = wire::SessionId::new("v2-structured"); + let handle = integration + .bind_session(AcpSessionBinding::new( + session_id.clone(), + SessionId::new("v2-structured-loop"), + sink.clone(), + )) + .unwrap(); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); let mut driver = Agent::builder() .model(ScriptAdapter { turns: Arc::clone(&turns), user_items_seen: Arc::new(AtomicUsize::new(0)), notification_items_seen: Arc::new(AtomicUsize::new(0)), }) + .observer(observer) .add_tool_source(tools) .task_manager(task_manager) .build() @@ -2981,23 +3063,16 @@ mod tests { .submit_input(vec![Item::text(ItemKind::User, "start background")]) .unwrap(); - let integration = AcpIntegration::default(); - let session_id = wire::SessionId::new("v2-structured"); - let handle = integration - .bind_session(AcpSessionBinding::new( - session_id.clone(), - SessionId::new("v2-structured-loop"), - RecordingSink::default(), - )) - .unwrap(); handle.prepare_injection_turn(); handle.start_injection_turn(); let generation = handle.cancellation_handle().generation(); let background_jobs = BackgroundJobs::default(); - let prompt = drive_prompt( + let prompt = run_active_turn( &session_id, - &mut driver, + &integration, &handle, + &mut driver, + &sink, generation, Some((&tasks, &background_jobs)), ); @@ -3030,6 +3105,16 @@ mod tests { .is_err() ); + assert_eq!( + recording + .updates + .lock() + .unwrap() + .iter() + .filter(|update| matches!(update.update, wire::SessionUpdate::StateUpdate(_))) + .count(), + 1 + ); background_jobs.finish_for_test("background-call"); assert!( timeout(Duration::from_millis(20), &mut prompt) @@ -3038,11 +3123,14 @@ mod tests { "structured v2 prompt crossed the terminal publication handoff early" ); release.notify_one(); - let reason = timeout(Duration::from_secs(1), &mut prompt) + timeout(Duration::from_secs(1), &mut prompt) .await .expect("structured v2 prompt did not synthesize") .unwrap(); - assert_eq!(reason, wire::StopReason::EndTurn); + assert_running_then_idle( + &recording.updates.lock().unwrap(), + wire::StopReason::EndTurn, + ); assert_eq!(turns.load(Ordering::SeqCst), 3); assert!( timeout(Duration::from_millis(20), tasks.next_event()) @@ -3128,19 +3216,38 @@ mod tests { } #[tokio::test] - async fn autonomous_no_work_emits_running_then_idle() { + async fn autonomous_no_work_emits_no_state_transition() { let integration = AcpIntegration::default(); - let sink = RecordingSink::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-no-work"); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), - SessionId::new("autonomous-no-work-loop"), + SessionId::new("autonomous_no_work_emits_no_state_transition-loop"), sink.clone(), )) .unwrap(); - let (mut driver, turns) = - test_driver(TestOutcome::Content, "autonomous-no-work-loop").await; + let turns = Arc::new(AtomicU64::new(0)); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::Content, + turns: turns.clone(), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start( + SessionConfig::new(SessionId::new( + "autonomous_no_work_emits_no_state_transition-loop", + )) + .without_cache(), + ) + .await + .unwrap(); let busy = AtomicBool::new(false); drive_autonomous( @@ -3156,8 +3263,8 @@ mod tests { assert_eq!(turns.load(Ordering::Relaxed), 0); assert!(!busy.load(Ordering::Relaxed)); - assert_eq!(sink.flushes.load(Ordering::Relaxed), 1); - assert_running_then_idle(&sink.updates.lock().unwrap(), wire::StopReason::EndTurn); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + assert!(recording.updates.lock().unwrap().is_empty()); } #[tokio::test] @@ -3208,6 +3315,21 @@ mod tests { assert_eq!(turns.load(Ordering::Relaxed), 1); assert!(!busy.load(Ordering::Relaxed)); assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + let update_count = recording.updates.lock().unwrap().len(); + for _ in 0..2 { + drive_autonomous( + &session_id, + &integration, + &handle, + &busy, + &mut driver, + &sink, + ) + .await + .unwrap(); + } + assert_eq!(turns.load(Ordering::Relaxed), 1); + assert_eq!(recording.updates.lock().unwrap().len(), update_count); let updates = recording.updates.lock().unwrap(); assert!(updates.iter().any(|update| { serde_json::to_string(&update.update) @@ -3220,17 +3342,36 @@ mod tests { #[tokio::test] async fn autonomous_provider_error_emits_running_error_idle() { let integration = AcpIntegration::default(); - let sink = RecordingSink::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-provider-error"); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), - SessionId::new("autonomous-provider-error-loop"), + SessionId::new("autonomous_provider_error_emits_running_error_idle-loop"), sink.clone(), )) .unwrap(); - let (mut driver, turns) = - test_driver(TestOutcome::ProviderError, "autonomous-provider-error-loop").await; + let turns = Arc::new(AtomicU64::new(0)); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::ProviderError, + turns: turns.clone(), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start( + SessionConfig::new(SessionId::new( + "autonomous_provider_error_emits_running_error_idle-loop", + )) + .without_cache(), + ) + .await + .unwrap(); driver .submit_input(vec![Item::notification("background event")]) .unwrap(); @@ -3249,8 +3390,8 @@ mod tests { assert!(matches!(result, Err(AcpRuntimeError::Loop(_)))); assert_eq!(turns.load(Ordering::Relaxed), 1); assert!(!busy.load(Ordering::Relaxed)); - assert_eq!(sink.flushes.load(Ordering::Relaxed), 1); - let updates = sink.updates.lock().unwrap(); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + let updates = recording.updates.lock().unwrap(); assert_eq!(updates.len(), 3); assert!( serde_json::to_string(&updates[1].update) @@ -3267,21 +3408,38 @@ mod tests { #[tokio::test] async fn autonomous_cancellation_has_no_error_diagnostic_or_continuation() { let integration = AcpIntegration::default(); - let sink = RecordingSink::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-cancel"); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), - SessionId::new("autonomous-cancel-loop"), + SessionId::new( + "autonomous_cancellation_has_no_error_diagnostic_or_continuation-loop", + ), sink.clone(), )) .unwrap(); - let (mut driver, turns) = test_driver_with_interrupt( - TestOutcome::ProviderError, - "autonomous-cancel-loop", - Some(handle.clone()), - ) - .await; + let turns = Arc::new(AtomicU64::new(0)); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::ProviderError, + turns: turns.clone(), + interrupt: Some(handle.clone()), + }) + .observer(observer) + .build() + .unwrap() + .start( + SessionConfig::new(SessionId::new( + "autonomous_cancellation_has_no_error_diagnostic_or_continuation-loop", + )) + .without_cache(), + ) + .await + .unwrap(); driver .submit_input(vec![Item::notification("background event")]) .unwrap(); @@ -3300,24 +3458,46 @@ mod tests { assert_eq!(turns.load(Ordering::Relaxed), 1); assert!(!busy.load(Ordering::Relaxed)); - assert_eq!(sink.flushes.load(Ordering::Relaxed), 1); - assert_running_then_idle(&sink.updates.lock().unwrap(), wire::StopReason::Cancelled); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + assert_running_then_idle( + &recording.updates.lock().unwrap(), + wire::StopReason::Cancelled, + ); } #[tokio::test] async fn autonomous_finish_error_emits_running_error_idle() { let integration = AcpIntegration::default(); - let sink = RecordingSink::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-error"); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), - SessionId::new("autonomous-error-loop"), + SessionId::new("autonomous_finish_error_emits_running_error_idle-loop"), sink.clone(), )) .unwrap(); - let (mut driver, turns) = - test_driver(TestOutcome::FinishError, "autonomous-error-loop").await; + let turns = Arc::new(AtomicU64::new(0)); + let observer = + ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::FinishError, + turns: turns.clone(), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start( + SessionConfig::new(SessionId::new( + "autonomous_finish_error_emits_running_error_idle-loop", + )) + .without_cache(), + ) + .await + .unwrap(); driver .submit_input(vec![Item::notification("background event")]) .unwrap(); @@ -3336,8 +3516,8 @@ mod tests { assert!(matches!(result, Err(AcpRuntimeError::Loop(_)))); assert_eq!(turns.load(Ordering::Relaxed), 1); assert!(!busy.load(Ordering::Relaxed)); - assert_eq!(sink.flushes.load(Ordering::Relaxed), 1); - let updates = sink.updates.lock().unwrap(); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + let updates = recording.updates.lock().unwrap(); assert_eq!(updates.len(), 3); assert!(matches!( updates[1].update, diff --git a/src/tui/app.rs b/src/tui/app.rs index d15ee5b..5dbb863 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -13,7 +13,11 @@ use std::ffi::OsStr; #[cfg(any(target_os = "macos", target_os = "linux"))] use std::process::{Command, Stdio}; -use agent_client_protocol::schema::v2::{AuthMethodTerminal, StopReason, ToolCallStatus, ToolKind}; +#[cfg(test)] +use agent_client_protocol::schema::v2::RunningStateUpdate; +use agent_client_protocol::schema::v2::{ + AuthMethodTerminal, StateUpdate, StopReason, ToolCallStatus, ToolKind, +}; #[cfg(test)] use agentkit_core::{DataRef, Item, ItemKind, Modality, Part, ToolOutput}; use crossterm::event::{ @@ -115,14 +119,8 @@ pub enum Update { ConfigOptions(Vec), /// Context window accounting. Usage { used: u64, size: u64 }, - /// Standard ACP v2 foreground state. - State { - active: bool, - steerable: bool, - cancelled: bool, - }, - /// An ACP v2 turn became idle with its exact terminal reason. - Stopped(Option), + /// The authoritative ACP v2 foreground lifecycle, preserved from the wire. + State(StateUpdate), /// A nested tool call started or finished inside a compose run. Runtime(RuntimeEvent), /// A diagnostic line from the agent process. @@ -1669,10 +1667,6 @@ impl App { self.message_blocks.insert(id, self.blocks.len() - 1); } - fn finish_turn(&mut self, cancelled: bool) { - self.finish_turn_with_outcome(!cancelled, cancelled.then_some("turn interrupted".into())); - } - fn finish_with_stop_reason(&mut self, reason: Option) { let (successful, notice) = match reason { Some(StopReason::EndTurn) => (true, None), @@ -1704,7 +1698,6 @@ impl App { } self.close_thought(); self.agent_stream_sealed = true; - let interrupted = self.phase == Phase::Cancelling; let turn_millis = self.stop_turn_timer(); self.phase = Phase::Idle; self.compacting = false; @@ -1727,9 +1720,7 @@ impl App { self.mark_block_dirty(index); self.reclassify_dynamic(index); } - if interrupted { - self.note("turn interrupted"); - } else if let Some(notice) = notice { + if let Some(notice) = notice { self.note(notice); } if let Some(millis) = turn_millis { @@ -1985,18 +1976,14 @@ impl App { } } Update::ConfigOptions(_) => {} - Update::State { - active, - steerable, - cancelled, - } => { - if active { + Update::State(state) => match state { + StateUpdate::Running(_) | StateUpdate::RequiresAction(_) => { if self.phase == Phase::Idle { self.agent_stream_sealed = true; self.turn_started = Some(Instant::now()); } if self.phase != Phase::Cancelling { - self.phase = if steerable { + self.phase = if matches!(state, StateUpdate::Running(_)) { Phase::Working } else { Phase::Blocked @@ -2004,11 +1991,10 @@ impl App { } self.follow = true; self.scroll = usize::MAX; - } else { - self.finish_turn(cancelled); } - } - Update::Stopped(reason) => self.finish_with_stop_reason(reason), + StateUpdate::Idle(idle) => self.finish_with_stop_reason(idle.stop_reason), + _ => {} + }, Update::ProcessExited(error) => { self.finish_turn_with_outcome(false, None); self.retire_active_agents_at(crate::events::now_millis()); @@ -2188,11 +2174,9 @@ impl App { images: Vec::new(), append: false, }); - self.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); + self.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); self.blocks.len() as u64 } @@ -3471,7 +3455,8 @@ mod tests { }; use agent_client_protocol::schema::v2::{ - AuthMethodTerminal, StopReason, ToolCallStatus, ToolKind, + AuthMethodTerminal, IdleStateUpdate, RequiresActionStateUpdate, RunningStateUpdate, + StateUpdate, StopReason, ToolCallStatus, ToolKind, }; use agentkit_core::{DataRef, Item, ItemKind, MediaPart, MetadataMap, Modality, Part}; use crossterm::event::{ @@ -3914,16 +3899,12 @@ mod tests { fn turn_end_clears_compaction_state() { let mut app = app(); app.compacting = true; - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); assert!(!app.compacting); } @@ -4066,12 +4047,12 @@ mod tests { fn closes_running_calls_when_the_turn_ends() { let mut app = app(); compose(&mut app, "a = shell({ command: \"ls\" })\nreturn a"); - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); - app.apply(Update::Stopped(Some(StopReason::EndTurn))); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); let call = app .blocks .iter() @@ -4104,12 +4085,12 @@ mod tests { ] { let mut app = app(); compose(&mut app, "a = shell({ command: \"ls\" })\nreturn a"); - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); - app.apply(Update::Stopped(Some(reason))); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(reason), + ))); let notice = app .blocks @@ -4134,31 +4115,115 @@ mod tests { } #[test] - fn duplicate_state_updates_are_idempotent() { + fn transcript_and_tool_activity_do_not_drive_foreground_lifecycle() { let mut app = app(); - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, + app.apply(Update::UserMessage { + id: "user".into(), + text: "hello".into(), + images: Vec::new(), + append: false, }); + compose(&mut app, "return 1"); + app.apply(Update::AgentMessage { + id: "agent".into(), + text: "hello".into(), + append: false, + }); + assert!(!app.working()); + assert!(app.turn_started.is_none()); + + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); let started = app.turn_started; - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, + app.apply(Update::ToolUpdated { + id: "call-1".into(), + status: Some(ToolCallStatus::Completed), + script: None, + output: Vec::new(), + backgrounded: false, }); + assert!(app.working()); assert_eq!(app.turn_started, started); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); + app.apply(Update::AgentMessage { + id: "late".into(), + text: "background result".into(), + append: false, }); + compose(&mut app, "return 2"); assert!(!app.working()); + assert!(app.turn_started.is_none()); + } + + #[test] + fn requires_action_preserves_the_running_turn_timer() { + let mut app = app(); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + let started = app.turn_started; + app.apply(Update::State(StateUpdate::RequiresAction( + RequiresActionStateUpdate::new(), + ))); + assert!(app.phase == Phase::Blocked); + assert_eq!(app.turn_started, started); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + assert!(app.phase == Phase::Working); + assert_eq!(app.turn_started, started); + } + + #[test] + fn cancellation_request_does_not_override_the_actual_stop_reason() { + let mut app = app(); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + let started = app.turn_started; + assert!(matches!(app.request_cancel(), Action::Cancel)); + assert_eq!(app.turn_started, started); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); + assert!(!app.working()); + assert!(app.turn_started.is_none()); + assert!( + !app.blocks + .iter() + .any(|block| matches!(block, Block::Notice(text) if text == "turn interrupted")) + ); + } + + #[test] + fn duplicate_state_updates_are_idempotent() { + let mut app = app(); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + let started = app.turn_started; + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); + assert_eq!(app.turn_started, started); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); + assert!(!app.working()); + assert!(app.turn_started.is_none()); + assert_eq!( + app.blocks + .iter() + .filter(|block| matches!(block, Block::TurnDuration(_))) + .count(), + 1 + ); } #[test] @@ -4167,11 +4232,9 @@ mod tests { app.push_user("hello".into()); app.turn_started = Some(Instant::now() - Duration::from_secs(65)); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); assert!(matches!( app.blocks.last(), @@ -4183,21 +4246,17 @@ mod tests { fn autonomous_turn_is_visible_and_cancellable() { let mut app = app(); - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); assert!(app.working()); assert!(matches!(app.request_cancel(), Action::Cancel)); assert!(app.phase == Phase::Cancelling); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::Cancelled), + ))); assert!(!app.working()); assert!(matches!( @@ -4224,11 +4283,9 @@ mod tests { backgrounded: true, }); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); let running = app .blocks @@ -4571,11 +4628,9 @@ mod tests { fn active_plain_text_is_submitted_as_steering_when_advertised() { let mut app = app(); app.can_steer = true; - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); app.paste("change direction"); app.last_key = Some(Instant::now() - Duration::from_millis(500)); @@ -4617,11 +4672,9 @@ mod tests { #[test] fn active_text_is_preserved_when_steering_is_not_advertised() { let mut app = app(); - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); app.paste("wait"); app.last_key = Some(Instant::now() - Duration::from_millis(500)); assert!(matches!( @@ -4742,11 +4795,9 @@ mod tests { fn copies_only_agent_text_after_the_latest_user_message() { let mut app = app(); app.apply(Update::test_text("old".into())); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); app.push_user("next".into()); app.apply(Update::AgentMessage { id: "next-agent".into(), @@ -4773,11 +4824,9 @@ mod tests { fn autonomous_text_starts_a_new_block_after_turn_end() { let mut app = app(); app.apply(Update::test_text("Started.".into())); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); app.apply(Update::AgentMessage { id: "autonomous".into(), text: "RAVENS_".into(), diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 4acc3b2..d2010c3 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -2215,20 +2215,7 @@ fn translate(notification: UpdateSessionNotification) -> (String, Vec) { used: usage.used, size: usage.size, }], - SessionUpdate::StateUpdate(state) => match state { - wire::StateUpdate::Running(_) => vec![Update::State { - active: true, - steerable: true, - cancelled: false, - }], - wire::StateUpdate::RequiresAction(_) => vec![Update::State { - active: true, - steerable: false, - cancelled: false, - }], - wire::StateUpdate::Idle(idle) => vec![Update::Stopped(idle.stop_reason)], - _ => Vec::new(), - }, + SessionUpdate::StateUpdate(state) => vec![Update::State(state)], _ => Vec::new(), }; (session_id, updates) @@ -2904,11 +2891,7 @@ mod tests { ); assert!(matches!( translate_for_session(running, "session").as_slice(), - [Update::State { - active: true, - steerable: true, - cancelled: false - }] + [Update::State(StateUpdate::Running(_))] )); let idle = UpdateSessionNotification::new( "session", @@ -2916,8 +2899,38 @@ mod tests { ); assert!(matches!( translate_for_session(idle, "session").as_slice(), - [Update::Stopped(None)] + [Update::State(StateUpdate::Idle(idle))] if idle.stop_reason.is_none() + )); + } + + #[test] + fn preserves_requires_action_and_terminal_state_reasons() { + let blocked = UpdateSessionNotification::new( + "session", + SessionUpdate::StateUpdate(StateUpdate::RequiresAction( + wire::RequiresActionStateUpdate::new(), + )), + ); + assert!(matches!( + translate_for_session(blocked, "session").as_slice(), + [Update::State(StateUpdate::RequiresAction(_))] )); + for reason in [ + wire::StopReason::EndTurn, + wire::StopReason::Cancelled, + wire::StopReason::Other("custom".into()), + ] { + let idle = UpdateSessionNotification::new( + "session", + SessionUpdate::StateUpdate(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(reason.clone()), + )), + ); + assert!(matches!( + translate_for_session(idle, "session").as_slice(), + [Update::State(StateUpdate::Idle(idle))] if idle.stop_reason.as_ref() == Some(&reason) + )); + } } #[test] diff --git a/src/tui/ui.rs b/src/tui/ui.rs index 4cf3471..bf09ba0 100644 --- a/src/tui/ui.rs +++ b/src/tui/ui.rs @@ -2219,6 +2219,9 @@ fn compact(value: u64) -> String { #[cfg(test)] mod tests { + use agent_client_protocol::schema::v2::{ + IdleStateUpdate, RunningStateUpdate, StateUpdate, StopReason, + }; use std::path::PathBuf; use agent_client_protocol::schema::v2::ToolKind; @@ -2940,11 +2943,9 @@ mod tests { "127.0.0.1:7331".into(), ); app.can_steer = true; - app.apply(Update::State { - active: true, - steerable: true, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Running( + RunningStateUpdate::new(), + ))); app.apply(Update::SteerAccepted { id: "first".into(), text: "first pending".into(), @@ -3372,22 +3373,18 @@ mod tests { output: vec!["exit code 1".into()], backgrounded: false, }); - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); app.apply(Update::Log("warn: retrying provider request".into())); app.show_logs = true; for index in 0..12 { app.push_user(format!("follow-up number {index}")); app.apply(Update::test_text(format!("answer number {index}"))); } - app.apply(Update::State { - active: false, - steerable: false, - cancelled: false, - }); + app.apply(Update::State(StateUpdate::Idle( + IdleStateUpdate::new().stop_reason(StopReason::EndTurn), + ))); let _ = render(&mut app, 100, 24); app.scroll_by(-6); let frame = render(&mut app, 100, 24); From a69e0a4fc8f9a601ec1165d2b44d7cc840dc629c Mon Sep 17 00:00:00 2001 From: daniel Date: Sat, 5 Sep 2026 11:20:41 +0100 Subject: [PATCH 2/5] refactor(acp): share session instrument and finalization --- src/protocols/acp.rs | 354 +++++++++++++++++------------- src/protocols/acp/activity.rs | 394 +++++++++++++++++++++++++++++----- src/protocols/acp/v2.rs | 356 +++++++++++++++++++----------- 3 files changed, 778 insertions(+), 326 deletions(-) diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index c5949c4..5573056 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -25,7 +25,7 @@ use agentkit_acp::{ SessionConfigOption, SessionConfigOptionCategory, SessionConfigSelectGroup, SessionConfigSelectOption, SessionForkCapabilities, SessionInfo, SessionListCapabilities, SessionNotification, SessionUpdate, SetSessionConfigOptionRequest, - SetSessionConfigOptionResponse, StopReason, TextContent, TextResourceContents, ToolCallStatus, + SetSessionConfigOptionResponse, TextContent, TextResourceContents, ToolCallStatus, ToolCallUpdateFields, }; use agentkit_core::{ @@ -952,86 +952,29 @@ struct SessionBindingGuard { session_id: agentkit_acp::SessionId, } -/// Legacy wire projection of the shared activity reducer. Foreground activity -/// is settled by the standard prompt response; only unsolicited activity needs -/// the extension notification. Both paths observe the same per-session reducer. -#[derive(Clone)] -struct LegacyActivity { - state: Arc>, +/// v1 retains standard prompt responses; autonomous intervals use the extension. +fn legacy_activity( session_id: agentkit_acp::SessionId, - notifications: mpsc::UnboundedSender, -} - -#[derive(Default)] -struct LegacyActivityProjection { - activity: activity::SessionActivity, - unsolicited: bool, - next_id: u64, - notification_id: Option, -} - -impl LegacyActivity { - fn new( - session_id: agentkit_acp::SessionId, - notifications: mpsc::UnboundedSender, - ) -> Self { - Self { - state: Arc::new(Mutex::new(LegacyActivityProjection::default())), - session_id, - notifications, - } - } - - fn begin_unsolicited(&self) { - self.state - .lock() - .unwrap_or_else(|e| e.into_inner()) - .unsolicited = true; - } - - fn settle(&self, error: Option) { - let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); - let transition = state.activity.settle(); - state.unsolicited = false; - if transition.is_some() - && let Some(turn_id) = state.notification_id.take() - { - let _ = self.notifications.send(TurnStateNotification { - session_id: self.session_id.clone(), - turn_id, - active: false, - error, - }); - } - } - - fn observe(&self, event: &AgentEvent) { - let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); - if state.activity.observe(event) == Some(true) && state.unsolicited { - state.next_id = state.next_id.wrapping_add(1); - let turn_id = state.next_id; - state.notification_id = Some(turn_id); - let _ = self.notifications.send(TurnStateNotification { - session_id: self.session_id.clone(), - turn_id, - active: true, - error: None, - }); + notify: impl Fn(TurnStateNotification) -> Result<(), AcpRuntimeError> + Send + Sync + 'static, +) -> activity::SessionActivity { + activity::SessionActivity::new(move |transition| { + if transition.origin == activity::ExecutionOrigin::Autonomous { + notify(TurnStateNotification { + session_id: session_id.clone(), + turn_id: transition.id, + active: transition.active, + error: transition.error, + })?; } - } -} - -impl LoopObserver for LegacyActivity { - fn handle_event(&self, event: ObservedEvent) { - self.observe(&event.event); - } + Ok(()) + }) } #[derive(Clone)] struct ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, - activity: LegacyActivity, + activity: activity::SessionActivity, session_id: agentkit_acp::SessionId, } @@ -1040,7 +983,7 @@ impl ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, session_id: agentkit_acp::SessionId, - activity: LegacyActivity, + activity: activity::SessionActivity, ) -> Self { Self { activity, @@ -1337,9 +1280,15 @@ impl Server { let cancellation = CancellationController::new(); let (client, messages) = AcpClientHandle::channel(); tokio::spawn(drain_client_messages(messages, connection.clone())); - let (turn_states, turn_state_messages) = mpsc::unbounded_channel(); - let activity = LegacyActivity::new(session_id.clone(), turn_states); - tokio::spawn(drain_turn_states(turn_state_messages, connection.clone())); + // Lifecycle notifications enter the SDK queue synchronously. Running is + // enqueued before the observer can produce content; finalization flushes + // the content drain before enqueuing Idle on this same SDK queue. + let activity_connection = connection.clone(); + let activity = legacy_activity(session_id.clone(), move |state| { + activity_connection + .send_notification(state) + .map_err(|error| AcpRuntimeError::Sdk(error.to_string())) + }); let mut metadata = MetadataMap::new(); metadata.insert("acp.cwd".into(), json!(cwd)); @@ -1656,7 +1605,7 @@ struct SessionActor { adapter: SelectableAdapter, catalog: Vec, commands: mpsc::Receiver, - activity: LegacyActivity, + activity: activity::SessionActivity, mcp_events: crate::tools::mcp::McpSubscription, } @@ -1685,7 +1634,7 @@ async fn session_actor(actor: SessionActor) { biased; command = commands.recv() => match command { Some(Command::Prompt { request, reply }) => { - let result = drive_runtime_prompt( + let result = activity.execute(activity::ExecutionOrigin::Prompt, drive_runtime_prompt( &session_id, &runtime, &integration, @@ -1695,9 +1644,11 @@ async fn session_actor(actor: SessionActor) { &tasks, &background_jobs, structured_completion, - ).await; - activity.settle(None); - let _ = reply.send(result); + ), |reason| Some(reason.clone())).await; + let response = result.and_then(|reason| { + agentkit_acp::finish_reason_to_stop_reason(&reason).map(PromptResponse::new) + }); + let _ = reply.send(response); } // The server already interrupted the shared controller; this // marker only establishes its serialized actor position. @@ -2060,7 +2011,7 @@ async fn drive_runtime_prompt( tasks: &TaskManagerHandle, background_jobs: &BackgroundJobs, structured_completion: bool, -) -> Result { +) -> Result { if structured_completion { let _ = settle_background_jobs(tasks, background_jobs).await?; } @@ -2081,17 +2032,17 @@ async fn drive_runtime_prompt( } })?; drop(current); - drive_submitted_prompt( + drive_finalized( session_id, integration, driver, - tasks, - background_jobs, - structured_completion, + true, + structured_completion.then_some((tasks, background_jobs)), ) .await } +#[cfg(test)] async fn drive_submitted_prompt( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, @@ -2100,48 +2051,34 @@ async fn drive_submitted_prompt( background_jobs: &BackgroundJobs, structured_completion: bool, ) -> Result { - let response = match drive_until_pause( + drive_until_pause( session_id, integration, driver, true, structured_completion.then_some((tasks, background_jobs)), ) - .await - { - Ok(Some(response)) => response, - Ok(None) => { - return Err(AcpRuntimeError::Loop( - "prompt ended without a response".into(), - )); - } - Err(error) => { - if structured_completion { - cancel_background_jobs(tasks, background_jobs).await; - let _ = settle_background_jobs(tasks, background_jobs).await; - } - return Err(error); - } - }; - if structured_completion && response.stop_reason == StopReason::Cancelled { - cancel_background_jobs(tasks, background_jobs).await; - let _ = settle_background_jobs(tasks, background_jobs).await?; - } - Ok(response) + .await? + .ok_or_else(|| AcpRuntimeError::Loop("prompt ended without a response".into())) } async fn drive_unsolicited( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, driver: &mut LoopDriver, - activity: &LegacyActivity, + activity: &activity::SessionActivity, ) -> Result<(), AcpRuntimeError> { - activity.begin_unsolicited(); - let result = drive_autonomous(session_id, integration, driver).await; - activity.settle(result.as_ref().err().map(ToString::to_string)); - result + activity + .execute( + activity::ExecutionOrigin::Autonomous, + drive_finalized(session_id, integration, driver, false, None), + |reason| Some(reason.clone()), + ) + .await + .map(|_| ()) } +#[cfg(test)] async fn drive_autonomous( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, @@ -2151,6 +2088,7 @@ async fn drive_autonomous( Ok(()) } +#[cfg(test)] async fn drive_until_pause( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, @@ -2158,12 +2096,49 @@ async fn drive_until_pause( answer_prompt: bool, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, ) -> Result, AcpRuntimeError> { + let reason = + drive_finalized(session_id, integration, driver, answer_prompt, structured).await?; + if answer_prompt { + Ok(Some(PromptResponse::new( + agentkit_acp::finish_reason_to_stop_reason(&reason)?, + ))) + } else { + Ok(None) + } +} + +async fn drive_finalized( + session_id: &agentkit_acp::SessionId, + integration: &AcpIntegration, + driver: &mut LoopDriver, + answer_prompt: bool, + structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, +) -> Result { + let cancellation = integration.cancellation_handle(session_id)?; + let generation = cancellation.generation(); + let result = + drive_domain_until_pause(session_id, integration, driver, answer_prompt, structured).await; + activity::finalize( + activity::ExecutionOutcome::new(result, cancellation.is_cancelled_since(generation)), + structured, + integration.flush_session_updates(session_id), + |_| Ok(()), + ) + .await +} + +async fn drive_domain_until_pause( + session_id: &agentkit_acp::SessionId, + integration: &AcpIntegration, + driver: &mut LoopDriver, + answer_prompt: bool, + structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, +) -> Result { loop { let step = match driver.next().await { Ok(step) => step, Err(LoopError::Cancelled) => { - integration.flush_session_updates(session_id).await?; - return Ok(answer_prompt.then(|| PromptResponse::new(StopReason::Cancelled))); + return Ok(FinishReason::Cancelled); } Err(error) => return Err(record_acp_loop_failure(session_id, &error)), }; @@ -2172,27 +2147,24 @@ async fn drive_until_pause( if result.finish_reason == FinishReason::ToolCall { continue; } - integration.flush_session_updates(session_id).await?; - if !answer_prompt { - return Ok(None); + if result.finish_reason == FinishReason::Error { + return Err(AcpRuntimeError::Loop("model turn failed".into())); } - let reason = agentkit_acp::finish_reason_to_stop_reason(&result.finish_reason)?; if let Some((tasks, background_jobs)) = structured && settle_background_jobs(tasks, background_jobs).await? { continue; } - return Ok(Some(PromptResponse::new(reason))); + return Ok(result.finish_reason); } LoopStep::Interrupt(LoopInterrupt::AwaitingInput(_)) => { - integration.flush_session_updates(session_id).await?; if answer_prompt && let Some((tasks, background_jobs)) = structured && settle_background_jobs(tasks, background_jobs).await? { continue; } - return Ok(answer_prompt.then(|| PromptResponse::new(StopReason::EndTurn))); + return Ok(FinishReason::Completed); } LoopStep::Interrupt(LoopInterrupt::AfterToolResult(_)) => continue, LoopStep::Interrupt(LoopInterrupt::ApprovalRequest(pending)) => { @@ -2541,26 +2513,20 @@ fn capabilities(logout_authentication: bool) -> agentkit_acp::AgentCapabilities } } -async fn drain_turn_states( - mut states: mpsc::UnboundedReceiver, - connection: ConnectionTo, -) { - while let Some(state) = states.recv().await { - let _ = connection.send_notification(state); - } -} - async fn drain_client_messages( mut messages: mpsc::UnboundedReceiver, connection: ConnectionTo, ) { + let mut failed = false; while let Some(message) = messages.recv().await { match message { AcpClientMessage::SessionNotification(notification) => { - let _ = connection.send_notification(notification); + failed |= connection.send_notification(notification).is_err(); } AcpClientMessage::Flush { response } => { - let _ = response.send(()); + if !failed { + let _ = response.send(()); + } } AcpClientMessage::PermissionRequest { request, response } => { let connection = connection.clone(); @@ -2608,6 +2574,104 @@ pub(super) mod tests { }; use super::*; + use agentkit_acp::StopReason; + + fn test_activity( + id: agentkit_acp::SessionId, + tx: mpsc::UnboundedSender, + ) -> activity::SessionActivity { + legacy_activity(id, move |state| { + tx.send(state).map_err(|_| AcpRuntimeError::ClientClosed) + }) + } + + #[tokio::test] + async fn lifecycle_transport_orders_running_partial_content_and_error_idle() { + let (client_transport, agent_transport) = Channel::duplex(); + let (events, mut received) = mpsc::unbounded_channel(); + let content_events = events.clone(); + let client = tokio::spawn(async move { + agent_client_protocol::Client + .builder() + .on_receive_notification( + async move |state: TurnStateNotification, _cx| { + events + .send(if state.active { "running" } else { "idle" }) + .unwrap(); + if !state.active { + assert!(state.error.is_some()); + } + Ok(()) + }, + agent_client_protocol::on_receive_notification!(), + ) + .on_receive_notification( + async move |_content: SessionNotification, _cx| { + content_events.send("content").unwrap(); + Ok(()) + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_to(client_transport) + .await + }); + agent_client_protocol::Agent + .builder() + .connect_with(agent_transport, async move |connection| { + let id = agentkit_acp::SessionId::new("ordered"); + let lifecycle_connection = connection.clone(); + let activity = legacy_activity(id.clone(), move |state| { + lifecycle_connection + .send_notification(state) + .map_err(|error| AcpRuntimeError::Sdk(error.to_string())) + }); + let (content, messages) = AcpClientHandle::channel(); + let result = activity + .execute( + activity::ExecutionOrigin::Autonomous, + async { + activity.observe(&AgentEvent::TurnStarted { + session_id: AgentkitSessionId::new("ordered"), + turn_id: agentkit_core::TurnId::new("first"), + }); + content.notify_session(SessionNotification::new( + id, + SessionUpdate::AgentMessageChunk(ContentChunk::new( + ContentBlock::Text(TextContent::new("partial")), + )), + ))?; + // Deliberately delay the content drain. Idle must await it, + // including when the operation fails after partial output. + tokio::spawn(drain_client_messages(messages, connection.clone())); + activity::finalize( + activity::ExecutionOutcome::new( + Err(AcpRuntimeError::Loop("approval failed".into())), + false, + ), + None, + content.flush(), + |_| Ok(()), + ) + .await + }, + |reason| Some(reason.clone()), + ) + .await; + assert!(result.is_err()); + for expected in ["running", "content", "idle"] { + assert_eq!( + timeout(Duration::from_secs(2), received.recv()) + .await + .unwrap(), + Some(expected) + ); + } + Ok(()) + }) + .await + .unwrap(); + client.abort(); + } #[test] fn terminal_auth_methods_require_client_support() { @@ -3275,7 +3339,7 @@ pub(super) mod tests { integration.clone(), client, session_id.clone(), - LegacyActivity::new(session_id.clone(), mpsc::unbounded_channel().0), + test_activity(session_id.clone(), mpsc::unbounded_channel().0), ); let emit = |event| { observer.handle_event(ObservedEvent { @@ -3973,7 +4037,7 @@ pub(super) mod tests { ) .await .unwrap(); - assert_eq!(response.stop_reason, StopReason::EndTurn); + assert_eq!(response, FinishReason::Completed); assert_eq!(notification_items_seen.load(Ordering::SeqCst), 1); drain.abort(); } @@ -4131,7 +4195,7 @@ pub(super) mod tests { } }); let (notifications, mut states) = mpsc::unbounded_channel(); - let activity = LegacyActivity::new(session_id.clone(), notifications); + let activity = test_activity(session_id.clone(), notifications); // Start the script at its plain content response, with no tool prerequisite. let turns = Arc::new(AtomicUsize::new(2)); let mut driver = Agent::builder() @@ -4179,7 +4243,7 @@ pub(super) mod tests { assert!(states.try_recv().is_err()); // Foreground uses the very same reducer, but retains standard v1 response - // semantics without extension notifications (or consuming extension IDs). + // semantics without extension notifications (while sharing session activity IDs). driver .submit_input(vec![Item::text(ItemKind::User, "foreground")]) .unwrap(); @@ -4187,13 +4251,13 @@ pub(super) mod tests { .await .unwrap() .unwrap(); - activity.settle(None); + activity.settle(None, None).unwrap(); assert_eq!(response.stop_reason, StopReason::EndTurn); assert!(states.try_recv().is_err()); // Multiple logical turns drained within one autonomous interval do not // publish an intermediate terminal state or allocate another turn ID. - activity.begin_unsolicited(); + activity.begin(activity::ExecutionOrigin::Autonomous); for text in ["first continuation", "second continuation"] { driver.submit_input(vec![Item::notification(text)]).unwrap(); drive_autonomous(&session_id, &integration, &mut driver) @@ -4202,14 +4266,16 @@ pub(super) mod tests { } let started = states.try_recv().unwrap(); assert!(started.active); - assert_eq!(started.turn_id, 2); + assert_eq!(started.turn_id, 3); assert!(states.try_recv().is_err()); - activity.settle(Some("terminal error".into())); + activity + .settle(None, Some("terminal error".into())) + .unwrap(); let ended = states.try_recv().unwrap(); assert!(!ended.active); assert_eq!(ended.turn_id, started.turn_id); assert_eq!(ended.error.as_deref(), Some("terminal error")); - activity.settle(None); + activity.settle(None, None).unwrap(); assert!(states.try_recv().is_err()); assert_eq!(turns.load(Ordering::SeqCst), 6); drain.abort(); @@ -4274,7 +4340,7 @@ pub(super) mod tests { release: Arc::clone(&release), }); let (turn_states_tx, mut turn_states_rx) = mpsc::unbounded_channel(); - let activity = LegacyActivity::new(acp_session_id.clone(), turn_states_tx); + let activity = test_activity(acp_session_id.clone(), turn_states_tx); let driver = Agent::builder() .model(ScriptAdapter { turns: Arc::clone(&turns), diff --git a/src/protocols/acp/activity.rs b/src/protocols/acp/activity.rs index 0a08e14..36628c3 100644 --- a/src/protocols/acp/activity.rs +++ b/src/protocols/acp/activity.rs @@ -1,88 +1,368 @@ -use agentkit_loop::AgentEvent; +use std::sync::{Arc, Mutex}; -/// Authoritative activity state for an attached session. A drive/wake-up is -/// not itself activity: only a logical turn started by the loop enters Running. -/// A finished logical turn remains Settling until the actor has drained steering -/// and structured background work. Those continuations belong to the same client-visible -/// activity interval, rather than producing an Idle/Running flicker. -#[derive(Default)] -pub(super) enum SessionActivity { +use agentkit_acp::AcpRuntimeError; +use agentkit_core::FinishReason; +use agentkit_loop::{AgentEvent, LoopObserver, ObservedEvent}; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) enum ExecutionOrigin { + #[default] + Prompt, + Autonomous, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum State { #[default] Idle, Running, Settling, } +/// An ordered snapshot, not another reducer. Projections select only wire format. +#[derive(Clone, Debug)] +pub(super) struct Transition { + pub id: u64, + pub origin: ExecutionOrigin, + pub active: bool, + pub reason: FinishReason, + pub error: Option, +} + +#[derive(Default)] +struct Activity { + state: State, + next_id: u64, + origin: ExecutionOrigin, + current: Option, +} + +/// Session-owned lifecycle instrument shared by the actor and its observer. +/// Admission alone is silent; TurnStarted allocates the activity identity. All +/// logical turns drained by an execution share that identity until settlement. +/// Projection runs synchronously under the state lock: Running precedes content, +/// and the actor must flush final content/diagnostics before settling to Idle. +/// Projections only enqueue wire notifications; they must not reenter this +/// instrument while its ordered transition is being projected. +#[derive(Clone)] +pub(super) struct SessionActivity { + state: Arc>, + project: Arc Result<(), AcpRuntimeError> + Send + Sync>, +} + impl SessionActivity { - pub(super) fn observe(&mut self, event: &AgentEvent) -> Option { + pub(super) fn new( + project: impl Fn(Transition) -> Result<(), AcpRuntimeError> + Send + Sync + 'static, + ) -> Self { + Self { + state: Arc::new(Mutex::new(Activity::default())), + project: Arc::new(project), + } + } + + pub(super) fn begin(&self, origin: ExecutionOrigin) { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + // Continuations cannot redefine the origin of an existing interval. + if state.state == State::Idle { + state.origin = origin; + } + } + + pub(super) fn observe(&self, event: &AgentEvent) { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); match event { AgentEvent::TurnStarted { .. } => { - let was_idle = matches!(self, Self::Idle); - *self = Self::Running; - was_idle.then_some(true) + let was_idle = state.state == State::Idle; + state.state = State::Running; + if was_idle { + state.next_id = state.next_id.wrapping_add(1); + let transition = Transition { + id: state.next_id, + origin: state.origin, + active: true, + reason: FinishReason::Completed, + error: None, + }; + state.current = Some(transition.clone()); + if let Err(error) = (self.project)(transition) { + tracing::debug!(%error, "failed to project session activity"); + } + } } - AgentEvent::TurnFinished(_) if !matches!(self, Self::Idle) => { - *self = Self::Settling; - None + AgentEvent::TurnFinished(result) if state.state != State::Idle => { + state.state = State::Settling; + if let Some(current) = &mut state.current { + current.reason = result.finish_reason.clone(); + } } - _ => None, + _ => {} } } - pub(super) fn settle(&mut self) -> Option { - if matches!(self, Self::Idle) { - return None; + pub(super) fn settle( + &self, + reason: Option, + error: Option, + ) -> Result<(), AcpRuntimeError> { + let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + state.origin = ExecutionOrigin::Prompt; + state.state = State::Idle; + if let Some(mut terminal) = state.current.take() { + terminal.active = false; + if let Some(reason) = reason { + terminal.reason = reason; + } + if error.is_some() { + terminal.reason = FinishReason::Error; + } + terminal.error = error; + (self.project)(terminal)?; + } + Ok(()) + } + + /// Both protocols use this boundary. The operation includes transport drain + /// and diagnostics; only then may the single session interval terminalize. + pub(super) async fn execute( + &self, + origin: ExecutionOrigin, + operation: impl std::future::Future>, + reason: impl FnOnce(&T) -> Option, + ) -> Result { + self.begin(origin); + let result = operation.await; + let terminal = result.as_ref().ok().and_then(reason); + let settled = self.settle(terminal, result.as_ref().err().map(ToString::to_string)); + match result { + Err(error) => Err(error), + Ok(value) => settled.map(|()| value), } - *self = Self::Idle; - Some(false) + } +} + +impl LoopObserver for SessionActivity { + fn handle_event(&self, event: ObservedEvent) { + self.observe(&event.event); } } #[cfg(test)] mod tests { use super::*; - use agentkit_core::{FinishReason, MetadataMap, SessionId}; + use agentkit_core::{MetadataMap, SessionId, TurnId}; + + fn started(id: &str) -> AgentEvent { + AgentEvent::TurnStarted { + session_id: SessionId::new("activity"), + turn_id: TurnId::new(id), + } + } + + fn finished(reason: FinishReason) -> AgentEvent { + AgentEvent::TurnFinished(agentkit_loop::TurnResult { + turn_id: TurnId::new("logical-turn"), + finish_reason: reason, + items: Vec::new(), + usage: None, + metadata: MetadataMap::new(), + }) + } - #[test] - fn session_activity_coalesces_continuations_and_settles_once() { - for reason in [ + #[tokio::test] + async fn execution_owns_identity_origin_and_ordered_settlement() { + let transitions = Arc::new(Mutex::new(Vec::new())); + let output = transitions.clone(); + let activity = SessionActivity::new(move |transition| { + output.lock().unwrap().push(transition); + Ok(()) + }); + activity + .execute(ExecutionOrigin::Autonomous, async { Ok(()) }, |_| None) + .await + .unwrap(); + assert!(transitions.lock().unwrap().is_empty()); + for (index, reason) in [ FinishReason::Completed, FinishReason::Cancelled, FinishReason::MaxTokens, FinishReason::Blocked, FinishReason::Error, - ] { - let mut state = SessionActivity::default(); - assert!(state.settle().is_none()); - let started = AgentEvent::TurnStarted { - session_id: SessionId::new("state-loop"), - turn_id: agentkit_core::TurnId::new("first"), - }; - assert!(matches!(state.observe(&started), Some(true))); - assert!(state.observe(&started).is_none()); - let finished = AgentEvent::TurnFinished(agentkit_loop::TurnResult { - turn_id: agentkit_core::TurnId::new("first"), - finish_reason: reason, - items: Vec::new(), - usage: None, - metadata: MetadataMap::new(), - }); - assert!(state.observe(&finished).is_none()); - assert!(matches!(state, SessionActivity::Settling)); - // A queued steer or structured synthesis continues before actor settlement. - assert!( - state - .observe(&AgentEvent::TurnStarted { - session_id: SessionId::new("state-loop"), - turn_id: agentkit_core::TurnId::new("continuation"), - }) - .is_none() - ); - assert!(matches!(state.settle(), Some(false))); - assert!(state.settle().is_none()); - assert!(state.observe(&finished).is_none()); - assert!(matches!(state, SessionActivity::Idle)); - assert!(matches!(state.observe(&started), Some(true))); + ] + .into_iter() + .enumerate() + { + activity + .execute( + ExecutionOrigin::Prompt, + async { + activity.observe(&started("first")); + assert!(transitions.lock().unwrap().last().unwrap().active); + activity.observe(&finished(FinishReason::ToolCall)); + // Steering and background synthesis cannot redefine the interval. + activity.begin(ExecutionOrigin::Autonomous); + activity.observe(&started("continuation")); + activity.observe(&finished(reason.clone())); + assert_eq!(transitions.lock().unwrap().len(), index * 2 + 1); + Ok(()) + }, + |_| None, + ) + .await + .unwrap(); + activity.settle(None, None).unwrap(); + activity.observe(&finished(FinishReason::Error)); + let events = transitions.lock().unwrap(); + let running = &events[index * 2]; + let idle = &events[index * 2 + 1]; + assert_eq!(running.id, index as u64 + 1); + assert_eq!(running.id, idle.id); + assert_eq!(idle.origin, ExecutionOrigin::Prompt); + assert!(!idle.active); + assert_eq!(idle.reason, reason); + } + activity + .execute( + ExecutionOrigin::Autonomous, + async { + activity.observe(&started("error")); + Err::<(), _>(AcpRuntimeError::Loop("terminal error".into())) + }, + |_| None, + ) + .await + .unwrap_err(); + activity.settle(None, None).unwrap(); + let events = transitions.lock().unwrap(); + assert_eq!(events.len(), 12); + let idle = events.last().unwrap(); + assert_eq!(idle.id, 6); + assert_eq!(idle.origin, ExecutionOrigin::Autonomous); + assert_eq!(idle.reason, FinishReason::Error); + assert!(idle.error.as_ref().unwrap().contains("terminal error")); + } + + #[tokio::test] + async fn finalization_normalizes_cancellation_and_never_skips_failed_drain() { + let diagnostic = Arc::new(Mutex::new(Vec::new())); + let reason = finalize( + ExecutionOutcome::new(Err(AcpRuntimeError::Loop("provider failed".into())), true), + None, + async { Ok(()) }, + |_| panic!("cancelled model failure is not an error"), + ) + .await + .unwrap(); + assert_eq!(reason, FinishReason::Cancelled); + + let order = Arc::new(Mutex::new(Vec::new())); + let output = order.clone(); + let activity = SessionActivity::new(move |transition| { + output + .lock() + .unwrap() + .push(if transition.active { "running" } else { "idle" }); + Ok(()) + }); + let result = activity + .execute( + ExecutionOrigin::Autonomous, + async { + activity.observe(&started("flush-failure")); + finalize( + ExecutionOutcome::new(Ok(FinishReason::Completed), false), + None, + async { + order.lock().unwrap().push("flush"); + Err(AcpRuntimeError::ClientClosed) + }, + |error| { + order.lock().unwrap().push("diagnostic"); + diagnostic.lock().unwrap().push(error.to_string()); + Ok(()) + }, + ) + .await + }, + |reason| Some(reason.clone()), + ) + .await; + assert!(matches!(result, Err(AcpRuntimeError::ClientClosed))); + activity.settle(None, None).unwrap(); + assert_eq!( + *order.lock().unwrap(), + ["running", "flush", "diagnostic", "idle"] + ); + assert_eq!(diagnostic.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn failed_terminal_projection_is_not_retried() { + let calls = Arc::new(Mutex::new(0)); + let output = calls.clone(); + let activity = SessionActivity::new(move |_| { + *output.lock().unwrap() += 1; + Err(AcpRuntimeError::ClientClosed) + }); + activity + .execute( + ExecutionOrigin::Autonomous, + async { + activity.observe(&started("first")); + Ok(()) + }, + |_| Some(FinishReason::Cancelled), + ) + .await + .unwrap_err(); + activity.settle(None, None).unwrap(); + assert_eq!(*calls.lock().unwrap(), 2); + } +} + +/// Domain outcome selected before any protocol representation. Cancellation wins +/// over model/approval errors; cleanup or delivery failure cannot become success. +pub(super) struct ExecutionOutcome { + result: Result, +} + +impl ExecutionOutcome { + pub(super) fn new(result: Result, cancelled: bool) -> Self { + Self { + result: if cancelled || matches!(result, Err(AcpRuntimeError::Cancelled)) { + Ok(FinishReason::Cancelled) + } else { + result + }, } } } + +/// Shared finalization order: cancel/drain structured work, drain all content even +/// on failure, render a diagnostic, then return the outcome for single settlement. +/// Hooks are transport only; neither hook chooses lifecycle or cleanup policy. +pub(super) async fn finalize( + outcome: ExecutionOutcome, + structured: Option<( + &agentkit_task_manager::TaskManagerHandle, + &crate::runtime::BackgroundJobs, + )>, + flush: impl std::future::Future>, + diagnostic: impl FnOnce(&AcpRuntimeError) -> Result<(), AcpRuntimeError>, +) -> Result { + let mut result = outcome.result; + if (result.is_err() || matches!(result, Ok(FinishReason::Cancelled))) + && let Some((tasks, jobs)) = structured + { + super::cancel_background_jobs(tasks, jobs).await; + if let Err(error) = super::settle_background_jobs(tasks, jobs).await { + result = Err(error); + } + } + if let Err(error) = flush.await { + result = Err(error); + } + if let Err(error) = &result { + diagnostic(error)?; + } + result +} diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 13f35f5..48e9dc6 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -32,7 +32,7 @@ use crate::{ runtime::{AcpDriverContext, BackgroundJobs, Runtime}, }; -use super::activity::SessionActivity; +use super::activity::{ExecutionOrigin, SessionActivity}; use super::{ AuthenticationRequiredData, CancelBackgroundRequest, CancelBackgroundResponse, @@ -131,9 +131,9 @@ fn map_loop_error(session_id: &wire::SessionId, error: &LoopError) -> AcpRuntime fn loop_error_stop_reason( session_id: &wire::SessionId, error: &LoopError, -) -> Result { +) -> Result { if matches!(error, LoopError::Cancelled) { - Ok(wire::StopReason::Cancelled) + Ok(FinishReason::Cancelled) } else { Err(map_loop_error(session_id, error)) } @@ -214,7 +214,6 @@ impl ReplacementGeneration { struct ResponseReplacementSink { inner: S, current: Arc>, - activity: Arc>, } impl ResponseReplacementSink { @@ -222,7 +221,6 @@ impl ResponseReplacementSink { Self { inner, current: Arc::new(Mutex::new(CurrentReplacementMessages::default())), - activity: Arc::new(Mutex::new(SessionActivity::default())), } } @@ -329,33 +327,22 @@ impl ResponseReplacementSink { } } -impl ResponseReplacementSink { - fn transition_activity( - &self, - session_id: &wire::SessionId, - transition: impl FnOnce(&mut SessionActivity) -> Option, - ) -> Result<(), AcpRuntimeError> { - let mut activity = self - .activity - .lock() - .unwrap_or_else(|error| error.into_inner()); - if let Some(state) = transition(&mut activity) { - send_state(&self.inner, session_id, state)?; - } - Ok(()) - } - - fn settle_activity( - &self, - session_id: &wire::SessionId, - reason: wire::StopReason, - ) -> Result<(), AcpRuntimeError> { - self.transition_activity(session_id, |activity| { - activity - .settle() - .map(|_| wire::StateUpdate::Idle(wire::IdleStateUpdate::new().stop_reason(reason))) - }) - } +/// Native v2 is a projection of the same session lifecycle as v1. +fn native_activity( + session_id: wire::SessionId, + sink: S, +) -> SessionActivity { + SessionActivity::new(move |transition| { + let state = if transition.active { + wire::StateUpdate::Running(wire::RunningStateUpdate::new()) + } else { + wire::StateUpdate::Idle( + wire::IdleStateUpdate::new() + .stop_reason(finish_reason_to_stop_reason(&transition.reason)), + ) + }; + send_state(&sink, &session_id, state) + }) } #[async_trait] @@ -385,6 +372,7 @@ impl AcpSessionUpdateSink for ResponseReplacementSink { inner: AcpIntegration, sink: ResponseReplacementSink, + activity: SessionActivity, session_id: wire::SessionId, } @@ -408,10 +396,12 @@ impl ResponseReplacementObserver { inner: AcpIntegration, sink: ResponseReplacementSink, session_id: wire::SessionId, + activity: SessionActivity, ) -> Self { Self { inner, sink, + activity, session_id, } } @@ -432,13 +422,7 @@ where S: AcpSessionUpdateSink + Clone, { fn handle_event(&self, event: ObservedEvent) { - if let Err(error) = self.sink.transition_activity(&self.session_id, |activity| { - activity - .observe(&event.event) - .map(|_| wire::StateUpdate::Running(wire::RunningStateUpdate::new())) - }) { - tracing::debug!(%error, "failed to queue ACP v2 activity update"); - } + self.activity.observe(&event.event); if let AgentEvent::UsageUpdated(usage) = &event.event { let Some(update) = usage_update(usage) else { return; @@ -807,6 +791,7 @@ impl Server { let session_id = wire::SessionId::new(claim.id()); let cancellation = CancellationController::new(); let sink = ResponseReplacementSink::new(ConnectionSink(connection)); + let activity = native_activity(session_id.clone(), sink.clone()); let binding = AcpSessionBinding::new(session_id.clone(), SessionId::new(claim.id()), sink.clone()) .cancellation(cancellation); @@ -819,6 +804,7 @@ impl Server { self.integration.as_ref().clone(), sink.clone(), session_id.clone(), + activity.clone(), ); let context = AcpDriverContext { cwd, @@ -851,6 +837,7 @@ impl Server { handle: handle.clone(), busy: Arc::clone(&busy), binding, + activity, sink, driver: driver.driver, tasks: driver.tasks, @@ -1122,6 +1109,7 @@ struct SessionActor { busy: Arc, binding: BindingGuard, sink: ResponseReplacementSink, + activity: SessionActivity, driver: LoopDriver, tasks: TaskManagerHandle, background_jobs: BackgroundJobs, @@ -1141,6 +1129,7 @@ async fn session_actor(actor: SessionActor) handle, busy, binding, + activity, sink, mut driver, tasks, @@ -1170,7 +1159,7 @@ async fn session_actor(actor: SessionActor) &tasks, &background_jobs, structured_completion, - ) + &activity,) .await; busy.store(false, Ordering::Release); if let Err(error) = result { @@ -1204,7 +1193,7 @@ async fn session_actor(actor: SessionActor) &busy, &mut driver, &sink, - ).await, + &activity,).await, Err(error) => Err(map_loop_error(&session_id, &error)), }; if let Err(error) = result { @@ -1223,7 +1212,7 @@ async fn session_actor(actor: SessionActor) &busy, &mut driver, &sink, - ).await + &activity,).await { eprintln!("ACP v2 autonomous turn failed for {session_id}: {error}"); } @@ -1257,6 +1246,7 @@ async fn prepare_prompt( tasks: &TaskManagerHandle, background_jobs: &BackgroundJobs, structured_completion: bool, + activity: &SessionActivity, ) -> Result<(), AcpRuntimeError> { let PromptCommand { request, @@ -1329,14 +1319,12 @@ async fn prepare_prompt( sink, cancellation_generation, structured_completion.then_some((tasks, background_jobs)), + activity, + ExecutionOrigin::Prompt, ) .await } .await; - if result.is_err() && structured_completion { - super::cancel_background_jobs(tasks, background_jobs).await; - let _ = super::settle_background_jobs(tasks, background_jobs).await; - } integration.finish_prompt(session_id); handle.stop_injection_turn(); result @@ -1378,7 +1366,7 @@ async fn drive_prompt( control: &C, cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, -) -> Result +) -> Result where S: ModelSession + Send + 'static, C: TurnControl, @@ -1389,13 +1377,13 @@ where Err(error) => { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return loop_error_stop_reason(session_id, &error); } }; if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } match step { LoopStep::Finished(result) => { @@ -1405,7 +1393,7 @@ where if result.finish_reason == FinishReason::Error { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return Err(AcpRuntimeError::Loop("model turn failed".into())); } @@ -1419,15 +1407,15 @@ where continue; } Ok(AcpInjectionBoundary::Stopped) => { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } Ok(AcpInjectionBoundary::Finished) => { - return Ok(finish_reason_to_stop_reason(&result.finish_reason)); + return Ok(result.finish_reason); } Err(error) => { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return Err(error); } @@ -1444,15 +1432,15 @@ where continue; } Ok(AcpInjectionBoundary::Stopped) => { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } Ok(AcpInjectionBoundary::Finished) => { - return Ok(wire::StopReason::EndTurn); + return Ok(FinishReason::Completed); } Err(error) => { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return Err(error); } @@ -1461,12 +1449,12 @@ where LoopStep::Interrupt(LoopInterrupt::AfterToolResult(_)) => { match control.handle_injection_boundary(driver, false).await { Ok(AcpInjectionBoundary::Stopped) => { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } Err(error) => { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return Err(error); } @@ -1477,7 +1465,7 @@ where if let Err(error) = driver.cancel_pending_approvals().await { control.stop_injection_turn(); if control.is_cancelled_since(cancellation_generation) { - return Ok(wire::StopReason::Cancelled); + return Ok(FinishReason::Cancelled); } return loop_error_stop_reason(session_id, &error); } @@ -1494,52 +1482,42 @@ async fn run_active_turn( sink: &ResponseReplacementSink, cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, + activity: &SessionActivity, + origin: ExecutionOrigin, ) -> Result<(), AcpRuntimeError> { - let mut result = match drive_prompt( - session_id, - driver, - handle, - cancellation_generation, - structured, - ) - .await - { - Err(_) - if handle - .cancellation_handle() - .is_cancelled_since(cancellation_generation) => - { - Ok(wire::StopReason::Cancelled) - } - result => result, - }; - if (result.is_err() || matches!(&result, Ok(wire::StopReason::Cancelled))) - && let Some((tasks, background_jobs)) = structured - { - super::cancel_background_jobs(tasks, background_jobs).await; - if let Err(error) = super::settle_background_jobs(tasks, background_jobs).await - && result.is_ok() - { - result = Err(error); - } - } - handle.stop_injection_turn(); - let _ = integration.flush_session_updates(session_id).await; - integration.finish_prompt(session_id); - - // Every exit goes through the same state settlement, including cleanup errors. - // Diagnostic delivery failure must not prevent the Idle transition. - let diagnostic = match &result { - Err(error) => sink.update(error_diagnostic_notification(session_id, error)), - Ok(_) => Ok(()), - }; - let stop_reason = result - .as_ref() - .cloned() - .unwrap_or_else(|_| error_stop_reason()); - let settled = sink.settle_activity(session_id, stop_reason); - diagnostic.and(settled)?; - result.map(|_| ()) + activity + .execute( + origin, + async { + let result = drive_prompt( + session_id, + driver, + handle, + cancellation_generation, + structured, + ) + .await; + let outcome = super::activity::ExecutionOutcome::new( + result, + handle + .cancellation_handle() + .is_cancelled_since(cancellation_generation), + ); + handle.stop_injection_turn(); + let result = super::activity::finalize( + outcome, + structured, + integration.flush_session_updates(session_id), + |error| sink.update(error_diagnostic_notification(session_id, error)), + ) + .await; + integration.finish_prompt(session_id); + result + }, + |reason| Some(reason.clone()), + ) + .await + .map(|_| ()) } async fn drive_autonomous( @@ -1549,6 +1527,7 @@ async fn drive_autonomous( busy: &AtomicBool, driver: &mut LoopDriver, sink: &ResponseReplacementSink, + activity: &SessionActivity, ) -> Result<(), AcpRuntimeError> { if claim_prompt(busy).is_err() { return Ok(()); @@ -1565,6 +1544,8 @@ async fn drive_autonomous( sink, cancellation_generation, None, + activity, + ExecutionOrigin::Autonomous, ) .await; integration.finish_prompt(session_id); @@ -2191,6 +2172,7 @@ mod tests { struct RecordingSink { updates: Arc>>, flushes: Arc, + fail_flush: Arc, } #[async_trait] @@ -2212,7 +2194,11 @@ mod tests { async fn flush(&self) -> Result<(), AcpRuntimeError> { self.flushes.fetch_add(1, Ordering::Relaxed); - Ok(()) + if self.fail_flush.load(Ordering::Relaxed) { + Err(AcpRuntimeError::ClientClosed) + } else { + Ok(()) + } } } @@ -2245,10 +2231,12 @@ mod tests { fn observer_reports_usage_with_a_known_context_window() { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); + let activity = native_activity(wire::SessionId::new("usage-session"), sink.clone()); let observer = ResponseReplacementObserver::new( AcpIntegration::default(), sink, wire::SessionId::new("usage-session"), + activity.clone(), ); let loop_session_id = SessionId::new("usage-loop"); let emit = |usage| { @@ -2283,6 +2271,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("thought-session"); + let activity = native_activity(session_id.clone(), sink.clone()); let loop_session_id = SessionId::new("thought-loop"); let _handle = integration .bind_session(AcpSessionBinding::new( @@ -2291,7 +2280,8 @@ mod tests { sink.clone(), )) .unwrap(); - let observer = ResponseReplacementObserver::new(integration, sink, session_id); + let observer = + ResponseReplacementObserver::new(integration, sink, session_id, activity.clone()); let emit = |delta| { observer.handle_event(ObservedEvent { session_id: Arc::new(loop_session_id.clone()), @@ -2339,6 +2329,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("replacement-session"); + let activity = native_activity(session_id.clone(), sink.clone()); let loop_session_id = SessionId::new("replacement-loop"); let _handle = integration .bind_session(AcpSessionBinding::new( @@ -2347,7 +2338,12 @@ mod tests { sink.clone(), )) .unwrap(); - let observer = ResponseReplacementObserver::new(integration, sink, session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration, + sink, + session_id.clone(), + activity.clone(), + ); let emit = |event| { observer.handle_event(ObservedEvent { session_id: Arc::new(loop_session_id.clone()), @@ -2481,6 +2477,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("cancelled-replacement-session"); + let activity = native_activity(session_id.clone(), sink.clone()); let loop_session_id = SessionId::new("cancelled-replacement-loop"); let _handle = integration .bind_session(AcpSessionBinding::new( @@ -2489,7 +2486,8 @@ mod tests { sink.clone(), )) .unwrap(); - let observer = ResponseReplacementObserver::new(integration, sink, session_id); + let observer = + ResponseReplacementObserver::new(integration, sink, session_id, activity.clone()); let emit = |event| { observer.handle_event(ObservedEvent { session_id: Arc::new(loop_session_id.clone()), @@ -2815,6 +2813,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("cancelled-marker-session"); + let activity = native_activity(session_id.clone(), sink.clone()); let loop_session_id = SessionId::new("cancelled-marker-loop"); let cancellation = CancellationController::new(); let handle = integration @@ -2823,7 +2822,8 @@ mod tests { .cancellation(cancellation), ) .unwrap(); - let observer = ResponseReplacementObserver::new(integration, sink, session_id); + let observer = + ResponseReplacementObserver::new(integration, sink, session_id, activity.clone()); let mut driver = Agent::builder() .model(StreamingCancellationAdapter { interrupt: handle.clone(), @@ -2904,7 +2904,7 @@ mod tests { )); assert_eq!( loop_error_stop_reason(&session_id, &LoopError::Cancelled).unwrap(), - wire::StopReason::Cancelled + FinishReason::Cancelled ); } @@ -2914,6 +2914,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("foreground-provider-error"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -2924,8 +2925,12 @@ mod tests { handle.prepare_injection_turn(); let cancellation_generation = handle.cancellation_handle().generation(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::ProviderError, @@ -2973,6 +2978,7 @@ mod tests { &tasks, &background_jobs, false, + &activity, ), acknowledge, ); @@ -3036,6 +3042,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("v2-structured"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -3043,8 +3050,12 @@ mod tests { sink.clone(), )) .unwrap(); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(ScriptAdapter { turns: Arc::clone(&turns), @@ -3075,6 +3086,8 @@ mod tests { &sink, generation, Some((&tasks, &background_jobs)), + &activity, + ExecutionOrigin::Prompt, ); tokio::pin!(prompt); @@ -3191,7 +3204,7 @@ mod tests { let result = drive_prompt(&session_id, &mut driver, &handle, generation, None).await; - assert_eq!(result.unwrap(), wire::StopReason::Cancelled); + assert_eq!(result.unwrap(), FinishReason::Cancelled); } #[tokio::test] @@ -3221,6 +3234,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-no-work"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -3229,8 +3243,12 @@ mod tests { )) .unwrap(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::Content, @@ -3257,6 +3275,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); @@ -3273,6 +3292,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-content"); + let activity = native_activity(session_id.clone(), sink.clone()); let loop_session_id = SessionId::new("autonomous-content-loop"); let handle = integration .bind_session(AcpSessionBinding::new( @@ -3282,8 +3302,12 @@ mod tests { )) .unwrap(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::Content, @@ -3308,6 +3332,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); @@ -3324,6 +3349,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); @@ -3345,6 +3371,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-provider-error"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -3353,8 +3380,12 @@ mod tests { )) .unwrap(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::ProviderError, @@ -3384,6 +3415,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await; @@ -3411,6 +3443,7 @@ mod tests { let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-cancel"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -3421,8 +3454,12 @@ mod tests { )) .unwrap(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::ProviderError, @@ -3452,6 +3489,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); @@ -3465,12 +3503,75 @@ mod tests { ); } + #[tokio::test] + async fn flush_failure_after_content_reports_error_before_idle_once() { + let integration = AcpIntegration::default(); + let recording = RecordingSink::default(); + recording.fail_flush.store(true, Ordering::Relaxed); + let sink = ResponseReplacementSink::new(recording.clone()); + let session_id = wire::SessionId::new("failed-flush"); + let activity = native_activity(session_id.clone(), sink.clone()); + let handle = integration + .bind_session(AcpSessionBinding::new( + session_id.clone(), + SessionId::new("failed-flush-loop"), + sink.clone(), + )) + .unwrap(); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::Content, + turns: Arc::new(AtomicU64::new(0)), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start(SessionConfig::new(SessionId::new("failed-flush-loop")).without_cache()) + .await + .unwrap(); + driver + .submit_input(vec![Item::notification("work")]) + .unwrap(); + let result = drive_autonomous( + &session_id, + &integration, + &handle, + &AtomicBool::new(false), + &mut driver, + &sink, + &activity, + ) + .await; + assert!(matches!(result, Err(AcpRuntimeError::ClientClosed))); + activity.settle(None, None).unwrap(); + assert_eq!(recording.flushes.load(Ordering::Relaxed), 1); + let updates = recording.updates.lock().unwrap(); + assert_running_then_idle(&updates, error_stop_reason()); + assert!(matches!( + updates[updates.len() - 2].update, + wire::SessionUpdate::AgentMessage(_) + )); + assert!( + serde_json::to_string(&updates[updates.len() - 2]) + .unwrap() + .contains(&AcpRuntimeError::ClientClosed.to_string()) + ); + } + #[tokio::test] async fn autonomous_finish_error_emits_running_error_idle() { let integration = AcpIntegration::default(); let recording = RecordingSink::default(); let sink = ResponseReplacementSink::new(recording.clone()); let session_id = wire::SessionId::new("autonomous-error"); + let activity = native_activity(session_id.clone(), sink.clone()); let handle = integration .bind_session(AcpSessionBinding::new( session_id.clone(), @@ -3479,8 +3580,12 @@ mod tests { )) .unwrap(); let turns = Arc::new(AtomicU64::new(0)); - let observer = - ResponseReplacementObserver::new(integration.clone(), sink.clone(), session_id.clone()); + let observer = ResponseReplacementObserver::new( + integration.clone(), + sink.clone(), + session_id.clone(), + activity.clone(), + ); let mut driver = Agent::builder() .model(TestAdapter { outcome: TestOutcome::FinishError, @@ -3510,6 +3615,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await; From 8fb5b9bdd7d3fbbc8314cc024724efbd4156d141 Mon Sep 17 00:00:00 2001 From: daniel Date: Sat, 5 Sep 2026 11:24:22 +0100 Subject: [PATCH 3/5] fix(acp): satisfy strict lifecycle Clippy checks --- src/protocols/acp/activity.rs | 96 +++++++++++++++++------------------ src/protocols/acp/v2.rs | 1 + 2 files changed, 49 insertions(+), 48 deletions(-) diff --git a/src/protocols/acp/activity.rs b/src/protocols/acp/activity.rs index 36628c3..e7672a3 100644 --- a/src/protocols/acp/activity.rs +++ b/src/protocols/acp/activity.rs @@ -146,6 +146,54 @@ impl LoopObserver for SessionActivity { } } +/// Domain outcome selected before any protocol representation. Cancellation wins +/// over model/approval errors; cleanup or delivery failure cannot become success. +pub(super) struct ExecutionOutcome { + result: Result, +} + +impl ExecutionOutcome { + pub(super) fn new(result: Result, cancelled: bool) -> Self { + Self { + result: if cancelled || matches!(result, Err(AcpRuntimeError::Cancelled)) { + Ok(FinishReason::Cancelled) + } else { + result + }, + } + } +} + +/// Shared finalization order: cancel/drain structured work, drain all content even +/// on failure, render a diagnostic, then return the outcome for single settlement. +/// Hooks are transport only; neither hook chooses lifecycle or cleanup policy. +pub(super) async fn finalize( + outcome: ExecutionOutcome, + structured: Option<( + &agentkit_task_manager::TaskManagerHandle, + &crate::runtime::BackgroundJobs, + )>, + flush: impl std::future::Future>, + diagnostic: impl FnOnce(&AcpRuntimeError) -> Result<(), AcpRuntimeError>, +) -> Result { + let mut result = outcome.result; + if (result.is_err() || matches!(result, Ok(FinishReason::Cancelled))) + && let Some((tasks, jobs)) = structured + { + super::cancel_background_jobs(tasks, jobs).await; + if let Err(error) = super::settle_background_jobs(tasks, jobs).await { + result = Err(error); + } + } + if let Err(error) = flush.await { + result = Err(error); + } + if let Err(error) = &result { + diagnostic(error)?; + } + result +} + #[cfg(test)] mod tests { use super::*; @@ -318,51 +366,3 @@ mod tests { assert_eq!(*calls.lock().unwrap(), 2); } } - -/// Domain outcome selected before any protocol representation. Cancellation wins -/// over model/approval errors; cleanup or delivery failure cannot become success. -pub(super) struct ExecutionOutcome { - result: Result, -} - -impl ExecutionOutcome { - pub(super) fn new(result: Result, cancelled: bool) -> Self { - Self { - result: if cancelled || matches!(result, Err(AcpRuntimeError::Cancelled)) { - Ok(FinishReason::Cancelled) - } else { - result - }, - } - } -} - -/// Shared finalization order: cancel/drain structured work, drain all content even -/// on failure, render a diagnostic, then return the outcome for single settlement. -/// Hooks are transport only; neither hook chooses lifecycle or cleanup policy. -pub(super) async fn finalize( - outcome: ExecutionOutcome, - structured: Option<( - &agentkit_task_manager::TaskManagerHandle, - &crate::runtime::BackgroundJobs, - )>, - flush: impl std::future::Future>, - diagnostic: impl FnOnce(&AcpRuntimeError) -> Result<(), AcpRuntimeError>, -) -> Result { - let mut result = outcome.result; - if (result.is_err() || matches!(result, Ok(FinishReason::Cancelled))) - && let Some((tasks, jobs)) = structured - { - super::cancel_background_jobs(tasks, jobs).await; - if let Err(error) = super::settle_background_jobs(tasks, jobs).await { - result = Err(error); - } - } - if let Err(error) = flush.await { - result = Err(error); - } - if let Err(error) = &result { - diagnostic(error)?; - } - result -} diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 48e9dc6..ce88f8d 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -1474,6 +1474,7 @@ where } } +#[allow(clippy::too_many_arguments)] async fn run_active_turn( session_id: &wire::SessionId, integration: &AcpIntegration, From d5a93efb38fa4f85b5f1de9e059b62617e1d0fd0 Mon Sep 17 00:00:00 2001 From: daniel Date: Sat, 5 Sep 2026 12:26:58 +0100 Subject: [PATCH 4/5] test(acp): reproduce cancellation at tool-result boundary --- src/protocols/acp/v2.rs | 154 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index ce88f8d..39e15b8 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -2579,6 +2579,7 @@ mod tests { #[derive(Clone, Copy)] enum TestOutcome { + ToolThenContent, Content, FinishError, ProviderError, @@ -2693,6 +2694,30 @@ mod tests { handle.interrupt(); } match self.outcome { + TestOutcome::ToolThenContent => { + self.outcome = TestOutcome::Content; + let call = agentkit_core::ToolCallPart::new( + "boundary-call", + "missing-tool", + serde_json::json!({}), + ); + Ok(TestTurn { + events: VecDeque::from([ + ModelTurnEvent::ToolCall(call.clone()), + ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::ToolCall, + output_items: vec![Item::new( + ItemKind::Assistant, + vec![agentkit_core::Part::ToolCall(call)], + )], + usage: None, + metadata: MetadataMap::new(), + }), + ]), + }) + } TestOutcome::Content => { let text = "autonomous content"; Ok(TestTurn { @@ -3154,6 +3179,135 @@ mod tests { handle.stop_injection_turn(); } + struct BoundaryCancellationControl { + before_boundary: bool, + boundaries: AtomicU64, + } + + #[async_trait] + impl TurnControl for BoundaryCancellationControl { + fn stop_injection_turn(&self) {} + + fn is_cancelled_since(&self, _generation: u64) -> bool { + self.before_boundary + } + + async fn handle_injection_boundary( + &self, + _driver: &mut LoopDriver, + terminal: bool, + ) -> Result { + assert!(!terminal, "cancellation must occur at AfterToolResult"); + self.boundaries.fetch_add(1, Ordering::Relaxed); + Ok(AcpInjectionBoundary::Stopped) + } + } + + async fn assert_boundary_cancellation_retires_turn(before_boundary: bool) { + let integration = AcpIntegration::default(); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); + let session_id = wire::SessionId::new("boundary-cancel"); + let loop_session_id = SessionId::new("boundary-cancel-loop"); + let activity = native_activity(session_id.clone(), sink.clone()); + let _handle = integration + .bind_session(AcpSessionBinding::new( + session_id.clone(), + loop_session_id.clone(), + sink.clone(), + )) + .unwrap(); + let turns = Arc::new(AtomicU64::new(0)); + let observer = ResponseReplacementObserver::new( + integration, + sink, + session_id.clone(), + activity.clone(), + ); + let mut driver = Agent::builder() + .model(TestAdapter { + outcome: TestOutcome::ToolThenContent, + turns: turns.clone(), + interrupt: None, + }) + .observer(observer) + .build() + .unwrap() + .start(SessionConfig::new(loop_session_id).without_cache()) + .await + .unwrap(); + driver + .submit_input(vec![Item::text(ItemKind::User, "first")]) + .unwrap(); + let control = BoundaryCancellationControl { + before_boundary, + boundaries: AtomicU64::new(0), + }; + let reason = activity + .execute( + ExecutionOrigin::Prompt, + drive_prompt(&session_id, &mut driver, &control, 0, None), + |reason| Some(reason.clone()), + ) + .await + .unwrap(); + assert_eq!(reason, FinishReason::Cancelled); + assert_eq!( + turns.load(Ordering::Relaxed), + 1, + "must not resume cancelled model work" + ); + assert_eq!( + control.boundaries.load(Ordering::Relaxed), + u64::from(!before_boundary) + ); + assert!( + driver + .snapshot() + .transcript + .iter() + .any(|item| item.kind == ItemKind::Tool) + ); + assert_running_then_idle( + &recording.updates.lock().unwrap(), + wire::StopReason::Cancelled, + ); + recording.updates.lock().unwrap().clear(); + + driver + .submit_input(vec![Item::text(ItemKind::User, "fresh")]) + .unwrap(); + activity + .execute( + ExecutionOrigin::Prompt, + drive_prompt( + &session_id, + &mut driver, + &TestTurnControl::new(false), + 0, + None, + ), + |reason| Some(reason.clone()), + ) + .await + .unwrap(); + assert_eq!(turns.load(Ordering::Relaxed), 2); + assert_running_then_idle( + &recording.updates.lock().unwrap(), + wire::StopReason::EndTurn, + ); + } + + #[tokio::test] + async fn cancellation_before_tool_boundary_retires_turn() { + assert_boundary_cancellation_retires_turn(true).await; + } + + #[tokio::test] + async fn cancellation_within_tool_boundary_retires_turn() { + assert_boundary_cancellation_retires_turn(false).await; + } + #[tokio::test] async fn finish_error_stops_before_delivering_pending_steer() { let (mut driver, turns) = test_driver(TestOutcome::FinishError, "finish-error").await; From 7e7745ab7c42cb0040e6f962e2166abd6b19e44b Mon Sep 17 00:00:00 2001 From: daniel Date: Sat, 5 Sep 2026 12:49:23 +0100 Subject: [PATCH 5/5] fix(acp): retire cancelled driver turns before settlement --- Cargo.lock | 4 ++-- Cargo.toml | 4 ++-- src/protocols/acp.rs | 9 +++++++++ src/protocols/acp/v2.rs | 30 ++++++++++++++++++++++++++++++ 4 files changed, 43 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 883994d..4ddcc0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,7 +134,7 @@ dependencies = [ [[package]] name = "agentkit-acp" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=7604f6c2ba1eed1de83fe3d9a0058192b5623bb4#7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" +source = "git+https://github.com/danielkov/agentkit.git?rev=513e92bb00465a86e5177084f33a130e85705bad#513e92bb00465a86e5177084f33a130e85705bad" dependencies = [ "agent-client-protocol", "agentkit-core", @@ -241,7 +241,7 @@ dependencies = [ [[package]] name = "agentkit-loop" version = "0.10.11" -source = "git+https://github.com/danielkov/agentkit.git?rev=7604f6c2ba1eed1de83fe3d9a0058192b5623bb4#7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" +source = "git+https://github.com/danielkov/agentkit.git?rev=513e92bb00465a86e5177084f33a130e85705bad#513e92bb00465a86e5177084f33a130e85705bad" dependencies = [ "agentkit-core", "agentkit-task-manager", diff --git a/Cargo.toml b/Cargo.toml index 3862410..07c44af 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,8 +90,8 @@ tempfile = "=3.27.0" tokio = { version = "=1.53.1", features = ["test-util"] } [patch.crates-io] -agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" } -agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "7604f6c2ba1eed1de83fe3d9a0058192b5623bb4" } +agentkit-loop = { git = "https://github.com/danielkov/agentkit.git", rev = "513e92bb00465a86e5177084f33a130e85705bad" } +agentkit-acp = { git = "https://github.com/danielkov/agentkit.git", rev = "513e92bb00465a86e5177084f33a130e85705bad" } agent-client-protocol = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } agent-client-protocol-http = { git = "https://github.com/danielkov/rust-sdk.git", rev = "2f039993d1d6ed8da35b38c31f54a7cbb7338c70" } diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index 5573056..17707c5 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -2134,6 +2134,8 @@ async fn drive_domain_until_pause( answer_prompt: bool, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, ) -> Result { + let cancellation = integration.cancellation_handle(session_id)?; + let generation = cancellation.generation(); loop { let step = match driver.next().await { Ok(step) => step, @@ -2142,6 +2144,13 @@ async fn drive_domain_until_pause( } Err(error) => return Err(record_acp_loop_failure(session_id, &error)), }; + if cancellation.is_cancelled_since(generation) { + driver + .retire_interrupted_turn() + .await + .map_err(|error| record_acp_loop_failure(session_id, &error))?; + return Ok(FinishReason::Cancelled); + } match step { LoopStep::Finished(result) => { if result.finish_reason == FinishReason::ToolCall { diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 39e15b8..748654d 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -1367,6 +1367,36 @@ async fn drive_prompt( cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, ) -> Result +where + S: ModelSession + Send + 'static, + C: TurnControl, +{ + let result = drive_prompt_inner( + session_id, + driver, + control, + cancellation_generation, + structured, + ) + .await; + if matches!(result, Ok(FinishReason::Cancelled)) { + // A cooperative interrupt is still a live logical turn. Retire it + // without another `next`, which could execute cancelled model work. + driver + .retire_interrupted_turn() + .await + .map_err(|error| AcpRuntimeError::Loop(error.to_string()))?; + } + result +} + +async fn drive_prompt_inner( + session_id: &wire::SessionId, + driver: &mut LoopDriver, + control: &C, + cancellation_generation: u64, + structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, +) -> Result where S: ModelSession + Send + 'static, C: TurnControl,