From be7c6e37c44d961895d6f5c2203096ac72b89734 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Thu, 13 Aug 2026 17:42:04 -0700 Subject: [PATCH] feat(acp): add proxy protocol router for v2 --- README.md | 8 +- md/conductor.md | 7 +- md/mcp-bridge.md | 15 +- md/migration_v2.0.md | 5 +- md/protocol-v2.md | 39 +- md/transport-architecture.md | 5 +- .../tests/proxy_protocol_router_v2.rs | 367 ++++++++++++++++++ src/agent-client-protocol-cookbook/src/lib.rs | 3 +- .../CHANGELOG.md | 2 + .../src/mcp_over_acp/mod.rs | 80 +++- src/agent-client-protocol/CHANGELOG.md | 4 + src/agent-client-protocol/README.md | 5 +- .../src/concepts/proxies.rs | 11 +- src/agent-client-protocol/src/lib.rs | 4 +- src/agent-client-protocol/src/role/acp.rs | 187 ++++++++- .../tests/proxy_protocol_router_v2.rs | 319 +++++++++++++++ 16 files changed, 988 insertions(+), 73 deletions(-) create mode 100644 src/agent-client-protocol-conductor/tests/proxy_protocol_router_v2.rs create mode 100644 src/agent-client-protocol/tests/proxy_protocol_router_v2.rs diff --git a/README.md b/README.md index 05b33cdd..201ede57 100644 --- a/README.md +++ b/README.md @@ -53,10 +53,10 @@ without coupling later session events to that response. [Protocol V2](./md/protocol-v2.md). `Client.builder()`, `Agent.builder()`, and `Proxy.builder()` remain stable-v1 -entry points; their `.v2()` counterparts select the draft-v2 API. Raw proxy -routing infrastructure that selects and validates a version itself can use -`without_acp_version_guard`, but ordinary v2 proxy implementations should use -`Proxy.v2()`. +entry points; their `.v2()` counterparts select the draft-v2 API. With +`unstable_protocol_v2`, use `Proxy.protocol_router()` to expose separate strict +v1 and v2 proxy implementations as one component. Custom raw routing +infrastructure can use `without_acp_version_guard`. ## Integrations diff --git a/md/conductor.md b/md/conductor.md index 0f1c83a2..d426d961 100644 --- a/md/conductor.md +++ b/md/conductor.md @@ -56,9 +56,12 @@ extension fields from being interpreted as a permissive v1 request and dropped. An exact-version request whose typed value is unchanged keeps its original raw parameters, including unknown extensions. A request for a later compatible protocol version selects v2 and is canonicalized through the selected v2 -schema, matching the core protocol router. +schema. The command-line component provider, `AgentOnly`, `ProxiesAndAgent`, and static -proxy vectors accept both versions. Custom instantiators can implement the +proxy vectors 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 instantiators can implement the feature-gated `instantiate_v2_proxies_and_agent` or `instantiate_v2_proxies` method; their default implementation rejects v2 with a JSON-RPC response and leaves the connection in a failed state that rejects later traffic. A modified diff --git a/md/mcp-bridge.md b/md/mcp-bridge.md index 53bb9943..93270749 100644 --- a/md/mcp-bridge.md +++ b/md/mcp-bridge.md @@ -30,8 +30,10 @@ agent-client-protocol-polyfill = { version = "...", features = ["unstable_protoc The feature makes this concrete compatibility proxy recognize v2 initialization, capability, session setup, and `mcp/*` wire types. It does not -add high-level v2 global MCP attachment or proxy-session helpers to the core -SDK; those remain v1-only. +change the core attachment API. `Proxy.v2().with_mcp_server(...)` provides +connection-global attachment, and `V2SessionBuilder::with_mcp_server(...)` +provides per-new-session attachment. The polyfill adapts their native +declarations when the final agent supports only HTTP MCP. ## Placement @@ -51,12 +53,13 @@ ConductorImpl::new_agent("conductor", components) .await?; ``` -For v1, the application proxy can attach a high-level +The application proxy can attach a high-level `agent_client_protocol::mcp_server::McpServer`. The SDK advertises it in session setup requests as `McpServer::Acp`; callers do not need to construct a transport -placeholder themselves. In a v2 chain, a version-aware proxy currently supplies -the `schema::v2::McpServer::Acp` declaration directly because the high-level -global proxy attachment helpers remain v1-only. +placeholder themselves. In v2, `Proxy.v2().with_mcp_server(...)` provides +connection-global attachment and `V2SessionBuilder::with_mcp_server(...)` +provides per-new-session attachment. The polyfill translates those native +declarations at the final compatibility boundary. During initialization, the polyfill forwards the request to its successor. When the successor advertises HTTP MCP support, the polyfill advertises native ACP diff --git a/md/migration_v2.0.md b/md/migration_v2.0.md index d9b90c6c..0b089451 100644 --- a/md/migration_v2.0.md +++ b/md/migration_v2.0.md @@ -339,8 +339,9 @@ Construct `Lines` and `ByteStreams` with `Lines::new(outgoing, incoming)` and ## Draft v2 schema updates The optional `unstable_protocol_v2` surface now tracks -`agent-client-protocol-schema` 1.5. Because this API is explicitly unstable, its source changes -are included in the SDK 2.0 migration rather than treated as stable-v1 wire changes. +`agent-client-protocol-schema` 1.6. The changes accumulated across schema 1.5 +and 1.6 are included in the SDK 2.0 migration because this API is explicitly +unstable, rather than treated as stable-v1 wire changes. - Many values that were plain `String` or `PathBuf` fields are semantic newtypes, including `AbsolutePath`, `MediaType`, session/message/tool/terminal IDs, and list cursors. Construct them diff --git a/md/protocol-v2.md b/md/protocol-v2.md index 66486c08..ce502aa4 100644 --- a/md/protocol-v2.md +++ b/md/protocol-v2.md @@ -265,8 +265,15 @@ the stable v1 builder, while `Proxy.v2()` is v2-only and requires `_proxy/initialize` to select protocol v2. A proxy built for one version rejects the other version instead of parsing it through a permissive schema. -Raw routing infrastructure is the exception. If a component deliberately -selects and validates the version itself, it can use +When one component must expose independently authored v1 and v2 proxies, +compose them with +`Proxy.protocol_router().with_v1(v1_proxy).with_v2(v2_proxy)`. The conductor +has already selected and canonicalized the protocol before +`_proxy/initialize`, so the proxy router requires an exact v1 or v2 match, +preserves the complete initial transport frame, and does not downgrade or +convert later traffic. + +Components implementing their own raw version selector can use `Proxy.builder().without_acp_version_guard()` and keep protocol-neutral `ConnectionTo` callbacks. This disables the SDK's automatic version guard and is not a substitute for selecting `Proxy.v2()` in an ordinary v2 proxy @@ -281,14 +288,17 @@ request, `info`, `capabilities`, metadata, and unknown extension fields therefore retain their wire shape across conductor-controlled rewrites. A proxy implementation can still deliberately replace the request it forwards. -As with the core protocol router, an exact v2 request can retain unknown raw -fields, while a request for a later compatible version is canonicalized through -the selected v2 schema before component instantiation. +An exact v2 request can retain unknown raw fields, while a request for a later +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 v2 `InitializeResponse`. The flat `schema::InitializeProxyRequest` remains the -stable v1 type. Static conductor component providers support both versions. +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 v2 rather than interpreting it as v1. Returning the initialize request @@ -331,8 +341,9 @@ The SDK handles the `initialize` negotiation at the JSON-RPC boundary: match. The SDK does not convert traffic between v1 and v2. That means v1 and v2 implementations still need separate handlers. -`Agent.v2()` and `Client.v2()` are v2-only. While protocol v2 stabilizes, the -`unstable_protocol_v2` crate feature also exposes `Agent.protocol_router()` and +`Agent.v2()`, `Client.v2()`, and `Proxy.v2()` are v2-only. While protocol v2 +stabilizes, the `unstable_protocol_v2` crate feature also exposes +`Agent.protocol_router()`, `Proxy.protocol_router()`, and `Client.protocol_connector()` for composing version-specific implementations. Agents can add protocol implementations independently, which makes it easy for @@ -380,7 +391,7 @@ agent # } ``` -The protocol router reads the initial `initialize` request, selects the +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 @@ -390,6 +401,12 @@ 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. +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 +the selected schema, then hands the complete, unchanged initial frame to that +strict implementation. + Clients use a connector because fallback may require opening a new transport. Both client implementations and the agent transport are factories: @@ -428,8 +445,8 @@ connection. That does not turn an otherwise valid v2 request into an error. 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 -types. The next schema dependency update removes the former schema-wide v1/v2 -conversion API: versioned implementations should remain separate, with +types. Schema 1.6 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 policy are available. These are draft API changes rather than stable v1 wire changes. See [Migrating to diff --git a/md/transport-architecture.md b/md/transport-architecture.md index 64e70469..a5f7dce3 100644 --- a/md/transport-architecture.md +++ b/md/transport-architecture.md @@ -199,7 +199,10 @@ Lifecycle-sensitive calls should normally be sent individually. As a compatibility measure, `AgentProtocolRouter` can select a v1 or v2 agent when the first call-shaped entry is `initialize`, while preserving the original frame for the selected implementation. Response-only frames received before -initialization are ignored. `ClientProtocolConnector` starts each attempted +initialization are ignored. `ProxyProtocolRouter` provides the analogous +boundary after conductor selection: it requires an exact v1 or v2 +`_proxy/initialize` match and preserves the complete initial frame without +cross-version conversion. `ClientProtocolConnector` starts each attempted client implementation with an individual `initialize` request. ## Message Flow diff --git a/src/agent-client-protocol-conductor/tests/proxy_protocol_router_v2.rs b/src/agent-client-protocol-conductor/tests/proxy_protocol_router_v2.rs new file mode 100644 index 00000000..8090e619 --- /dev/null +++ b/src/agent-client-protocol-conductor/tests/proxy_protocol_router_v2.rs @@ -0,0 +1,367 @@ +#![cfg(feature = "unstable_protocol_v2")] + +use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, +}; + +use agent_client_protocol::schema::{InitializeProxyRequest, ProtocolVersion, v1, v2}; +use agent_client_protocol::{ + Agent, ByteStreams, Client, Conductor, ConnectionTo, Error, Proxy, V2ConnectionTo, +}; +use agent_client_protocol_conductor::{ConductorImpl, ProxiesAndAgent}; +use tokio::io::duplex; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +#[derive(Default)] +struct ProtocolObservations { + proxy_initializations: AtomicUsize, + proxy_list_sessions: AtomicUsize, + agent_initializations: AtomicUsize, + agent_list_sessions: AtomicUsize, + proxy_versions: Mutex>, + agent_versions: Mutex>, +} + +#[derive(Default)] +struct Observations { + v1: ProtocolObservations, + v2: ProtocolObservations, +} + +impl ProtocolObservations { + fn record_proxy_initialize(&self, version: ProtocolVersion) { + self.proxy_initializations.fetch_add(1, Ordering::SeqCst); + self.proxy_versions + .lock() + .expect("proxy versions lock should not be poisoned") + .push(version); + } + + fn record_agent_initialize(&self, version: ProtocolVersion) { + self.agent_initializations.fetch_add(1, Ordering::SeqCst); + self.agent_versions + .lock() + .expect("agent versions lock should not be poisoned") + .push(version); + } + + fn assert_selected(&self, version: ProtocolVersion) { + assert_eq!(self.proxy_initializations.load(Ordering::SeqCst), 1); + assert_eq!(self.proxy_list_sessions.load(Ordering::SeqCst), 1); + assert_eq!(self.agent_initializations.load(Ordering::SeqCst), 1); + assert_eq!(self.agent_list_sessions.load(Ordering::SeqCst), 1); + assert_eq!( + *self + .proxy_versions + .lock() + .expect("proxy versions lock should not be poisoned"), + [version] + ); + assert_eq!( + *self + .agent_versions + .lock() + .expect("agent versions lock should not be poisoned"), + [version] + ); + } + + fn assert_not_selected(&self) { + assert_eq!(self.proxy_initializations.load(Ordering::SeqCst), 0); + assert_eq!(self.proxy_list_sessions.load(Ordering::SeqCst), 0); + assert_eq!(self.agent_initializations.load(Ordering::SeqCst), 0); + assert_eq!(self.agent_list_sessions.load(Ordering::SeqCst), 0); + assert!( + self.proxy_versions + .lock() + .expect("proxy versions lock should not be poisoned") + .is_empty() + ); + assert!( + self.agent_versions + .lock() + .expect("agent versions lock should not be poisoned") + .is_empty() + ); + } +} + +fn components(observations: Arc) -> ProxiesAndAgent { + let v1_proxy_initialize = Arc::clone(&observations); + let v1_proxy_list = Arc::clone(&observations); + let v1_proxy = Proxy + .builder() + .name("v1-proxy") + .on_receive_request_from( + Client, + async move |request: InitializeProxyRequest, responder, cx| { + v1_proxy_initialize + .v1 + .record_proxy_initialize(request.initialize.protocol_version); + cx.send_request_to(Agent, request.initialize) + .forward_response_to(responder) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request_from( + Client, + async move |request: v1::ListSessionsRequest, responder, cx| { + v1_proxy_list + .v1 + .proxy_list_sessions + .fetch_add(1, Ordering::SeqCst); + cx.send_request_to(Agent, request) + .forward_response_to(responder) + }, + agent_client_protocol::on_receive_request!(), + ); + + let v2_proxy_initialize = Arc::clone(&observations); + let v2_proxy_list = Arc::clone(&observations); + let v2_proxy = Proxy + .v2() + .name("v2-proxy") + .on_receive_request_from( + Client, + async move |request: v2::InitializeProxyRequest, + responder, + cx: V2ConnectionTo| { + v2_proxy_initialize + .v2 + .record_proxy_initialize(request.initialize.protocol_version); + cx.send_request_to(Agent, request.initialize) + .forward_response_to(responder) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request_from( + Client, + async move |request: v2::ListSessionsRequest, + responder, + cx: V2ConnectionTo| { + v2_proxy_list + .v2 + .proxy_list_sessions + .fetch_add(1, Ordering::SeqCst); + cx.send_request_to(Agent, request) + .forward_response_to(responder) + }, + agent_client_protocol::on_receive_request!(), + ); + + let v1_agent_initialize = Arc::clone(&observations); + let v1_agent_list = Arc::clone(&observations); + let v1_agent = Agent + .builder() + .name("v1-agent") + .on_receive_request( + async move |request: v1::InitializeRequest, responder, _cx| { + v1_agent_initialize + .v1 + .record_agent_initialize(request.protocol_version); + responder.respond(v1::InitializeResponse::new(request.protocol_version)) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |_request: v1::ListSessionsRequest, responder, _cx| { + v1_agent_list + .v1 + .agent_list_sessions + .fetch_add(1, Ordering::SeqCst); + responder.respond(v1::ListSessionsResponse::new(Vec::new())) + }, + agent_client_protocol::on_receive_request!(), + ); + + let v2_agent_initialize = Arc::clone(&observations); + let v2_agent_list = observations; + let v2_agent = Agent + .v2() + .name("v2-agent") + .on_receive_request( + async move |request: v2::InitializeRequest, responder, _cx| { + v2_agent_initialize + .v2 + .record_agent_initialize(request.protocol_version); + responder.respond(v2::InitializeResponse::new( + request.protocol_version, + v2::Implementation::new("v2-agent", "1.0.0"), + )) + }, + agent_client_protocol::on_receive_request!(), + ) + .on_receive_request( + async move |_request: v2::ListSessionsRequest, responder, _cx| { + v2_agent_list + .v2 + .agent_list_sessions + .fetch_add(1, Ordering::SeqCst); + responder.respond(v2::ListSessionsResponse::new(Vec::new())) + }, + agent_client_protocol::on_receive_request!(), + ); + + let proxy = Proxy.protocol_router().with_v1(v1_proxy).with_v2(v2_proxy); + let agent = Agent.protocol_router().with_v1(v1_agent).with_v2(v2_agent); + + ProxiesAndAgent::new(agent).proxy(proxy) +} + +async fn run_v1_client( + components: ProxiesAndAgent, + client_task: impl AsyncFnOnce(ConnectionTo) -> Result<(), Error>, +) -> Result<(), Error> { + let (client_out, conductor_in) = duplex(4096); + let (conductor_out, client_in) = duplex(4096); + + Client + .builder() + .name("v1-client") + .with_spawned(|_cx| async move { + ConductorImpl::new_agent("conductor", components) + .run(ByteStreams::new( + conductor_out.compat_write(), + conductor_in.compat(), + )) + .await + }) + .connect_with( + ByteStreams::new(client_out.compat_write(), client_in.compat()), + client_task, + ) + .await +} + +async fn run_v2_client( + components: ProxiesAndAgent, + client_task: impl AsyncFnOnce(V2ConnectionTo) -> Result<(), Error>, +) -> Result<(), Error> { + let (client_out, conductor_in) = duplex(4096); + let (conductor_out, client_in) = duplex(4096); + + Client + .v2() + .name("v2-client") + .with_spawned(|_cx| async move { + ConductorImpl::new_agent("conductor", components) + .run(ByteStreams::new( + conductor_out.compat_write(), + conductor_in.compat(), + )) + .await + }) + .connect_with( + ByteStreams::new(client_out.compat_write(), client_in.compat()), + client_task, + ) + .await +} + +async fn run_raw_client( + components: ProxiesAndAgent, + client_task: impl AsyncFnOnce(ConnectionTo) -> Result<(), Error>, +) -> Result<(), Error> { + let (client_out, conductor_in) = duplex(4096); + let (conductor_out, client_in) = duplex(4096); + + Client + .builder() + .without_acp_version_guard() + .name("future-client") + .with_spawned(|_cx| async move { + ConductorImpl::new_agent("conductor", components) + .run(ByteStreams::new( + conductor_out.compat_write(), + conductor_in.compat(), + )) + .await + }) + .connect_with( + ByteStreams::new(client_out.compat_write(), client_in.compat()), + client_task, + ) + .await +} + +#[tokio::test] +async fn v1_conductor_routes_initialize_and_later_requests_to_v1() -> Result<(), Error> { + let observations = Arc::new(Observations::default()); + + run_v1_client(components(Arc::clone(&observations)), async |cx| { + let response = cx + .send_request(v1::InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + assert_eq!(response.protocol_version, ProtocolVersion::V1); + + let response = cx + .send_request(v1::ListSessionsRequest::new()) + .block_task() + .await?; + assert!(response.sessions.is_empty()); + Ok(()) + }) + .await?; + + observations.v1.assert_selected(ProtocolVersion::V1); + observations.v2.assert_not_selected(); + Ok(()) +} + +#[tokio::test] +async fn v2_conductor_routes_initialize_and_later_requests_to_v2() -> Result<(), Error> { + let observations = Arc::new(Observations::default()); + + run_v2_client(components(Arc::clone(&observations)), async |cx| { + let response = cx + .send_request(v2::InitializeRequest::new( + ProtocolVersion::V2, + v2::Implementation::new("v2-client", "1.0.0"), + )) + .block_task() + .await?; + assert_eq!(response.protocol_version, ProtocolVersion::V2); + + let response = cx + .send_request(v2::ListSessionsRequest::new()) + .block_task() + .await?; + assert!(response.sessions.is_empty()); + Ok(()) + }) + .await?; + + observations.v1.assert_not_selected(); + observations.v2.assert_selected(ProtocolVersion::V2); + Ok(()) +} + +#[tokio::test] +async fn conductor_canonicalizes_future_version_before_proxy_routing() -> Result<(), Error> { + let observations = Arc::new(Observations::default()); + + run_raw_client(components(Arc::clone(&observations)), async |cx| { + let response = cx + .send_request(v2::InitializeRequest::new( + ProtocolVersion::from(3_u16), + v2::Implementation::new("future-client", "1.0.0"), + )) + .block_task() + .await?; + assert_eq!(response.protocol_version, ProtocolVersion::V2); + + let response = cx + .send_request(v2::ListSessionsRequest::new()) + .block_task() + .await?; + assert!(response.sessions.is_empty()); + Ok(()) + }) + .await?; + + observations.v1.assert_not_selected(); + observations.v2.assert_selected(ProtocolVersion::V2); + Ok(()) +} diff --git a/src/agent-client-protocol-cookbook/src/lib.rs b/src/agent-client-protocol-cookbook/src/lib.rs index 8eb0fe22..570c119f 100644 --- a/src/agent-client-protocol-cookbook/src/lib.rs +++ b/src/agent-client-protocol-cookbook/src/lib.rs @@ -23,7 +23,8 @@ //! messages. The most common use case is adding MCP tools. Use //! [`Proxy.builder()`](agent_client_protocol::Proxy) for stable protocol v1 //! proxy connections. With the core SDK's `unstable_protocol_v2` feature, use -//! `Proxy.v2()` for a draft-v2-only proxy. +//! `Proxy.v2()` for a draft-v2-only proxy, or `Proxy.protocol_router()` to +//! expose separate strict v1 and v2 implementations as one component. //! //! **Important:** Proxies don't run standalone—they need the [`agent-client-protocol-conductor`] to //! orchestrate the connection between client, proxies, and agent. See diff --git a/src/agent-client-protocol-polyfill/CHANGELOG.md b/src/agent-client-protocol-polyfill/CHANGELOG.md index 8098b691..f7f5a14c 100644 --- a/src/agent-client-protocol-polyfill/CHANGELOG.md +++ b/src/agent-client-protocol-polyfill/CHANGELOG.md @@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Route v1 and v2 initialization through strict version-specific proxy + builders instead of disabling the ACP version guard. - Forward MCP request cancellation hop by hop instead of tunneling the loopback connection's `$/cancel_request` ID inside `mcp/message`. diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs index 7dc7ca97..feeb6624 100644 --- a/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs +++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/mod.rs @@ -127,28 +127,68 @@ impl ConnectTo for McpOverAcpPolyfill { self, client: impl ConnectTo, ) -> Result<(), agent_client_protocol::Error> { - let (bridge_tx, bridge_rx) = mpsc::channel(128); - - let proxy = Proxy.builder(); #[cfg(feature = "unstable_protocol_v2")] - let proxy = proxy.without_acp_version_guard(); + { + Proxy + .protocol_router() + .with_v1(McpOverAcpProxy(PolyfillProtocol::V1)) + .with_v2(McpOverAcpProxy(PolyfillProtocol::V2)) + .connect_to(client) + .await + } - proxy - .name("mcp-over-acp-polyfill") - .with_runner(BridgeRunner { - bridge_tx: bridge_tx.clone(), - bridge_rx, - protocol: None, - downstream_mode: DownstreamMcpMode::Unknown, - listeners: BridgeListeners::default(), - bridge_connections: HashMap::new(), - }) - .with_handler(PolyfillHandler { - protocol: None, - bridge_tx, - }) - .connect_to(client) - .await + #[cfg(not(feature = "unstable_protocol_v2"))] + { + McpOverAcpProxy(PolyfillProtocol::V1) + .connect_to(client) + .await + } + } +} + +#[derive(Debug)] +struct McpOverAcpProxy(PolyfillProtocol); + +impl ConnectTo for McpOverAcpProxy { + async fn connect_to( + self, + client: impl ConnectTo, + ) -> Result<(), agent_client_protocol::Error> { + let (bridge_tx, bridge_rx) = mpsc::channel(128); + let runner = BridgeRunner { + bridge_tx: bridge_tx.clone(), + bridge_rx, + protocol: None, + downstream_mode: DownstreamMcpMode::Unknown, + listeners: BridgeListeners::default(), + bridge_connections: HashMap::new(), + }; + let handler = PolyfillHandler { + protocol: None, + bridge_tx, + }; + + match self.0 { + PolyfillProtocol::V1 => { + Proxy + .builder() + .name("mcp-over-acp-polyfill") + .with_runner(runner) + .with_handler(handler) + .connect_to(client) + .await + } + #[cfg(feature = "unstable_protocol_v2")] + PolyfillProtocol::V2 => { + Proxy + .v2() + .name("mcp-over-acp-polyfill") + .with_runner(runner) + .with_handler(handler) + .connect_to(client) + .await + } + } } } diff --git a/src/agent-client-protocol/CHANGELOG.md b/src/agent-client-protocol/CHANGELOG.md index 5ec93e8e..8f989623 100644 --- a/src/agent-client-protocol/CHANGELOG.md +++ b/src/agent-client-protocol/CHANGELOG.md @@ -4,6 +4,10 @@ ### Added +- *(unstable-v2)* Add `Proxy::protocol_router` and `ProxyProtocolRouter` to + compose strict v1 and v2 proxy implementations behind one connection. + Routing requires the exact version selected by the conductor and preserves + the complete initial transport frame. - *(unstable)* Expose programmatic tool-call names through the `unstable_tool_call_name` feature. - *(unstable)* Expose v1 and draft-v2 plan operations through the diff --git a/src/agent-client-protocol/README.md b/src/agent-client-protocol/README.md index 69ea1a64..2fa13b3d 100644 --- a/src/agent-client-protocol/README.md +++ b/src/agent-client-protocol/README.md @@ -46,8 +46,9 @@ Draft protocol v2 is opt-in through `unstable_protocol_v2`. `Client.v2()`, high-level, command-only session helpers because prompt acceptance and inbound traffic are independent. Session updates and interactive requests use typed connection handlers. See [Protocol V2](https://agentclientprotocol.github.io/rust-sdk/protocol-v2.html#high-level-v2-sessions). -`Proxy.builder()` remains the stable v1 entry point; raw routing infrastructure -that selects and validates the version itself can use +`Proxy.builder()` remains the stable v1 entry point. Use +`Proxy.protocol_router()` to expose separate strict v1 and v2 proxy +implementations as one component; custom raw routing infrastructure can use `Proxy.builder().without_acp_version_guard()`. ## MCP Server Attachment diff --git a/src/agent-client-protocol/src/concepts/proxies.rs b/src/agent-client-protocol/src/concepts/proxies.rs index 8479afc7..866adabb 100644 --- a/src/agent-client-protocol/src/concepts/proxies.rs +++ b/src/agent-client-protocol/src/concepts/proxies.rs @@ -22,8 +22,12 @@ //! validates `_proxy/initialize` and later traffic against the selected //! version. //! -//! Low-level routing infrastructure that deliberately selects and validates a -//! raw protocol version itself can use +//! Use `Proxy.protocol_router()` to package separate v1 and v2 implementations +//! behind one `ConnectTo` component. It dispatches the +//! conductor-selected `_proxy/initialize` version exactly and performs no +//! cross-version conversion. +//! +//! Low-level infrastructure implementing custom raw version routing can use //! `Proxy.builder().without_acp_version_guard()`. Disabling the guard is an //! explicit version-neutral escape hatch, not the ordinary way to author a v2 //! proxy. @@ -196,7 +200,8 @@ //! |------|----------| //! | Forward everything | Just `connect_to(transport)` | //! | Author a v1 or v2 proxy | `Proxy.builder()` or `Proxy.v2()` | -//! | Route versions yourself | `without_acp_version_guard` on the raw proxy builder | +//! | Package v1 and v2 implementations | `Proxy.protocol_router().with_v1(...).with_v2(...)` | +//! | Implement custom raw routing | `without_acp_version_guard` on the raw proxy builder | //! | Intercept specific messages | `on_receive_*_from` with explicit peers | //! | Add global tools | `with_mcp_server` on builder | //! | Add per-session tools | `with_mcp_server` on session builder | diff --git a/src/agent-client-protocol/src/lib.rs b/src/agent-client-protocol/src/lib.rs index 3cb8edd6..002973ce 100644 --- a/src/agent-client-protocol/src/lib.rs +++ b/src/agent-client-protocol/src/lib.rs @@ -130,9 +130,7 @@ pub use jsonrpc::{RequestCancellation, is_cancel_request_notification}; pub use jsonrpc::{V2Builder, V2ConnectionContext, V2ConnectionTo}; #[cfg(feature = "unstable_protocol_v2")] -pub use role::acp::AgentProtocolRouter; -#[cfg(feature = "unstable_protocol_v2")] -pub use role::acp::ClientProtocolConnector; +pub use role::acp::{AgentProtocolRouter, ClientProtocolConnector, ProxyProtocolRouter}; pub use role::{ Role, RoleId, UntypedRole, acp::{Agent, Client, Conductor, Proxy}, diff --git a/src/agent-client-protocol/src/role/acp.rs b/src/agent-client-protocol/src/role/acp.rs index c603a279..754eb57b 100644 --- a/src/agent-client-protocol/src/role/acp.rs +++ b/src/agent-client-protocol/src/role/acp.rs @@ -406,7 +406,7 @@ impl AgentProtocolRouter { #[cfg(feature = "unstable_protocol_v2")] impl ConnectTo for AgentProtocolRouter { async fn connect_to(self, client: impl ConnectTo) -> Result<(), crate::Error> { - let supported = SupportedAgentProtocols { + let supported = SupportedProtocols { v1: self.v1.is_some(), v2: self.v2.is_some(), }; @@ -442,13 +442,13 @@ impl ConnectTo for AgentProtocolRouter { #[cfg(feature = "unstable_protocol_v2")] #[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum AgentProtocol { +enum SelectedProtocol { V1, V2, } #[cfg(feature = "unstable_protocol_v2")] -impl AgentProtocol { +impl SelectedProtocol { fn take_agent(self, agent: AgentProtocolRouter) -> Option> { match self { Self::V1 => agent.v1, @@ -470,7 +470,7 @@ impl AgentProtocol { } } - fn unsupported_error(self, supported: SupportedAgentProtocols) -> crate::Error { + fn unsupported_error(self, supported: SupportedProtocols) -> crate::Error { crate::Error::invalid_request().data(format!( "ACP protocol version {} is not configured; this endpoint supports {}", self.name(), @@ -481,25 +481,35 @@ impl AgentProtocol { #[cfg(feature = "unstable_protocol_v2")] #[derive(Debug, Clone, Copy, PartialEq, Eq)] -struct SupportedAgentProtocols { +struct SupportedProtocols { v1: bool, v2: bool, } #[cfg(feature = "unstable_protocol_v2")] -impl SupportedAgentProtocols { - fn highest_compatible(self, requested: ProtocolVersion) -> Option { +impl SupportedProtocols { + fn highest_compatible(self, requested: ProtocolVersion) -> Option { if self.v2 && requested >= ProtocolVersion::V2 { - return Some(AgentProtocol::V2); + return Some(SelectedProtocol::V2); } if self.v1 && requested >= ProtocolVersion::V1 { - return Some(AgentProtocol::V1); + return Some(SelectedProtocol::V1); } None } + fn exact(self, requested: ProtocolVersion) -> Option { + if self.v1 && requested == ProtocolVersion::V1 { + Some(SelectedProtocol::V1) + } else if self.v2 && requested == ProtocolVersion::V2 { + Some(SelectedProtocol::V2) + } else { + None + } + } + fn description(self) -> String { match (self.v1, self.v2) { (true, true) => "ACP protocol versions 1 and 2".into(), @@ -513,8 +523,8 @@ impl SupportedAgentProtocols { #[cfg(feature = "unstable_protocol_v2")] fn select_agent_protocol( message: &mut RawJsonRpcMessage, - supported: SupportedAgentProtocols, -) -> Result { + supported: SupportedProtocols, +) -> Result { let RawJsonRpcMessage::Request(request) = message else { return Err( crate::Error::invalid_request().data("first ACP message must be an initialize request") @@ -596,17 +606,17 @@ fn normalize_v2_initialize_params_for_reuse( fn rewrite_initialize_params( params: &mut serde_json::Map, requested: ProtocolVersion, - selected: AgentProtocol, + selected: SelectedProtocol, ) -> Result<(), crate::Error> { // Validate exact-version initialization without replacing its raw // parameters. Reserializing through the SDK's pinned schema would discard // fields added by newer compatible peers. if requested == selected.version() { match selected { - AgentProtocol::V1 => { + SelectedProtocol::V1 => { parse_initialize_params::(params)?; } - AgentProtocol::V2 => { + SelectedProtocol::V2 => { parse_initialize_params::(params)?; } } @@ -614,12 +624,12 @@ fn rewrite_initialize_params( } match selected { - AgentProtocol::V1 => { + SelectedProtocol::V1 => { debug_assert!(requested >= ProtocolVersion::V2); *params = normalize_v2_initialize_params_for_v1(params, false)?; Ok(()) } - AgentProtocol::V2 => { + SelectedProtocol::V2 => { let mut initialize = parse_initialize_params::(params)?; initialize.protocol_version = ProtocolVersion::V2; *params = serialize_initialize_params(initialize)?; @@ -855,8 +865,8 @@ fn json_object_contains( #[cfg(feature = "unstable_protocol_v2")] fn highest_compatible_agent_protocol( requested: ProtocolVersion, - supported: SupportedAgentProtocols, -) -> Result { + supported: SupportedProtocols, +) -> Result { supported.highest_compatible(requested).ok_or_else(|| { crate::Error::invalid_request().data(format!( "unsupported ACP protocol version {requested}; this endpoint supports {}", @@ -1336,6 +1346,147 @@ impl Proxy { pub fn v2(self) -> V2Builder { self.builder().v2_proxy() } + + /// Create a router that chooses between configured proxy implementations. + /// + /// Add implementations with [`ProxyProtocolRouter::with_v1`] and + /// [`ProxyProtocolRouter::with_v2`]. The router reads the initial + /// `_proxy/initialize` request, selects the implementation for that exact + /// protocol version, and hands over the complete initial transport frame. + /// It does not downgrade proxy traffic or convert later messages. + /// + /// Requires the `unstable_protocol_v2` crate feature while protocol v2 + /// stabilizes. + #[cfg(feature = "unstable_protocol_v2")] + #[must_use] + pub fn protocol_router(self) -> ProxyProtocolRouter { + ProxyProtocolRouter::new() + } +} + +/// Proxy component that routes each connection to a configured protocol implementation. +/// +/// Use [`Proxy::protocol_router`] to start the builder, then add stable-v1 and +/// draft-v2 proxy implementations independently. Unlike +/// [`AgentProtocolRouter`], this router requires an exact version match: the +/// conductor has already selected and canonicalized the wire protocol before +/// sending `_proxy/initialize` to a proxy. +#[cfg(feature = "unstable_protocol_v2")] +#[derive(Debug, Default)] +pub struct ProxyProtocolRouter { + v1: Option>, + v2: Option>, +} + +#[cfg(feature = "unstable_protocol_v2")] +impl ProxyProtocolRouter { + /// Create an empty proxy protocol router. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Return this router with a stable ACP v1 proxy implementation. + #[must_use] + pub fn with_v1(mut self, proxy: impl ConnectTo) -> Self { + self.v1 = Some(DynConnectTo::new(proxy)); + self + } + + /// Return this router with a draft ACP v2 proxy implementation. + #[must_use] + pub fn with_v2(mut self, proxy: impl ConnectTo) -> Self { + self.v2 = Some(DynConnectTo::new(proxy)); + self + } +} + +#[cfg(feature = "unstable_protocol_v2")] +impl ConnectTo for ProxyProtocolRouter { + async fn connect_to(self, conductor: impl ConnectTo) -> Result<(), crate::Error> { + let supported = SupportedProtocols { + v1: self.v1.is_some(), + v2: self.v2.is_some(), + }; + let mut conductor = RunningProtocolPeer::new(conductor); + let (first_frame, conductor, selected) = loop { + let Some((mut frame, next_conductor)) = conductor.next_frame().await? else { + return Ok(()); + }; + let message = match initialize_message_mut(&mut frame) { + Ok(Some(message)) => message, + Ok(None) => { + conductor = next_conductor; + continue; + } + Err(error) => return reject_initialize(next_conductor, &frame, error).await, + }; + let selected = match select_proxy_protocol(message, supported) { + Ok(selected) => selected, + Err(error) => return reject_initialize(next_conductor, &frame, error).await, + }; + break (frame, next_conductor, selected); + }; + let Some(proxy) = selected.take_proxy(self) else { + let error = selected.unsupported_error(supported); + return reject_initialize(conductor, &first_frame, error).await; + }; + + let proxy = RunningProtocolPeer::new(proxy); + proxy.send_frame(first_frame)?; + pipe_protocol_peers_until_closed(conductor, proxy).await + } +} + +#[cfg(feature = "unstable_protocol_v2")] +impl SelectedProtocol { + fn take_proxy(self, proxy: ProxyProtocolRouter) -> Option> { + match self { + Self::V1 => proxy.v1, + Self::V2 => proxy.v2, + } + } +} + +#[cfg(feature = "unstable_protocol_v2")] +fn select_proxy_protocol( + message: &RawJsonRpcMessage, + supported: SupportedProtocols, +) -> Result { + let RawJsonRpcMessage::Request(request) = message else { + return Err(crate::Error::invalid_request() + .data("first ACP proxy message must be an `_proxy/initialize` request")); + }; + + if request.method.as_ref() != METHOD_INITIALIZE_PROXY { + return Err(crate::Error::invalid_request() + .data("first ACP proxy request must be `_proxy/initialize`")); + } + + let Some(RawJsonRpcParams::Object(params)) = &request.params else { + return Err(invalid_initialize_protocol_version()); + }; + let Some(protocol_version) = params.get("protocolVersion") else { + return Err(invalid_initialize_protocol_version()); + }; + let requested = serde_json::from_value::(protocol_version.clone()) + .map_err(|_| invalid_initialize_protocol_version())?; + let selected = supported.exact(requested).ok_or_else(|| { + crate::Error::invalid_request().data(format!( + "unsupported ACP protocol version {requested}; this proxy supports {}", + supported.description() + )) + })?; + + match selected { + SelectedProtocol::V1 => { + parse_initialize_params::(params)?; + } + SelectedProtocol::V2 => { + parse_initialize_params::(params)?; + } + } + Ok(selected) } impl HasPeer for Proxy { diff --git a/src/agent-client-protocol/tests/proxy_protocol_router_v2.rs b/src/agent-client-protocol/tests/proxy_protocol_router_v2.rs new file mode 100644 index 00000000..fab29293 --- /dev/null +++ b/src/agent-client-protocol/tests/proxy_protocol_router_v2.rs @@ -0,0 +1,319 @@ +#![cfg(feature = "unstable_protocol_v2")] + +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + +use agent_client_protocol::schema::{InitializeProxyRequest, ProtocolVersion, v1, v2}; +use agent_client_protocol::{ + ByteStreams, Channel, Client, Conductor, ConnectTo, Error, Proxy, ProxyProtocolRouter, + RawJsonRpcMessage, TransportFrame, +}; +use futures::StreamExt as _; +use serde_json::Value; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +fn v1_proxy_initialize(protocol_version: ProtocolVersion) -> InitializeProxyRequest { + InitializeProxyRequest::from( + v1::InitializeRequest::new(protocol_version) + .client_info(v1::Implementation::new("router-test-client", "1.0.0")), + ) +} + +fn v2_proxy_initialize(protocol_version: ProtocolVersion) -> v2::InitializeProxyRequest { + v2::InitializeProxyRequest::new(v2::InitializeRequest::new( + protocol_version, + v2::Implementation::new("router-test-client", "1.0.0"), + )) +} + +fn router(v1_hits: Arc, v2_hits: Arc) -> ProxyProtocolRouter { + Proxy + .protocol_router() + .with_v1(v1_proxy(v1_hits)) + .with_v2(v2_proxy(v2_hits)) +} + +fn v1_proxy(v1_hits: Arc) -> impl ConnectTo { + Proxy.builder().on_receive_request_from( + Client, + async move |request: InitializeProxyRequest, responder, _cx| { + v1_hits.fetch_add(1, Ordering::SeqCst); + responder.respond(v1::InitializeResponse::new( + request.initialize.protocol_version, + )) + }, + agent_client_protocol::on_receive_request!(), + ) +} + +fn v2_proxy(v2_hits: Arc) -> impl ConnectTo { + Proxy.v2().on_receive_request_from( + Client, + async move |request: v2::InitializeProxyRequest, responder, _cx| { + v2_hits.fetch_add(1, Ordering::SeqCst); + responder.respond(v2::InitializeResponse::new( + request.initialize.protocol_version, + v2::Implementation::new("router-test-proxy", "1.0.0"), + )) + }, + agent_client_protocol::on_receive_request!(), + ) +} + +async fn initialize( + router: ProxyProtocolRouter, + params: Value, +) -> Result, Error> { + request(router, "_proxy/initialize", params).await +} + +async fn request( + router: ProxyProtocolRouter, + method: &str, + params: Value, +) -> Result, Error> { + let (Channel { mut rx, tx }, future) = ConnectTo::::into_channel_and_future(router); + let task = tokio::spawn(future); + let request_id = v1::RequestId::Number(1); + + tx.unbounded_send(TransportFrame::Single(RawJsonRpcMessage::request( + method.into(), + params, + request_id.clone(), + )?)) + .map_err(Error::into_internal_error)?; + + let result = loop { + let frame = rx.next().await.ok_or_else(|| { + Error::internal_error().data("proxy router closed before initialize response") + })?; + let TransportFrame::Single(RawJsonRpcMessage::Response(response)) = frame else { + continue; + }; + match response { + v1::Response::Result { id, result } if id == request_id => break Ok(result), + v1::Response::Error { id, error } if id == request_id => break Err(error), + _ => {} + } + }; + + drop(tx); + drop(rx); + task.await.map_err(Error::into_internal_error)??; + Ok(result) +} + +#[tokio::test(flavor = "current_thread")] +async fn routes_the_exact_conductor_selected_protocol() -> Result<(), Error> { + let v1_hits = Arc::new(AtomicUsize::new(0)); + let v2_hits = Arc::new(AtomicUsize::new(0)); + + let response = initialize( + router(Arc::clone(&v1_hits), Arc::clone(&v2_hits)), + serde_json::to_value(v1_proxy_initialize(ProtocolVersion::V1)) + .map_err(Error::into_internal_error)?, + ) + .await??; + assert_eq!(response["protocolVersion"], 1); + + let response = initialize( + router(Arc::clone(&v1_hits), Arc::clone(&v2_hits)), + serde_json::to_value(v2_proxy_initialize(ProtocolVersion::V2)) + .map_err(Error::into_internal_error)?, + ) + .await??; + assert_eq!(response["protocolVersion"], 2); + + assert_eq!(v1_hits.load(Ordering::SeqCst), 1); + assert_eq!(v2_hits.load(Ordering::SeqCst), 1); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn rejects_an_unconfigured_exact_version() -> Result<(), Error> { + let v1_hits = Arc::new(AtomicUsize::new(0)); + let error = initialize( + Proxy + .protocol_router() + .with_v1(v1_proxy(Arc::clone(&v1_hits))), + serde_json::to_value(v2_proxy_initialize(ProtocolVersion::V2)) + .map_err(Error::into_internal_error)?, + ) + .await? + .expect_err("a v1-only router must reject v2"); + assert_eq!(i32::from(error.code), -32600); + assert_eq!(v1_hits.load(Ordering::SeqCst), 0); + + let v2_hits = Arc::new(AtomicUsize::new(0)); + let error = initialize( + Proxy + .protocol_router() + .with_v2(v2_proxy(Arc::clone(&v2_hits))), + serde_json::to_value(v1_proxy_initialize(ProtocolVersion::V1)) + .map_err(Error::into_internal_error)?, + ) + .await? + .expect_err("a v2-only router must reject v1"); + assert_eq!(i32::from(error.code), -32600); + assert_eq!(v2_hits.load(Ordering::SeqCst), 0); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn requires_proxy_initialize_as_the_first_call() -> Result<(), Error> { + let error = request( + Proxy + .protocol_router() + .with_v1(v1_proxy(Arc::new(AtomicUsize::new(0)))), + "initialize", + serde_json::to_value(v1_proxy_initialize(ProtocolVersion::V1)) + .map_err(Error::into_internal_error)?, + ) + .await? + .expect_err("an ordinary initialize request must not select a proxy"); + + assert_eq!(i32::from(error.code), -32600); + assert!( + error + .data + .as_ref() + .and_then(Value::as_str) + .is_some_and(|data| data.contains("must be `_proxy/initialize`")), + "{error:?}" + ); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn rejects_future_versions_instead_of_downgrading_them() -> Result<(), Error> { + let v1_hits = Arc::new(AtomicUsize::new(0)); + let v2_hits = Arc::new(AtomicUsize::new(0)); + + let error = initialize( + router(Arc::clone(&v1_hits), Arc::clone(&v2_hits)), + serde_json::to_value(v2_proxy_initialize(ProtocolVersion::from(3_u16))) + .map_err(Error::into_internal_error)?, + ) + .await? + .expect_err("a proxy router must not canonicalize a future protocol version"); + + assert_eq!(i32::from(error.code), -32600); + assert!( + error + .data + .as_ref() + .and_then(Value::as_str) + .is_some_and(|data| data.contains("unsupported ACP protocol version 3")), + "{error:?}" + ); + assert_eq!(v1_hits.load(Ordering::SeqCst), 0); + assert_eq!(v2_hits.load(Ordering::SeqCst), 0); + Ok(()) +} + +fn initialize_params_with_extensions(protocol_version: ProtocolVersion) -> Result { + let mut params = if protocol_version == ProtocolVersion::V1 { + serde_json::to_value(v1_proxy_initialize(protocol_version)) + } else { + serde_json::to_value(v2_proxy_initialize(protocol_version)) + } + .map_err(Error::into_internal_error)?; + + let params = params + .as_object_mut() + .expect("serialized initialize params should be an object"); + params.insert( + "_futureInitializeField".into(), + serde_json::json!({ "preserved": true }), + ); + Ok(Value::Object(params.clone())) +} + +async fn write_wire_json( + writer: &mut (impl tokio::io::AsyncWrite + Unpin), + value: &Value, +) -> Result<(), Error> { + use tokio::io::AsyncWriteExt as _; + + let mut bytes = serde_json::to_vec(value).map_err(Error::into_internal_error)?; + bytes.push(b'\n'); + writer + .write_all(&bytes) + .await + .map_err(Error::into_internal_error)?; + writer.flush().await.map_err(Error::into_internal_error) +} + +async fn read_wire_json( + reader: &mut (impl tokio::io::AsyncBufRead + Unpin), +) -> Result { + use tokio::io::AsyncBufReadExt as _; + + let mut line = String::new(); + reader + .read_line(&mut line) + .await + .map_err(Error::into_internal_error)?; + serde_json::from_str(line.trim()).map_err(Error::into_internal_error) +} + +#[tokio::test(flavor = "current_thread")] +async fn preserves_the_complete_initial_transport_frame() -> Result<(), Error> { + use tokio::io::{AsyncWriteExt as _, BufReader}; + + for protocol_version in [ProtocolVersion::V1, ProtocolVersion::V2] { + let (mut conductor_writer, router_reader) = tokio::io::duplex(4096); + let (router_writer, _conductor_reader) = tokio::io::duplex(4096); + let conductor_transport = + ByteStreams::new(router_writer.compat_write(), router_reader.compat()); + + let (router_to_proxy_writer, proxy_reader) = tokio::io::duplex(4096); + let (mut proxy_writer, proxy_to_router_reader) = tokio::io::duplex(4096); + let external_proxy_transport = ByteStreams::new( + router_to_proxy_writer.compat_write(), + proxy_to_router_reader.compat(), + ); + let router = if protocol_version == ProtocolVersion::V1 { + Proxy.protocol_router().with_v1(external_proxy_transport) + } else { + Proxy.protocol_router().with_v2(external_proxy_transport) + }; + let router_task = tokio::spawn(router.connect_to(conductor_transport)); + let mut proxy_reader = BufReader::new(proxy_reader); + + let initialize = serde_json::json!([ + { + "jsonrpc": "2.0", + "id": 99, + "result": null, + }, + { + "jsonrpc": "2.0", + "id": 1, + "method": "_proxy/initialize", + "params": initialize_params_with_extensions(protocol_version)?, + }, + { + "jsonrpc": "2.0", + "method": "_future/notification", + "params": { "preserved": true }, + }, + ]); + write_wire_json(&mut conductor_writer, &initialize).await?; + assert_eq!(read_wire_json(&mut proxy_reader).await?, initialize); + + conductor_writer + .shutdown() + .await + .map_err(Error::into_internal_error)?; + proxy_writer + .shutdown() + .await + .map_err(Error::into_internal_error)?; + router_task.await.map_err(Error::into_internal_error)??; + } + + Ok(()) +}