From 1c77932f42705f87a65a592523b81837e2c9c99a Mon Sep 17 00:00:00 2001 From: daniel Date: Tue, 13 Oct 2026 15:04:59 +0100 Subject: [PATCH 1/2] fix(a2a)!: admit messages into the interactive session Route session-bound A2A messages through existing ACP prompt and steering admission instead of transient model runs. BREAKING CHANGE: stdio v2 session-bound A2A returns an admission acknowledgment rather than execution output and rejects task tracking modes. Standalone and ACP v1 A2A execution are unchanged. --- src/main.rs | 34 +- src/protocols/a2a.rs | 269 +++++++++- src/protocols/acp.rs | 155 ++++++ src/protocols/acp/v2.rs | 788 +++++++++++++++++++++++++++++- src/protocols/acp/v2/a2a_tests.rs | 228 +++++++++ src/protocols/http.rs | 182 ++++++- 6 files changed, 1631 insertions(+), 25 deletions(-) create mode 100644 src/protocols/acp/v2/a2a_tests.rs diff --git a/src/main.rs b/src/main.rs index 652a9c15..b216f65e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1964,6 +1964,17 @@ async fn termination_signal() -> io::Result<()> { } } +fn serve_a2a_execution_mode( + no_stdio: bool, + protocol: AcpProtocolVersion, +) -> kit::protocols::http::A2aExecutionMode { + if !no_stdio && matches!(protocol, AcpProtocolVersion::V2) { + kit::protocols::http::A2aExecutionMode::SessionAdmission + } else { + kit::protocols::http::A2aExecutionMode::Standalone + } +} + async fn supervise_serve( runtime: std::sync::Arc, sessions: kit::protocols::acp::SessionRegistry, @@ -2298,13 +2309,14 @@ async fn run_cli(cli: Cli) -> Result<(), Box> { let address = a2a.unwrap_or_else(|| "127.0.0.1:0".into()); let serve_a2a = !no_a2a; let sessions = kit::protocols::acp::SessionRegistry::new(); - let http = kit::protocols::http::start_with_registry( + let http = kit::protocols::http::start_with_registry_and_mode( runtime.clone(), address, serve_a2a, remote_acp, server_credential_file.as_deref(), sessions.clone(), + serve_a2a_execution_mode(no_stdio, stdio_protocol_version), ) .await?; let bound = http.address(); @@ -2597,6 +2609,26 @@ mod tests { supervise_serve_with_trigger, validate_auth_storage, }; + #[test] + fn serve_a2a_execution_mode_preserves_no_stdio_and_v1_execution() { + use super::{AcpProtocolVersion, serve_a2a_execution_mode}; + use kit::protocols::http::A2aExecutionMode; + assert_eq!( + serve_a2a_execution_mode(false, AcpProtocolVersion::V2), + A2aExecutionMode::SessionAdmission + ); + for (no_stdio, protocol) in [ + (true, AcpProtocolVersion::V1), + (true, AcpProtocolVersion::V2), + (false, AcpProtocolVersion::V1), + ] { + assert_eq!( + serve_a2a_execution_mode(no_stdio, protocol), + A2aExecutionMode::Standalone + ); + } + } + #[test] fn usage_command_accepts_optional_provider_and_credential_overrides() { for provider in [ diff --git a/src/protocols/a2a.rs b/src/protocols/a2a.rs index 2c7e9085..d97ed369 100644 --- a/src/protocols/a2a.rs +++ b/src/protocols/a2a.rs @@ -7,7 +7,7 @@ use a2a_protocol_server::{ use a2a_protocol_types::{ agent_card::{AgentCapabilities, AgentCard, AgentInterface, AgentSkill}, error::A2aResult, - message::Part, + message::{Message, MessageId, MessageRole, Part}, security::{HttpAuthSecurityScheme, SecurityRequirement, SecurityScheme, StringList}, task::TaskState, }; @@ -17,7 +17,167 @@ use tracing::Instrument as _; use crate::runtime::Runtime; -struct KitAgent(Arc); +pub(crate) struct Dispatcher { + inner: JsonRpcDispatcher, + sessions: Option, +} + +impl Dispatcher { + pub(crate) async fn dispatch( + &self, + request: axum::http::Request, + ) -> axum::http::Response { + use axum::{ + body::{Body, to_bytes}, + http::{Method, Response, StatusCode}, + }; + let Some(sessions) = &self.sessions else { + return self.inner.dispatch(request).await.map(Body::new); + }; + if request.method() != Method::POST { + return self.inner.dispatch(request).await.map(Body::new); + } + let bytes = match to_bytes(Body::new(request.into_body()), 4 * 1024 * 1024).await { + Ok(bytes) => bytes, + Err(_) => { + let mut response = + Response::new(Body::from("A2A request body exceeds admission limit")); + *response.status_mut() = StatusCode::PAYLOAD_TOO_LARGE; + return response; + } + }; + let value = match serde_json::from_slice::(&bytes) { + Ok(value) => value, + Err(_) => { + return rpc_response( + serde_json::Value::Null, + Err((-32700, "Invalid JSON".into())), + ); + } + }; + let id = value + .get("id") + .filter(|id| id.is_null() || id.is_string() || id.is_number()) + .cloned() + .unwrap_or_default(); + rpc_response(id, admit_message(sessions, value).await) + } +} + +fn rpc_response( + id: serde_json::Value, + result: Result, +) -> axum::http::Response { + use serde_json::{Map, Value}; + let result = result.and_then(|message| { + serde_json::to_value(a2a_protocol_types::responses::SendMessageResponse::Message( + message, + )) + .map_err(|error| (-32603, error.to_string())) + }); + let mut value = Map::from_iter([ + ("jsonrpc".into(), Value::String("2.0".into())), + ("id".into(), id), + ]); + match result { + Ok(message) => { + value.insert("result".into(), message); + } + Err((code, message)) => { + value.insert( + "error".into(), + Value::Object(Map::from_iter([ + ("code".into(), Value::from(code)), + ("message".into(), Value::String(message)), + ])), + ); + } + } + let value = Value::Object(value); + let mut response = axum::http::Response::new(axum::body::Body::from(value.to_string())); + response.headers_mut().insert( + axum::http::header::CONTENT_TYPE, + axum::http::HeaderValue::from_static("application/json"), + ); + response +} + +async fn admit_message( + sessions: &crate::protocols::acp::SessionRegistry, + request: serde_json::Value, +) -> Result { + if request["jsonrpc"] != "2.0" + || !request.is_object() + || !request + .get("id") + .is_some_and(|id| id.is_null() || id.is_string() || id.is_number()) + { + return Err((-32600, "Invalid JSON-RPC request".into())); + } + let Some(method) = request["method"].as_str() else { + return Err((-32600, "JSON-RPC method must be a string".into())); + }; + if !matches!(method, "SendMessage" | "message/send") { + return Err((-32601, "Session-bound A2A supports SendMessage admission acknowledgments, not task tracking or streaming".into())); + } + let params: a2a_protocol_types::params::MessageSendParams = + serde_json::from_value(request["params"].clone()) + .map_err(|error| (-32602, error.to_string()))?; + if params.configuration.as_ref().is_some_and(|configuration| { + configuration.return_immediately == Some(true) + || configuration.task_push_notification_config.is_some() + }) { + return Err((-32602, "Session-bound A2A requires a direct acknowledgment: returnImmediately=false and no push notification configuration".into())); + } + if params.message.role != MessageRole::User || params.message.task_id.is_some() { + return Err(( + -32602, + "Admission requires a new user message without a taskId".into(), + )); + } + if params + .message + .parts + .iter() + .any(|part| part.text_content().is_none()) + { + return Err(( + -32602, + "Session admission supports text parts only; nontext parts are not accepted".into(), + )); + } + let prompt = params + .message + .parts + .iter() + .filter_map(Part::text_content) + .collect::>() + .join("\n"); + if prompt.trim().is_empty() { + return Err((-32602, "A2A request must contain a text part".into())); + } + let acknowledgment = sessions + .admit_a2a(prompt) + .await + .map_err(|error| (-32603, error))?; + Ok(Message { + id: MessageId::new(format!("{}-admission", params.message.id)), + role: MessageRole::Agent, + parts: vec![Part::text(acknowledgment)], + task_id: None, + context_id: params.message.context_id, + reference_task_ids: None, + extensions: None, + metadata: Some(serde_json::Value::Object(serde_json::Map::from_iter([( + "kind".into(), + serde_json::Value::String("admission_acknowledgment".into()), + )]))), + }) +} + +struct KitAgent { + runtime: Arc, +} impl AgentExecutor for KitAgent { fn execute<'a>( @@ -28,7 +188,7 @@ impl AgentExecutor for KitAgent { Box::pin( async move { let emit = EventEmitter::new(context, queue); - emit.status(TaskState::Working).await?; + let prompt = context .message .parts @@ -47,8 +207,9 @@ impl AgentExecutor for KitAgent { emit.status(TaskState::Failed).await?; return Ok(()); } + emit.status(TaskState::Working).await?; match self - .0 + .runtime .run_cancelled(prompt, 0, Some(context.cancellation_token.clone())) .await { @@ -109,7 +270,8 @@ pub(crate) fn dispatcher( runtime: Arc, bound: std::net::SocketAddr, authenticated: bool, -) -> Result> { + sessions: Option, +) -> Result> { let url = format!("http://{bound}"); let (security_schemes, security_requirements) = if authenticated { let name = "bearer".to_string(); @@ -132,7 +294,11 @@ pub(crate) fn dispatcher( let card = AgentCard { url: None, name: "Kit".into(), - description: "Coding agent runtime".into(), + description: if sessions.is_some() { + "Admit text into the bound interactive ACP v2 session. SendMessage returns an admission acknowledgment, not task completion or model output.".into() + } else { + "Coding agent runtime".into() + }, version: env!("CARGO_PKG_VERSION").into(), supported_interfaces: vec![AgentInterface { url, @@ -161,9 +327,96 @@ pub(crate) fn dispatcher( signatures: None, }; let handler = Arc::new( - RequestHandlerBuilder::new(KitAgent(runtime)) + RequestHandlerBuilder::new(KitAgent { runtime }) .with_agent_card(card) .build()?, ); - Ok(JsonRpcDispatcher::new(handler)) + Ok(Dispatcher { + inner: JsonRpcDispatcher::new(handler), + sessions, + }) +} + +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::unreachable, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +mod tests { + use super::*; + use serde_json::json; + + fn send_request() -> serde_json::Value { + json!({"jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": { + "message": {"messageId": "inbound", "role": "ROLE_USER", "parts": [{"text": "request"}]} + }}) + } + + #[tokio::test] + async fn a2a_direct_acknowledgment_is_a_message_not_a_completed_task() { + let registry = super::super::acp::a2a_test_support::bound_registry(Arc::new(|_prompt| { + Box::pin(async { Ok("Accepted into the bound ACP session; not completion".into()) }) + })); + let acknowledgment = admit_message(®istry, send_request()).await.unwrap(); + assert_eq!(acknowledgment.role, MessageRole::Agent); + assert_eq!(acknowledgment.task_id, None); + assert!(acknowledgment.text().unwrap().contains("not completion")); + assert_eq!( + acknowledgment.metadata.unwrap()["kind"], + "admission_acknowledgment" + ); + } + + #[tokio::test] + async fn a2a_admission_rejects_task_modes_and_missing_session_without_execution() { + let registry = crate::protocols::acp::SessionRegistry::new(); + let mut request = send_request(); + request["params"]["configuration"] = json!({"returnImmediately": true}); + assert_eq!( + admit_message(®istry, request).await.unwrap_err().0, + -32602 + ); + let mut request = send_request(); + request["method"] = json!("SendStreamingMessage"); + assert_eq!( + admit_message(®istry, request).await.unwrap_err().0, + -32601 + ); + let mut request = send_request(); + request["params"]["message"]["taskId"] = json!("existing-task"); + assert_eq!( + admit_message(®istry, request).await.unwrap_err().0, + -32602 + ); + assert!( + admit_message(®istry, send_request()) + .await + .unwrap_err() + .1 + .contains("no bound") + ); + let mut mixed = send_request(); + mixed["params"]["message"]["parts"] + .as_array_mut() + .unwrap() + .push(json!({"data": {"ignored": true}})); + assert_eq!(admit_message(®istry, mixed).await.unwrap_err().0, -32602); + let mut malformed = send_request(); + malformed["method"] = json!(42); + assert_eq!( + admit_message(®istry, malformed).await.unwrap_err().0, + -32600 + ); + let registry = super::super::acp::a2a_test_support::bound_registry(Arc::new(|_prompt| { + Box::pin(async { Err("ACP admission failed".into()) }) + })); + assert_eq!( + admit_message(®istry, send_request()).await.unwrap_err(), + (-32603, "ACP admission failed".into()) + ); + } } diff --git a/src/protocols/acp.rs b/src/protocols/acp.rs index 4ba0e85f..cb34bf2c 100644 --- a/src/protocols/acp.rs +++ b/src/protocols/acp.rs @@ -582,6 +582,40 @@ struct RegisteredSession { completed: watch::Receiver, } +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::unreachable, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +pub(super) mod a2a_test_support { + use super::*; + + pub(in crate::protocols) fn bound_registry(admit: AdmitA2a) -> SessionRegistry { + let registry = SessionRegistry::new(); + let token = registry.next_token().unwrap(); + let mut admission = registry.begin_attachment().unwrap(); + let actor = tokio::spawn(std::future::pending::<()>()); + actor.abort(); + let (_completed, completion) = watch::channel(true); + registry + .register_v2( + &mut admission, + token, + Arc::new(|| {}), + Arc::new(|| Box::pin(async {})), + actor.abort_handle(), + completion, + ) + .unwrap(); + registry.bind_a2a(token, admit); + registry + } +} + type CloseV2Session = Arc Pin + Send + 'static>> + Send + Sync + 'static>; @@ -594,11 +628,16 @@ struct RegisteredV2Session { completed: watch::Receiver, } +type AdmitA2a = Arc< + dyn Fn(String) -> Pin> + Send>> + Send + Sync, +>; + struct RegistryState { accepting: bool, permanently_closed: bool, generation: u64, pending_attachments: usize, + a2a_target: Option<(u64, AdmitA2a)>, sessions: HashMap, v2_sessions: HashMap, } @@ -632,6 +671,8 @@ impl SessionRegistryInner { // awaits: admission counts are checked before incrementing; completion // consumes a validated admission; insertion rejects existing tokens; // removal extracts ownership; gate closure cannot wrap its generation. + // A2A binding moves a prepared Arc into an empty slot and never replaces + // it; rejected callback ownership is dropped after releasing the guard. // Snapshots clone only concrete Arc/channel/handle types. Reopening // happens only after teardown and the external reset have succeeded; // cancellation or panic before then leaves admission closed. Thus unwind @@ -689,6 +730,7 @@ impl SessionRegistry { permanently_closed: false, generation: 0, pending_attachments: 0, + a2a_target: None, sessions: HashMap::new(), v2_sessions: HashMap::new(), }), @@ -705,6 +747,42 @@ impl SessionRegistry { .map_err(|_| AcpRuntimeError::Loop("ACP session registry token space exhausted".into())) } + // Only the explicitly designated stdio v2 component binds the listener. + // Retain its token after removal: a closed default must never retarget. + pub(super) fn bind_a2a(&self, token: u64, admit: AdmitA2a) { + let mut state = self.inner.lock_state(); + if state.accepting && state.v2_sessions.contains_key(&token) && state.a2a_target.is_none() { + state.a2a_target = Some((token, admit)); + return; + } + drop(state); + drop(admit); + } + + pub(crate) async fn admit_a2a(&self, prompt: String) -> Result { + // Reuse the lifecycle admission claim so shutdown also waits for accepted + // ingress. Drop rolls back the count on rejection, cancellation, or unwind. + let admission = self + .begin_attachment() + .map_err(|()| "ACP session admission is closed".to_owned())?; + let admit = { + let state = self.inner.lock_state(); + if !state.accepting || state.generation != admission.generation { + return Err("ACP session admission is closed".into()); + } + let (token, admit) = state + .a2a_target + .as_ref() + .ok_or("A2A listener has no bound interactive ACP v2 session")?; + if !state.v2_sessions.contains_key(token) { + return Err("The bound interactive ACP session is closed".into()); + } + Arc::clone(admit) + }; + // Never hold the registry lock across session callbacks or awaits. + admit(prompt).await + } + fn begin_attachment(&self) -> Result { let mut state = self.inner.lock_state(); if !state.accepting { @@ -4006,6 +4084,83 @@ pub(super) mod tests { late_actor.await.unwrap_err(); } + #[tokio::test] + async fn a2a_registry_requires_explicit_sticky_binding_and_open_admission() { + let registry = SessionRegistry::new(); + assert!(registry.admit_a2a("missing".into()).await.is_err()); + let mut admission = registry.begin_attachment().unwrap(); + let token = registry.next_token().unwrap(); + register_test_v2(®istry, &mut admission, token, Arc::new(|| {})).unwrap(); + // Merely publishing an HTTP session must not select it as the default. + assert!(registry.admit_a2a("unbound".into()).await.is_err()); + registry.bind_a2a( + token, + Arc::new(|prompt| Box::pin(async move { Ok(prompt) })), + ); + assert_eq!( + registry.admit_a2a("accepted".into()).await.unwrap(), + "accepted" + ); + let mut second_admission = registry.begin_attachment().unwrap(); + let second = registry.next_token().unwrap(); + register_test_v2(®istry, &mut second_admission, second, Arc::new(|| {})).unwrap(); + registry.remove(token); + registry.bind_a2a( + second, + Arc::new(|prompt| Box::pin(async move { Ok(prompt) })), + ); + assert!( + registry + .admit_a2a("must not retarget".into()) + .await + .unwrap_err() + .contains("closed") + ); + registry.close_gate_and_snapshot(false); + assert!( + registry + .admit_a2a("closed gate".into()) + .await + .unwrap_err() + .contains("admission is closed") + ); + registry.remove(second); + } + + #[tokio::test] + async fn a2a_ingress_cancellation_and_unwind_release_lifecycle_claim() { + let (entered, mut received) = mpsc::unbounded_channel(); + let registry = a2a_test_support::bound_registry(Arc::new(move |_prompt| { + let entered = entered.clone(); + Box::pin(async move { + entered.send(()).unwrap(); + std::future::pending().await + }) + })); + let ingress = tokio::spawn({ + let registry = registry.clone(); + async move { registry.admit_a2a("pending".into()).await } + }); + received.recv().await.unwrap(); + assert_eq!(registry.inner.lock_state().pending_attachments, 1); + ingress.abort(); + assert!(ingress.await.unwrap_err().is_cancelled()); + assert_eq!(registry.inner.lock_state().pending_attachments, 0); + registry.shutdown().await; + + let registry = a2a_test_support::bound_registry(Arc::new(|_prompt| { + Box::pin(async { panic!("admission backend panic") }) + })); + let ingress = tokio::spawn({ + let registry = registry.clone(); + async move { registry.admit_a2a("panic".into()).await } + }); + assert!(ingress.await.unwrap_err().is_panic()); + assert_eq!(registry.inner.lock_state().pending_attachments, 0); + assert!(!registry.inner.state.is_poisoned()); + registry.shutdown().await; + } + fn register_test_v2( registry: &SessionRegistry, admission: &mut SessionAdmission, diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 658a1b66..a0622743 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -44,6 +44,125 @@ use super::{ SessionRegistry, skill_catalog, terminal_auth_method_specs, }; +// The injection API deliberately requires an ACP responder and tracked receipt. +// This local transport reuses that admission boundary without a second session. +async fn inject_a2a( + integration: Arc, + request: wire::InjectSessionRequest, +) -> Result { + let (client, agent) = agent_client_protocol::Channel::duplex(); + inject_a2a_over(integration, request, client, agent).await +} + +async fn finish_a2a_injection( + integration: Arc, + request: wire::InjectSessionRequest, + responder: Responder, + cancellation: tokio_util::sync::CancellationToken, +) -> Result<(), agent_client_protocol::Error> { + // The pinned SDK allocates a FIFO sequence before awaiting its permit and + // installs Drop cleanup only afterward. Never cancel/drop this await: even + // abandoned ingress must reach the permit and advance the global queue. + let Some(reserved) = integration + .reserve_inject_request(request, responder) + .await? + else { + return Ok(()); + }; + if cancellation.is_cancelled() { + drop(reserved); + return Err(agent_client_protocol::Error::request_cancelled()); + } + let Some(acceptance) = reserved.respond_tracked()? else { + return Ok(()); + }; + // FIFO ownership is now released. Dropping pending activation is safe and + // discards its unactivated reservation; prefer cancellation on a tie. + match select( + Box::pin(cancellation.cancelled()), + Box::pin(acceptance.activate_after_response()), + ) + .await + { + Either::Left(((), pending)) => { + drop(pending); + Err(agent_client_protocol::Error::request_cancelled()) + } + Either::Right((result, _)) => result, + } +} + +// Keep the receipt paired with the SDK transport-queue acknowledgment. +async fn inject_a2a_over( + integration: Arc, + request: wire::InjectSessionRequest, + client: impl ConnectTo, + agent: impl ConnectTo, +) -> Result { + // Cancel commitment/activation on ingress drop, but let an already-started + // SDK FIFO waiter drain to its permit so later steering cannot be stranded. + let cancellation = tokio_util::sync::CancellationToken::new(); + let _cancel_on_drop = cancellation.clone().drop_guard(); + let (activated, mut activation) = mpsc::unbounded_channel(); + let host = agent_client_protocol::Agent + .v2() + .on_receive_request( + async move |request: wire::InitializeRequest, responder, _cx| { + responder.respond(wire::InitializeResponse::new( + request.protocol_version, + wire::Implementation::new("kit-a2a-admission", "0"), + )) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |request: wire::InjectSessionRequest, responder, cx| { + let integration = Arc::clone(&integration); + let activated = activated.clone(); + let cancellation = cancellation.clone(); + // Keep FIFO cleanup alive even if the SDK connection task is + // dropped. This task owns ingress bookkeeping only, never a model + // turn; cancellation fences commitment after permit acquisition. + let reservation = tokio::spawn(finish_a2a_injection( + integration, + request, + responder, + cancellation, + )); + cx.spawn(async move { + let result = reservation.await.map_err(|error| { + agent_client_protocol::util::internal_error(error.to_string()) + })?; + let _ = activated.send(result.clone()); + result + })?; + Ok(()) + }, + agent_client_protocol::on_receive_request!(), + ); + // Keep the client transport alive until the tracked receipt activates. + let client = agent_client_protocol::Client + .v2() + .connect_with(client, async move |cx| { + cx.send_request(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("kit-a2a", "0"), + )) + .block_task() + .await?; + let response = cx.send_request(request).block_task().await?; + activation.recv().await.ok_or_else(|| { + agent_client_protocol::util::internal_error("injection activation closed") + })??; + Ok(response.message_id) + }); + let (client, _) = futures_util::future::join(client, host.connect_to(agent)).await; + client +} + +#[cfg(test)] +mod a2a_tests; + const PAGE_SIZE: usize = 100; static NEXT_ERROR_MESSAGE_ID: AtomicU64 = AtomicU64::new(1); @@ -616,7 +735,25 @@ struct PendingSessionPublication { session: SessionHandle, } +// Busy is returned only before sending a command or admitting input. +enum PromptAdmissionError { + Busy, + Runtime(AcpRuntimeError), +} + +impl PromptAdmissionError { + fn into_runtime(self) -> AcpRuntimeError { + match self { + Self::Busy => { + AcpRuntimeError::Unsupported("session is already running a prompt".into()) + } + Self::Runtime(error) => error, + } + } +} + struct Server { + bind_a2a: bool, runtime: Arc, integration: Arc, registry: SessionRegistry, @@ -644,6 +781,7 @@ impl Server { fn new(runtime: Arc, registry: SessionRegistry) -> Self { Self { runtime, + bind_a2a: false, integration: Arc::new(AcpIntegration::default()), registry, sessions: Mutex::new(HashMap::new()), @@ -1091,6 +1229,23 @@ impl Server { SessionPublicationError::Commit(error) => error, }); } + if self.bind_a2a { + let server = Arc::downgrade(self); + let bound_id = session_id.clone(); + self.registry.bind_a2a( + token, + Arc::new(move |prompt| { + let server = server.clone(); + let session_id = bound_id.clone(); + Box::pin(async move { + let server = server + .upgrade() + .ok_or("Interactive ACP connection is closed")?; + server.admit_a2a(session_id, prompt).await + }) + }), + ); + } crate::events::emit(&crate::events::RuntimeEvent::SessionStarted { session_id: session_id.to_string(), }); @@ -1103,24 +1258,100 @@ impl Server { }) } + async fn admit_a2a( + &self, + session_id: wire::SessionId, + prompt: String, + ) -> Result { + let content = vec![wire::ContentBlock::Text(wire::TextContent::new(prompt))]; + // Try the real idle admission, not a racy observation of its busy flag. + let prepared = self + .try_prepare_prompt(wire::PromptRequest::new( + session_id.clone(), + content.clone(), + )) + .await; + let message_id = match prepared { + Ok((start, message_id)) => { + start + .send(()) + .map_err(|_| "Interactive ACP session closed before activation")?; + message_id + } + Err(PromptAdmissionError::Busy) => { + match inject_a2a( + Arc::clone(&self.integration), + wire::InjectSessionRequest::new( + session_id.clone(), + wire::SessionInjectMode::Steer, + content.clone(), + ), + ) + .await + { + Ok(message_id) => message_id, + // The SDK rejects this before reservation/commit. The turn may + // have ended while the local transport was initializing. Retry + // idle admission exactly once, never an already-admitted input. + Err(error) + if i32::from(error.code) + == i32::from(wire::Error::inject_no_running_turn().code) + && error + .data + .as_ref() + .and_then(|data| data.get("reason")) + .and_then(Value::as_str) + == Some("no_running_turn") => + { + let (start, message_id) = self + .try_prepare_prompt(wire::PromptRequest::new( + session_id.clone(), + content, + )) + .await + .map_err(|error| error.into_runtime().to_string())?; + start + .send(()) + .map_err(|_| "Interactive ACP session closed before activation")?; + message_id + } + Err(error) => return Err(error.to_string()), + } + } + Err(PromptAdmissionError::Runtime(error)) => return Err(error.to_string()), + }; + Ok(format!( + "Accepted into ACP session {session_id} as message {message_id}. This acknowledges admission, not completion; output is delivered through that ACP session." + )) + } + async fn prepare_prompt( &self, request: wire::PromptRequest, ) -> Result<(oneshot::Sender<()>, wire::MessageId), AcpRuntimeError> { - let (sender, busy, handle) = self.prompt_route(&request.session_id)?; + self.try_prepare_prompt(request) + .await + .map_err(PromptAdmissionError::into_runtime) + } + + async fn try_prepare_prompt( + &self, + request: wire::PromptRequest, + ) -> Result<(oneshot::Sender<()>, wire::MessageId), PromptAdmissionError> { + let (sender, busy, handle) = self + .prompt_route(&request.session_id) + .map_err(PromptAdmissionError::Runtime)?; // Reject overlaps before waiting: mailbox pressure must not queue another turn. if busy.load(Ordering::Acquire) { - return Err(AcpRuntimeError::Unsupported( - "session is already running a prompt".into(), - )); + return Err(PromptAdmissionError::Busy); } // Claim only after capacity is available so cancellation cannot strand // ownership. Recheck atomically afterward in case another request now owns it. let permit = sender .reserve() .await - .map_err(|_| AcpRuntimeError::ClientClosed)?; - claim_prompt(&busy)?; + .map_err(|_| PromptAdmissionError::Runtime(AcpRuntimeError::ClientClosed))?; + claim_prompt(&busy).map_err(|_| PromptAdmissionError::Busy)?; handle.prepare_injection_turn(); let cancellation_generation = handle.cancellation_handle().generation(); let (reply, response) = oneshot::channel(); @@ -1130,11 +1361,11 @@ impl Server { reply, })); match response.await { - Ok(response) => response, + Ok(response) => response.map_err(PromptAdmissionError::Runtime), Err(_) => { handle.stop_injection_turn(); busy.store(false, Ordering::Release); - Err(AcpRuntimeError::ClientClosed) + Err(PromptAdmissionError::Runtime(AcpRuntimeError::ClientClosed)) } } } @@ -2224,7 +2455,12 @@ pub async fn serve_with_registry( runtime: Arc, registry: SessionRegistry, ) -> Result<(), AcpRuntimeError> { - super::connect_stdio(v2_router(runtime, registry)).await + super::connect_stdio( + agent_client_protocol::Agent + .protocol_router() + .with_v2(stdio_component(runtime, registry)), + ) + .await } pub(crate) fn http_router(runtime: Arc, registry: SessionRegistry) -> axum::Router { @@ -2252,7 +2488,24 @@ pub(crate) fn component( runtime: Arc, registry: SessionRegistry, ) -> impl ConnectTo { - let state = Arc::new(Server::new(runtime, registry)); + component_with_binding(runtime, registry, false) +} + +pub(super) fn stdio_component( + runtime: Arc, + registry: SessionRegistry, +) -> impl ConnectTo { + component_with_binding(runtime, registry, true) +} + +fn component_with_binding( + runtime: Arc, + registry: SessionRegistry, + bind_a2a: bool, +) -> impl ConnectTo { + let mut server = Server::new(runtime, registry); + server.bind_a2a = bind_a2a; + let state = Arc::new(server); agent_client_protocol::Agent .v2() .name("kit") @@ -3768,6 +4021,474 @@ mod tests { handle } + #[tokio::test] + async fn a2a_busy_admission_steers_the_bound_driver() { + assert_a2a_busy_admission(false).await; + } + + #[tokio::test] + async fn a2a_idle_to_busy_race_steers_without_duplicate_admission() { + assert_a2a_busy_admission(true).await; + } + + async fn assert_a2a_busy_admission(competing_turn: bool) { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new(root.path(), "gpt-5.4").unwrap(); + let server = Server::new(runtime, SessionRegistry::new()); + let session_id = wire::SessionId::new("a2a-busy"); + let recording = RecordingSink::default(); + let handle = bind_test_session(&server.integration, &session_id, recording.clone()); + let (commands, mut received) = mpsc::channel(1); + let sender = commands.clone(); + let busy = Arc::new(AtomicBool::new(!competing_turn)); + server.sessions.lock().unwrap().insert( + session_id.clone(), + SessionHandle { + subagents: None, + voice_state: crate::runtime::voice_state::VoiceState::default(), + token: 1, + commands, + integration: handle.clone(), + busy: busy.clone(), + background_jobs: BackgroundJobs::default(), + structured_completion: false, + tasks: AsyncTaskManager::new().handle(), + }, + ); + if competing_turn { + handle.stop_injection_turn(); + } + let (reply, _closed) = oneshot::channel(); + if competing_turn { + sender.send(Command::Close { reply }).await.unwrap(); + } + let admission = server.admit_a2a(session_id.clone(), "a2a steer".into()); + tokio::pin!(admission); + if competing_turn { + assert!(futures_util::poll!(admission.as_mut()).is_pending()); + claim_prompt(&busy).unwrap(); + handle.prepare_injection_turn(); + handle.start_injection_turn(); + let _ = received.recv().await.unwrap(); + } + let acknowledgment = timeout(Duration::from_secs(2), admission) + .await + .unwrap() + .unwrap(); + assert!(acknowledgment.contains("not completion")); + let (mut driver, _) = test_driver(TestOutcome::Content, "a2a-busy").await; + driver + .submit_input(vec![Item::text(ItemKind::User, "existing turn")]) + .unwrap(); + drive_prompt( + &session_id, + &mut driver, + &handle, + handle.cancellation_handle().generation(), + None, + ) + .await + .unwrap(); + assert!(driver.snapshot().transcript.iter().any(|item| { + item.kind == ItemKind::User + && item + .parts + .iter() + .any(|part| matches!(part, Part::Text(text) if text.text == "a2a steer")) + })); + assert!(recording.updates.lock().unwrap().iter().any(|update| update.session_id == session_id && matches!(&update.update, wire::SessionUpdate::UserMessage(message) if serde_json::to_string(message).unwrap().contains("a2a steer")))); + handle.close(); + assert!( + server + .admit_a2a(session_id, "closed steer".into()) + .await + .is_err() + ); + } + + #[tokio::test] + async fn a2a_idle_admission_prompts_the_bound_driver() { + assert_a2a_idle_admission(false).await; + } + + #[tokio::test] + async fn a2a_busy_to_idle_race_retries_only_before_commit() { + assert_a2a_idle_admission(true).await; + } + + async fn assert_a2a_idle_admission(ending_turn: bool) { + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new(root.path(), "gpt-5.4").unwrap(); + let server = Server::new(runtime.clone(), SessionRegistry::new()); + let session_id = wire::SessionId::new("a2a-idle"); + let recording = RecordingSink::default(); + let sink = ResponseReplacementSink::new(recording.clone()); + let handle = server + .integration + .bind_session(AcpSessionBinding::new( + session_id.clone(), + SessionId::new("a2a-idle"), + sink.clone(), + )) + .unwrap(); + let activity = native_activity(session_id.clone(), sink.clone()); + let busy = Arc::new(AtomicBool::new(ending_turn)); + if ending_turn { + handle.prepare_injection_turn(); + handle.start_injection_turn(); + } + let (commands, mut received) = mpsc::channel(1); + let tasks = AsyncTaskManager::new().handle(); + let background_jobs = BackgroundJobs::default(); + server.sessions.lock().unwrap().insert( + session_id.clone(), + SessionHandle { + subagents: None, + voice_state: crate::runtime::voice_state::VoiceState::default(), + token: 1, + commands, + integration: handle.clone(), + busy: busy.clone(), + background_jobs: background_jobs.clone(), + structured_completion: false, + tasks: tasks.clone(), + }, + ); + let (mut driver, _) = test_driver(TestOutcome::Content, "a2a-idle").await; + let mut catalog = skill_catalog::SkillCatalogMonitor::new(&[]).unwrap(); + let mut voice = crate::runtime::voice_state::VoiceState::default().monitor(&[]); + let admission = server.admit_a2a(session_id.clone(), "idle a2a message".into()); + tokio::pin!(admission); + if ending_turn { + assert!(futures_util::poll!(admission.as_mut()).is_pending()); + handle.stop_injection_turn(); + busy.store(false, Ordering::Release); + } + let execution = async { + let Command::Prompt(command) = received.recv().await.unwrap() else { + panic!("expected prompt") + }; + prepare_prompt( + &session_id, + &runtime, + &server.integration, + &handle, + &mut catalog, + &mut voice, + &mut driver, + command, + &sink, + &tasks, + &background_jobs, + false, + &activity, + ) + .await + .unwrap(); + }; + let (acknowledgment, ()) = timeout(Duration::from_secs(2), async { + tokio::join!(admission, execution) + }) + .await + .unwrap(); + assert!(acknowledgment.unwrap().contains("a2a-idle")); + assert!(driver.snapshot().transcript.iter().any(|item| { + item.kind == ItemKind::User + && item + .parts + .iter() + .any(|part| matches!(part, Part::Text(text) if text.text == "idle a2a message")) + })); + assert!( + recording + .updates + .lock() + .unwrap() + .iter() + .any(|update| update.session_id == session_id + && matches!(update.update, wire::SessionUpdate::UserMessage(_))) + ); + assert!( + server + .admit_a2a(wire::SessionId::new("missing"), "message".into()) + .await + .is_err() + ); + } + + // The SDK receipt commits at queue admission, before physical transport writes. + mod a2a_transport_support { + use super::*; + use futures_util::StreamExt as _; + + pub(super) async fn inbound_inject( + session_id: &wire::SessionId, + text: &str, + ) -> ( + wire::InjectSessionRequest, + Responder, + agent_client_protocol::Channel, + tokio::task::JoinHandle>, + ) { + let (mut client, agent) = agent_client_protocol::Channel::duplex(); + let (received, mut requests) = mpsc::unbounded_channel(); + let server = tokio::spawn(async move { + agent_client_protocol::Agent + .v2() + .on_receive_request( + async move |request: wire::InitializeRequest, responder, _cx| { + responder.respond(wire::InitializeResponse::new( + request.protocol_version, + wire::Implementation::new("fifo-boundary-test", "0"), + )) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |request: wire::InjectSessionRequest, responder, _cx| { + received.send((request, responder)).map_err(|error| { + agent_client_protocol::util::internal_error(error.to_string()) + })?; + Ok(()) + }, + agent_client_protocol::on_receive_request!(), + ) + .connect_to(agent) + .await + }); + send_wire( + &client, + "initialize", + 1, + serde_json::to_value(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("fifo-client", "0"), + )) + .unwrap(), + ); + assert!( + receive_wire(&mut client, "FIFO initialize") + .await + .get("result") + .is_some() + ); + send_wire( + &client, + "session/inject", + 2, + serde_json::to_value(wire::InjectSessionRequest::new( + session_id.clone(), + wire::SessionInjectMode::Steer, + vec![wire::ContentBlock::Text(wire::TextContent::new(text))], + )) + .unwrap(), + ); + let (request, responder) = timeout(Duration::from_secs(2), requests.recv()) + .await + .unwrap() + .unwrap(); + (request, responder, client, server) + } + + pub(super) fn suspended_bridge( + integration: Arc, + session_id: wire::SessionId, + ) -> ( + tokio::task::JoinHandle>, + mpsc::UnboundedReceiver<()>, + Arc, + ) { + let (client, agent) = agent_client_protocol::Channel::duplex(); + let (entered, received) = mpsc::unbounded_channel(); + let release = Arc::new(Notify::new()); + let outgoing = futures_util::sink::unfold( + (agent.tx, entered, release.clone()), + async move |(tx, entered, release), line: String| { + let value: Value = + serde_json::from_str(&line).map_err(std::io::Error::other)?; + if value["result"]["messageId"].is_string() { + entered.send(()).map_err(std::io::Error::other)?; + release.notified().await; + } + let message = serde_json::from_str(&line).map_err(std::io::Error::other)?; + tx.unbounded_send(agent_client_protocol::TransportFrame::Single(message)) + .map_err(std::io::Error::other)?; + Ok::<_, std::io::Error>((tx, entered, release)) + }, + ); + let incoming = agent.rx.map(|frame| { + let agent_client_protocol::TransportFrame::Single(message) = frame else { + panic!("single bridge request expected") + }; + serde_json::to_string(&message).map_err(std::io::Error::other) + }); + let transport = + agent_client_protocol::Lines::new(Box::pin(outgoing), Box::pin(incoming)); + let bridge = tokio::spawn(inject_a2a_over( + integration, + wire::InjectSessionRequest::new( + session_id, + wire::SessionInjectMode::Steer, + vec![wire::ContentBlock::Text(wire::TextContent::new( + "receipt-gated steer", + ))], + ), + client, + transport, + )); + (bridge, received, release) + } + } + + #[tokio::test] + async fn a2a_cancelled_fifo_waiter_advances_before_later_a2a_and_tui_steering() { + let integration = Arc::new(AcpIntegration::default()); + let session_id = wire::SessionId::new("a2a-fifo-cancellation"); + let handle = bind_test_session(&integration, &session_id, RecordingSink::default()); + let (a_request, a_responder, _a_client, a_server) = + a2a_transport_support::inbound_inject(&session_id, "held A").await; + let held = integration + .reserve_inject_request(a_request, a_responder) + .await + .unwrap() + .unwrap(); + let (b_request, b_responder, _b_client, b_server) = + a2a_transport_support::inbound_inject(&session_id, "cancelled B").await; + let cancellation = tokio_util::sync::CancellationToken::new(); + let mut waiting = Box::pin(finish_a2a_injection( + integration.clone(), + b_request, + b_responder, + cancellation.clone(), + )); + // A holds the real permit; this first poll allocates B's SDK sequence + // and suspends at its FIFO wait, without implementation instrumentation. + assert!(futures_util::poll!(waiting.as_mut()).is_pending()); + cancellation.cancel(); + assert!( + futures_util::poll!(waiting.as_mut()).is_pending(), + "cancelled ingress must still drain to its FIFO permit" + ); + let drain = tokio::spawn(waiting); + drop(held); + let error = timeout(Duration::from_secs(2), drain) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!( + error.code, + agent_client_protocol::Error::request_cancelled().code + ); + timeout( + Duration::from_secs(2), + inject_a2a( + integration.clone(), + wire::InjectSessionRequest::new( + session_id.clone(), + wire::SessionInjectMode::Steer, + vec![wire::ContentBlock::Text(wire::TextContent::new("later C"))], + ), + ), + ) + .await + .unwrap() + .unwrap(); + let (ordinary, _tui_client, tui_server) = + staged_injection((*integration).clone(), &session_id).await; + ordinary.activate_after_response().await.unwrap(); + let (mut driver, _) = test_driver(TestOutcome::Content, &session_id.to_string()).await; + driver + .submit_input(vec![Item::text(ItemKind::User, "existing request")]) + .unwrap(); + timeout( + Duration::from_secs(2), + drive_prompt( + &session_id, + &mut driver, + &handle, + handle.cancellation_handle().generation(), + None, + ), + ) + .await + .unwrap() + .unwrap(); + let transcript = driver.snapshot().transcript; + for text in ["later C", "pending steer"] { + assert!(transcript.iter().any(|item| { + item.kind == ItemKind::User + && item + .parts + .iter() + .any(|part| matches!(part, Part::Text(part) if part.text == text)) + })); + } + assert!(!transcript.iter().any(|item| item.parts.iter().any(|part| matches!(part, Part::Text(part) if part.text == "cancelled B" || part.text == "held A")))); + for server in [a_server, b_server, tui_server] { + server.abort(); + let _ = server.await; + } + } + + #[tokio::test] + async fn a2a_queue_activated_injection_survives_later_transport_cancellation() { + for cancel in [false, true] { + let integration = Arc::new(AcpIntegration::default()); + let session_id = wire::SessionId::new(if cancel { + "a2a-receipt-cancel" + } else { + "a2a-receipt-success" + }); + let recording = RecordingSink::default(); + let handle = bind_test_session(&integration, &session_id, recording); + let (bridge, mut entered, release) = + a2a_transport_support::suspended_bridge(integration, session_id.clone()); + timeout(Duration::from_secs(2), entered.recv()) + .await + .unwrap() + .unwrap(); + assert!( + !bridge.is_finished(), + "the client response still waits for the physical transport write" + ); + if cancel { + bridge.abort(); + assert!(bridge.await.unwrap_err().is_cancelled()); + release.notify_one(); + } else { + release.notify_one(); + timeout(Duration::from_secs(2), bridge) + .await + .unwrap() + .unwrap() + .unwrap(); + } + let (mut driver, _) = test_driver(TestOutcome::Content, &session_id.to_string()).await; + driver + .submit_input(vec![Item::text(ItemKind::User, "existing request")]) + .unwrap(); + timeout( + Duration::from_secs(2), + drive_prompt( + &session_id, + &mut driver, + &handle, + handle.cancellation_handle().generation(), + None, + ), + ) + .await + .unwrap() + .unwrap(); + let delivered = driver.snapshot().transcript.iter().any(|item| item.kind == ItemKind::User && item.parts.iter().any(|part| matches!(part, Part::Text(text) if text.text == "receipt-gated steer"))); + assert!( + delivered, + "queue-activated steers survive later transport cancellation" + ); + } + } + // Hold a real response-committed receipt at the external ACP transport boundary. // Until activation, the real injection coordinator must wait rather than lose the steer. async fn staged_injection( @@ -4011,6 +4732,53 @@ mod tests { let _ = server.await; } + #[tokio::test] + async fn a2a_pending_activation_is_discarded_on_owner_drop_and_session_close() { + for close in [false, true] { + let integration = AcpIntegration::default(); + let session_id = wire::SessionId::new(if close { + "a2a-pending-close" + } else { + "a2a-pending-drop" + }); + let handle = bind_test_session(&integration, &session_id, RecordingSink::default()); + let (receipt, _client, server) = staged_injection(integration, &session_id).await; + let (mut driver, _) = test_driver(TestOutcome::Content, &session_id.to_string()).await; + driver + .submit_input(vec![Item::text(ItemKind::User, "existing request")]) + .unwrap(); + let generation = handle.cancellation_handle().generation(); + { + let turn = drive_prompt(&session_id, &mut driver, &handle, generation, None); + tokio::pin!(turn); + assert!( + futures_util::poll!(turn.as_mut()).is_pending(), + "a committed but unactivated steer must hold the boundary" + ); + if close { + handle.close(); + receipt.activate_after_response().await.unwrap(); + } else { + // This is the real ownership cleanup used when scoped ingress + // cancellation drops its suspended activation future. + drop(receipt); + } + let result = timeout(Duration::from_secs(2), turn).await.unwrap(); + if !close { + assert_eq!(result.unwrap(), FinishReason::Completed); + } + } + assert!(!driver.snapshot().transcript.iter().any(|item| { + item.kind == ItemKind::User + && item.parts.iter().any( + |part| matches!(part, Part::Text(text) if text.text == "pending steer"), + ) + })); + server.abort(); + let _ = server.await; + } + } + #[tokio::test] async fn terminal_injection_waits_for_response_activation_and_delivers_input() { let integration = AcpIntegration::default(); diff --git a/src/protocols/acp/v2/a2a_tests.rs b/src/protocols/acp/v2/a2a_tests.rs new file mode 100644 index 00000000..73ca93d0 --- /dev/null +++ b/src/protocols/acp/v2/a2a_tests.rs @@ -0,0 +1,228 @@ +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] + +use super::*; +use futures_util::StreamExt as _; +use serde_json::{Value, json}; +use tokio::time::{Duration, timeout}; + +const WAIT: Duration = Duration::from_secs(30); +const PROMPT: &str = "a2a-real-session-execution-sentinel"; +const ANSWER: &str = "answer-from-local-provider"; + +fn send(channel: &agent_client_protocol::Channel, method: &str, id: i64, params: Value) { + channel + .tx + .unbounded_send(agent_client_protocol::TransportFrame::Single( + agent_client_protocol::RawJsonRpcMessage::request(method.into(), params, id.into()) + .unwrap(), + )) + .unwrap(); +} + +async fn receive(channel: &mut agent_client_protocol::Channel) -> Value { + let frame = timeout(WAIT, channel.rx.next()) + .await + .expect("ACP frame timed out") + .expect("ACP transport closed"); + let agent_client_protocol::TransportFrame::Single(message) = frame else { + panic!("expected a single ACP message"); + }; + serde_json::to_value(message).unwrap() +} + +/// The subprocess isolates provider URL and HOME, following runtime's provider +/// boundary tests. Only external HTTP inference is fake: ACP publication, +/// registry admission, HTTP A2A handling, and the session actor all remain real. +#[tokio::test] +async fn stdio_session_a2a_admission_executes_on_real_actor() { + const CHILD: &str = "KIT_TEST_STDIO_A2A_EXECUTION"; + if std::env::var_os(CHILD).is_none() { + async fn inference( + axum::extract::State(requests): axum::extract::State< + tokio::sync::mpsc::UnboundedSender, + >, + axum::Json(body): axum::Json, + ) -> impl axum::response::IntoResponse { + requests.send(body).unwrap(); + let chunk = json!({"id": "local-a2a-test", "choices": [{ + "index": 0, + "delta": {"role": "assistant", "content": ANSWER}, + "finish_reason": "stop" + }]}); + ( + [("content-type", "text/event-stream")], + format!("data: {chunk}\n\ndata: [DONE]\n\n"), + ) + } + let home = tempfile::tempdir().unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (requests, mut received) = tokio::sync::mpsc::unbounded_channel(); + let app = axum::Router::new() + .fallback( + axum::routing::post(inference).get(|| async { axum::Json(json!({"data": []})) }), + ) + .with_state(requests); + let provider = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let mut command = tokio::process::Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "protocols::acp::v2::a2a_tests::stdio_session_a2a_admission_executes_on_real_actor", + "--nocapture", + ]) + .env_clear() + .env(CHILD, "1") + .env("HOME", home.path()) + .env("PATH", "/usr/bin:/bin") + .env("OPENROUTER_API_KEY", "local-test-key") + .env("OPENROUTER_BASE_URL", format!("http://{address}/stream")) + .env("NO_PROXY", "127.0.0.1") + .kill_on_drop(true); + let output = timeout(Duration::from_secs(60), command.output()) + .await + .expect("isolated ACP/A2A test timed out") + .unwrap(); + provider.abort(); + let _ = provider.await; + assert!( + output.status.success(), + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + let body = received + .try_recv() + .expect("A2A prompt never reached inference"); + assert_eq!(body["model"], "test/model"); + assert!( + body["messages"].as_array().unwrap().iter().any(|message| { + message["role"] == "user" && message["content"].to_string().contains(PROMPT) + }), + "provider request did not contain the admitted user prompt" + ); + return; + } + + let root = tempfile::tempdir().unwrap(); + let runtime = Runtime::new_with_provider_credentials_and_effort( + root.path(), + "test/model", + crate::ProviderKind::OpenRouter, + crate::credentials::CredentialStorage::Memory, + None, + ) + .unwrap(); + let registry = SessionRegistry::new(); + let (mut client, agent) = agent_client_protocol::Channel::duplex(); + let router = agent_client_protocol::Agent + .protocol_router() + .with_v2(stdio_component(runtime.clone(), registry.clone())); + let acp = tokio::spawn(async move { router.connect_to(agent).await }); + send( + &client, + "initialize", + 1, + serde_json::to_value(wire::InitializeRequest::new( + wire::ProtocolVersion::V2, + wire::Implementation::new("a2a-test", "0"), + )) + .unwrap(), + ); + let initialized = receive(&mut client).await; + assert_eq!(initialized["id"], 1); + assert!(initialized.get("result").is_some(), "{initialized}"); + send( + &client, + "session/new", + 2, + serde_json::to_value(wire::NewSessionRequest::new(root.path().to_path_buf())).unwrap(), + ); + let created = receive(&mut client).await; + assert_eq!(created["id"], 2); + let session: wire::NewSessionResponse = + serde_json::from_value(created["result"].clone()).unwrap(); + + let mut http = crate::protocols::http::start_with_registry( + runtime, + "127.0.0.1:0".into(), + true, + false, + None, + registry, + ) + .await + .unwrap(); + let admitted: Value = timeout( + WAIT, + reqwest::Client::new() + .post(format!("http://{}/", http.address())) + .json(&json!({ + "jsonrpc": "2.0", "id": 3, "method": "SendMessage", + "params": {"message": { + "messageId": "stdio-a2a-test", "role": "ROLE_USER", + "parts": [{"text": PROMPT}] + }} + })) + .send(), + ) + .await + .expect("A2A admission timed out") + .unwrap() + .error_for_status() + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(admitted["id"], 3); + assert_eq!( + admitted["result"]["message"]["metadata"]["kind"], + "admission_acknowledgment" + ); + + // An admission acknowledgment alone cannot prove execution. Require the + // external provider's answer on this exact ACP session and its idle boundary. + let mut answer = String::new(); + let mut running = false; + timeout(WAIT, async { + loop { + let frame = receive(&mut client).await; + assert_eq!(frame["method"], "session/update", "{frame}"); + let update: wire::UpdateSessionNotification = + serde_json::from_value(frame["params"].clone()).unwrap(); + assert_eq!(update.session_id, session.session_id); + match update.update { + wire::SessionUpdate::AgentMessageChunk(chunk) => { + if let wire::ContentBlock::Text(text) = chunk.content { + answer.push_str(&text.text); + } + } + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Running(_)) => running = true, + wire::SessionUpdate::StateUpdate(wire::StateUpdate::Idle(_)) if running => break, + _ => {} + } + } + }) + .await + .expect("admitted A2A turn never completed on ACP"); + assert_eq!(answer, ANSWER); + send( + &client, + "session/close", + 4, + serde_json::to_value(wire::CloseSessionRequest::new(session.session_id)).unwrap(), + ); + let closed = receive(&mut client).await; + assert_eq!(closed["id"], 4); + assert!(closed.get("result").is_some(), "{closed}"); + http.shutdown(); + http.join().await.unwrap(); + acp.abort(); + let _ = acp.await; +} diff --git a/src/protocols/http.rs b/src/protocols/http.rs index 03ed2a11..ad9d7c11 100644 --- a/src/protocols/http.rs +++ b/src/protocols/http.rs @@ -107,6 +107,13 @@ impl Drop for HttpServer { } } +/// Select A2A behavior at startup, never as a fallback after failed admission. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum A2aExecutionMode { + Standalone, + SessionAdmission, +} + pub async fn start( runtime: Arc, address: String, @@ -114,13 +121,14 @@ pub async fn start( serve_remote_acp: bool, credential_file: Option<&Path>, ) -> Result> { - start_with_registry( + start_with_registry_and_mode( runtime, address, serve_a2a, serve_remote_acp, credential_file, crate::protocols::acp::SessionRegistry::new(), + A2aExecutionMode::Standalone, ) .await } @@ -132,6 +140,27 @@ pub async fn start_with_registry( serve_remote_acp: bool, credential_file: Option<&Path>, sessions: crate::protocols::acp::SessionRegistry, +) -> Result> { + start_with_registry_and_mode( + runtime, + address, + serve_a2a, + serve_remote_acp, + credential_file, + sessions, + A2aExecutionMode::SessionAdmission, + ) + .await +} + +pub async fn start_with_registry_and_mode( + runtime: Arc, + address: String, + serve_a2a: bool, + serve_remote_acp: bool, + credential_file: Option<&Path>, + sessions: crate::protocols::acp::SessionRegistry, + a2a_mode: A2aExecutionMode, ) -> Result> { if !serve_a2a && !serve_remote_acp { return Err(io::Error::new( @@ -152,7 +181,14 @@ pub async fn start_with_registry( let listener = tokio::net::TcpListener::bind(&address).await?; let bound = listener.local_addr()?; let a2a = serve_a2a - .then(|| crate::protocols::a2a::dispatcher(runtime.clone(), bound, credential.is_some())) + .then(|| { + crate::protocols::a2a::dispatcher( + runtime.clone(), + bound, + credential.is_some(), + (a2a_mode == A2aExecutionMode::SessionAdmission).then(|| sessions.clone()), + ) + }) .transpose()? .map(Arc::new); let acp = serve_remote_acp.then(|| { @@ -182,7 +218,7 @@ pub async fn start_with_registry( async fn serve_bound( listener: tokio::net::TcpListener, - a2a: Option>, + a2a: Option>, acp: Option, credential: Option, stop_accepting: CancellationToken, @@ -316,7 +352,7 @@ fn report_connection(result: Result<(), tokio::task::JoinError>) { async fn dispatch( request: Request, - a2a: Option>, + a2a: Option>, acp: Option, credential: Option, ) -> Result, Infallible> { @@ -340,7 +376,7 @@ async fn dispatch( } if let Some(dispatcher) = a2a { - return Ok(dispatcher.dispatch(request).await.map(Body::new)); + return Ok(dispatcher.dispatch(request).await); } let mut response = Response::new(Body::from("not found")); @@ -360,7 +396,10 @@ async fn dispatch( mod tests { use std::time::Duration; - use super::{BearerToken, start}; + use super::{ + A2aExecutionMode, BearerToken, start, start_with_registry, start_with_registry_and_mode, + }; + use std::sync::Arc; use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; #[test] @@ -397,6 +436,137 @@ mod tests { } } + #[tokio::test] + async fn explicit_standalone_mode_does_not_require_an_interactive_default() { + let directory = tempfile::tempdir().unwrap(); + let runtime = crate::Runtime::new(directory.path(), "gpt-5.4").unwrap(); + let sessions = crate::protocols::acp::SessionRegistry::new(); + let mut server = start_with_registry_and_mode( + runtime, + "127.0.0.1:0".into(), + true, + true, + None, + sessions.clone(), + A2aExecutionMode::Standalone, + ) + .await + .unwrap(); + // Empty input exercises the standalone executor without contacting a provider. + let response: serde_json::Value = reqwest::Client::new().post(format!("http://{}/", server.address())) + .header("A2A-Version", "1.0").json(&serde_json::json!({ + "jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": { + "message": {"messageId": "standalone", "role": "ROLE_USER", "parts": [{"text": " "}]} + } + })).send().await.unwrap().json().await.unwrap(); + assert!(response.get("error").is_none(), "{response}"); + assert_eq!( + response["result"]["task"]["status"]["state"], + "TASK_STATE_FAILED" + ); + assert!( + response["result"]["task"]["artifacts"] + .to_string() + .contains("must contain a text part") + ); + sessions.shutdown().await; + server.shutdown(); + server.join().await.unwrap(); + } + + #[tokio::test] + async fn a2a_shared_listener_returns_admission_message_and_never_falls_back() { + let directory = tempfile::tempdir().unwrap(); + let runtime = crate::Runtime::new(directory.path(), "gpt-5.4").unwrap(); + let client = reqwest::Client::new(); + let request = serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": { + "message": {"messageId": "http-admission", "role": "ROLE_USER", "parts": [{"text": "admit"}]} + }}); + let mut unbound = start_with_registry( + runtime.clone(), + "127.0.0.1:0".into(), + true, + false, + None, + crate::protocols::acp::SessionRegistry::new(), + ) + .await + .unwrap(); + let response: serde_json::Value = client + .post(format!("http://{}/", unbound.address())) + .json(&request) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!( + response["error"]["message"] + .as_str() + .unwrap() + .contains("no bound") + ); + unbound.shutdown(); + unbound.join().await.unwrap(); + + let sessions = + crate::protocols::acp::a2a_test_support::bound_registry(Arc::new(|_prompt| { + Box::pin(async { Ok("Admission only, not completion".into()) }) + })); + let mut server = + start_with_registry(runtime, "127.0.0.1:0".into(), true, false, None, sessions) + .await + .unwrap(); + let response: serde_json::Value = client + .post(format!("http://{}/", server.address())) + .json(&request) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(response["id"], 1); + assert_eq!(response["result"]["message"]["role"], "ROLE_AGENT"); + assert_eq!( + response["result"]["message"]["metadata"]["kind"], + "admission_acknowledgment" + ); + assert!(response["result"]["message"].get("status").is_none()); + assert!(response["result"]["message"].get("taskId").is_none()); + let a2a_client = + a2a_protocol_client::ClientBuilder::new(format!("http://{}/", server.address())) + .build() + .unwrap(); + let decoded = a2a_client + .send_message(serde_json::from_value(request["params"].clone()).unwrap()) + .await + .unwrap(); + let a2a_protocol_types::responses::SendMessageResponse::Message(message) = decoded else { + panic!("expected an admission Message, not a Task") + }; + assert_eq!( + message.metadata.unwrap()["kind"], + "admission_acknowledgment" + ); + let mut malformed = request.clone(); + malformed["id"] = serde_json::json!({"invalid": "id"}); + let response: serde_json::Value = client + .post(format!("http://{}/", server.address())) + .json(&malformed) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(response["id"].is_null()); + assert_eq!(response["error"]["code"], -32600); + server.shutdown(); + server.join().await.unwrap(); + } + #[tokio::test] async fn port_zero_is_retained_and_bearer_protects_all_protocols() { let directory = tempfile::tempdir().unwrap(); From 7ea16149191e151287a7c1b7740a25e01bcd11ff Mon Sep 17 00:00:00 2001 From: daniel Date: Tue, 13 Oct 2026 15:16:24 +0100 Subject: [PATCH 2/2] test(a2a): adapt prompt preparation to current main --- src/protocols/acp/v2.rs | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/src/protocols/acp/v2.rs b/src/protocols/acp/v2.rs index 29308f34..5f229f4b 100644 --- a/src/protocols/acp/v2.rs +++ b/src/protocols/acp/v2.rs @@ -4494,19 +4494,23 @@ mod tests { panic!("expected prompt") }; prepare_prompt( - &session_id, - &runtime, - &server.integration, - &handle, - &mut catalog, - &mut voice, + TurnContext { + session_id: &session_id, + integration: &server.integration, + handle: &handle, + sink: &sink, + activity: &activity, + }, + PromptPreparation { + runtime: &runtime, + skill_catalog: &mut catalog, + voice_monitor: &mut voice, + tasks: &tasks, + background_jobs: &background_jobs, + structured_completion: false, + }, &mut driver, command, - &sink, - &tasks, - &background_jobs, - false, - &activity, ) .await .unwrap();