From aa5e73c84a262df941c0895ff4c276016b8b45cf Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Fri, 16 Oct 2026 12:29:20 +0200 Subject: [PATCH] fix(acp): align protocol version negotiation --- justfile | 1 + md/SUMMARY.md | 2 +- md/protocol-v2.md | 115 +++++--- .../src/conductor.rs | 6 +- .../tests/initialization_v2.rs | 17 +- src/agent-client-protocol/src/jsonrpc.rs | 2 + .../src/jsonrpc/protocol_compat.rs | 263 +++++++++--------- src/agent-client-protocol/src/role/acp.rs | 67 +++-- src/agent-client-protocol/src/util.rs | 13 + .../tests/protocol_negotiation.rs | 213 ++++++++++++++ .../tests/protocol_v2.rs | 251 +++++++++++++---- 11 files changed, 704 insertions(+), 246 deletions(-) create mode 100644 src/agent-client-protocol/tests/protocol_negotiation.rs diff --git a/justfile b/justfile index efa207f6..bc39dbc4 100644 --- a/justfile +++ b/justfile @@ -6,3 +6,4 @@ prep-tests: # Prepare subprocess fixtures, then run all tests, including doctests. test: prep-tests cargo test --workspace --all-features --locked + cargo test -p agent-client-protocol --no-default-features --locked diff --git a/md/SUMMARY.md b/md/SUMMARY.md index 85e4c9cb..623bfad6 100644 --- a/md/SUMMARY.md +++ b/md/SUMMARY.md @@ -12,7 +12,7 @@ - [Request Cancellation](./request-cancellation.md) - [Ordered Application Dispatch](./ordered-application-dispatch.md) - [Configurable LLM Providers](./llm-providers.md) -- [Protocol V2](./protocol-v2.md) +- [Protocol V2 and Version Routing](./protocol-v2.md) - [Runnable V2 Quickstart](./protocol-v2-quickstart.md) - [Session Operation Coordination](./session-operation-coordination.md) diff --git a/md/protocol-v2.md b/md/protocol-v2.md index d243e8c4..b3c8c62a 100644 --- a/md/protocol-v2.md +++ b/md/protocol-v2.md @@ -79,8 +79,8 @@ Agent When v2 mode is enabled, application code should use types from `agent_client_protocol::schema::v2`. The flat `agent_client_protocol::schema::*` exports remain the stable v1 schema. This will likely change as v2 gets closer -to release. The preceding agent fragment demonstrates version negotiation only; -an agent that advertises session support must also implement the complete +to release. The preceding agent fragment demonstrates version negotiation only. +An agent that advertises session support must also implement the complete baseline session surface shown in the runnable quickstart. ## High-level v2 sessions @@ -92,7 +92,7 @@ the v2 `build_session*` and `resume_session*` helpers, plus feature-gated `fork_session*` helpers when `unstable_session_fork` is enabled. The shared names describe the same lifecycle operations while the connection type selects their schema and return types at compile time. Resume and fork return -`V2ResumeSessionBuilder` and `V2ForkSessionBuilder`; call `start_session` to +`V2ResumeSessionBuilder` and `V2ForkSessionBuilder`. Call `start_session` to publish the request and obtain an `OpenedV2Session` containing the command handle and complete operation-specific response. @@ -108,7 +108,7 @@ the same low-level `HandleDispatchFrom` interface and therefore also receive Nested connections preserve the stable `ConnectionTo` API while still typing the child implementation's callbacks. On a raw `ConnectionTo<_>`, -`spawn_connection(Client.v2(), transport)` returns a raw `ConnectionTo<_>`; +`spawn_connection(Client.v2(), transport)` returns a raw `ConnectionTo<_>`, and callbacks installed on that v2 child builder still receive `V2ConnectionTo<_>`. Existing `spawn_connection::` calls therefore remain source-compatible. @@ -133,10 +133,10 @@ V2 deliberately separates prompt submission from session observation: - `session/prompt` returns a `PromptResponse` as soon as the agent accepts the prompt. Its `message_id` identifies the inserted user message, and live - `UserMessage` updates for that prompt use the same ID; persisting the message + `UserMessage` updates for that prompt use the same ID. Persisting the message in resumable history remains optional. The response does not indicate turn completion. `V2Session::send_prompt` returns that request as a - `SentRequest`; callers must explicitly await it, register a + `SentRequest`. Callers must explicitly await it, register a response callback, or detach it. - `V2SessionBuilder::start_session` likewise returns a mapped `SentRequest`. Its `OpenedV2Session` result keeps the command handle separate from the @@ -154,17 +154,17 @@ V2 deliberately separates prompt submission from session observation: connection. It does not own, buffer, or unregister inbound messages. - Register typed `UpdateSessionNotification` and `RequestPermissionRequest` handlers on `Client.v2()` before connecting. Updates and interactive requests - are separate protocol lanes; permission handlers should transfer responders + are separate protocol lanes. Permission handlers should transfer responders to application-owned work rather than waiting for user input inside the connection dispatch loop. Without matching handlers, unhandled v2 notifications are ignored and unhandled requests receive a method-not-found - response; they are not retained for a later per-session receiver. + response. They are not retained for a later per-session receiver. - `session/update` events can arrive before, during, or after a prompt request. They carry a session ID and entity IDs, but no prompt or turn ID. The SDK therefore does not attribute intervening events to a locally submitted prompt or provide a prompt-scoped text accumulator. - `state_update` describes the session-wide foreground state. `idle` means the - session can accept ordinary new foreground work; it is not a wire-level + session can accept ordinary new foreground work. It is not a wire-level boundary assigning previous events to one prompt, and background updates may continue while idle. - `cancel_active_work` sends session-wide `session/cancel`. Cancellation @@ -197,12 +197,12 @@ session configured with `V2SessionBuilder::with_mcp_server` or `V2ForkSessionBuilder::with_mcp_server`, the SDK installs the MCP routes and initially polls their runner tasks before publishing the corresponding setup request, so the agent can connect to those servers during setup or resume -replay. Runners may continue asynchronous initialization; custom connectors +replay. Runners may continue asynchronous initialization, so custom connectors must be able to queue connections and messages once constructed. A successful -setup promotes the attachment to the connection lifetime; a setup failure, +setup promotes the attachment to the connection lifetime. A setup failure, including an error response after cancellation, cleans up the pending attachment. This attachment requires both `unstable_protocol_v2` and -`unstable_mcp_over_acp`; fork additionally requires +`unstable_mcp_over_acp`. Fork additionally requires `unstable_session_fork`. A v2 proxy can instead attach one server globally with @@ -246,7 +246,7 @@ source session ID. For `session/resume`, the ID is already known, so routing and MCP attachment are ready before the downstream request is published. Replay updates can therefore be forwarded upstream before the complete `ResumeSessionResponse`, as required by the protocol. A failed or cancelled -downstream response drops pending routing and MCP attachment; successful setup +downstream response drops pending routing and MCP attachment. Successful setup keeps them for the connection lifetime. A cancellation signal itself remains advisory: it is forwarded downstream while the helper awaits that response. Each helper then spawns the callback outside the ordering barrier with an @@ -292,17 +292,17 @@ compatible version is canonicalized through the selected v2 schema before component instantiation. Proxy implementations use -`agent_client_protocol::schema::v2::InitializeProxyRequest`; its response is the +`agent_client_protocol::schema::v2::InitializeProxyRequest`. Its response is the v2 `InitializeResponse`. The flat `schema::InitializeProxyRequest` remains the stable v1 type. Static conductor component providers can carry either selected schema, but each supplied component must support that version. Use `Agent.protocol_router()` or `Proxy.protocol_router()` when a static component has separate implementations. Custom `InstantiateProxiesAndAgent` and `InstantiateProxies` implementations -opt into v2 by implementing their feature-gated v2 method; the default rejects +opt into v2 by implementing their feature-gated v2 method. The default rejects v2 rather than interpreting it as v1. Returning the initialize request unchanged preserves its complete raw parameters for an exact-version request, -including unknown extensions; returning a modified typed request makes that +including unknown extensions. Returning a modified typed request makes that serialized request authoritative. The conductor pins `protocolVersion` to its selected implementation even if an instantiator attempts to change it, and validates the final agent's initialize response against that selection. @@ -325,12 +325,14 @@ This feature extends the concrete compatibility proxy only. The core SDK's global MCP attachment and proxy-session helpers support v1 and v2 independently, as described above. +## Version routing + The SDK handles the `initialize` negotiation at the JSON-RPC boundary: - Native `Client.v2()` and `Agent.v2()` connections reject ordinary protocol - traffic until the initialization response completes; `$/cancel_request` + traffic until the initialization response completes. `$/cancel_request` remains available while initialization is in progress. The client is the - initializer and the agent is the responder; attempts in the opposite + initializer and the agent is the responder. Attempts in the opposite direction are rejected. An initialization error leaves the connection uninitialized so the client can retry, while a second initialization after a successful handshake is rejected. @@ -338,9 +340,20 @@ The SDK handles the `initialize` negotiation at the JSON-RPC boundary: - A v2 client requires a v2 agent. If the agent responds with v1, the `initialize` request resolves with an error and the caller must explicitly fall back to a v1 client implementation if that is acceptable. -- A v2 agent requires a v2 client. If a client initializes with v1, the - `initialize` request resolves with an error and the caller must use a v1 - agent implementation instead. +- A native v1 agent selects v1 for requests advertising v1 or a newer version, + with or without `unstable_protocol_v2` enabled. Newer initialization metadata + is normalized into v1 identity and shared capabilities. The SDK never enables + v1 file-system or terminal-execution capabilities from unknown newer fields. + V1's original fallback-to-latest rule is retained, including a v1 selection + when the client requests an unsupported older version. +- A standalone v2 agent accepts a request for v2 or a newer version with + compatible initialization parameters and selects v2. The request delivered + to the handler has `protocol_version` normalized to v2, and the response + selects v2. If a client initializes with v1, the request resolves with + `-32022` (`Unsupported protocol version`) and + `data: { "requested": 1, "supported": [2] }`. The v2 initialization handler + is not invoked. Use an agent router with a v1 implementation to serve v1 + sessions. - If the agent responds with any other unsupported version, the request resolves with an error so the client can close the connection. - After initialization, the local API version and negotiated wire version must @@ -352,6 +365,14 @@ stabilizes, the `unstable_protocol_v2` crate feature also exposes `Agent.protocol_router()`, `Proxy.protocol_router()`, and `Client.protocol_connector()` for composing version-specific implementations. +Keep these protocol handlers as adapters around shared application logic, not +as general-purpose conversions between wire types. In particular, v1 prompt +responses end a turn, while v2 prompt responses acknowledge insertion and +completion arrives through session state updates. Shared storage also does not +automatically make sessions resumable through both versions. See the protocol's +[dual-stack migration guidance](https://agentclientprotocol.com/protocol/v2/migration#supporting-v1-and-v2-side-by-side) +for adapter boundaries and a validation checklist. + Agents can add protocol implementations independently, which makes it easy for applications built with v2 support to control v2 rollout with a runtime feature flag: @@ -397,16 +418,44 @@ agent # } ``` -The agent protocol router reads the initial `initialize` request, selects the -highest configured protocol version that is compatible with the requested -version, and then hands the connection to that implementation. If only v2 is -configured, v1 clients are rejected without changing the fluent API. The router +The agent protocol router reads the initial `initialize` request and selects +a version-specific implementation. Endpoints exposing v2 select the highest +configured version at or below the requested version, or reject initialization +with a structured version-mismatch error if none exists. V1-only endpoints +retain v1's original fallback-to-latest rule. The router normalizes a v2 initialize request when selecting a v1 implementation, but does not convert messages between v1 and v2 after routing. For compatibility, the -initial frame may be a batch whose first call-shaped entry is `initialize`; the +initial frame may be a batch whose first call-shaped entry is `initialize`. The router preserves the complete frame when handing it to the selected implementation. Response-only frames before initialization are ignored. +### Version mismatch errors + +A v2-only endpoint rejects a v1 request without requiring v2's Client identity +or manufacturing a higher-version success response. The error code is +`-32022`, and its data contains `requested` and the endpoint's `supported` +versions. This diagnostic is separate from malformed initialization parameters, +which remain `-32602`. The router preserves response IDs and batch framing, +flushes the rejection, and closes without starting the selected application. + +Custom endpoints can construct the same wire error with +`agent_client_protocol::util::unsupported_protocol_version(requested, supported)`. +The code is defined by v2 negotiation, not by a change to the v1 protocol. +The shared SDK error model can preserve it as an unknown numeric code, so a +legacy v1 client can handle the initialization error without new v1 schema +definitions. + +`just test` runs the negotiation tests in both the all-features workspace and +the core crate without default features, so v1 negotiation does not depend on +whether the draft v2 API is compiled in. + +Clients should close after a version mismatch and display the supported +versions. A successful downgrade must also be checked against the Client's +actual supported versions, because advertising a latest version does not imply +support for every older version. + +### Proxy and client routing + The proxy protocol router reads `_proxy/initialize` after the conductor has selected the chain's wire version. It therefore requires an exact configured v1 or v2 implementation instead of negotiating or downgrading. It validates @@ -449,8 +498,8 @@ connection. That does not turn an otherwise valid v2 request into an error. ## Schema changes in schema 1.5 through 1.11 The `unstable_protocol_v2` API follows the moving draft schema. Schema 1.5 added -semantic newtypes for paths, media types, IDs, and cursors; renamed -`DiffPatch.diff` to `DiffPatch.text`; and added terminal state and output update +semantic newtypes for paths, media types, IDs, and cursors, renamed +`DiffPatch.diff` to `DiffPatch.text`, and added terminal state and output update types. Schema 1.7 removed the former schema-wide v1/v2 conversion API: versioned implementations should remain separate, with purpose-specific adapters at runtime boundaries where the required state and @@ -462,25 +511,25 @@ Schema 1.6 adds `Cancelled` tool-call and plan-entry statuses to draft v2. Schema 1.7 stabilizes elicitation and terminal authentication, so neither surface requires its former SDK feature flag. It also adds context compaction -updates (stable since schema 1.11); the SDK carries them through its +updates (stable since schema 1.11). The SDK carries them through its existing typed `session/update` routing in both protocol versions. V1 clients advertise compaction support through `ClientSessionCapabilities::compaction`. Schema 1.8 stabilizes optional programmatic tool-call names in both protocol -versions. No tool-name-specific feature is required; draft v2 users only need +versions. No tool-name-specific feature is required. Draft v2 users only need `unstable_protocol_v2`. In v1, an omitted or `null` name leaves the existing value unchanged on updates. In v2, omission leaves it unchanged, `null` clears it, and a string replaces it. V1 cannot express the explicit v2 `null` clear operation. Schema 1.11 stabilizes session notices and compaction. No surface-specific -feature is required; draft v2 users still need `unstable_protocol_v2`. +feature is required. Draft v2 users still need `unstable_protocol_v2`. In draft v2, failures after a prompt's user message is inserted end active work with `StopReason::Error(ErrorStopReason)` on the idle state update, optionally including a JSON-RPC error object. Failures before insertion remain error responses to `session/prompt`. `StopReason` now serializes as an object tagged -by `stopReason`, flattened into `IdleStateUpdate`; its `Other` variant takes +by `stopReason`, flattened into `IdleStateUpdate`. Its `Other` variant takes `OtherStopReason` rather than a string and preserves unknown payload fields. V1 prompt responses are unchanged and do not accept an `error` stop reason. @@ -489,6 +538,6 @@ silently dropping invalid entries or substituting defaults. This includes session setup's MCP servers and additional directories, terminal launch arguments and environment, and v2 replay cursors. Typed request decoding reports invalid parameters (`-32602`) without invoking the handler. `null` -still means empty for the defaultable acted-on lists; omission is accepted +still means empty for the defaultable acted-on lists. Omission is accepted only where the field is optional. Capability and display metadata retain their lenient decoding. diff --git a/src/agent-client-protocol-conductor/src/conductor.rs b/src/agent-client-protocol-conductor/src/conductor.rs index 4b5024b2..fb714191 100644 --- a/src/agent-client-protocol-conductor/src/conductor.rs +++ b/src/agent-client-protocol-conductor/src/conductor.rs @@ -162,8 +162,10 @@ impl InitializeProtocol { } else if requested == ProtocolVersion::V1 { Self::V1 } else { - return Err(Error::invalid_request() - .data(format!("unsupported ACP protocol version {requested}"))); + return Err(agent_client_protocol::util::unsupported_protocol_version( + requested, + &[ProtocolVersion::V1, ProtocolVersion::V2], + )); }; Ok(InitializeProtocolSelection { diff --git a/src/agent-client-protocol-conductor/tests/initialization_v2.rs b/src/agent-client-protocol-conductor/tests/initialization_v2.rs index cd9a0fdc..2a04a229 100644 --- a/src/agent-client-protocol-conductor/tests/initialization_v2.rs +++ b/src/agent-client-protocol-conductor/tests/initialization_v2.rs @@ -321,8 +321,10 @@ async fn assert_invalid_final_agent_initialize_response( async fn assert_raw_initialize_rejected( params: serde_json::Value, expected_error: &str, + expected_code: i32, ) -> Result<(), Error> { run_raw_client_with_conductor(ProxiesAndAgent::new(Agent.v2()), async move |cx| { + let requested = params.get("protocolVersion").cloned(); let request = UntypedMessage { method: "initialize".to_string(), params, @@ -334,6 +336,16 @@ async fn assert_raw_initialize_rejected( .await .expect("invalid initialize request should not hang") .expect_err("invalid initialize request must be rejected"); + assert_eq!(i32::from(error.code), expected_code); + if expected_code == -32022 { + assert_eq!( + error.data, + Some(serde_json::json!({ + "requested": requested.expect("a version mismatch has a requested version"), + "supported": [1, 2], + })) + ); + } assert!( error.to_string().contains(expected_error), "unexpected initialize rejection: {error:?}" @@ -473,16 +485,19 @@ async fn v2_conductor_rejects_invalid_protocol_versions_with_responses() -> Resu assert_raw_initialize_rejected( serde_json::json!({}), "protocolVersion must be a valid ACP protocol version", + -32602, ) .await?; assert_raw_initialize_rejected( serde_json::json!({ "protocolVersion": "2" }), "protocolVersion must be a valid ACP protocol version", + -32602, ) .await?; assert_raw_initialize_rejected( serde_json::json!({ "protocolVersion": 0 }), - "unsupported ACP protocol version 0", + "Unsupported protocol version", + -32022, ) .await } diff --git a/src/agent-client-protocol/src/jsonrpc.rs b/src/agent-client-protocol/src/jsonrpc.rs index 17cc1290..85ac9199 100644 --- a/src/agent-client-protocol/src/jsonrpc.rs +++ b/src/agent-client-protocol/src/jsonrpc.rs @@ -45,6 +45,8 @@ pub use crate::jsonrpc::handlers::NullHandler; use crate::jsonrpc::handlers::{ChainedHandler, NamedHandler}; use crate::jsonrpc::handlers::{MessageHandler, NotificationHandler, RequestHandler}; use crate::jsonrpc::outgoing_actor::{OutgoingMessageTx, send_raw_message}; +#[cfg(feature = "unstable_protocol_v2")] +pub(crate) use crate::jsonrpc::protocol_compat::normalize_v1_initialize_request; use crate::jsonrpc::protocol_compat::{ProtocolCompat, ProtocolMode}; pub use crate::jsonrpc::raw_error::{RawJsonRpcError, RawJsonRpcResponse}; use crate::jsonrpc::run::SpawnedRun; diff --git a/src/agent-client-protocol/src/jsonrpc/protocol_compat.rs b/src/agent-client-protocol/src/jsonrpc/protocol_compat.rs index 536a74c7..8c963c0a 100644 --- a/src/agent-client-protocol/src/jsonrpc/protocol_compat.rs +++ b/src/agent-client-protocol/src/jsonrpc/protocol_compat.rs @@ -1,107 +1,57 @@ -#[cfg(not(feature = "unstable_protocol_v2"))] -mod imp { - #![allow(clippy::unused_self, clippy::unnecessary_wraps)] - use crate::schema::v1::RequestId; - use crate::{UntypedMessage, role::RemoteStyle}; - - #[derive(Clone, Copy, Debug, Default)] - pub(crate) struct ProtocolMode; - - impl ProtocolMode { - pub(crate) fn disabled() -> Self { - Self - } - - pub(crate) fn v1_agent() -> Self { - Self - } - - pub(crate) fn v1_client() -> Self { - Self - } - - pub(crate) fn v1_proxy() -> Self { - Self - } - - pub(crate) fn merge(self, _other: Self) -> Self { - self +/// Normalize a newer peer's initialization into the stable v1 surface. +/// +/// Only shared capabilities are carried across. In particular, unknown newer +/// fields must never enable the v1 file-system or terminal-execution surfaces. +pub(crate) fn normalize_v1_initialize_request( + params: &serde_json::Value, +) -> Result, crate::Error> { + use crate::schema::{ProtocolVersion, v1}; + + let info: v1::Implementation = + serde_json::from_value(params.get("info").cloned().unwrap_or_default())?; + let mut capabilities = serde_json::Map::new(); + if let Some(source) = params + .get("capabilities") + .and_then(serde_json::Value::as_object) + { + for key in ["auth", "elicitation", "nes", "positionEncodings", "_meta"] { + if let Some(value) = source.get(key) { + capabilities.insert(key.into(), value.clone()); + } } } - - #[derive(Clone, Debug, Default)] - pub(crate) struct ProtocolCompat; - - impl ProtocolCompat { - pub(crate) fn new(_mode: ProtocolMode) -> Self { - Self - } - - pub(crate) fn incoming_message( - &self, - message: UntypedMessage, - ) -> Result { - Ok(message) - } - - pub(crate) fn incoming_request( - &self, - _id: &RequestId, - message: UntypedMessage, - ) -> Result { - self.incoming_message(message) - } - - pub(crate) fn outgoing_message( - &self, - message: UntypedMessage, - _remote_style: RemoteStyle, - ) -> Result { - Ok(message) - } - - pub(crate) fn incoming_notification( - &self, - message: UntypedMessage, - ) -> Result, crate::Error> { - Ok(vec![message]) - } - - pub(crate) fn outgoing_notification( - &self, - message: UntypedMessage, - ) -> Result, crate::Error> { - Ok(vec![message]) - } - - pub(crate) fn incoming_response( - &self, - _method: &str, - result: Result, - ) -> Result { - result - } - - pub(crate) fn outgoing_response( - &self, - _method: &str, - result: Result, - ) -> Result { - result - } - - pub(crate) fn outgoing_response_to( - &self, - _id: &RequestId, - method: &str, - result: Result, - ) -> Result { - self.outgoing_response(method, result) - } + if let Some(auth) = capabilities + .get_mut("auth") + .and_then(serde_json::Value::as_object_mut) + { + let terminal = auth + .remove("terminal") + .is_some_and(|value| value.is_object()); + auth.insert("terminal".into(), terminal.into()); } + capabilities.insert( + "session".into(), + serde_json::json!({ "configOptions": { "boolean": {} } }), + ); + let mut target = serde_json::json!({ + "protocolVersion": ProtocolVersion::V1, + "clientInfo": info, + "clientCapabilities": capabilities, + }); + if let Some(meta) = params.get("_meta") { + target["_meta"] = meta.clone(); + } + let initialize: v1::InitializeRequest = serde_json::from_value(target)?; + let serde_json::Value::Object(normalized) = + serde_json::to_value(initialize).map_err(crate::Error::into_internal_error)? + else { + return Err(crate::util::internal_error( + "initialize params did not serialize to an object", + )); + }; + Ok(normalized) } -#[cfg(feature = "unstable_protocol_v2")] mod imp { use std::sync::{Arc, Mutex}; @@ -128,6 +78,7 @@ mod imp { } #[derive(Clone, Copy, Debug, PartialEq, Eq)] + #[cfg_attr(not(feature = "unstable_protocol_v2"), allow(dead_code))] enum InitializationRole { /// Preserve the v1 and proxy initialization behavior. A v2 proxy has /// both an incoming predecessor initialization and an outgoing @@ -190,6 +141,7 @@ mod imp { }) } + #[cfg(feature = "unstable_protocol_v2")] pub(crate) fn v2_agent() -> Self { Self::Acp(AcpProtocolMode { api: ProtocolVersionKind::V2, @@ -198,6 +150,7 @@ mod imp { }) } + #[cfg(feature = "unstable_protocol_v2")] pub(crate) fn v2_client() -> Self { Self::Acp(AcpProtocolMode { api: ProtocolVersionKind::V2, @@ -206,6 +159,7 @@ mod imp { }) } + #[cfg(feature = "unstable_protocol_v2")] pub(crate) fn v2_proxy() -> Self { Self::Acp(AcpProtocolMode { api: ProtocolVersionKind::V2, @@ -239,6 +193,7 @@ mod imp { } } + #[cfg(feature = "unstable_protocol_v2")] pub(crate) fn api_protocol_version(self) -> Option { match self { Self::Disabled => None, @@ -279,14 +234,14 @@ mod imp { fn as_protocol_version(self) -> ProtocolVersion { match self { Self::V1 => ProtocolVersion::V1, - Self::V2 => ProtocolVersion::V2, + Self::V2 => ProtocolVersion::from(2_u16), } } fn from_protocol_version(version: ProtocolVersion) -> Option { if version == ProtocolVersion::V1 { Some(Self::V1) - } else if version == ProtocolVersion::V2 { + } else if version == ProtocolVersion::from(2_u16) { Some(Self::V2) } else { None @@ -319,7 +274,7 @@ mod imp { } } - #[cfg(test)] + #[cfg(all(test, feature = "unstable_protocol_v2"))] pub(crate) fn incoming_message( &self, message: UntypedMessage, @@ -432,7 +387,7 @@ mod imp { Ok(value) } - #[cfg(test)] + #[cfg(all(test, feature = "unstable_protocol_v2"))] pub(crate) fn outgoing_response( &self, method: &str, @@ -503,9 +458,23 @@ mod imp { mut message: UntypedMessage, ) -> Result { let requested = required_protocol_version_from_value(message.params())?; - let requested_kind = ProtocolVersionKind::from_protocol_version(requested) - .ok_or_else(|| unsupported_protocol_version(requested, mode.api))?; - if requested_kind != mode.api { + if mode.initialize_surface == InitializeSurface::Peer { + // V1 retains its original fallback-to-latest bootstrap. Only + // v2 endpoints reject a version below their supported range. + if mode.api == ProtocolVersionKind::V2 && requested < mode.api.as_protocol_version() + { + return Err(crate::util::unsupported_protocol_version( + requested, + &[mode.api.as_protocol_version()], + )); + } + if mode.api == ProtocolVersionKind::V1 && requested > ProtocolVersion::V1 { + message.params = serde_json::Value::Object( + super::normalize_v1_initialize_request(message.params())?, + ); + } + } else if ProtocolVersionKind::from_protocol_version(requested) != Some(mode.api) { + // The conductor has already selected a proxy's wire version. return Err(unsupported_protocol_version(requested, mode.api)); } @@ -810,31 +779,44 @@ mod imp { Ok(()) } + // The shared guard stays fallible when v2 validation is compiled out. + #[cfg_attr( + not(feature = "unstable_protocol_v2"), + allow(clippy::unnecessary_wraps) + )] fn validate_native_v2_initialize_request( mode: AcpProtocolMode, value: &serde_json::Value, ) -> Result<(), crate::Error> { - if mode.initialization_role == InitializationRole::Unchecked { - return Ok(()); + #[cfg(feature = "unstable_protocol_v2")] + if mode.initialization_role != InitializationRole::Unchecked { + ::parse_message( + "initialize", + value, + )?; } - ::parse_message( - "initialize", - value, - )?; + #[cfg(not(feature = "unstable_protocol_v2"))] + let _ = (mode, value); Ok(()) } + #[cfg_attr( + not(feature = "unstable_protocol_v2"), + allow(clippy::unnecessary_wraps) + )] fn validate_native_v2_initialize_response( mode: AcpProtocolMode, value: &serde_json::Value, ) -> Result<(), crate::Error> { - if mode.initialization_role == InitializationRole::Unchecked { - return Ok(()); + #[cfg(feature = "unstable_protocol_v2")] + if mode.initialization_role != InitializationRole::Unchecked { + ::from_value( + "initialize", + value.clone(), + )?; } - ::from_value( - "initialize", - value.clone(), - )?; + #[cfg(not(feature = "unstable_protocol_v2"))] + let _ = (mode, value); Ok(()) } @@ -875,7 +857,7 @@ mod imp { )) } - #[cfg(test)] + #[cfg(all(test, feature = "unstable_protocol_v2"))] mod tests { use super::*; use agent_client_protocol_schema::v2; @@ -1423,6 +1405,31 @@ mod imp { Ok(()) } + #[test] + fn incoming_initialize_request_negotiates_newer_protocol_versions() + -> Result<(), crate::Error> { + for requested in [2_u16, 3, u16::MAX] { + let compat = ProtocolCompat::new(ProtocolMode::v2_agent()); + let request = compat.incoming_message(UntypedMessage::new( + "initialize", + v2_initialize_request(ProtocolVersion::from(requested)), + )?)?; + assert_eq!(request.params()["protocolVersion"], serde_json::json!(2)); + assert_eq!(pending_initialize(&compat), Some(ProtocolVersionKind::V2)); + + let response = compat.outgoing_response( + "initialize", + Ok(serde_json::to_value(v2_initialize_response( + ProtocolVersion::V2, + ))?), + )?; + assert_eq!(response["protocolVersion"], serde_json::json!(2)); + assert_eq!(initialization_state(&compat), InitializationState::Ready); + assert_eq!(compat.active_wire_version(), ProtocolVersionKind::V2); + } + Ok(()) + } + #[test] fn incoming_initialize_request_rejects_unsupported_protocol_version() -> Result<(), crate::Error> { @@ -1430,20 +1437,20 @@ mod imp { let error = compat .incoming_message(UntypedMessage::new( "initialize", - v2_initialize_request(ProtocolVersion::V1), + serde_json::json!({ "protocolVersion": 1, "clientCapabilities": {} }), )?) .expect_err("v2 agents should reject v1 initialization without a v1 handler"); - let data = error - .data - .as_ref() - .and_then(|data| data.as_str()) - .unwrap_or_default(); - assert!( - data.contains("only supports ACP protocol version 2"), - "{error:?}" + assert_eq!(i32::from(error.code), -32022); + assert_eq!( + error.data, + Some(serde_json::json!({ "requested": 1, "supported": [2] })) ); assert_eq!(negotiated(&compat), ProtocolVersionKind::V2); assert_eq!(compat.active_wire_version(), ProtocolVersionKind::V2); + assert_eq!( + initialization_state(&compat), + InitializationState::Uninitialized + ); Ok(()) } diff --git a/src/agent-client-protocol/src/role/acp.rs b/src/agent-client-protocol/src/role/acp.rs index cb504821..8e696e2d 100644 --- a/src/agent-client-protocol/src/role/acp.rs +++ b/src/agent-client-protocol/src/role/acp.rs @@ -343,8 +343,9 @@ impl Agent { /// Create an agent builder that uses the ACP protocol v2 API. /// - /// This builder requires clients to negotiate protocol v2 during - /// initialization. Use a v1 builder for v1 clients. + /// This builder selects protocol v2 for clients requesting v2 or a newer + /// version with compatible initialization parameters. Use + /// [`Self::protocol_router`] with a v1 implementation for v1 clients. /// /// Requires the `unstable_protocol_v2` crate feature. #[cfg(feature = "unstable_protocol_v2")] @@ -357,10 +358,10 @@ impl Agent { /// Add implementations with [`AgentProtocolRouter::with_v1`] and /// [`AgentProtocolRouter::with_v2`]. /// The resulting router reads the initial - /// `initialize` request, selects the highest configured implementation - /// compatible with the client's requested protocol version, then forwards - /// the connection to that implementation. It does not convert traffic - /// between protocol versions after routing. + /// `initialize` request and selects a version-specific implementation. An + /// endpoint exposing v2 selects the highest configured version at or below + /// the requested version. A v1-only endpoint retains v1's fallback-to-latest + /// rule. The router does not convert traffic after initialization. /// /// Requires the `unstable_protocol_v2` crate feature while protocol v2 /// stabilizes. @@ -491,7 +492,11 @@ struct SupportedProtocols { #[cfg(feature = "unstable_protocol_v2")] impl SupportedProtocols { - fn highest_compatible(self, requested: ProtocolVersion) -> Option { + fn select_for_agent(self, requested: ProtocolVersion) -> Option { + if self.v1 && !self.v2 { + // A v1-only endpoint retains the legacy fallback-to-latest rule. + return Some(SelectedProtocol::V1); + } if self.v2 && requested >= ProtocolVersion::V2 { return Some(SelectedProtocol::V2); } @@ -513,6 +518,16 @@ impl SupportedProtocols { } } + fn versions(self) -> Vec { + [ + self.v1.then_some(ProtocolVersion::V1), + self.v2.then_some(ProtocolVersion::V2), + ] + .into_iter() + .flatten() + .collect() + } + fn description(self) -> String { match (self.v1, self.v2) { (true, true) => "ACP protocol versions 1 and 2".into(), @@ -547,7 +562,7 @@ fn select_agent_protocol( let requested = serde_json::from_value::(protocol_version.clone()) .map_err(|_| invalid_initialize_protocol_version())?; - let selected = highest_compatible_agent_protocol(requested, supported)?; + let selected = select_configured_agent_protocol(requested, supported)?; rewrite_initialize_params(params, requested, selected)?; Ok(selected) @@ -628,8 +643,13 @@ fn rewrite_initialize_params( match selected { SelectedProtocol::V1 => { - debug_assert!(requested >= ProtocolVersion::V2); - *params = normalize_v2_initialize_params_for_v1(params, false)?; + if requested > ProtocolVersion::V1 { + *params = normalize_v2_initialize_params_for_v1(params, false)?; + } else { + let mut initialize = parse_initialize_params::(params)?; + initialize.protocol_version = ProtocolVersion::V1; + *params = serialize_initialize_params(initialize)?; + } Ok(()) } SelectedProtocol::V2 => { @@ -646,10 +666,14 @@ fn normalize_v2_initialize_params_for_v1( params: &serde_json::Map, require_lossless: bool, ) -> Result, crate::Error> { - // Canonicalize through v2 first so tolerant field semantics are applied. - // A lossless result is only required by the connection-reuse probe. Normal - // v1 routing may discard fields that have no meaning in the selected - // protocol version. + if !require_lossless { + return crate::jsonrpc::normalize_v1_initialize_request(&serde_json::Value::Object( + params.clone(), + )); + } + // The connection-reuse probe additionally requires lossless v2 + // canonicalization. Normal v1 routing above uses the same bootstrap as a + // standalone v1 agent, even when the v2 schema is not compiled in. let initialize = parse_initialize_params::(params)?; let mut target = serialize_initialize_params(initialize)?; if require_lossless && target != *params { @@ -864,16 +888,13 @@ fn json_object_contains( } #[cfg(feature = "unstable_protocol_v2")] -fn highest_compatible_agent_protocol( +fn select_configured_agent_protocol( requested: ProtocolVersion, supported: SupportedProtocols, ) -> Result { - supported.highest_compatible(requested).ok_or_else(|| { - crate::Error::invalid_request().data(format!( - "unsupported ACP protocol version {requested}; this endpoint supports {}", - supported.description() - )) - }) + supported + .select_for_agent(requested) + .ok_or_else(|| crate::util::unsupported_protocol_version(requested, &supported.versions())) } #[cfg(feature = "unstable_protocol_v2")] @@ -974,8 +995,8 @@ async fn reject_initialize( let drain_incoming = async move { // Later input has no protocol meaning once initialization is rejected. - // Keep draining it only so the transport can flush the queued rejection; - // treating a malformed trailing frame as fatal would cancel that flush. + // Keep draining it only so the transport can flush the queued rejection. + // Treating a malformed trailing frame as fatal would cancel that flush. while rx.next().await.is_some() {} Ok::<_, crate::Error>(()) }; diff --git a/src/agent-client-protocol/src/util.rs b/src/agent-client-protocol/src/util.rs index ed704bcb..fb495347 100644 --- a/src/agent-client-protocol/src/util.rs +++ b/src/agent-client-protocol/src/util.rs @@ -67,6 +67,19 @@ pub fn parse_error(message: impl ToString) -> crate::Error { crate::Error::parse_error().data(message.to_string()) } +/// No version supported by a v2 endpoint is at or below the requested version. +/// +/// Uses the protocol wire code so legacy v1 error models can carry the v2 +/// diagnostic as an unknown error code without changing the v1 contract. +#[must_use] +pub fn unsupported_protocol_version( + requested: crate::schema::ProtocolVersion, + supported: &[crate::schema::ProtocolVersion], +) -> crate::Error { + crate::Error::new(-32022, "Unsupported protocol version") + .data(serde_json::json!({ "requested": requested, "supported": supported })) +} + pub(crate) fn instrumented_with_connection_name( name: String, task: F, diff --git a/src/agent-client-protocol/tests/protocol_negotiation.rs b/src/agent-client-protocol/tests/protocol_negotiation.rs new file mode 100644 index 00000000..5486c996 --- /dev/null +++ b/src/agent-client-protocol/tests/protocol_negotiation.rs @@ -0,0 +1,213 @@ +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + +use agent_client_protocol::{ + Agent, Channel, Client, ConnectTo, Error, RawJsonRpcMessage, RawJsonRpcResponse, + TransportFrame, + schema::{ProtocolVersion, v1}, +}; +use futures::StreamExt as _; +use serde_json::{Value, json}; + +async fn request( + channel: &mut Channel, + id: i64, + method: &str, + params: Value, +) -> Result { + channel + .tx + .unbounded_send(TransportFrame::Single(RawJsonRpcMessage::request( + method.into(), + params, + v1::RequestId::Number(id), + )?)) + .map_err(Error::into_internal_error)?; + let frame = tokio::time::timeout(std::time::Duration::from_secs(5), channel.rx.next()) + .await + .map_err(Error::into_internal_error)? + .expect("agent must respond before closing"); + match frame { + TransportFrame::Single(RawJsonRpcMessage::Response(RawJsonRpcResponse::Result { + id: response_id, + result, + .. + })) => { + assert_eq!(response_id, v1::RequestId::Number(id)); + Ok(json!({ "result": result })) + } + TransportFrame::Single(RawJsonRpcMessage::Response(RawJsonRpcResponse::Error { + id: response_id, + error, + .. + })) => { + assert_eq!(response_id, v1::RequestId::Number(id)); + Ok(json!({ + "error": { + "code": error.code, + "message": error.message, + "data": error.data.value(), + } + })) + } + other => panic!("expected a response for {method}, got {other:?}"), + } +} + +// This file deliberately runs both with and without unstable_protocol_v2. +#[tokio::test(flavor = "current_thread")] +async fn standalone_v1_keeps_legacy_selection_and_serves_v1_sessions() -> Result<(), Error> { + for requested in [0_u16, 1, 2, 3, u16::MAX] { + let initialized = Arc::new(AtomicUsize::new(0)); + let calls = initialized.clone(); + let agent = Agent + .builder() + .on_receive_request( + async move |initialize: v1::InitializeRequest, responder, _cx| { + calls.fetch_add(1, Ordering::SeqCst); + assert_eq!(initialize.protocol_version, ProtocolVersion::V1); + if requested >= 2 { + assert_eq!( + initialize + .client_info + .as_ref() + .map(|info| info.name.as_str()), + Some("newer-client") + ); + assert!(initialize.client_capabilities.auth.terminal); + assert!(!initialize.client_capabilities.fs.read_text_file); + assert!(!initialize.client_capabilities.fs.write_text_file); + assert!(!initialize.client_capabilities.terminal); + let wire = serde_json::to_value(&initialize)?; + assert_eq!(wire["_meta"], json!({ "request": true })); + assert_eq!( + wire["clientCapabilities"]["_meta"], + json!({ "capability": true }) + ); + assert_eq!( + wire["clientCapabilities"]["session"]["configOptions"]["boolean"], + json!({}) + ); + } + responder.respond(v1::InitializeResponse::new(initialize.protocol_version)) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async |_request: v1::NewSessionRequest, responder, _cx| { + responder.respond(v1::NewSessionResponse::new(v1::SessionId::new("native-v1"))) + }, + agent_client_protocol::on_receive_request!(), + ); + let (mut channel, driver) = ConnectTo::::into_channel_and_future(agent); + let task = tokio::spawn(driver.expect("agent owns a driver")); + let params = if requested < 2 { + json!({ "protocolVersion": requested, "clientCapabilities": {} }) + } else { + json!({ + "protocolVersion": requested, + "info": { "name": "newer-client", "version": "1" }, + "capabilities": { + "auth": { "terminal": { "_meta": { "marker": true } } }, + "_meta": { "capability": true }, + "fs": { "readTextFile": true, "writeTextFile": true }, + "terminal": true, + }, + "_meta": { "request": true }, + }) + }; + let initialize = request(&mut channel, 1, "initialize", params).await?; + assert_eq!(initialize["result"]["protocolVersion"], 1); + assert_eq!(initialized.load(Ordering::SeqCst), 1); + if requested >= 1 { + let params = serde_json::to_value(v1::NewSessionRequest::new( + std::env::current_dir().map_err(Error::into_internal_error)?, + ))?; + let session = request(&mut channel, 2, "session/new", params).await?; + assert_eq!(session["result"]["sessionId"], "native-v1"); + } + task.abort(); + } + Ok(()) +} + +#[cfg(feature = "unstable_protocol_v2")] +#[tokio::test(flavor = "current_thread")] +async fn v1_only_router_preserves_legacy_fallback() -> Result<(), Error> { + let agent = Agent + .protocol_router() + .with_v1(Agent.builder().on_receive_request( + async |initialize: v1::InitializeRequest, responder, _cx| { + assert_eq!(initialize.protocol_version, ProtocolVersion::V1); + responder.respond(v1::InitializeResponse::new(initialize.protocol_version)) + }, + agent_client_protocol::on_receive_request!(), + )); + let (mut channel, driver) = ConnectTo::::into_channel_and_future(agent); + let task = tokio::spawn(driver.expect("router owns a driver")); + let response = request( + &mut channel, + 1, + "initialize", + json!({ "protocolVersion": 0, "clientCapabilities": {} }), + ) + .await?; + assert_eq!(response["result"]["protocolVersion"], 1); + task.abort(); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn standalone_v1_rejects_malformed_initialization_before_dispatch() -> Result<(), Error> { + for params in [ + json!({}), + json!({ "protocolVersion": "1" }), + json!({ "protocolVersion": -1 }), + json!({ "protocolVersion": 1.5 }), + json!({ "protocolVersion": 65536 }), + json!({ "protocolVersion": 2 }), + json!({ "protocolVersion": 2, "info": { "name": 123, "version": "1" } }), + ] { + let agent = Agent.builder().on_receive_request( + async |_initialize: v1::InitializeRequest, responder, _cx| { + responder + .respond_with_internal_error("malformed initialization reached the handler") + }, + agent_client_protocol::on_receive_request!(), + ); + let (mut channel, driver) = ConnectTo::::into_channel_and_future(agent); + let task = tokio::spawn(driver.expect("agent owns a driver")); + let response = request(&mut channel, 1, "initialize", params).await?; + assert_eq!(response["error"]["code"], -32602, "{response}"); + task.abort(); + } + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn standalone_v1_proxy_requires_exact_selected_version() -> Result<(), Error> { + for requested in [0, 2, 3] { + let (mut channel, driver) = + ConnectTo::::into_channel_and_future( + agent_client_protocol::Proxy.builder(), + ); + let task = tokio::spawn(driver.expect("proxy owns a driver")); + let response = request( + &mut channel, + 1, + "_proxy/initialize", + json!({ "protocolVersion": requested }), + ) + .await?; + assert_eq!(response["error"]["code"], -32600); + assert!( + response["error"]["data"] + .as_str() + .is_some_and(|data| data.contains("only supports ACP protocol version 1")) + ); + task.abort(); + } + Ok(()) +} diff --git a/src/agent-client-protocol/tests/protocol_v2.rs b/src/agent-client-protocol/tests/protocol_v2.rs index 60ad937d..04f28fb1 100644 --- a/src/agent-client-protocol/tests/protocol_v2.rs +++ b/src/agent-client-protocol/tests/protocol_v2.rs @@ -68,6 +68,15 @@ fn cwd() -> Result { std::env::current_dir().map_err(Error::into_internal_error) } +fn assert_version_mismatch(error: &Error, requested: u16, supported: &[u16]) { + assert_eq!(i32::from(error.code), -32022); + assert_eq!(error.message, "Unsupported protocol version"); + assert_eq!( + error.data, + Some(serde_json::json!({ "requested": requested, "supported": supported })) + ); +} + #[allow(dead_code)] fn spawn_child_from_generic_context( parent: &ConnectionTo, @@ -329,7 +338,7 @@ impl ConnectTo for ExpectingV2InitializeErrorClient { .and_then(|data| data.as_str()) .unwrap_or_default(); assert!( - data.contains("only supports ACP protocol version 1"), + data.contains("deliberate initialize rejection"), "{error:?}" ); Ok(()) @@ -530,7 +539,7 @@ async fn assert_malformed_initialize_rejected(params: Map) -> Res )) } -async fn assert_v2_client_rejected_by_v1_agent(agent: impl ConnectTo) -> Result<(), Error> { +async fn assert_v2_client_rejects_v1_selection(agent: impl ConnectTo) -> Result<(), Error> { Client .v2() .connect_with(agent, async |cx| { @@ -538,16 +547,13 @@ async fn assert_v2_client_rejected_by_v1_agent(agent: impl ConnectTo) -> .send_request(v2_initialize_request(ProtocolVersion::V2)) .block_task() .await - .expect_err("v1 agent protocol mode should reject v2 clients"); + .expect_err("a v2-only client should reject a negotiated v1 response"); let data = error .data .as_ref() .and_then(|data| data.as_str()) .unwrap_or_default(); - assert!( - data.contains("only supports ACP protocol version 1"), - "{error:?}" - ); + assert!(data.contains("peer negotiated 1"), "{error:?}"); Ok(()) }) .await @@ -603,7 +609,7 @@ async fn v2_agent_rejects_initialize_with_malformed_protocol_version() -> Result } #[tokio::test(flavor = "current_thread")] -async fn role_builder_v1_agent_rejects_v2_client_negotiation() -> Result<(), Error> { +async fn role_builder_v1_agent_negotiates_v1_for_v2_client() -> Result<(), Error> { let agent = ::builder(Agent).on_receive_request( async |initialize: v1::InitializeRequest, responder, _cx| { assert_eq!(initialize.protocol_version, ProtocolVersion::V1); @@ -612,11 +618,11 @@ async fn role_builder_v1_agent_rejects_v2_client_negotiation() -> Result<(), Err agent_client_protocol::on_receive_request!(), ); - assert_v2_client_rejected_by_v1_agent(agent).await + assert_v2_client_rejects_v1_selection(agent).await } #[tokio::test(flavor = "current_thread")] -async fn builder_new_v1_agent_rejects_v2_client_negotiation() -> Result<(), Error> { +async fn builder_new_v1_agent_negotiates_v1_for_v2_client() -> Result<(), Error> { let agent = Builder::new(Agent).on_receive_request( async |initialize: v1::InitializeRequest, responder, _cx| { assert_eq!(initialize.protocol_version, ProtocolVersion::V1); @@ -625,11 +631,11 @@ async fn builder_new_v1_agent_rejects_v2_client_negotiation() -> Result<(), Erro agent_client_protocol::on_receive_request!(), ); - assert_v2_client_rejected_by_v1_agent(agent).await + assert_v2_client_rejects_v1_selection(agent).await } #[tokio::test(flavor = "current_thread")] -async fn builder_new_with_v1_agent_rejects_v2_client_negotiation() -> Result<(), Error> { +async fn builder_new_with_v1_agent_negotiates_v1_for_v2_client() -> Result<(), Error> { let agent = Builder::new_with(Agent, NullHandler).on_receive_request( async |initialize: v1::InitializeRequest, responder, _cx| { assert_eq!(initialize.protocol_version, ProtocolVersion::V1); @@ -638,7 +644,7 @@ async fn builder_new_with_v1_agent_rejects_v2_client_negotiation() -> Result<(), agent_client_protocol::on_receive_request!(), ); - assert_v2_client_rejected_by_v1_agent(agent).await + assert_v2_client_rejects_v1_selection(agent).await } #[tokio::test(flavor = "current_thread")] @@ -657,15 +663,8 @@ async fn role_builder_v1_client_is_rejected_by_v2_agent() -> Result<(), Error> { .block_task() .await .expect_err("v2 agents require a v2 client implementation"); - let data = error - .data - .as_ref() - .and_then(|data| data.as_str()) - .unwrap_or_default(); - assert!( - data.contains("only supports ACP protocol version 2"), - "{error:?}" - ); + assert_version_mismatch(&error, 1, &[2]); + assert_eq!(error.code, agent_client_protocol::ErrorCode::Other(-32022)); Ok(()) }) .await @@ -1143,10 +1142,7 @@ async fn v2_client_rejects_v1_agent() -> Result<(), Error> { .as_ref() .and_then(|data| data.as_str()) .unwrap_or_default(); - assert!( - data.contains("only supports ACP protocol version 1"), - "{error:?}" - ); + assert!(data.contains("peer negotiated 1"), "{error:?}"); Ok(()) }) .await @@ -1567,7 +1563,26 @@ async fn client_protocol_connector_does_not_retry_after_v2_initialize_rejection( .protocol_connector() .with_v1(|| RejectingV1Client) .with_v2(|| ExpectingV2InitializeErrorClient) - .connect_to(|| v1_agent_with_session("v1-client-connector-session")) + .connect_to(|| { + Agent.builder().on_receive_request( + async |_request: v1::InitializeRequest, responder, _cx| { + responder.respond_with_error( + Error::invalid_request().data("deliberate initialize rejection"), + ) + }, + agent_client_protocol::on_receive_request!(), + ) + }) + .await +} + +#[tokio::test(flavor = "current_thread")] +async fn client_protocol_connector_falls_back_to_standalone_v1_agent() -> Result<(), Error> { + Client + .protocol_connector() + .with_v1(|| InitializingV1Client::new("v1-native-session")) + .with_v2(|| InitializingV2Client::new("v2-unused-session")) + .connect_to(|| v1_agent_with_session("v1-native-session")) .await } @@ -2041,14 +2056,10 @@ async fn protocol_router_v2_only_rejects_v1_client() -> Result<(), Error> { let RawJsonRpcMessage::Response(RawJsonRpcResponse::Error { error, .. }) = message else { continue; }; - let data = error - .data - .value() - .and_then(|data| data.as_str()) - .unwrap_or_default(); - assert!( - data.contains("supports ACP protocol version 2"), - "{error:?}" + assert_eq!(error.code, -32022); + assert_eq!( + error.data.value(), + Some(&serde_json::json!({ "requested": 1, "supported": [2] })) ); agent_task.abort(); return Ok(()); @@ -2079,15 +2090,7 @@ async fn protocol_router_rejection_is_initialize_request_error() -> Result<(), E .block_task() .await .expect_err("v1 initialize should be rejected by the v2-only router"); - let data = error - .data - .as_ref() - .and_then(|data| data.as_str()) - .unwrap_or_default(); - assert!( - data.contains("supports ACP protocol version 2"), - "{error:?}" - ); + assert_version_mismatch(&error, 1, &[2]); Ok(()) }) .await @@ -2119,15 +2122,7 @@ async fn protocol_router_rejection_flushes_over_byte_streams() -> Result<(), Err .block_task() .await .expect_err("v1 initialize should be rejected by the v2-only router"); - let data = error - .data - .as_ref() - .and_then(|data| data.as_str()) - .unwrap_or_default(); - assert!( - data.contains("supports ACP protocol version 2"), - "{error:?}" - ); + assert_version_mismatch(&error, 1, &[2]); Ok(()) }) .await?; @@ -2179,11 +2174,10 @@ async fn protocol_router_rejection_flushes_with_trailing_malformed_input() -> Re let response = read_wire_json(&mut client_reader).await?; assert_eq!(response["id"], 1); - assert!( - response["error"]["data"] - .as_str() - .is_some_and(|data| data.contains("supports ACP protocol version 2")), - "{response:?}" + assert_eq!(response["error"]["code"], -32022); + assert_eq!( + response["error"]["data"], + serde_json::json!({ "requested": 1, "supported": [2] }) ); agent_task @@ -2734,6 +2728,147 @@ async fn v2_protocol_router_accepts_inbound_json_rpc_batches() -> Result<(), Err Ok(()) } +#[tokio::test(flavor = "current_thread")] +async fn protocol_router_mismatch_flushes_without_starting_application() -> Result<(), Error> { + use tokio::io::BufReader; + + for batched in [false, true] { + let started = Arc::new(AtomicBool::new(false)); + let runner_started = started.clone(); + let handler_started = started.clone(); + let agent = Agent.protocol_router().with_v2( + Agent + .v2() + .with_spawned(async move |_cx| { + runner_started.store(true, Ordering::SeqCst); + Ok(()) + }) + .on_receive_request( + async move |_request: v2::NewSessionRequest, responder, _cx| { + handler_started.store(true, Ordering::SeqCst); + responder.respond_with_internal_error("session handler must not run") + }, + agent_client_protocol::on_receive_request!(), + ), + ); + let (mut client_writer, server_reader) = tokio::io::duplex(4096); + let (server_writer, client_reader) = tokio::io::duplex(1); + let transport = ByteStreams::new(server_writer.compat_write(), server_reader.compat()); + let agent_task = tokio::spawn(agent.connect_to(transport)); + let mut client_reader = BufReader::new(client_reader); + + let initialize = serde_json::json!({ + "jsonrpc": "2.0", + "id": "initialize-id", + "method": "initialize", + "params": { "protocolVersion": 1, "clientCapabilities": {} }, + }); + let request = if batched { + serde_json::json!([ + initialize, + { + "jsonrpc": "2.0", + "id": "session-id", + "method": "session/new", + "params": { "cwd": cwd()?, "mcpServers": [] }, + }, + ]) + } else { + initialize + }; + write_wire_json(&mut client_writer, &request).await?; + let wire_response = read_wire_json(&mut client_reader).await?; + let initialize_response = if batched { + assert_eq!(wire_response[1]["id"], "session-id"); + assert_eq!(wire_response[1]["error"]["code"], -32022); + &wire_response[0] + } else { + &wire_response + }; + assert_eq!(initialize_response["id"], "initialize-id"); + assert_eq!( + initialize_response["error"], + serde_json::json!({ + "code": -32022, + "message": "Unsupported protocol version", + "data": { "requested": 1, "supported": [2] }, + }) + ); + tokio::time::timeout(tokio::time::Duration::from_secs(10), agent_task) + .await + .map_err(Error::into_internal_error)? + .map_err(Error::into_internal_error)??; + assert!(!started.load(Ordering::SeqCst)); + } + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn standalone_v2_agent_negotiates_future_protocol_version() -> Result<(), Error> { + let agent = Agent + .v2() + .on_receive_request( + async |initialize: v2::InitializeRequest, responder, _cx| { + assert_eq!(initialize.protocol_version, ProtocolVersion::V2); + responder.respond(v2_initialize_response_with_session(ProtocolVersion::V2)) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async |_request: v2::NewSessionRequest, responder, _cx| { + responder.respond(v2::NewSessionResponse::new(v2::SessionId::new( + "standalone-v2-session", + ))) + }, + agent_client_protocol::on_receive_request!(), + ); + let (mut channel, agent_future) = ConnectTo::::into_channel_and_future(agent); + let agent_task = tokio::spawn(agent_future.expect("v2 agent owns a connection driver")); + + for (id, method, params) in [ + ( + 1, + "initialize", + json_value(v2_initialize_request(ProtocolVersion::from(3_u16)))?, + ), + ( + 2, + "session/new", + json_value(v2::NewSessionRequest::new(cwd()?))?, + ), + ] { + channel + .tx + .unbounded_send(TransportFrame::Single(RawJsonRpcMessage::request( + method.into(), + params, + v1::RequestId::Number(id), + )?)) + .map_err(Error::into_internal_error)?; + let frame = tokio::time::timeout(tokio::time::Duration::from_secs(10), channel.rx.next()) + .await + .map_err(Error::into_internal_error)? + .expect("agent must respond before closing"); + let TransportFrame::Single(RawJsonRpcMessage::Response(RawJsonRpcResponse::Result { + id: response_id, + result, + .. + })) = frame + else { + panic!("expected a successful {method} response, got {frame:?}"); + }; + assert_eq!(response_id, v1::RequestId::Number(id)); + if method == "initialize" { + assert_eq!(result["protocolVersion"], serde_json::json!(2)); + } else { + assert_eq!(result["sessionId"], "standalone-v2-session"); + } + } + + agent_task.abort(); + Ok(()) +} + #[tokio::test(flavor = "current_thread")] async fn protocol_router_routes_future_protocol_version_to_v2() -> Result<(), Error> { let agent = Agent