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 f525899..17707c5 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::{ @@ -46,6 +46,7 @@ use tokio::{ time::timeout, }; +mod activity; mod skill_catalog; pub mod v2; @@ -951,10 +952,29 @@ struct SessionBindingGuard { session_id: agentkit_acp::SessionId, } +/// v1 retains standard prompt responses; autonomous intervals use the extension. +fn legacy_activity( + session_id: agentkit_acp::SessionId, + 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, + })?; + } + Ok(()) + }) +} + #[derive(Clone)] struct ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, + activity: activity::SessionActivity, session_id: agentkit_acp::SessionId, } @@ -963,8 +983,10 @@ impl ResponseInterruptionNoticeObserver { inner: AcpIntegration, client: AcpClientHandle, session_id: agentkit_acp::SessionId, + activity: activity::SessionActivity, ) -> Self { Self { + activity, inner, client, session_id, @@ -974,6 +996,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(), @@ -1257,8 +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(); - 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)); @@ -1281,6 +1311,7 @@ impl Server { self.integration.as_ref().clone(), client, session_id.clone(), + activity.clone(), ); let context = AcpDriverContext { cwd, @@ -1333,7 +1364,7 @@ impl Server { adapter: driver.adapter, catalog, commands: rx, - turn_states, + activity, mcp_events, }; let token = self.registry.next_token(); @@ -1574,7 +1605,7 @@ struct SessionActor { adapter: SelectableAdapter, catalog: Vec, commands: mpsc::Receiver, - turn_states: mpsc::UnboundedSender, + activity: activity::SessionActivity, mcp_events: crate::tools::mcp::McpSubscription, } @@ -1592,11 +1623,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 @@ -1604,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, @@ -1614,8 +1644,11 @@ async fn session_actor(actor: SessionActor) { &tasks, &background_jobs, structured_completion, - ).await; - 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. @@ -1665,8 +1698,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 +1717,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}"); @@ -1980,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?; } @@ -2001,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, @@ -2020,62 +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, - turn_states: &mpsc::UnboundedSender, - next_turn_id: &mut u64, + activity: &activity::SessionActivity, ) -> 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, - }); - 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, - }); - 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, @@ -2085,6 +2088,7 @@ async fn drive_autonomous( Ok(()) } +#[cfg(test)] async fn drive_until_pause( session_id: &agentkit_acp::SessionId, integration: &AcpIntegration, @@ -2092,41 +2096,84 @@ 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 { + let cancellation = integration.cancellation_handle(session_id)?; + let generation = cancellation.generation(); 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)), }; + 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 { 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)) => { @@ -2475,26 +2522,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(); @@ -2542,6 +2583,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() { @@ -3209,6 +3348,7 @@ pub(super) mod tests { integration.clone(), client, session_id.clone(), + test_activity(session_id.clone(), mpsc::unbounded_channel().0), ); let emit = |event| { observer.handle_event(ObservedEvent { @@ -3906,7 +4046,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(); } @@ -4039,6 +4179,117 @@ 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 = 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() + .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 (while sharing session activity 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, 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(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) + .await + .unwrap(); + } + let started = states.try_recv().unwrap(); + assert!(started.active); + assert_eq!(started.turn_id, 3); + assert!(states.try_recv().is_err()); + 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, None).unwrap(); + 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 +4348,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 = test_activity(acp_session_id.clone(), turn_states_tx); let driver = Agent::builder() .model(ScriptAdapter { turns: Arc::clone(&turns), @@ -4106,6 +4359,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 +4367,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 +4387,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..e7672a3 --- /dev/null +++ b/src/protocols/acp/activity.rs @@ -0,0 +1,368 @@ +use std::sync::{Arc, Mutex}; + +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 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 = 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(result) if state.state != State::Idle => { + state.state = State::Settling; + if let Some(current) = &mut state.current { + current.reason = result.finish_reason.clone(); + } + } + _ => {} + } + } + + 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), + } + } +} + +impl LoopObserver for SessionActivity { + fn handle_event(&self, event: ObservedEvent) { + self.observe(&event.event); + } +} + +/// 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::*; + 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(), + }) + } + + #[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, + ] + .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); + } +} diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 5548a46..748654d 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::{ExecutionOrigin, SessionActivity}; + use super::{ AuthenticationRequiredData, CancelBackgroundRequest, CancelBackgroundResponse, DetachComposeRequest, DetachComposeResponse, FileSearchRequest, FileSearchResponse, @@ -129,14 +131,17 @@ 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)) } } +// 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(|_| ()) @@ -322,6 +327,24 @@ impl ResponseReplacementSink { } } +/// 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] impl AcpSessionUpdateSink for ResponseReplacementSink { fn update( @@ -349,6 +372,7 @@ impl AcpSessionUpdateSink for ResponseReplacementSink { inner: AcpIntegration, sink: ResponseReplacementSink, + activity: SessionActivity, session_id: wire::SessionId, } @@ -372,10 +396,12 @@ impl ResponseReplacementObserver { inner: AcpIntegration, sink: ResponseReplacementSink, session_id: wire::SessionId, + activity: SessionActivity, ) -> Self { Self { inner, sink, + activity, session_id, } } @@ -396,6 +422,7 @@ where S: AcpSessionUpdateSink + Clone, { fn handle_event(&self, event: ObservedEvent) { + self.activity.observe(&event.event); if let AgentEvent::UsageUpdated(usage) = &event.event { let Some(update) = usage_update(usage) else { return; @@ -764,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); @@ -776,6 +804,7 @@ impl Server { self.integration.as_ref().clone(), sink.clone(), session_id.clone(), + activity.clone(), ); let context = AcpDriverContext { cwd, @@ -808,6 +837,7 @@ impl Server { handle: handle.clone(), busy: Arc::clone(&busy), binding, + activity, sink, driver: driver.driver, tasks: driver.tasks, @@ -1079,6 +1109,7 @@ struct SessionActor { busy: Arc, binding: BindingGuard, sink: ResponseReplacementSink, + activity: SessionActivity, driver: LoopDriver, tasks: TaskManagerHandle, background_jobs: BackgroundJobs, @@ -1098,6 +1129,7 @@ async fn session_actor(actor: SessionActor) handle, busy, binding, + activity, sink, mut driver, tasks, @@ -1127,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 { @@ -1161,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 { @@ -1180,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}"); } @@ -1210,10 +1242,11 @@ 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, + activity: &SessionActivity, ) -> Result<(), AcpRuntimeError> { let PromptCommand { request, @@ -1286,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 @@ -1335,7 +1366,37 @@ async fn drive_prompt( control: &C, cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, -) -> Result +) -> 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, @@ -1346,13 +1407,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) => { @@ -1362,7 +1423,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())); } @@ -1376,15 +1437,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); } @@ -1401,15 +1462,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); } @@ -1418,12 +1479,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); } @@ -1434,7 +1495,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); } @@ -1443,59 +1504,51 @@ where } } +#[allow(clippy::too_many_arguments)] async fn run_active_turn( session_id: &wire::SessionId, integration: &AcpIntegration, handle: &AcpSessionHandle, driver: &mut LoopDriver, - sink: &impl AcpSessionUpdateSink, + sink: &ResponseReplacementSink, cancellation_generation: u64, structured: Option<(&TaskManagerHandle, &BackgroundJobs)>, + activity: &SessionActivity, + origin: ExecutionOrigin, ) -> Result<(), AcpRuntimeError> { - send_state( - sink, - session_id, - wire::StateUpdate::Running(wire::RunningStateUpdate::new()), - )?; - let stop_reason = match drive_prompt( - session_id, - driver, - handle, - cancellation_generation, - structured, - ) - .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); - } - }; - if stop_reason == 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?; - } - 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)), - ) + 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( @@ -1504,7 +1557,8 @@ async fn drive_autonomous( handle: &AcpSessionHandle, busy: &AtomicBool, driver: &mut LoopDriver, - sink: &impl AcpSessionUpdateSink, + sink: &ResponseReplacementSink, + activity: &SessionActivity, ) -> Result<(), AcpRuntimeError> { if claim_prompt(busy).is_err() { return Ok(()); @@ -1521,6 +1575,8 @@ async fn drive_autonomous( sink, cancellation_generation, None, + activity, + ExecutionOrigin::Autonomous, ) .await; integration.finish_prompt(session_id); @@ -1529,26 +1585,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, @@ -2167,6 +2203,7 @@ mod tests { struct RecordingSink { updates: Arc>>, flushes: Arc, + fail_flush: Arc, } #[async_trait] @@ -2188,7 +2225,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(()) + } } } @@ -2221,10 +2262,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| { @@ -2259,6 +2302,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( @@ -2267,7 +2311,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()), @@ -2315,6 +2360,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( @@ -2323,7 +2369,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()), @@ -2369,7 +2420,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 +2456,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 +2467,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 +2492,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] @@ -2447,6 +2508,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( @@ -2455,7 +2517,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()), @@ -2490,7 +2553,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 +2570,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 +2578,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 +2592,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, @@ -2536,6 +2609,7 @@ mod tests { #[derive(Clone, Copy)] enum TestOutcome { + ToolThenContent, Content, FinishError, ProviderError, @@ -2650,6 +2724,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 { @@ -2771,6 +2869,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 @@ -2779,7 +2878,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(), @@ -2800,7 +2900,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, @@ -2855,26 +2960,50 @@ mod tests { )); assert_eq!( loop_error_stop_reason(&session_id, &LoopError::Cancelled).unwrap(), - wire::StopReason::Cancelled + FinishReason::Cancelled ); } #[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 activity = native_activity(session_id.clone(), sink.clone()); 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(), + activity.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( @@ -2905,14 +3034,15 @@ mod tests { &tasks, &background_jobs, false, + &activity, ), acknowledge, ); 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 +3094,31 @@ 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 activity = native_activity(session_id.clone(), sink.clone()); + 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(), + activity.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,25 +3130,20 @@ 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)), + &activity, + ExecutionOrigin::Prompt, ); tokio::pin!(prompt); @@ -3030,6 +3174,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 +3192,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()) @@ -3052,6 +3209,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; @@ -3103,7 +3389,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] @@ -3128,19 +3414,43 @@ 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 activity = native_activity(session_id.clone(), sink.clone()); 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(), + activity.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( @@ -3150,14 +3460,15 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); 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] @@ -3166,6 +3477,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( @@ -3175,8 +3487,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, @@ -3201,6 +3517,7 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); @@ -3208,6 +3525,22 @@ 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, + &activity, + ) + .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 +3553,41 @@ 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 activity = native_activity(session_id.clone(), sink.clone()); 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(), + activity.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(); @@ -3243,14 +3600,15 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await; 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 +3625,43 @@ 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 activity = native_activity(session_id.clone(), sink.clone()); 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(), + activity.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(); @@ -3294,30 +3674,120 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await .unwrap(); 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 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 sink = RecordingSink::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(), - 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(), + activity.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(); @@ -3330,14 +3800,15 @@ mod tests { &busy, &mut driver, &sink, + &activity, ) .await; 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);