diff --git a/Cargo.lock b/Cargo.lock index 65d999e..edfe1fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -50,6 +50,7 @@ checksum = "d5c231915b4ab578c722eca2d1bd7df4d300bfd6cac3b8e9f0d1e3ddc95b187c" dependencies = [ "anyhow", "derive_more", + "diffy", "schemars 1.2.2", "serde", "serde_json", @@ -91,7 +92,7 @@ dependencies = [ [[package]] name = "agentkit-acp" -version = "0.10.7" +version = "0.10.8" dependencies = [ "agent-client-protocol", "agentkit-core", @@ -1227,6 +1228,15 @@ version = "0.1.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" +[[package]] +name = "diffy" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10aec8f7f9393bd6a4f2762be0ceb012d3cbe2478987258cc9960de148561914" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "digest" version = "0.10.7" @@ -1397,6 +1407,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -1603,6 +1619,9 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "foldhash", +] [[package]] name = "hashlink" diff --git a/book/src/acp.md b/book/src/acp.md index 2c4de0f..02d0769 100644 --- a/book/src/acp.md +++ b/book/src/acp.md @@ -9,6 +9,50 @@ Like `agentkit-mcp`, this crate does not define a parallel protocol vocabulary. - **Protocol docs:** [agentclientprotocol.com](https://agentclientprotocol.com/protocol/v1/overview) - **Rust SDK:** [`agent-client-protocol` on crates.io](https://crates.io/crates/agent-client-protocol) +## Opt-in ACP v2 runtime + +ACP v2 support is additive and disabled by default. Enable it explicitly: + +```toml +agentkit-acp = { version = "0.10.8", features = ["protocol-v2"] } +``` + +`protocol-v2` enables the official upstream +`agent-client-protocol/unstable_protocol_v2` feature. The root API and +`agentkit_acp::wire` continue to expose stable v1 behavior. Experimental v2 +runtime APIs and official v2 wire types are isolated under +`agentkit_acp::v2` and `agentkit_acp::v2::wire`; v1 wire types are not part of +that namespace. + +Build a v2 server with `agentkit_acp::v2::AcpHeadlessRuntime`. Its factory is +called once for each `session/new` and receives a v2 session ID, an agentkit +session ID, the v2 output observer, and a cancellation handle. Install the +observer and cancellation handle on the returned agent loop in the same way as +the v1 factory. + +The v2 prompt lifecycle differs from v1: `session/prompt` acknowledges +acceptance immediately instead of waiting for the turn to finish. The runtime +then emits, in order: + +1. a `user_message` update with a generated stable message ID; +2. a `running` state update; +3. streamed agent message or thought chunks with distinct stable message IDs; +4. tool-call lifecycle updates when tools run; +5. an `idle` state update with the final stop reason. + +Each session has its own worker and loop driver. Independent sessions can make +progress concurrently, while a second prompt for a running session is rejected. +`session/cancel` interrupts only the selected session and produces an idle +`cancelled` update after loop cleanup. `session/close` cooperatively cancels +work and drops the session worker. `session/list` and `session/resume` cover +active in-memory sessions; replay is not supported. + +The initial v2 foundation routes text, reasoning, and tool lifecycle updates. +ACP v2 permission callbacks are intentionally deferred; an unsupported approval +interrupt retains the transcript and ends the prompt with the custom `_error` +stop reason rather than `refusal`. Upstream labels the v2 protocol unstable, so +opt-in callers should expect the `v2` namespace to track official SDK changes. + ## Two integration shapes `agentkit-acp` exposes the same functionality at two levels: diff --git a/crates/agentkit-acp/Cargo.toml b/crates/agentkit-acp/Cargo.toml index 6a1a018..5ee9288 100644 --- a/crates/agentkit-acp/Cargo.toml +++ b/crates/agentkit-acp/Cargo.toml @@ -4,7 +4,7 @@ homepage.workspace = true name = "agentkit-acp" readme = "README.md" repository.workspace = true -version = "0.10.7" +version = "0.10.8" edition.workspace = true license.workspace = true rust-version.workspace = true @@ -13,9 +13,10 @@ rust-version.workspace = true default = ["stdio"] stdio = [] unstable-acp = ["agent-client-protocol/unstable"] +protocol-v2 = ["agent-client-protocol/unstable_protocol_v2"] [dependencies] -agent-client-protocol = "2.0.0" +agent-client-protocol = "=2.0.0" agentkit-core = { version = "0.10.5", path = "../agentkit-core" } agentkit-loop = { version = "0.10.5", path = "../agentkit-loop" } agentkit-tools-core = { version = "0.10.5", path = "../agentkit-tools-core" } @@ -23,7 +24,7 @@ async-trait.workspace = true base64.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["sync", "time"] } tracing.workspace = true [dev-dependencies] diff --git a/crates/agentkit-acp/README.md b/crates/agentkit-acp/README.md index 50594a5..07a1e58 100644 --- a/crates/agentkit-acp/README.md +++ b/crates/agentkit-acp/README.md @@ -15,7 +15,36 @@ upstream SDK transport. It handles initialize, session lifecycle, prompt conversion, streaming updates, cancellation, and ACP permission requests for agentkit approval interrupts. -Run the in-memory end-to-end example with: +## Opt-in ACP v2 foundation + +The crate root, default features, and `wire` module remain ACP v1. To use the +experimental upstream ACP v2 protocol, enable the additive feature: + +```toml +agentkit-acp = { version = "0.10.8", features = ["protocol-v2"] } +``` + +The feature maps directly to the official +`agent-client-protocol/unstable_protocol_v2` feature. V2 APIs and official v2 +wire types live only under `agentkit_acp::v2` (and +`agentkit_acp::v2::wire`). + +`v2::AcpHeadlessRuntime` supports ACP v2 initialize, new/list/resume +session, prompt, cancel, session updates, and close. Listing and resume cover +the runtime's active in-memory sessions; replay is not supported. Each session +owns a worker and loop driver, so work in one session does not block request +handling for another. A prompt is accepted before model work completes, then +the runtime emits ordered `UserMessage`, `Running`, streamed output, and `Idle` +updates. User, visible-agent, and thought message IDs are distinct and stable +for the lifetime of a prompt. + +This first v2 foundation streams text, reasoning, and tool lifecycle updates. +The v1 permission bridge is not exposed through v2 wire types. Unsupported +approval interrupts retain the transcript and therefore end with the custom +`_error` stop reason rather than `Refusal`. Because upstream marks protocol v2 +unstable, all APIs in the `v2` namespace can evolve with the official SDK. + +Run the stable v1 in-memory end-to-end example with: ```sh cargo run -p openrouter-acp-trio diff --git a/crates/agentkit-acp/src/lib.rs b/crates/agentkit-acp/src/lib.rs index 5c49b2d..8f53c13 100644 --- a/crates/agentkit-acp/src/lib.rs +++ b/crates/agentkit-acp/src/lib.rs @@ -41,6 +41,10 @@ pub mod wire { pub use agent_client_protocol::schema::v1::*; } +/// Opt-in ACP protocol v2 runtime and upstream wire types. +#[cfg(feature = "protocol-v2")] +pub mod v2; + const ALLOW_ONCE_OPTION: &str = "allow_once"; const ALLOW_ALWAYS_OPTION: &str = "allow_always"; const REJECT_ONCE_OPTION: &str = "reject_once"; diff --git a/crates/agentkit-acp/src/v2.rs b/crates/agentkit-acp/src/v2.rs new file mode 100644 index 0000000..ace92cb --- /dev/null +++ b/crates/agentkit-acp/src/v2.rs @@ -0,0 +1,2368 @@ +//! Opt-in runtime foundation for the experimental ACP protocol v2. +//! +//! Enable the `protocol-v2` crate feature to use this module. The feature maps +//! directly to the official `agent-client-protocol/unstable_protocol_v2` +//! feature. Root-level APIs remain the stable ACP v1 integration. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; + +use agent_client_protocol::{Client, ConnectionTo, Handled}; +use agentkit_core::{ + CancellationController, CancellationHandle, DataRef, Delta, FilePart, FinishReason, Item, + ItemKind, MediaPart, MetadataMap, Modality, Part, PartId, PartKind, + SessionId as AgentkitSessionId, StructuredPart, TextPart, ToolCallPart, ToolOutput, + ToolResultPart, +}; +use agentkit_loop::{ + AgentEvent, LoopInterrupt, LoopObserver, LoopStep, ModelAdapter, ModelSession, ObservedEvent, +}; +use async_trait::async_trait; +use serde_json::json; +use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; + +use crate::AcpRuntimeError; + +/// Official upstream ACP v2 wire types. +/// +/// These are gated by upstream's `unstable_protocol_v2` feature and can change +/// while ACP v2 is under development. No stable v1 wire type is re-exported +/// from this namespace. +pub use agent_client_protocol::schema::ProtocolVersion; +pub use agent_client_protocol::schema::v2::*; + +/// Explicit namespace for the official upstream ACP v2 wire types. +pub mod wire { + pub use agent_client_protocol::schema::ProtocolVersion; + pub use agent_client_protocol::schema::v2::*; +} + +enum ClientMessage { + Update(Box), + Flush(oneshot::Sender<()>), +} + +#[derive(Clone)] +struct ClientHandle { + tx: mpsc::UnboundedSender, +} + +impl ClientHandle { + fn channel() -> (Self, mpsc::UnboundedReceiver) { + let (tx, rx) = mpsc::unbounded_channel(); + (Self { tx }, rx) + } + + fn update( + &self, + session_id: wire::SessionId, + update: wire::SessionUpdate, + ) -> Result<(), AcpRuntimeError> { + self.tx + .send(ClientMessage::Update(Box::new( + wire::UpdateSessionNotification::new(session_id, update), + ))) + .map_err(|_| AcpRuntimeError::ClientClosed) + } + + async fn flush(&self) -> Result<(), AcpRuntimeError> { + let (tx, rx) = oneshot::channel(); + self.tx + .send(ClientMessage::Flush(tx)) + .map_err(|_| AcpRuntimeError::ClientClosed)?; + rx.await.map_err(|_| AcpRuntimeError::ClientClosed) + } +} + +async fn drain_client_messages( + mut rx: mpsc::UnboundedReceiver, + cx: ConnectionTo, +) { + while let Some(message) = rx.recv().await { + match message { + ClientMessage::Update(notification) => { + if let Err(error) = cx.send_notification(*notification) { + tracing::debug!(%error, "failed to send ACP v2 session update"); + break; + } + } + ClientMessage::Flush(response) => { + let _ = response.send(()); + } + } + } +} + +#[derive(Clone)] +struct CurrentMessageIds { + agent: wire::MessageId, + thought: wire::MessageId, +} + +struct IntegrationSession { + acp_session_id: wire::SessionId, + client: ClientHandle, + next_message: AtomicU64, + current_messages: Mutex>, + part_kinds: Mutex>, +} + +#[derive(Default)] +struct IntegrationInner { + by_acp: HashMap>, + by_agentkit: HashMap, +} + +/// Routes agentkit loop output to ACP v2 `session/update` notifications. +/// +/// Agent factories should install this value as their loop observer. The +/// headless runtime binds and unbinds sessions automatically. +#[derive(Clone, Default)] +pub struct AcpIntegration { + inner: Arc>, +} + +impl AcpIntegration { + fn bind( + &self, + acp_session_id: wire::SessionId, + agentkit_session_id: AgentkitSessionId, + client: ClientHandle, + ) -> Result<(), AcpRuntimeError> { + let mut inner = self + .inner + .write() + .unwrap_or_else(|error| error.into_inner()); + if inner.by_acp.contains_key(&acp_session_id) { + return Err(AcpRuntimeError::SessionAlreadyBound( + acp_session_id.to_string(), + )); + } + if inner.by_agentkit.contains_key(&agentkit_session_id) { + return Err(AcpRuntimeError::SessionAlreadyBound( + agentkit_session_id.to_string(), + )); + } + inner + .by_agentkit + .insert(agentkit_session_id, acp_session_id.clone()); + inner.by_acp.insert( + acp_session_id.clone(), + Arc::new(IntegrationSession { + acp_session_id, + client, + next_message: AtomicU64::new(1), + current_messages: Mutex::new(None), + part_kinds: Mutex::new(HashMap::new()), + }), + ); + Ok(()) + } + + fn unbind(&self, session_id: &wire::SessionId) -> Result<(), AcpRuntimeError> { + let mut inner = self + .inner + .write() + .unwrap_or_else(|error| error.into_inner()); + let session = inner + .by_acp + .remove(session_id) + .ok_or_else(|| AcpRuntimeError::SessionNotFound(session_id.to_string()))?; + inner + .by_agentkit + .retain(|_, mapped| mapped != &session.acp_session_id); + Ok(()) + } + + fn session( + &self, + session_id: &wire::SessionId, + ) -> Result, AcpRuntimeError> { + self.inner + .read() + .unwrap_or_else(|error| error.into_inner()) + .by_acp + .get(session_id) + .cloned() + .ok_or_else(|| AcpRuntimeError::SessionNotFound(session_id.to_string())) + } + + fn begin_prompt( + &self, + session_id: &wire::SessionId, + ) -> Result { + let session = self.session(session_id)?; + let sequence = session.next_message.fetch_add(1, Ordering::Relaxed); + finish_model_message(&session); + Ok(wire::MessageId::new(format!( + "{session_id}-user-{sequence}" + ))) + } + + fn finish_prompt(&self, session_id: &wire::SessionId) { + if let Ok(session) = self.session(session_id) { + finish_model_message(&session); + } + } + + fn route_event(&self, session_id: &AgentkitSessionId, event: AgentEvent) { + let session = { + let inner = self.inner.read().unwrap_or_else(|error| error.into_inner()); + let Some(acp_session_id) = inner.by_agentkit.get(session_id) else { + return; + }; + let Some(session) = inner.by_acp.get(acp_session_id) else { + return; + }; + Arc::clone(session) + }; + + match &event { + AgentEvent::TurnStarted { .. } => { + start_model_message(&session); + return; + } + AgentEvent::TurnFinished(_) => { + finish_model_message(&session); + return; + } + AgentEvent::ToolExecutionStarted(_) | AgentEvent::ToolResultReceived(_) => { + finish_model_message(&session); + } + AgentEvent::ContentDelta(_) | AgentEvent::ToolCallRequested(_) => { + let has_message = session + .current_messages + .lock() + .unwrap_or_else(|error| error.into_inner()) + .is_some(); + if !has_message { + start_model_message(&session); + } + } + _ => {} + } + + let message_ids = session + .current_messages + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone(); + let mut part_kinds = session + .part_kinds + .lock() + .unwrap_or_else(|error| error.into_inner()); + let Some(update) = event_to_update(&event, message_ids.as_ref(), &mut part_kinds) else { + return; + }; + if let Err(error) = session + .client + .update(session.acp_session_id.clone(), update) + { + tracing::debug!(%error, "failed to queue ACP v2 session update"); + } + } +} + +fn start_model_message(session: &IntegrationSession) { + let sequence = session.next_message.fetch_add(1, Ordering::Relaxed); + *session + .current_messages + .lock() + .unwrap_or_else(|error| error.into_inner()) = Some(CurrentMessageIds { + agent: wire::MessageId::new(format!("{}-agent-{sequence}", session.acp_session_id)), + thought: wire::MessageId::new(format!("{}-thought-{sequence}", session.acp_session_id)), + }); + session + .part_kinds + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clear(); +} + +fn finish_model_message(session: &IntegrationSession) { + *session + .current_messages + .lock() + .unwrap_or_else(|error| error.into_inner()) = None; + session + .part_kinds + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clear(); +} + +impl LoopObserver for AcpIntegration { + fn handle_event(&self, event: ObservedEvent) { + self.route_event(&event.session_id, event.event); + } +} + +/// Context passed to an ACP v2 agent factory for each new session. +#[derive(Clone)] +pub struct AcpAgentFactoryContext { + /// ACP v2 session id visible to the client. + pub acp_session_id: wire::SessionId, + /// Agentkit loop session id. + pub agentkit_session_id: AgentkitSessionId, + /// Current working directory. + pub cwd: PathBuf, + /// Additional workspace roots. + pub additional_directories: Vec, + /// ACP v2 output observer to install on the agent loop. + pub integration: Arc, + /// Cancellation handle to install on the agent loop. + pub cancellation: CancellationHandle, + /// Session metadata. + pub metadata: MetadataMap, +} + +/// Creates one agentkit loop driver for each ACP v2 session. +#[async_trait] +pub trait AcpAgentFactory: Send + Sync + 'static +where + M: ModelAdapter, +{ + /// Builds and starts a loop driver for a new ACP v2 session. + async fn start( + &self, + ctx: AcpAgentFactoryContext, + ) -> Result, AcpRuntimeError>; +} + +/// Headless ACP v2 runtime. +pub struct AcpHeadlessRuntime +where + M: ModelAdapter, +{ + _marker: std::marker::PhantomData, +} + +impl AcpHeadlessRuntime +where + M: ModelAdapter + Send + Sync + 'static, + M::Session: Send + 'static, +{ + /// Starts building an ACP v2 runtime. + #[must_use] + pub fn builder() -> AcpHeadlessRuntimeBuilder { + AcpHeadlessRuntimeBuilder::default() + } +} + +/// Builder for [`AcpHeadlessRuntime`]. +pub struct AcpHeadlessRuntimeBuilder +where + M: ModelAdapter, +{ + factory: Option>>, + name: String, + version: String, +} + +impl Default for AcpHeadlessRuntimeBuilder +where + M: ModelAdapter, +{ + fn default() -> Self { + Self { + factory: None, + name: "agentkit".into(), + version: env!("CARGO_PKG_VERSION").into(), + } + } +} + +struct ServeGuard { + shutdown: Option>, + task: tokio::task::JoinHandle>, +} + +impl Drop for ServeGuard { + fn drop(&mut self) { + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + } +} + +impl AcpHeadlessRuntimeBuilder +where + M: ModelAdapter + Send + Sync + 'static, + M::Session: Send + 'static, +{ + /// Sets the per-session agent factory. + #[must_use] + pub fn agent_factory(mut self, factory: impl AcpAgentFactory) -> Self { + self.factory = Some(Arc::new(factory)); + self + } + + /// Sets the implementation name reported by `initialize`. + #[must_use] + pub fn name(mut self, name: impl Into) -> Self { + self.name = name.into(); + self + } + + /// Sets the implementation version reported by `initialize`. + #[must_use] + pub fn version(mut self, version: impl Into) -> Self { + self.version = version.into(); + self + } + + /// Serves ACP v2 over stdio. + #[cfg(feature = "stdio")] + pub async fn serve_stdio(self) -> Result<(), AcpRuntimeError> { + self.serve(agent_client_protocol::Stdio::new()).await + } + + /// Serves ACP v2 over a custom upstream SDK transport. + pub async fn serve( + self, + transport: impl agent_client_protocol::ConnectTo + 'static, + ) -> Result<(), AcpRuntimeError> { + let factory = self + .factory + .ok_or(AcpRuntimeError::MissingField("agent_factory"))?; + let state = Arc::new(RuntimeState::new(factory, self.name, self.version)); + let (shutdown, mut shutdown_rx) = oneshot::channel(); + let connection = agent_client_protocol::Agent + .v2() + .name(state.name.as_str()) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::InitializeRequest, responder, _cx| { + responder.respond_with_result( + state.initialize(request).map_err(crate::sdk_error), + ) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::NewSessionRequest, responder, cx| { + let state = Arc::clone(&state); + let connection = cx.clone(); + cx.spawn(async move { + responder.respond_with_result( + state + .new_session(request, connection) + .await + .map_err(crate::sdk_error), + ) + })?; + Ok(()) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::ListSessionsRequest, responder, _cx| { + responder.respond_with_result( + state.list_sessions(request).await.map_err(crate::sdk_error), + ) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::ResumeSessionRequest, responder, _cx| { + responder.respond_with_result( + state + .resume_session(request) + .await + .map_err(crate::sdk_error), + ) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::PromptRequest, responder, cx| { + let state = Arc::clone(&state); + cx.spawn(async move { + match state.prompt(request).await { + Ok(start) => { + responder.respond(wire::PromptResponse::new())?; + let _ = start.send(()); + Ok(()) + } + Err(error) => { + responder.respond_with_result(Err(crate::sdk_error(error))) + } + } + })?; + Ok(()) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_notification( + { + let state = Arc::clone(&state); + async move |notification: wire::CancelSessionNotification, _cx| { + state.cancel(notification).await.map_err(crate::sdk_error)?; + Ok(Handled::Yes) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .on_receive_request( + { + let state = Arc::clone(&state); + async move |request: wire::CloseSessionRequest, responder, _cx| { + responder.respond_with_result( + state.close(request).await.map_err(crate::sdk_error), + ) + } + }, + agent_client_protocol::on_receive_request!(), + ) + .connect_to(transport); + let task = tokio::spawn(async move { + tokio::pin!(connection); + let result = tokio::select! { + result = &mut connection => { + result.map_err(|error| AcpRuntimeError::Sdk(error.to_string())) + } + _ = &mut shutdown_rx => Ok(()), + }; + state.shutdown().await; + result + }); + let mut guard = ServeGuard { + shutdown: Some(shutdown), + task, + }; + let result = (&mut guard.task) + .await + .map_err(|error| AcpRuntimeError::Sdk(error.to_string())); + guard.shutdown.take(); + result? + } +} + +struct SessionEntry { + commands: mpsc::UnboundedSender, + cancellation: CancellationController, + info: wire::SessionInfo, + busy: Arc, + closed: AtomicBool, + lifecycle: Mutex<()>, + task: Mutex>>, + drain_task: Mutex>>, +} + +enum SessionCommand { + Prompt { + request: wire::PromptRequest, + items: Vec, + cancellation_generation: u64, + response: oneshot::Sender, AcpRuntimeError>>, + }, + Shutdown, +} + +struct RuntimeState +where + M: ModelAdapter, +{ + factory: Arc>, + integration: Arc, + sessions: AsyncMutex>>, + next_session: AtomicU64, + name: String, + version: String, +} + +impl RuntimeState +where + M: ModelAdapter + Send + Sync + 'static, + M::Session: Send + 'static, +{ + fn new(factory: Arc>, name: String, version: String) -> Self { + Self { + factory, + integration: Arc::new(AcpIntegration::default()), + sessions: AsyncMutex::new(HashMap::new()), + next_session: AtomicU64::new(1), + name, + version, + } + } + + fn initialize( + &self, + request: wire::InitializeRequest, + ) -> Result { + if request.protocol_version != wire::ProtocolVersion::V2 { + return Err(AcpRuntimeError::Unsupported( + "ACP v2 runtime requires protocol version 2".into(), + )); + } + Ok(wire::InitializeResponse::new( + wire::ProtocolVersion::V2, + wire::Implementation::new(self.name.clone(), self.version.clone()), + ) + .capabilities(headless_capabilities())) + } + + async fn new_session( + self: &Arc, + request: wire::NewSessionRequest, + cx: ConnectionTo, + ) -> Result { + let sequence = self.next_session.fetch_add(1, Ordering::Relaxed); + let acp_session_id = wire::SessionId::new(format!("session-{sequence}")); + let agentkit_session_id = AgentkitSessionId::new(acp_session_id.to_string()); + let cancellation = CancellationController::new(); + let (client, client_messages) = ClientHandle::channel(); + let info = wire::SessionInfo::new(acp_session_id.clone(), request.cwd.clone()) + .additional_directories(request.additional_directories.clone()); + + let mut metadata = MetadataMap::new(); + metadata.insert("acp.protocol_version".into(), json!(2)); + metadata.insert("acp.cwd".into(), json!(request.cwd)); + metadata.insert( + "acp.additional_directories".into(), + json!(request.additional_directories), + ); + + self.integration.bind( + acp_session_id.clone(), + agentkit_session_id.clone(), + client.clone(), + )?; + let drain_task = tokio::spawn(drain_client_messages(client_messages, cx)); + let ctx = AcpAgentFactoryContext { + acp_session_id: acp_session_id.clone(), + agentkit_session_id, + cwd: request.cwd.into_inner(), + additional_directories: request + .additional_directories + .into_iter() + .map(wire::AbsolutePath::into_inner) + .collect(), + integration: Arc::clone(&self.integration), + cancellation: cancellation.handle(), + metadata, + }; + let driver = match self.factory.start(ctx).await { + Ok(driver) => driver, + Err(error) => { + let _ = self.integration.unbind(&acp_session_id); + drain_task.abort(); + let _ = drain_task.await; + return Err(error); + } + }; + + let (commands, rx) = mpsc::unbounded_channel(); + let busy = Arc::new(AtomicBool::new(false)); + let worker_busy = Arc::clone(&busy); + let integration = Arc::clone(&self.integration); + let worker_session_id = acp_session_id.clone(); + let worker_cancellation = cancellation.handle(); + let task = tokio::spawn(async move { + session_worker( + worker_session_id, + driver, + client, + integration, + worker_cancellation, + worker_busy, + rx, + ) + .await; + }); + let entry = Arc::new(SessionEntry { + commands, + cancellation, + info, + busy, + closed: AtomicBool::new(false), + lifecycle: Mutex::new(()), + task: Mutex::new(Some(task)), + drain_task: Mutex::new(Some(drain_task)), + }); + self.sessions + .lock() + .await + .insert(acp_session_id.clone(), entry); + Ok(wire::NewSessionResponse::new(acp_session_id)) + } + + async fn list_sessions( + &self, + request: wire::ListSessionsRequest, + ) -> Result { + if request.cursor.is_some() { + return Err(AcpRuntimeError::Unsupported( + "ACP v2 session list cursors are not supported".into(), + )); + } + let sessions = self.sessions.lock().await; + let mut infos = sessions + .values() + .filter(|entry| { + !entry.closed.load(Ordering::Acquire) + && request + .cwd + .as_ref() + .is_none_or(|cwd| cwd == &entry.info.cwd) + }) + .map(|entry| entry.info.clone()) + .collect::>(); + infos.sort_by(|left, right| { + left.session_id + .to_string() + .cmp(&right.session_id.to_string()) + }); + Ok(wire::ListSessionsResponse::new(infos)) + } + + async fn resume_session( + &self, + request: wire::ResumeSessionRequest, + ) -> Result { + if request.replay_from.is_some() { + return Err(AcpRuntimeError::Unsupported( + "ACP v2 session replay is not supported".into(), + )); + } + let entry = self + .sessions + .lock() + .await + .get(&request.session_id) + .cloned() + .ok_or_else(|| AcpRuntimeError::SessionNotFound(request.session_id.to_string()))?; + if entry.closed.load(Ordering::Acquire) || entry.info.cwd != request.cwd { + return Err(AcpRuntimeError::SessionNotFound( + request.session_id.to_string(), + )); + } + if entry.info.additional_directories != request.additional_directories { + return Err(AcpRuntimeError::Unsupported( + "changing ACP v2 session directories on resume is not supported".into(), + )); + } + Ok(wire::ResumeSessionResponse::new()) + } + + async fn prompt( + &self, + request: wire::PromptRequest, + ) -> Result, AcpRuntimeError> { + let items = prompt_to_items(&request)?; + let entry = self + .sessions + .lock() + .await + .get(&request.session_id) + .cloned() + .ok_or_else(|| AcpRuntimeError::SessionNotFound(request.session_id.to_string()))?; + let (tx, rx) = oneshot::channel(); + { + let _lifecycle = entry + .lifecycle + .lock() + .unwrap_or_else(|error| error.into_inner()); + if entry.closed.load(Ordering::Acquire) { + return Err(AcpRuntimeError::SessionNotFound( + request.session_id.to_string(), + )); + } + if entry + .busy + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return Err(AcpRuntimeError::Unsupported( + "session is already running a prompt".into(), + )); + } + let cancellation_generation = entry.cancellation.handle().generation(); + if entry + .commands + .send(SessionCommand::Prompt { + request, + items, + cancellation_generation, + response: tx, + }) + .is_err() + { + entry.busy.store(false, Ordering::Release); + return Err(AcpRuntimeError::ClientClosed); + } + } + rx.await.map_err(|_| AcpRuntimeError::ClientClosed)? + } + + async fn cancel( + &self, + notification: wire::CancelSessionNotification, + ) -> Result<(), AcpRuntimeError> { + let entry = self + .sessions + .lock() + .await + .get(¬ification.session_id) + .cloned(); + if let Some(entry) = entry { + let _lifecycle = entry + .lifecycle + .lock() + .unwrap_or_else(|error| error.into_inner()); + if !entry.closed.load(Ordering::Acquire) && entry.busy.load(Ordering::Acquire) { + entry.cancellation.interrupt(); + } + } + Ok(()) + } + + async fn close( + &self, + request: wire::CloseSessionRequest, + ) -> Result { + let entry = self + .sessions + .lock() + .await + .remove(&request.session_id) + .ok_or_else(|| AcpRuntimeError::SessionNotFound(request.session_id.to_string()))?; + stop_session(Arc::clone(&entry)).await; + self.integration.unbind(&request.session_id)?; + stop_client(entry).await; + Ok(wire::CloseSessionResponse::new()) + } + + async fn shutdown(&self) { + let sessions = { + let mut sessions = self.sessions.lock().await; + sessions.drain().collect::>() + }; + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); + let mut session_tasks = Vec::with_capacity(sessions.len()); + for (session_id, entry) in &sessions { + signal_session_stop(entry); + if let Some(task) = take_task(&entry.task) { + session_tasks.push(task); + } + let _ = self.integration.unbind(session_id); + } + join_tasks_until(deadline, session_tasks).await; + + let drain_tasks = sessions + .iter() + .filter_map(|(_, entry)| take_task(&entry.drain_task)) + .collect(); + drop(sessions); + join_tasks_until(deadline, drain_tasks).await; + } +} + +fn signal_session_stop(entry: &Arc) { + let _lifecycle = entry + .lifecycle + .lock() + .unwrap_or_else(|error| error.into_inner()); + entry.closed.store(true, Ordering::Release); + entry.cancellation.interrupt(); + let _ = entry.commands.send(SessionCommand::Shutdown); +} + +fn take_task( + task: &Mutex>>, +) -> Option> { + task.lock() + .unwrap_or_else(|error| error.into_inner()) + .take() +} + +async fn join_tasks_until( + deadline: tokio::time::Instant, + mut tasks: Vec>, +) { + if tokio::time::timeout_at(deadline, async { + for task in &mut tasks { + let _ = task.await; + } + }) + .await + .is_err() + { + for task in tasks { + task.abort(); + } + } +} + +async fn stop_session(entry: Arc) { + signal_session_stop(&entry); + if let Some(task) = take_task(&entry.task) { + join_tasks_until( + tokio::time::Instant::now() + std::time::Duration::from_secs(2), + vec![task], + ) + .await; + } +} + +async fn stop_client(entry: Arc) { + if let Some(task) = take_task(&entry.drain_task) { + let _ = task.await; + } +} + +async fn session_worker( + session_id: wire::SessionId, + mut driver: agentkit_loop::LoopDriver, + client: ClientHandle, + integration: Arc, + cancellation: CancellationHandle, + busy: Arc, + mut commands: mpsc::UnboundedReceiver, +) where + S: ModelSession + Send + 'static, +{ + while let Some(command) = commands.recv().await { + let SessionCommand::Prompt { + request, + items, + cancellation_generation, + response, + } = command + else { + break; + }; + if let Err(error) = driver + .submit_input(items) + .map_err(|error| AcpRuntimeError::Loop(error.to_string())) + { + busy.store(false, Ordering::Release); + let _ = response.send(Err(error)); + continue; + } + let user_message_id = match integration.begin_prompt(&session_id) { + Ok(message_id) => message_id, + Err(error) => { + busy.store(false, Ordering::Release); + let _ = response.send(Err(error)); + continue; + } + }; + let (start_tx, start_rx) = oneshot::channel(); + if response.send(Ok(start_tx)).is_err() || start_rx.await.is_err() { + integration.finish_prompt(&session_id); + busy.store(false, Ordering::Release); + continue; + } + + if client + .update( + session_id.clone(), + wire::SessionUpdate::UserMessage( + wire::UserMessage::new(user_message_id).content(request.prompt), + ), + ) + .and_then(|()| { + client.update( + session_id.clone(), + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running( + wire::RunningStateUpdate::new(), + )), + ) + }) + .is_err() + { + integration.finish_prompt(&session_id); + busy.store(false, Ordering::Release); + continue; + } + + let stop_reason = drive_prompt(&mut driver, &cancellation, cancellation_generation).await; + if let Err(error) = client.flush().await { + tracing::debug!(%error, "failed to flush ACP v2 output"); + } + integration.finish_prompt(&session_id); + busy.store(false, Ordering::Release); + let _ = client.update( + session_id.clone(), + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle( + wire::IdleStateUpdate::new().stop_reason(stop_reason), + )), + ); + } +} + +async fn drive_prompt( + driver: &mut agentkit_loop::LoopDriver, + cancellation: &CancellationHandle, + generation: u64, +) -> wire::StopReason +where + S: ModelSession + Send + 'static, +{ + loop { + // Cancellation is installed on the driver and its model/tool work. Keep + // polling the driver so it can close interrupted tool calls and leave a + // resumable transcript before the session becomes idle. + match driver.next().await { + Ok(LoopStep::Finished(result)) => { + if result.finish_reason == FinishReason::ToolCall { + continue; + } + return if cancellation.is_cancelled_since(generation) { + wire::StopReason::Cancelled + } else { + finish_reason_to_stop_reason(&result.finish_reason) + }; + } + Ok(LoopStep::Interrupt(LoopInterrupt::AwaitingInput(_))) => { + return if cancellation.is_cancelled_since(generation) { + wire::StopReason::Cancelled + } else { + wire::StopReason::EndTurn + }; + } + Ok(LoopStep::Interrupt(LoopInterrupt::AfterToolResult(_))) => continue, + Ok(LoopStep::Interrupt(LoopInterrupt::ApprovalRequest(_))) => { + if let Err(error) = driver.cancel_pending_approvals().await { + tracing::debug!(%error, "failed to cancel unsupported ACP v2 approval"); + } + return if cancellation.is_cancelled_since(generation) { + wire::StopReason::Cancelled + } else { + error_stop_reason() + }; + } + Err(error) => { + tracing::debug!(%error, "ACP v2 agent loop failed"); + return if cancellation.is_cancelled_since(generation) { + wire::StopReason::Cancelled + } else { + error_stop_reason() + }; + } + } + } +} + +fn prompt_to_items(request: &wire::PromptRequest) -> Result, AcpRuntimeError> { + let mut user_parts = Vec::new(); + let mut context_items = Vec::new(); + + for block in &request.prompt { + match block { + wire::ContentBlock::Text(text) => { + user_parts.push(Part::Text(TextPart::new(text.text.clone()))); + } + wire::ContentBlock::Image(image) => { + let mime_type = image.mime_type.as_ref(); + user_parts.push(Part::media( + Modality::Image, + mime_type, + crate::data_url_ref(mime_type, &image.data), + )); + } + wire::ContentBlock::Audio(audio) => { + let mime_type = audio.mime_type.as_ref(); + user_parts.push(Part::media( + Modality::Audio, + mime_type, + crate::data_url_ref(mime_type, &audio.data), + )); + } + wire::ContentBlock::ResourceLink(link) => { + context_items.push(resource_link_item(link)); + } + wire::ContentBlock::Resource(resource) => { + context_items.push(resource_item(resource)?); + } + wire::ContentBlock::Other(other) => { + return Err(AcpRuntimeError::UnsupportedContent(format!( + "unknown ACP v2 content block: {}", + other.type_ + ))); + } + _ => { + return Err(AcpRuntimeError::UnsupportedContent( + "unknown ACP v2 content block".into(), + )); + } + } + } + + let mut items = context_items; + if !user_parts.is_empty() { + items.push(Item::new(ItemKind::User, user_parts)); + } else if !items.is_empty() { + items.push(Item::text(ItemKind::User, "Use the provided context.")); + } + Ok(items) +} + +fn resource_link_item(link: &wire::ResourceLink) -> Item { + let mut metadata = MetadataMap::new(); + metadata.insert("acp.resource.uri".into(), json!(link.uri)); + metadata.insert("acp.resource.name".into(), json!(link.name)); + if let Some(description) = &link.description { + metadata.insert("acp.resource.description".into(), json!(description)); + } + if let Some(mime_type) = &link.mime_type { + metadata.insert("acp.resource.mime_type".into(), json!(mime_type)); + } + Item::new( + ItemKind::Context, + vec![Part::file(DataRef::uri(link.uri.clone()))], + ) + .with_metadata(metadata) +} + +fn resource_item(resource: &wire::EmbeddedResource) -> Result { + match &resource.resource { + wire::EmbeddedResourceResource::TextResourceContents(text) => { + let mut metadata = MetadataMap::new(); + metadata.insert("acp.resource.uri".into(), json!(text.uri)); + if let Some(mime_type) = &text.mime_type { + metadata.insert("acp.resource.mime_type".into(), json!(mime_type)); + } + Ok(Item::text(ItemKind::Context, text.text.clone()).with_metadata(metadata)) + } + wire::EmbeddedResourceResource::BlobResourceContents(blob) => { + let mime_type = blob + .mime_type + .as_ref() + .map(|mime_type| mime_type.as_ref()) + .unwrap_or("application/octet-stream"); + let mut metadata = MetadataMap::new(); + metadata.insert("acp.resource.uri".into(), json!(blob.uri)); + metadata.insert("acp.resource.mime_type".into(), json!(mime_type)); + Ok(Item::new( + ItemKind::Context, + vec![Part::media( + Modality::Binary, + mime_type, + crate::data_url_ref(mime_type, &blob.blob), + )], + ) + .with_metadata(metadata)) + } + _ => Err(AcpRuntimeError::UnsupportedContent( + "unknown ACP v2 embedded resource".into(), + )), + } +} + +fn event_to_update( + event: &AgentEvent, + message_ids: Option<&CurrentMessageIds>, + part_kinds: &mut HashMap, +) -> Option { + match event { + AgentEvent::ContentDelta(delta) => delta_to_update(delta, message_ids, part_kinds), + AgentEvent::ToolCallRequested(call) => { + Some(wire::SessionUpdate::ToolCallUpdate(tool_call_update(call))) + } + AgentEvent::ToolExecutionStarted(call) => Some(wire::SessionUpdate::ToolCallUpdate( + tool_status_update(&call.id, wire::ToolCallStatus::InProgress), + )), + AgentEvent::ToolExecutionProgress(result) => Some(wire::SessionUpdate::ToolCallUpdate( + tool_result_update(result, wire::ToolCallStatus::InProgress), + )), + AgentEvent::ToolResultReceived(result) => { + Some(wire::SessionUpdate::ToolCallUpdate(tool_result_update( + result, + if result.is_error { + wire::ToolCallStatus::Failed + } else { + wire::ToolCallStatus::Completed + }, + ))) + } + AgentEvent::Warning { message } => { + tracing::warn!(%message, "agentkit warning while routing ACP v2 event"); + None + } + AgentEvent::RunFailed { message } => { + tracing::debug!(%message, "agentkit run failed while routing ACP v2 event"); + None + } + _ => None, + } +} + +fn delta_to_update( + delta: &Delta, + message_ids: Option<&CurrentMessageIds>, + part_kinds: &mut HashMap, +) -> Option { + match delta { + Delta::BeginPart { part_id, kind } => { + part_kinds.insert(part_id.clone(), *kind); + None + } + Delta::AppendText { part_id, chunk } => { + let message_ids = message_ids?; + let content = wire::ContentBlock::Text(wire::TextContent::new(chunk.clone())); + match part_kinds.get(part_id) { + Some(PartKind::Reasoning) => Some(wire::SessionUpdate::AgentThoughtChunk( + wire::ContentChunk::new(content, message_ids.thought.clone()), + )), + Some(PartKind::Text) | None => Some(wire::SessionUpdate::AgentMessageChunk( + wire::ContentChunk::new(content, message_ids.agent.clone()), + )), + Some(_) => None, + } + } + Delta::CommitPart { .. } + | Delta::AppendBytes { .. } + | Delta::ReplaceStructured { .. } + | Delta::SetMetadata { .. } => None, + } +} + +fn tool_call_update(call: &ToolCallPart) -> wire::ToolCallUpdate { + wire::ToolCallUpdate::new(call.id.to_string()) + .title(call.name.clone()) + .status(wire::ToolCallStatus::Pending) + .raw_input(call.input.clone()) +} + +fn tool_status_update( + call_id: &agentkit_core::ToolCallId, + status: wire::ToolCallStatus, +) -> wire::ToolCallUpdate { + wire::ToolCallUpdate::new(call_id.to_string()).status(status) +} + +fn tool_result_update( + result: &ToolResultPart, + status: wire::ToolCallStatus, +) -> wire::ToolCallUpdate { + wire::ToolCallUpdate::new(result.call_id.to_string()) + .status(status) + .raw_output(crate::tool_output_raw(&result.output)) + .content(tool_output_content(&result.output)) +} + +fn tool_output_content(output: &ToolOutput) -> Option> { + let content = match output { + ToolOutput::Text(text) => vec![text_to_tool_content(text.clone())], + ToolOutput::Structured(value) => vec![text_to_tool_content(value.to_string())], + ToolOutput::Parts(parts) => parts.iter().filter_map(part_to_tool_content).collect(), + ToolOutput::Files(files) => files.iter().map(file_to_tool_content).collect(), + }; + (!content.is_empty()).then_some(content) +} + +fn part_to_tool_content(part: &Part) -> Option { + match part { + Part::Text(text) => Some(text_to_tool_content(text.text.clone())), + Part::Structured(value) => Some(structured_to_tool_content(value)), + Part::Media(media) => Some(media_to_tool_content(media)), + Part::File(file) => Some(file_to_tool_content(file)), + Part::Reasoning(reasoning) => reasoning + .summary + .as_ref() + .map(|summary| text_to_tool_content(summary.clone())), + Part::Custom(custom) => Some(text_to_tool_content( + custom + .value + .as_ref() + .map(ToString::to_string) + .or_else(|| custom.data.as_ref().map(crate::data_ref_payload)) + .unwrap_or_else(|| custom.kind.clone()), + )), + Part::ToolCall(_) | Part::ToolResult(_) => None, + } +} + +fn text_to_tool_content(text: String) -> wire::ToolCallContent { + wire::ToolCallContent::Content(Box::new(wire::Content::new(wire::ContentBlock::Text( + wire::TextContent::new(text), + )))) +} + +fn structured_to_tool_content(part: &StructuredPart) -> wire::ToolCallContent { + text_to_tool_content(part.value.to_string()) +} + +fn media_to_tool_content(media: &MediaPart) -> wire::ToolCallContent { + match media.modality { + Modality::Image + if matches!(media.data, DataRef::InlineText(_) | DataRef::InlineBytes(_)) => + { + let mut image = wire::ImageContent::new( + crate::data_ref_base64_payload(&media.data), + media.mime_type.clone(), + ); + if let Some(uri) = crate::data_ref_uri(&media.data) { + image = image.uri(uri); + } + wire::ToolCallContent::Content(Box::new(wire::Content::new(wire::ContentBlock::Image( + image, + )))) + } + Modality::Audio + if matches!(media.data, DataRef::InlineText(_) | DataRef::InlineBytes(_)) => + { + wire::ToolCallContent::Content(Box::new(wire::Content::new(wire::ContentBlock::Audio( + wire::AudioContent::new( + crate::data_ref_base64_payload(&media.data), + media.mime_type.clone(), + ), + )))) + } + Modality::Image | Modality::Audio | Modality::Video | Modality::Binary => { + data_ref_to_resource_content(None, Some(&media.mime_type), &media.data) + } + } +} + +fn file_to_tool_content(file: &FilePart) -> wire::ToolCallContent { + data_ref_to_resource_content(file.name.as_deref(), file.mime_type.as_deref(), &file.data) +} + +fn data_ref_to_resource_content( + name: Option<&str>, + mime_type: Option<&str>, + data: &DataRef, +) -> wire::ToolCallContent { + let content = match data { + DataRef::Uri(uri) => { + let mut link = wire::ResourceLink::new(name.unwrap_or(uri), uri.clone()); + if let Some(mime_type) = mime_type { + link = link.mime_type(mime_type.to_string()); + } + wire::ContentBlock::ResourceLink(link) + } + DataRef::Handle(handle) => { + let uri = format!("artifact://{handle}"); + let link_name = name.map(str::to_owned).unwrap_or_else(|| uri.clone()); + let mut link = wire::ResourceLink::new(link_name, uri); + if let Some(mime_type) = mime_type { + link = link.mime_type(mime_type.to_string()); + } + wire::ContentBlock::ResourceLink(link) + } + DataRef::InlineText(text) if mime_type.is_none_or(|mime| mime.starts_with("text/")) => { + let mut resource = wire::TextResourceContents::new( + text.clone(), + crate::inline_resource_uri(name.unwrap_or("tool-output")), + ); + if let Some(mime_type) = mime_type { + resource = resource.mime_type(mime_type.to_string()); + } + wire::ContentBlock::Resource(wire::EmbeddedResource::new( + wire::EmbeddedResourceResource::TextResourceContents(resource), + )) + } + _ => { + let mut resource = wire::BlobResourceContents::new( + crate::data_ref_base64_payload(data), + crate::inline_resource_uri(name.unwrap_or("tool-output")), + ); + if let Some(mime_type) = mime_type { + resource = resource.mime_type(mime_type.to_string()); + } + wire::ContentBlock::Resource(wire::EmbeddedResource::new( + wire::EmbeddedResourceResource::BlobResourceContents(resource), + )) + } + }; + wire::ToolCallContent::Content(Box::new(wire::Content::new(content))) +} + +fn error_stop_reason() -> wire::StopReason { + wire::StopReason::Other("_error".into()) +} + +fn finish_reason_to_stop_reason(reason: &FinishReason) -> wire::StopReason { + match reason { + FinishReason::Completed | FinishReason::ToolCall | FinishReason::Other(_) => { + wire::StopReason::EndTurn + } + FinishReason::MaxTokens => wire::StopReason::MaxTokens, + FinishReason::Cancelled => wire::StopReason::Cancelled, + FinishReason::Blocked | FinishReason::Error => error_stop_reason(), + } +} + +fn headless_capabilities() -> wire::AgentCapabilities { + wire::AgentCapabilities::new().session( + wire::SessionCapabilities::new().prompt( + wire::PromptCapabilities::new() + .image(wire::PromptImageCapabilities::new()) + .audio(wire::PromptAudioCapabilities::new()) + .embedded_context(wire::PromptEmbeddedContextCapabilities::new()), + ), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::AtomicUsize; + use std::time::Duration; + + use agent_client_protocol::Channel; + use agentkit_core::{ItemKind, ToolCallId, ToolOutput, ToolResultPart, TurnCancellation}; + use agentkit_integration_tests::mock_model::{MockAdapter, TurnScript}; + use agentkit_loop::{ + Agent, LoopError, ModelSession, ModelTurn, ModelTurnEvent, ModelTurnResult, SessionConfig, + TurnRequest, + }; + use agentkit_tools_core::{ + Tool, ToolContext, ToolError, ToolRegistry, ToolRequest, ToolResult, ToolSpec, + }; + + #[derive(Clone)] + struct TestFactory { + adapter: A, + } + + #[async_trait] + impl AcpAgentFactory for TestFactory + where + A: ModelAdapter + Clone + Send + Sync + 'static, + A::Session: Send + 'static, + { + async fn start( + &self, + ctx: AcpAgentFactoryContext, + ) -> Result, AcpRuntimeError> { + Agent::builder() + .model(self.adapter.clone()) + .observer(ctx.integration.as_ref().clone()) + .cancellation(ctx.cancellation) + .build() + .map_err(|error| AcpRuntimeError::Loop(error.to_string()))? + .start(SessionConfig::new(ctx.agentkit_session_id).with_metadata(ctx.metadata)) + .await + .map_err(|error| AcpRuntimeError::Loop(error.to_string())) + } + } + + #[derive(Clone)] + struct CancellationAwareTool { + spec: ToolSpec, + entered: Arc, + cleaned: Arc, + } + + impl CancellationAwareTool { + fn new() -> Self { + Self { + spec: ToolSpec::new( + "blocking_tool", + "waits for turn cancellation", + json!({ "type": "object" }), + ), + entered: Arc::new(AtomicUsize::new(0)), + cleaned: Arc::new(AtomicUsize::new(0)), + } + } + + async fn wait_for_entered(&self, count: usize) { + tokio::time::timeout(Duration::from_secs(2), async { + while self.entered.load(Ordering::Acquire) < count { + tokio::task::yield_now().await; + } + }) + .await + .expect("tool did not start"); + } + } + + #[async_trait] + impl Tool for CancellationAwareTool { + fn spec(&self) -> &ToolSpec { + &self.spec + } + + async fn invoke( + &self, + request: ToolRequest, + ctx: &mut ToolContext<'_>, + ) -> Result { + self.entered.fetch_add(1, Ordering::AcqRel); + ctx.cancellation + .as_ref() + .expect("turn cancellation installed") + .cancelled() + .await; + self.cleaned.fetch_add(1, Ordering::AcqRel); + Ok(ToolResult::new(ToolResultPart::error( + request.call_id, + ToolOutput::text("cancelled"), + ))) + } + } + + #[derive(Clone)] + struct ToolTestFactory { + adapter: MockAdapter, + tool: CancellationAwareTool, + } + + #[async_trait] + impl AcpAgentFactory for ToolTestFactory { + async fn start( + &self, + ctx: AcpAgentFactoryContext, + ) -> Result< + agentkit_loop::LoopDriver<::Session>, + AcpRuntimeError, + > { + Agent::builder() + .model(self.adapter.clone()) + .add_tool_source(ToolRegistry::new().with(self.tool.clone())) + .observer(ctx.integration.as_ref().clone()) + .cancellation(ctx.cancellation) + .build() + .map_err(|error| AcpRuntimeError::Loop(error.to_string()))? + .start(SessionConfig::new(ctx.agentkit_session_id).with_metadata(ctx.metadata)) + .await + .map_err(|error| AcpRuntimeError::Loop(error.to_string())) + } + } + + fn tool_turn(call_id: &str) -> TurnScript { + let call = ToolCallPart::new(ToolCallId::new(call_id), "blocking_tool", json!({})); + TurnScript::new([ + ModelTurnEvent::ToolCall(call.clone()), + ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::ToolCall, + output_items: vec![Item::new(ItemKind::Assistant, vec![Part::ToolCall(call)])], + usage: None, + metadata: MetadataMap::new(), + }), + ]) + } + + fn streamed_text_and_tool(text: &str, call_id: &str) -> TurnScript { + let call = ToolCallPart::new(ToolCallId::new(call_id), "missing_tool", json!({})); + TurnScript::new([ + ModelTurnEvent::Delta(Delta::BeginPart { + part_id: PartId::new("part-1"), + kind: PartKind::Text, + }), + ModelTurnEvent::Delta(Delta::AppendText { + part_id: PartId::new("part-1"), + chunk: text.to_string(), + }), + ModelTurnEvent::ToolCall(call.clone()), + ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::ToolCall, + output_items: vec![Item::new( + ItemKind::Assistant, + vec![Part::text(text), Part::ToolCall(call)], + )], + usage: None, + metadata: MetadataMap::new(), + }), + ]) + } + + fn streamed_text(text: &str) -> TurnScript { + TurnScript::new([ + ModelTurnEvent::Delta(Delta::BeginPart { + part_id: PartId::new("part-1"), + kind: PartKind::Text, + }), + ModelTurnEvent::Delta(Delta::AppendText { + part_id: PartId::new("part-1"), + chunk: text.to_string(), + }), + ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::Completed, + output_items: vec![Item::text(ItemKind::Assistant, text)], + usage: None, + metadata: MetadataMap::new(), + }), + ]) + } + + fn streamed_thought_and_text(thought: &str, text: &str) -> TurnScript { + TurnScript::new([ + ModelTurnEvent::Delta(Delta::BeginPart { + part_id: PartId::new("thought-part"), + kind: PartKind::Reasoning, + }), + ModelTurnEvent::Delta(Delta::AppendText { + part_id: PartId::new("thought-part"), + chunk: thought.to_string(), + }), + ModelTurnEvent::Delta(Delta::BeginPart { + part_id: PartId::new("text-part"), + kind: PartKind::Text, + }), + ModelTurnEvent::Delta(Delta::AppendText { + part_id: PartId::new("text-part"), + chunk: text.to_string(), + }), + ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::Completed, + output_items: vec![Item::text(ItemKind::Assistant, text)], + usage: None, + metadata: MetadataMap::new(), + }), + ]) + } + + async fn wait_for_idle( + updates: &Arc>>, + session_id: &wire::SessionId, + ) { + wait_for_idle_count(updates, session_id, 1).await; + } + + async fn wait_for_idle_count( + updates: &Arc>>, + session_id: &wire::SessionId, + count: usize, + ) { + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let idle_count = updates + .lock() + .unwrap() + .iter() + .filter(|(id, update)| { + id == session_id + && matches!( + update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle(_)) + ) + }) + .count(); + if idle_count >= count { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("idle update timed out"); + } + + #[tokio::test] + async fn runtime_rotates_message_ids_per_model_turn_and_session() { + let adapter = MockAdapter::new(); + adapter.enqueue(streamed_text_and_tool("before tool", "call-1")); + adapter.enqueue(streamed_thought_and_text("first thought", "first output")); + adapter.enqueue(streamed_thought_and_text("second thought", "second output")); + let updates = Arc::new(Mutex::new(Vec::new())); + let (client_transport, agent_transport) = Channel::duplex(); + + let server = tokio::spawn({ + let factory = TestFactory { + adapter: adapter.clone(), + }; + async move { + AcpHeadlessRuntime::::builder() + .name("agentkit-v2-test") + .agent_factory(factory) + .serve(agent_transport) + .await + } + }); + + let client = agent_client_protocol::Client + .v2() + .on_receive_notification( + { + let updates = Arc::clone(&updates); + async move |notification: wire::UpdateSessionNotification, _cx| { + updates + .lock() + .unwrap() + .push((notification.session_id, notification.update)); + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, { + let updates = Arc::clone(&updates); + async move |cx| { + let initialize = cx + .send_request(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("test-client", "1"), + )) + .block_task() + .await?; + assert_eq!(initialize.protocol_version, wire::ProtocolVersion::V2); + assert_eq!(initialize.info.name, "agentkit-v2-test"); + + let cwd = std::env::current_dir() + .map_err(agent_client_protocol::Error::into_internal_error)?; + let first = cx + .send_request(wire::NewSessionRequest::new(cwd.clone())) + .block_task() + .await?; + let second = cx + .send_request(wire::NewSessionRequest::new(cwd)) + .block_task() + .await?; + assert_ne!(first.session_id, second.session_id); + + cx.send_request(wire::PromptRequest::new( + first.session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new("first"))], + )) + .block_task() + .await?; + wait_for_idle(&updates, &first.session_id).await; + + cx.send_request(wire::PromptRequest::new( + second.session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new("second"))], + )) + .block_task() + .await?; + wait_for_idle(&updates, &second.session_id).await; + + cx.send_request(wire::CloseSessionRequest::new(first.session_id)) + .block_task() + .await?; + cx.send_request(wire::CloseSessionRequest::new(second.session_id)) + .block_task() + .await?; + Ok(()) + } + }); + + tokio::time::timeout(Duration::from_secs(5), client) + .await + .expect("client timed out") + .expect("client run"); + server.abort(); + let _ = server.await; + + let updates = updates.lock().unwrap(); + for session_id in ["session-1", "session-2"] { + let session_updates = updates + .iter() + .filter(|(id, _)| id.to_string() == session_id) + .map(|(_, update)| update) + .collect::>(); + assert!(matches!( + session_updates[0], + wire::SessionUpdate::UserMessage(_) + )); + assert!(matches!( + session_updates[1], + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) + )); + let user_id = match session_updates[0] { + wire::SessionUpdate::UserMessage(message) => message.message_id.to_string(), + _ => unreachable!(), + }; + let agent_ids = session_updates + .iter() + .filter_map(|update| match update { + wire::SessionUpdate::AgentMessageChunk(chunk) => { + Some(chunk.message_id.to_string()) + } + _ => None, + }) + .collect::>(); + let thought_ids = session_updates + .iter() + .filter_map(|update| match update { + wire::SessionUpdate::AgentThoughtChunk(chunk) => { + Some(chunk.message_id.to_string()) + } + _ => None, + }) + .collect::>(); + assert_eq!(user_id, format!("{session_id}-user-1")); + if session_id == "session-1" { + assert_eq!( + agent_ids, + [ + "session-1-agent-2".to_string(), + "session-1-agent-3".to_string() + ] + ); + assert_eq!(thought_ids, ["session-1-thought-3"]); + assert_ne!(agent_ids[0], agent_ids[1]); + } else { + assert_eq!(agent_ids, ["session-2-agent-2"]); + assert_eq!(thought_ids, ["session-2-thought-2"]); + } + assert_ne!(agent_ids.last(), thought_ids.last()); + assert!(matches!( + session_updates.last(), + Some(wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle(_))) + )); + } + } + + #[test] + fn v2_tool_updates_include_visible_text_structured_parts_and_files() { + let outputs = [ + ToolOutput::text("plain text"), + ToolOutput::structured(json!({ "ok": true })), + ToolOutput::parts(vec![ + Part::text("part text"), + Part::structured(json!({ "part": true })), + ]), + ToolOutput::files(vec![ + FilePart::named("artifact.txt", DataRef::inline_text("artifact body")) + .with_mime_type("text/plain"), + FilePart::named("remote.txt", DataRef::uri("file:///tmp/remote.txt")), + ]), + ]; + let contents = outputs + .into_iter() + .map(|output| { + let result = ToolResultPart::success(ToolCallId::new("call"), output); + let update = tool_result_update(&result, wire::ToolCallStatus::Completed); + serde_json::to_value(update).expect("serialize tool update")["content"].clone() + }) + .collect::>(); + + assert_eq!(contents[0][0]["content"]["text"], "plain text"); + assert_eq!(contents[1][0]["content"]["text"], r#"{"ok":true}"#); + assert_eq!(contents[2].as_array().map(Vec::len), Some(2)); + assert_eq!(contents[2][0]["content"]["text"], "part text"); + assert_eq!(contents[3].as_array().map(Vec::len), Some(2)); + assert_eq!( + contents[3][0]["content"]["resource"]["text"], + "artifact body" + ); + assert_eq!(contents[3][1]["content"]["uri"], "file:///tmp/remote.txt"); + } + + #[tokio::test] + async fn joining_stuck_session_tasks_uses_one_shared_deadline() { + struct DropMarker(Arc); + impl Drop for DropMarker { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::AcqRel); + } + } + + let dropped = Arc::new(AtomicUsize::new(0)); + let tasks = (0..3) + .map(|_| { + let dropped = Arc::clone(&dropped); + tokio::spawn(async move { + let _marker = DropMarker(dropped); + std::future::pending::<()>().await; + }) + }) + .collect(); + tokio::task::yield_now().await; + let started = std::time::Instant::now(); + join_tasks_until( + tokio::time::Instant::now() + Duration::from_millis(50), + tasks, + ) + .await; + + assert!(started.elapsed() < Duration::from_millis(250)); + tokio::time::timeout(Duration::from_millis(250), async { + while dropped.load(Ordering::Acquire) != 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("aborted session tasks were not dropped"); + } + + #[tokio::test] + async fn aborting_serve_cleans_up_active_sessions() { + let adapter = MockAdapter::new(); + adapter.enqueue(tool_turn("serve-cancel")); + let tool = CancellationAwareTool::new(); + let (client_transport, agent_transport) = Channel::duplex(); + let server = tokio::spawn({ + let factory = ToolTestFactory { + adapter, + tool: tool.clone(), + }; + async move { + AcpHeadlessRuntime::::builder() + .agent_factory(factory) + .serve(agent_transport) + .await + } + }); + let client = tokio::spawn(agent_client_protocol::Client.v2().connect_with( + client_transport, + async move |cx| { + cx.send_request(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("test-client", "1"), + )) + .block_task() + .await?; + let cwd = std::env::current_dir() + .map_err(agent_client_protocol::Error::into_internal_error)?; + let session = cx + .send_request(wire::NewSessionRequest::new(cwd)) + .block_task() + .await?; + cx.send_request(wire::PromptRequest::new( + session.session_id, + vec![wire::ContentBlock::Text(wire::TextContent::new("run"))], + )) + .block_task() + .await?; + std::future::pending::>().await + }, + )); + + tool.wait_for_entered(1).await; + server.abort(); + let _ = server.await; + tokio::time::timeout(Duration::from_secs(2), async { + while tool.cleaned.load(Ordering::Acquire) != 1 { + tokio::task::yield_now().await; + } + }) + .await + .expect("serve drop did not clean the active tool"); + client.abort(); + let _ = client.await; + } + + #[test] + fn tool_execution_boundary_rotates_message_ids_without_a_terminal_result() { + let integration = AcpIntegration::default(); + let (client, mut messages) = ClientHandle::channel(); + let acp_id = wire::SessionId::new("acp-session"); + let agentkit_id = AgentkitSessionId::new("agentkit-session"); + integration + .bind(acp_id.clone(), agentkit_id.clone(), client) + .expect("bind session"); + integration.begin_prompt(&acp_id).expect("begin prompt"); + + for event in [ + AgentEvent::ContentDelta(Delta::BeginPart { + part_id: PartId::new("before-tool"), + kind: PartKind::Text, + }), + AgentEvent::ContentDelta(Delta::AppendText { + part_id: PartId::new("before-tool"), + chunk: "before".into(), + }), + AgentEvent::ToolExecutionStarted(ToolCallPart::new( + ToolCallId::new("call"), + "background_tool", + json!({}), + )), + AgentEvent::ContentDelta(Delta::BeginPart { + part_id: PartId::new("after-tool"), + kind: PartKind::Text, + }), + AgentEvent::ContentDelta(Delta::AppendText { + part_id: PartId::new("after-tool"), + chunk: "after".into(), + }), + ] { + integration.route_event(&agentkit_id, event); + } + + let message_ids = std::iter::from_fn(|| messages.try_recv().ok()) + .filter_map(|message| match message { + ClientMessage::Update(notification) => match notification.update { + wire::SessionUpdate::AgentMessageChunk(chunk) => Some(chunk.message_id), + _ => None, + }, + ClientMessage::Flush(_) => None, + }) + .collect::>(); + assert_eq!( + message_ids, + [ + wire::MessageId::new("acp-session-agent-2"), + wire::MessageId::new("acp-session-agent-3"), + ] + ); + } + + #[test] + fn integration_rejects_duplicate_agentkit_session_ids() { + let integration = AcpIntegration::default(); + let (first_client, _first_rx) = ClientHandle::channel(); + let (second_client, _second_rx) = ClientHandle::channel(); + let agentkit_id = AgentkitSessionId::new("shared-agentkit-session"); + let first_acp = wire::SessionId::new("first-acp-session"); + let second_acp = wire::SessionId::new("second-acp-session"); + integration + .bind(first_acp.clone(), agentkit_id.clone(), first_client) + .expect("first binding"); + let error = integration + .bind(second_acp.clone(), agentkit_id, second_client) + .expect_err("duplicate AgentKit session id must be rejected"); + assert!(matches!(error, AcpRuntimeError::SessionAlreadyBound(_))); + assert!(matches!( + integration.session(&second_acp), + Err(AcpRuntimeError::SessionNotFound(_)) + )); + integration.unbind(&first_acp).expect("unbind first"); + } + + #[test] + fn retained_transcript_failures_use_custom_error_stop_reason() { + for reason in [FinishReason::Blocked, FinishReason::Error] { + assert_eq!( + finish_reason_to_stop_reason(&reason), + wire::StopReason::Other("_error".into()) + ); + } + } + + #[tokio::test] + async fn cancel_then_prompt_and_close_cleanup_active_tools() { + let adapter = MockAdapter::new(); + adapter.enqueue(tool_turn("call-cancel")); + adapter.enqueue(streamed_text("ready again")); + adapter.enqueue(tool_turn("call-close")); + let tool = CancellationAwareTool::new(); + let updates = Arc::new(Mutex::new(Vec::new())); + let (client_transport, agent_transport) = Channel::duplex(); + + let server = tokio::spawn({ + let factory = ToolTestFactory { + adapter: adapter.clone(), + tool: tool.clone(), + }; + async move { + AcpHeadlessRuntime::::builder() + .agent_factory(factory) + .serve(agent_transport) + .await + } + }); + + let client = agent_client_protocol::Client + .v2() + .on_receive_notification( + { + let updates = Arc::clone(&updates); + async move |notification: wire::UpdateSessionNotification, _cx| { + updates + .lock() + .unwrap() + .push((notification.session_id, notification.update)); + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, { + let updates = Arc::clone(&updates); + let tool = tool.clone(); + async move |cx| { + let initialize = cx + .send_request(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("test-client", "1"), + )) + .block_task() + .await?; + assert!(initialize.capabilities.session.is_some()); + let cwd = std::env::current_dir() + .map_err(agent_client_protocol::Error::into_internal_error)?; + let first = cx + .send_request(wire::NewSessionRequest::new(cwd.clone())) + .block_task() + .await?; + + let listed = cx + .send_request(wire::ListSessionsRequest::new()) + .block_task() + .await?; + assert_eq!(listed.sessions.len(), 1); + assert_eq!(listed.sessions[0].session_id, first.session_id); + cx.send_request(wire::ResumeSessionRequest::new( + first.session_id.clone(), + cwd.clone(), + )) + .block_task() + .await?; + + cx.send_request(wire::PromptRequest::new( + first.session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new( + "start tool", + ))], + )) + .block_task() + .await?; + tool.wait_for_entered(1).await; + cx.send_notification(wire::CancelSessionNotification::new( + first.session_id.clone(), + ))?; + wait_for_idle_count(&updates, &first.session_id, 1).await; + assert_eq!(tool.cleaned.load(Ordering::Acquire), 1); + + tokio::time::timeout( + Duration::from_millis(250), + cx.send_request(wire::PromptRequest::new( + first.session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new("again"))], + )) + .block_task(), + ) + .await + .expect("session was not ready when Idle was emitted")?; + wait_for_idle_count(&updates, &first.session_id, 2).await; + + let second = cx + .send_request(wire::NewSessionRequest::new(cwd)) + .block_task() + .await?; + cx.send_request(wire::PromptRequest::new( + second.session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new( + "close tool", + ))], + )) + .block_task() + .await?; + tool.wait_for_entered(2).await; + tokio::time::timeout( + Duration::from_secs(3), + cx.send_request(wire::CloseSessionRequest::new(second.session_id.clone())) + .block_task(), + ) + .await + .expect("close did not complete after cooperative cleanup")?; + assert_eq!(tool.cleaned.load(Ordering::Acquire), 2); + cx.send_request(wire::CloseSessionRequest::new(first.session_id)) + .block_task() + .await?; + Ok(()) + } + }); + + tokio::time::timeout(Duration::from_secs(8), client) + .await + .expect("client timed out") + .expect("client run"); + server.abort(); + let _ = server.await; + + let updates = updates.lock().unwrap(); + for call_id in ["call-cancel", "call-close"] { + let statuses = updates + .iter() + .filter_map(|(_, update)| match update { + wire::SessionUpdate::ToolCallUpdate(update) + if update.tool_call_id.to_string() == call_id => + { + Some(update.status.clone()) + } + _ => None, + }) + .collect::>(); + assert!(statuses.iter().any(|status| matches!( + status, + agent_client_protocol::schema::MaybeUndefined::Value(wire::ToolCallStatus::Pending) + ))); + assert!(statuses.iter().any(|status| matches!( + status, + agent_client_protocol::schema::MaybeUndefined::Value( + wire::ToolCallStatus::InProgress + ) + ))); + assert!(statuses.iter().any(|status| matches!( + status, + agent_client_protocol::schema::MaybeUndefined::Value(wire::ToolCallStatus::Failed) + ))); + } + } + + #[derive(Clone, Default)] + struct BlockingAdapter; + + struct BlockingSession; + + struct BlockingTurn { + finished: bool, + } + + #[async_trait] + impl ModelAdapter for BlockingAdapter { + type Session = BlockingSession; + + async fn start_session(&self, _config: SessionConfig) -> Result { + Ok(BlockingSession) + } + } + + #[async_trait] + impl ModelSession for BlockingSession { + type Turn = BlockingTurn; + + async fn begin_turn( + &mut self, + _request: TurnRequest, + _cancellation: Option, + ) -> Result { + Ok(BlockingTurn { finished: false }) + } + } + + #[async_trait] + impl ModelTurn for BlockingTurn { + async fn next_event( + &mut self, + cancellation: Option, + ) -> Result, LoopError> { + if self.finished { + return Ok(None); + } + cancellation + .expect("cancellation installed") + .cancelled() + .await; + self.finished = true; + Ok(Some(ModelTurnEvent::Finished(ModelTurnResult { + model: None, + response_id: None, + finish_reason: FinishReason::Cancelled, + output_items: Vec::new(), + usage: None, + metadata: MetadataMap::new(), + }))) + } + } + + #[tokio::test] + async fn independent_prompts_are_accepted_immediately_and_cancel_separately() { + let updates = Arc::new(Mutex::new(Vec::new())); + let (client_transport, agent_transport) = Channel::duplex(); + let server = tokio::spawn(async move { + AcpHeadlessRuntime::::builder() + .agent_factory(TestFactory { + adapter: BlockingAdapter, + }) + .serve(agent_transport) + .await + }); + + let client = + agent_client_protocol::Client + .v2() + .on_receive_notification( + { + let updates = Arc::clone(&updates); + async move |notification: wire::UpdateSessionNotification, _cx| { + updates + .lock() + .unwrap() + .push((notification.session_id, notification.update)); + Ok(()) + } + }, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(client_transport, { + let updates = Arc::clone(&updates); + async move |cx| { + cx.send_request(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("test-client", "1"), + )) + .block_task() + .await?; + let cwd = std::env::current_dir() + .map_err(agent_client_protocol::Error::into_internal_error)?; + let first = cx + .send_request(wire::NewSessionRequest::new(cwd.clone())) + .block_task() + .await?; + let second = cx + .send_request(wire::NewSessionRequest::new(cwd)) + .block_task() + .await?; + + for session_id in [&first.session_id, &second.session_id] { + tokio::time::timeout( + Duration::from_millis(250), + cx.send_request(wire::PromptRequest::new( + session_id.clone(), + vec![wire::ContentBlock::Text(wire::TextContent::new("wait"))], + )) + .block_task(), + ) + .await + .expect("prompt acceptance was blocked by another session")?; + } + + cx.send_notification(wire::CancelSessionNotification::new( + first.session_id.clone(), + ))?; + wait_for_idle(&updates, &first.session_id).await; + assert!(!updates.lock().unwrap().iter().any(|(id, update)| { + id == &second.session_id + && matches!( + update, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle(_)) + ) + })); + + cx.send_notification(wire::CancelSessionNotification::new( + second.session_id.clone(), + ))?; + wait_for_idle(&updates, &second.session_id).await; + for session_id in [&first.session_id, &second.session_id] { + let stop_reason = updates.lock().unwrap().iter().find_map( + |(id, update)| match update { + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle( + idle, + )) if id == session_id => idle.stop_reason.clone(), + _ => None, + }, + ); + assert_eq!(stop_reason, Some(wire::StopReason::Cancelled)); + } + cx.send_request(wire::CloseSessionRequest::new(first.session_id)) + .block_task() + .await?; + cx.send_request(wire::CloseSessionRequest::new(second.session_id)) + .block_task() + .await?; + Ok(()) + } + }); + + tokio::time::timeout(Duration::from_secs(5), client) + .await + .expect("client timed out") + .expect("client run"); + server.abort(); + let _ = server.await; + } +} diff --git a/docs/acp.md b/docs/acp.md index 74a630b..bfa4c3e 100644 --- a/docs/acp.md +++ b/docs/acp.md @@ -1,5 +1,11 @@ # agentkit-acp design +> **Implementation status:** Root APIs and default features implement stable ACP +> v1. Version 0.10.8 also provides an opt-in ACP v2 runtime foundation under +> `agentkit_acp::v2`; enable it with `protocol-v2`. It uses only the official +> `agent-client-protocol` 2.0.0 `unstable_protocol_v2` feature. See the crate +> README or book chapter for the supported v2 lifecycle and current limits. + ## Purpose `agentkit-acp` is the Agent Client Protocol integration crate for agentkit. @@ -69,6 +75,7 @@ tracing = { workspace = true } default = ["stdio"] stdio = [] unstable-acp = ["agent-client-protocol/unstable"] +protocol-v2 = ["agent-client-protocol/unstable_protocol_v2"] ``` The umbrella crate should later add: