diff --git a/md/conductor.md b/md/conductor.md index 50c89a47..b03a0e07 100644 --- a/md/conductor.md +++ b/md/conductor.md @@ -81,8 +81,11 @@ The command-line component provider, `AgentOnly`, `ProxiesAndAgent`, and static proxy vectors use their components for every version, so each supplied component must support the selected version. Use `Agent.protocol_router()` or `Proxy.protocol_router()` when a static component has separate -implementations. A custom instantiator rejects a version it does not support -(for example with `SelectedInitializeRequest::into_v1`); the error answers the +implementations. A custom instantiator that supports only a lower version +negotiates down by returning the request for that version (for example with +`SelectedInitializeRequest::into_v1`), and the client receives an `initialize` +response for it. An instantiator returns `InstantiatedChain` (agent mode) or +`InstantiatedProxies` (proxy mode). An instantiator error answers the `initialize` request and leaves the connection in a failed state that rejects later traffic. A first message that is not a valid initialization is answered with an error and leaves the conductor uninitialized, so the client can retry. @@ -187,9 +190,11 @@ orchestration from silently changing session MCP declarations. See [MCP Bridge](./mcp-bridge.md). The polyfill supports v1 by default. For a draft-v2 chain, enable -`unstable_protocol_v2` on both the conductor and polyfill crates; without the -polyfill feature, it rejects v2 initialization instead of interpreting v2 -traffic as v1. +`unstable_protocol_v2` on both the conductor and the polyfill crate; the +conductor's feature does not enable the polyfill's. A polyfill built without +the feature negotiates its `proxy/initialize` down to v1 like any v1-only +proxy, so the whole chain, and the `initialize` response the client +receives, use v1. ## Tracing diff --git a/md/mcp-bridge.md b/md/mcp-bridge.md index 5aee9643..d02dcb63 100644 --- a/md/mcp-bridge.md +++ b/md/mcp-bridge.md @@ -100,10 +100,11 @@ polyfill: into the corresponding HTTP JSON-RPC response. 5. Cancels only that operation if its HTTP response stream is dropped. -Enable the polyfill crate's `unstable_session_fork` feature when adapting fork -requests. Stable v1 setup includes `session/new`, `session/load`, and -`session/resume`; draft v2 includes `session/new` and `session/resume`. Both -versions include `session/fork` when `unstable_session_fork` is enabled. +The polyfill adapts the declarations of every session setup request: v1 +`session/new`, `session/load`, `session/resume`, and `session/fork`, and draft +v2 `session/new`, `session/resume`, and `session/fork`. With the polyfill +crate's `unstable_session_fork` feature it also validates `session/fork` +requests against the schema's fork type. Without it, the agent validates them. Declarations using another transport are left unchanged, including extension transports represented by v2's `McpServer::Other`. diff --git a/md/migration-v5.md b/md/migration-v5.md index 895947df..9531b166 100644 --- a/md/migration-v5.md +++ b/md/migration-v5.md @@ -930,6 +930,72 @@ implemented outside the conductor crate. Its methods take a `ConductorRunner`, whose fields are private, so no external implementation could be written before either. Use the provided `Agent` and `Proxy` host roles. +`ConductorHostRole` is now a marker trait: its `Instantiator` type and its +`initialize_method`, `select_protocol_version`, `instantiate` and +`handle_dispatch` methods moved to a private supertrait and take an argument +only the conductor crate can create, so code outside it cannot call them. +`ConductorImpl::new` +is private. Create conductors with `ConductorImpl::new_agent` or +`ConductorImpl::new_proxy`. + +### Conductor internals are no longer exported + +The conductor crate exports its API explicitly instead of through a glob. +`ConductorRunner`, `ConductorMessage`, `ConductorSuccessor`, `ComponentIndex` +and `SourceComponentIndex` are no longer public; they only described the +conductor's internal routing. + +### Instantiators return `InstantiatedChain` and `InstantiatedProxies` + +`InstantiateProxiesAndAgent` and `InstantiateProxies`, including instantiator +closures, return `#[non_exhaustive]` structs instead of tuples. The +`ProxiesAndAgentComponents` type alias is removed. + +| Before | After | +| --- | --- | +| `Ok((request, proxies, agent))` | `Ok(InstantiatedChain::new(request, proxies, agent))` | +| `Ok((request, proxies))` | `Ok(InstantiatedProxies::new(request, proxies))` | +| `ProxiesAndAgentComponents` | `InstantiatedChain` | + +`new` takes `impl Into`, so a `v1::InitializeRequest` +needs no `.into()`. + +### Instantiators can negotiate down + +`SelectedInitializeRequest::into_v1` now converts a v2 request to v1 the way +the SDK's agents negotiate `initialize`, instead of failing. An instantiator +may return a request for a lower version than the one the conductor selected, +and the chain then initializes with that version. A v1-only instantiator that +calls `into_v1()?` therefore accepts a v2 client and answers it with a v1 +`initialize` response, where it used to reject it with `-32600`. Returning a +request for a version above the selected one is still rejected. + +### Other conductor changes + +- `ConductorArgs`, `ConductorCommand` and its variants are `#[non_exhaustive]`. + Create `ConductorArgs` by parsing arguments with `clap::Parser`. +- `trace::Protocol` is renamed to `trace::MessageProtocol`, so it no longer + shares a name with the core `Protocol` trait. The serialized trace format is + unchanged. +- The conductor's `unstable_protocol_v2` feature no longer enables the + polyfill's feature (it only did so for the conductor's own tests). Enable + `unstable_protocol_v2` on `agent-client-protocol-polyfill` yourself for a v2 + chain. +- `McpOverAcpPolyfill` is `#[non_exhaustive]`. Create it with + `McpOverAcpPolyfill::http()`. +- The polyfill rewrites `McpServer::Acp` entries in `session/fork` in every + build. It used to forward them unchanged unless `unstable_session_fork` was + enabled. +- The polyfill forwards a repeated `proxy/initialize` on the connection's + protocol version instead of closing its connection. An invalid + `protocolVersion` is answered with `-32602` instead of `-32603`, and a + request for a different version than the first one with `-32600`. +- Conductor traces record the messages between the last proxy and the agent. + Trace snapshots and tools that read traces see these extra events. +- `agent_client_protocol_trace_viewer::TraceViewerConfig` is + `#[non_exhaustive]`. Start from `TraceViewerConfig::default()` and use the + `port` and `open_browser` setters. `TraceSource` is no longer public. + ## The rmcp integration moves behind the `rmcp_3` feature `agent-client-protocol-rmcp` now has one module per supported rmcp major diff --git a/md/protocol-v2.md b/md/protocol-v2.md index 181ae89e..648859ad 100644 --- a/md/protocol-v2.md +++ b/md/protocol-v2.md @@ -368,13 +368,16 @@ 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 -receive a `SelectedInitializeRequest` whose variant is the selected version, -and reject the versions they do not support (for example with -`SelectedInitializeRequest::into_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 -serialized request authoritative. The conductor pins `protocolVersion` to its -selected version even if an instantiator attempts to change it. +receive a `SelectedInitializeRequest` whose variant is the selected version. +An instantiator that supports only a lower version negotiates down by +returning the request for that version (for example with +`SelectedInitializeRequest::into_v1`, which keeps what parses under 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 serialized request +authoritative. The conductor sets `protocolVersion` to the version of the +returned variant even if an instantiator attempts to change the field, and +rejects a variant above the selected version. ### MCP compatibility polyfill @@ -383,7 +386,7 @@ The concrete participate in a v2 conductor chain when its `unstable_protocol_v2` feature is enabled. It selects v1 or v2 from `proxy/initialize`, uses that version's MCP capability and wire types, and adapts native `McpServer::Acp` declarations in -v2 `session/new`, `session/resume`, and feature-gated `session/fork` requests. +v2 `session/new`, `session/resume`, and `session/fork` requests. Other declarations and unrelated request fields remain unchanged. See [MCP-over-ACP Compatibility Bridge](./mcp-bridge.md) for placement and feature configuration. diff --git a/src/agent-client-protocol-conductor/Cargo.toml b/src/agent-client-protocol-conductor/Cargo.toml index 2d8ca99f..8ad9143e 100644 --- a/src/agent-client-protocol-conductor/Cargo.toml +++ b/src/agent-client-protocol-conductor/Cargo.toml @@ -10,16 +10,25 @@ description = "Conductor for orchestrating Agent Client Protocol proxy chains" keywords = ["acp", "agent", "conductor", "ai"] categories = ["development-tools"] +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + [[bin]] name = "agent-client-protocol-conductor" path = "src/main.rs" +# Skipped unless the polyfill's own v2 feature is enabled too, as +# `cargo test --workspace --all-features` does. +[[test]] +name = "mcp_over_acp_polyfill_v2" +required-features = ["unstable_protocol_v2", "agent-client-protocol-polyfill/unstable_protocol_v2"] + [features] default = [] -unstable_protocol_v2 = [ - "agent-client-protocol/unstable_protocol_v2", - "agent-client-protocol-polyfill/unstable_protocol_v2", -] +# The polyfill is a separate crate: enable its own `unstable_protocol_v2` +# feature to use it in a v2 chain. +unstable_protocol_v2 = ["agent-client-protocol/unstable_protocol_v2"] [dependencies] agent-client-protocol = { workspace = true, features = ["process", "stdio", "unstable_mcp_over_acp"] } diff --git a/src/agent-client-protocol-conductor/src/conductor.rs b/src/agent-client-protocol-conductor/src/conductor.rs index f1d2bf75..3fc6434a 100644 --- a/src/agent-client-protocol-conductor/src/conductor.rs +++ b/src/agent-client-protocol-conductor/src/conductor.rs @@ -96,10 +96,13 @@ //! //! ```ignore //! use agent_client_protocol::{Client, Conductor, DynConnectTo}; -//! use agent_client_protocol_conductor::{ConductorImpl, SelectedInitializeRequest}; +//! use agent_client_protocol_conductor::{ +//! ConductorImpl, InstantiatedChain, SelectedInitializeRequest, +//! }; //! //! let conductor = ConductorImpl::new_agent("my-conductor", |init_req: SelectedInitializeRequest| async move { -//! // This chain supports only v1; reject any other selected version. +//! // This chain supports only v1: a request for a newer version is +//! // negotiated down to v1. //! let init_req = init_req.into_v1()?; //! let mut proxies: Vec> = Vec::new(); //! if has_auth_capability(&init_req) { @@ -107,7 +110,7 @@ //! } //! //! let agent: DynConnectTo = DynConnectTo::new(make_agent()); -//! Ok((init_req.into(), proxies, agent)) +//! Ok(InstantiatedChain::new(init_req, proxies, agent)) //! }); //! ``` //! @@ -155,7 +158,9 @@ use tracing::{debug, info}; /// chain's components. The conductor selects the version before /// instantiation, as an agent does: v2 for a request for v2 or newer (with the /// `unstable_protocol_v2` feature) and v1 otherwise. A conductor acting as a -/// proxy negotiates its `proxy/initialize` the same way. +/// proxy negotiates its `proxy/initialize` the same way. An instantiator may +/// return the request for a lower version than the one selected, for example +/// with [`Self::into_v1`], to negotiate the chain down to that version. /// /// The selected version is the highest the chain may speak. The conductor /// forwards the request at that version, and each proxy negotiates its @@ -197,18 +202,33 @@ impl SelectedInitializeRequest { } } - /// The v1 request, for an instantiator that only supports v1. + /// The request as a v1 request, for an instantiator that only supports + /// v1. + /// + /// A request for a newer version is negotiated down to v1 the way the + /// SDK's agents negotiate `initialize`: only what parses under v1, such + /// as the capabilities v1 shares with the newer version, is kept. An + /// instantiator that returns the v1 request initializes the chain with + /// v1, and the client receives a v1 `initialize` response. /// /// # Errors /// - /// Returns an invalid-request error naming the selected version when the - /// conductor selected another version, so a v1-only instantiator can - /// reject it with `request.into_v1()?`. + /// Returns an error when the request cannot be represented as a v1 + /// request. pub fn into_v1(self) -> Result { match self { Self::V1(request) => Ok(request), #[cfg(feature = "unstable_protocol_v2")] - other => Err(unsupported_by_instantiator(other.protocol_version())), + Self::V2(request) => { + let params = serde_json::to_value(&request).map_err(Error::into_internal_error)?; + let params = serde_json::Value::Object( + agent_client_protocol::util::initialize_params_for_version( + ¶ms, + ProtocolVersion::V1, + )?, + ); + v1::InitializeRequest::parse_message("initialize", ¶ms) + } } } @@ -259,15 +279,6 @@ impl From for SelectedInitializeRequest { } } -/// The error an instantiator returns for a protocol version it does not -/// support. -#[cfg(feature = "unstable_protocol_v2")] -fn unsupported_by_instantiator(version: ProtocolVersion) -> Error { - Error::invalid_request().data(format!( - "this conductor instantiator does not support ACP protocol version {version}" - )) -} - /// The protocol version requested by an `initialize` or `proxy/initialize` /// request. fn requested_protocol_version(request: &UntypedMessage) -> Result { @@ -285,9 +296,11 @@ fn requested_protocol_version(request: &UntypedMessage) -> Result Result { let selected = original.protocol_version(); - if modified.protocol_version() != selected { + if modified.protocol_version() > selected { return Err(Error::invalid_request().data(format!( - "the conductor instantiator must return an initialize request for the selected ACP protocol version {selected}, not {}", + "the conductor instantiator must return an initialize request for the selected ACP protocol version {selected} or lower, not {}", modified.protocol_version(), ))); } @@ -327,12 +340,15 @@ struct AcceptedInitialize { /// Returns `None` after answering a message that cannot initialize the /// conductor: the conductor stays uninitialized and the client may retry. /// Notifications before initialization are dropped. -fn accept_initialize( +fn accept_initialize( host: &Host, message: Dispatch, ) -> Result, Error> { let expected = || { - Error::invalid_request().data(format!("expected `{}` request", host.initialize_method())) + Error::invalid_request().data(format!( + "expected `{}` request", + host.initialize_method(Token::new()) + )) }; let (request, responder) = match message { Dispatch::Request(request, responder) => (request, responder), @@ -348,9 +364,9 @@ fn accept_initialize( return Ok(None); } }; - let accepted = if request.method() == host.initialize_method() { + let accepted = if request.method() == host.initialize_method(Token::new()) { requested_protocol_version(&request).and_then(|requested| { - let selected = host.select_protocol_version(requested)?; + let selected = host.select_protocol_version(Token::new(), requested)?; let parsed = SelectedInitializeRequest::parse(selected, request.params())?; Ok((requested, parsed)) }) @@ -376,6 +392,8 @@ fn accept_initialize( /// It maintains connections to all components in the chain and routes messages /// bidirectionally between the editor, components, and agent. /// +/// Create one with [`ConductorImpl::new_agent`] or +/// [`ConductorImpl::new_proxy`]. #[derive(Debug)] pub struct ConductorImpl { host: Host, @@ -385,7 +403,7 @@ pub struct ConductorImpl { } impl ConductorImpl { - pub fn new(host: Host, name: impl ToString, instantiator: Host::Instantiator) -> Self { + fn new(host: Host, name: impl ToString, instantiator: Host::Instantiator) -> Self { ConductorImpl { name: name.to_string(), host, @@ -522,21 +540,19 @@ impl ConnectTo for ConductorImpl { +struct ConductorMessageHandler { conductor_tx: mpsc::Sender, host: Host, } -impl HandleDispatchFrom - for ConductorMessageHandler -{ +impl HandleDispatchFrom for ConductorMessageHandler { async fn handle_dispatch_from( &mut self, message: Dispatch, connection: agent_client_protocol::ConnectionTo, ) -> Result, agent_client_protocol::Error> { self.host - .handle_dispatch(message, connection, &mut self.conductor_tx) + .handle_dispatch(Token::new(), message, connection, &mut self.conductor_tx) .await } @@ -545,14 +561,15 @@ impl HandleDispatchFrom } } -/// The conductor manages the proxy chain lifecycle and message routing. -/// -/// It maintains connections to all components in the chain and routes messages -/// bidirectionally between the editor, components, and agent. +/// The conductor's central actor: it owns the chain's connections and routes +/// every [`ConductorMessage`]. /// +/// Declared `pub` only because the sealed [`HostRole`] trait names it. The +/// `conductor` module is private and does not re-export it, so it is not part +/// of the crate's public API. pub struct ConductorRunner where - Host: ConductorHostRole, + Host: HostRole, { conductor_rx: mpsc::Receiver, @@ -588,7 +605,7 @@ enum InitializationState { impl std::fmt::Debug for ConductorRunner where - Host: ConductorHostRole, + Host: HostRole, { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ConductorRunner") @@ -603,7 +620,7 @@ where impl RunWithConnectionTo for ConductorRunner where - Host: ConductorHostRole, + Host: HostRole, { async fn run_with_connection_to( mut self, @@ -624,7 +641,7 @@ where impl ConductorRunner where - Host: ConductorHostRole, + Host: HostRole, { /// Recursively spawns components and builds the proxy chain. /// @@ -871,7 +888,10 @@ where "ensure_initialized: instantiating chain" ); let original = request.clone(); - let forwarded = match host.instantiate(request, client, instantiator, self).await { + let forwarded = match host + .instantiate(Token::new(), request, client, instantiator, self) + .await + { Ok(modified) => { forwarded_initialize_request(params, requested, &original, modified) } @@ -1148,14 +1168,16 @@ fn forward_message_to_agent( /// The error for a `proxy/initialize` request that arrives from outside the /// conductor: only the conductor initializes its proxies. fn initialize_proxy_from_outside() -> Error { - Error::invalid_request().data("initialize/proxy requests are only sent by the conductor") + Error::invalid_request().data(format!( + "`{METHOD_INITIALIZE_PROXY}` requests are only sent by the conductor" + )) } /// Identifies a component in the conductor's chain for tracing purposes. /// /// Used to track message sources and destinations through the proxy chain. -#[derive(Debug, Clone, Copy)] -pub enum ComponentIndex { +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ComponentIndex { /// The client (editor) at the start of the chain. Client, @@ -1168,18 +1190,17 @@ pub enum ComponentIndex { impl ComponentIndex { /// Return the index for the predecessor of `proxy_index`, which might be `Client`. - #[must_use] - pub fn predecessor_of(proxy_index: usize) -> Self { + pub(crate) fn predecessor_of(proxy_index: usize) -> Self { match proxy_index.checked_sub(1) { Some(p_i) => ComponentIndex::Proxy(p_i), None => ComponentIndex::Client, } } - /// Return the index for the predecessor of `proxy_index`, which might be `Client`. - #[must_use] - pub fn successor_of(proxy_index: usize, num_proxies: usize) -> Self { - if proxy_index == num_proxies { + /// Return the index for the successor of `proxy_index` in a chain of + /// `num_proxies` proxies, which is `Agent` for the last proxy. + pub(crate) fn successor_of(proxy_index: usize, num_proxies: usize) -> Self { + if proxy_index + 1 == num_proxies { ComponentIndex::Agent } else { ComponentIndex::Proxy(proxy_index + 1) @@ -1192,6 +1213,9 @@ impl ComponentIndex { /// This enum handles the fact that the conductor may receive messages from two different sources: /// 1. From one of its managed components (identified by index) /// 2. From the conductor's own successor in a larger proxy chain (when in proxy mode) +/// +/// Declared `pub` only because [`ConductorMessage`] names it; it is not part +/// of the crate's public API. #[derive(Debug, Clone, Copy)] pub enum SourceComponentIndex { /// Message from a specific component at the given index in the managed chain. @@ -1201,37 +1225,93 @@ pub enum SourceComponentIndex { Successor, } +/// The result of [`InstantiateProxies::instantiate_proxies`]: the (possibly +/// modified) `initialize` request and the proxy components of a conductor +/// acting as a proxy. +/// +/// New fields may be added in later releases, so create one with +/// [`InstantiatedProxies::new`]. +#[derive(Debug)] +#[non_exhaustive] +pub struct InstantiatedProxies { + /// The request to forward into the chain. + /// + /// See [`SelectedInitializeRequest`] for how the conductor forwards it. + pub request: SelectedInitializeRequest, + /// The proxy components, in chain order. + pub proxies: Vec>, +} + +impl InstantiatedProxies { + /// Create the result from the request to forward and the proxies, in + /// chain order. + pub fn new( + request: impl Into, + proxies: Vec>, + ) -> Self { + Self { + request: request.into(), + proxies, + } + } +} + +/// The result of [`InstantiateProxiesAndAgent::instantiate_proxies_and_agent`]: +/// the (possibly modified) `initialize` request, the proxy components, and +/// the agent component of a conductor acting as an agent. +/// +/// New fields may be added in later releases, so create one with +/// [`InstantiatedChain::new`]. +#[derive(Debug)] +#[non_exhaustive] +pub struct InstantiatedChain { + /// The request to forward into the chain. + /// + /// See [`SelectedInitializeRequest`] for how the conductor forwards it. + pub request: SelectedInitializeRequest, + /// The proxy components, in chain order. + pub proxies: Vec>, + /// The agent at the end of the chain. + pub agent: DynConnectTo, +} + +impl InstantiatedChain { + /// Create the result from the request to forward, the proxies in chain + /// order, and the agent. + pub fn new( + request: impl Into, + proxies: Vec>, + agent: DynConnectTo, + ) -> Self { + Self { + request: request.into(), + proxies, + agent, + } + } +} + /// Trait for lazy proxy instantiation (proxy mode). /// -/// Used by conductors in proxy mode (`ConductorToConductor`) where all components -/// are proxies that forward to an outer conductor. +/// Used by conductors created with [`ConductorImpl::new_proxy`], where all +/// components are proxies and the conductor's own successor is the agent. pub trait InstantiateProxies: Send { /// Instantiate proxy components based on the `initialize` request. /// /// The request is parsed for the protocol version the conductor selected; /// see [`SelectedInitializeRequest`]. An implementation that supports only - /// some versions rejects the others with an error, for example with + /// lower versions negotiates down, for example with /// [`SelectedInitializeRequest::into_v1`]. /// - /// Returns the (possibly modified) request and the proxy components, typed - /// as `DynConnectTo` since proxies communicate with the - /// conductor. + /// Returns the (possibly modified) request and the proxy components. fn instantiate_proxies( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result< - (SelectedInitializeRequest, Vec>), - agent_client_protocol::Error, - >, - >; + ) -> BoxFuture<'static, Result>; } /// Simple implementation: provide all proxy components unconditionally, for /// every protocol version. -/// -/// Requires `T: ConnectTo`. impl InstantiateProxies for Vec where T: ConnectTo + 'static, @@ -1239,17 +1319,10 @@ where fn instantiate_proxies( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result< - (SelectedInitializeRequest, Vec>), - agent_client_protocol::Error, - >, - > { + ) -> BoxFuture<'static, Result> { Box::pin(async move { - let components: Vec> = - (*self).into_iter().map(|c| DynConnectTo::new(c)).collect(); - Ok((req, components)) + let components = (*self).into_iter().map(DynConnectTo::new).collect(); + Ok(InstantiatedProxies::new(req, components)) }) } } @@ -1258,59 +1331,34 @@ where impl InstantiateProxies for F where F: FnOnce(SelectedInitializeRequest) -> Fut + Send + 'static, - Fut: std::future::Future< - Output = Result< - (SelectedInitializeRequest, Vec>), - agent_client_protocol::Error, - >, - > + Send - + 'static, + Fut: Future> + Send + 'static, { fn instantiate_proxies( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result< - (SelectedInitializeRequest, Vec>), - agent_client_protocol::Error, - >, - > { + ) -> BoxFuture<'static, Result> { Box::pin(async move { (*self)(req).await }) } } -/// The result of [`InstantiateProxiesAndAgent::instantiate_proxies_and_agent`]: -/// the (possibly modified) request, the proxy components, and the agent -/// component. -pub type ProxiesAndAgentComponents = ( - SelectedInitializeRequest, - Vec>, - DynConnectTo, -); - /// Trait for lazy proxy and agent instantiation (agent mode). /// -/// Used by conductors in agent mode (`ConductorToClient`) where there are -/// zero or more proxies followed by an agent component. +/// Used by conductors created with [`ConductorImpl::new_agent`], where zero +/// or more proxies are followed by an agent component. pub trait InstantiateProxiesAndAgent: Send { /// Instantiate proxy and agent components based on the `initialize` request. /// /// The request is parsed for the protocol version the conductor selected; /// see [`SelectedInitializeRequest`]. An implementation that supports only - /// some versions rejects the others with an error, for example with + /// lower versions negotiates down, for example with /// [`SelectedInitializeRequest::into_v1`]. /// - /// Returns the (possibly modified) request, a vector of proxy components - /// (typed as `DynConnectTo`), and the agent component - /// (typed as `DynConnectTo`). + /// Returns the (possibly modified) request, the proxy components, and the + /// agent component. fn instantiate_proxies_and_agent( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result, - >; + ) -> BoxFuture<'static, Result>; } /// Wrapper to convert a single agent component (no proxies) into @@ -1322,11 +1370,14 @@ impl + 'static> InstantiateProxiesAndAgent for AgentOnly fn instantiate_proxies_and_agent( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result, - > { - Box::pin(async move { Ok((req, Vec::new(), DynConnectTo::new(self.0))) }) + ) -> BoxFuture<'static, Result> { + Box::pin(async move { + Ok(InstantiatedChain::new( + req, + Vec::new(), + DynConnectTo::new(self.0), + )) + }) } } @@ -1379,11 +1430,8 @@ impl InstantiateProxiesAndAgent for ProxiesAndAgent { fn instantiate_proxies_and_agent( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result, - > { - Box::pin(async move { Ok((req, self.proxies, self.agent)) }) + ) -> BoxFuture<'static, Result> { + Box::pin(async move { Ok(InstantiatedChain::new(req, self.proxies, self.agent)) }) } } @@ -1391,18 +1439,12 @@ impl InstantiateProxiesAndAgent for ProxiesAndAgent { impl InstantiateProxiesAndAgent for F where F: FnOnce(SelectedInitializeRequest) -> Fut + Send + 'static, - Fut: std::future::Future< - Output = Result, - > + Send - + 'static, + Fut: Future> + Send + 'static, { fn instantiate_proxies_and_agent( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result, - > { + ) -> BoxFuture<'static, Result> { Box::pin(async move { (*self)(req).await }) } } @@ -1416,6 +1458,9 @@ where /// /// All spawned tasks send messages via this enum through a shared channel, /// allowing centralized routing logic in the conductor runner's event loop. +/// +/// Declared `pub` only because the sealed [`HostRole`] trait names it; it is +/// not part of the crate's public API. #[derive(Debug)] pub enum ConductorMessage { /// If this message is a request or notification, then it is going "left-to-right" @@ -1439,52 +1484,18 @@ pub enum ConductorMessage { }, } -/// Trait implemented for the two links the conductor can use: -/// -/// * ConductorToClient -- conductor is acting as an agent, so when its last proxy sends to its successor, the conductor sends that message to its agent component -/// * ConductorToConductor -- conductor is acting as a proxy, so when its last proxy sends to its successor, the (inner) conductor sends that message to its successor, via the outer conductor +/// The role a conductor plays toward its own client: [`Agent`] for a +/// conductor created with [`ConductorImpl::new_agent`], whose last proxy talks +/// to the conductor's agent component, or [`Proxy`] for one created with +/// [`ConductorImpl::new_proxy`], whose last proxy talks to the conductor's own +/// successor. /// /// This trait is sealed: it is implemented for [`Agent`] and [`Proxy`] and /// cannot be implemented outside this crate. -pub trait ConductorHostRole: sealed::Sealed + Role> { - /// The type used to instantiate components for this link type. - type Instantiator: Send; - - /// The method of the request that initializes the conductor: - /// `initialize`, or `proxy/initialize` for a conductor acting as a proxy. - fn initialize_method(&self) -> &'static str; - - /// Select the chain's protocol version for an initialization that - /// requested `requested`. - /// - /// # Errors - /// - /// Returns the error that answers the initialization when the conductor - /// supports no suitable version. - fn select_protocol_version(&self, requested: ProtocolVersion) - -> Result; +pub trait ConductorHostRole: sealed::HostRole + Role> {} - /// Instantiate and spawn the chain's components for `request`. - /// - /// Takes ownership of the instantiator and returns the (possibly - /// modified) request to forward into the chain. An error answers the - /// initialization and every later message. - fn instantiate( - &self, - request: SelectedInitializeRequest, - connection: ConnectionTo, - instantiator: Self::Instantiator, - runner: &mut ConductorRunner, - ) -> impl Future> + Send; - - /// Handle an incoming message from the client or conductor, depending on `Self` - fn handle_dispatch( - &self, - message: Dispatch, - connection: ConnectionTo, - conductor_tx: &mut mpsc::Sender, - ) -> impl Future, agent_client_protocol::Error>> + Send; -} +impl ConductorHostRole for Agent {} +impl ConductorHostRole for Proxy {} /// Select the highest version this build supports at or below `requested`, /// as an agent does, keeping v1's fallback-to-latest rule for a request below @@ -1498,19 +1509,80 @@ fn select_like_an_agent(requested: ProtocolVersion) -> ProtocolVersion { ProtocolVersion::V1 } +pub(crate) use sealed::{HostRole, Token}; + mod sealed { - /// Prevents implementations of [`super::ConductorHostRole`] outside this crate. - pub trait Sealed {} + use super::*; - impl Sealed for agent_client_protocol::Agent {} - impl Sealed for agent_client_protocol::Proxy {} + /// A value only this crate can create. + /// + /// Every [`HostRole`] method takes one, so code outside this crate cannot + /// call those methods through a [`ConductorHostRole`] bound. + #[derive(Debug, Clone, Copy)] + pub struct Token(()); + + impl Token { + pub(crate) const fn new() -> Self { + Self(()) + } + } + + /// The conductor's behavior for each [`ConductorHostRole`]. + /// + /// Private to this crate, so [`ConductorHostRole`] cannot be implemented + /// outside it. Its methods take a [`Token`], so they cannot be called + /// outside it either. `Instantiator` is not a stable extension point. + pub trait HostRole: Role> { + /// The type used to instantiate components for this role. + type Instantiator: Send; + + /// The method of the request that initializes the conductor: + /// `initialize`, or `proxy/initialize` for a conductor acting as a proxy. + fn initialize_method(&self, token: Token) -> &'static str; + + /// Select the chain's protocol version for an initialization that + /// requested `requested`. + /// + /// # Errors + /// + /// Returns the error that answers the initialization when the conductor + /// supports no suitable version. + fn select_protocol_version( + &self, + token: Token, + requested: ProtocolVersion, + ) -> Result; + + /// Instantiate and spawn the chain's components for `request`. + /// + /// Takes ownership of the instantiator and returns the (possibly + /// modified) request to forward into the chain. An error answers the + /// initialization and every later message. + fn instantiate( + &self, + token: Token, + request: SelectedInitializeRequest, + connection: ConnectionTo, + instantiator: Self::Instantiator, + runner: &mut ConductorRunner, + ) -> impl Future> + Send; + + /// Handle an incoming message from the client or conductor, depending on `Self` + fn handle_dispatch( + &self, + token: Token, + message: Dispatch, + connection: ConnectionTo, + conductor_tx: &mut mpsc::Sender, + ) -> impl Future, Error>> + Send; + } } /// Conductor acting as an agent -impl ConductorHostRole for Agent { +impl HostRole for Agent { type Instantiator = Box; - fn initialize_method(&self) -> &'static str { + fn initialize_method(&self, _: Token) -> &'static str { "initialize" } @@ -1519,6 +1591,7 @@ impl ConductorHostRole for Agent { /// v1. fn select_protocol_version( &self, + _: Token, requested: ProtocolVersion, ) -> Result { Ok(select_like_an_agent(requested)) @@ -1526,13 +1599,17 @@ impl ConductorHostRole for Agent { async fn instantiate( &self, + _: Token, request: SelectedInitializeRequest, client_connection: ConnectionTo, instantiator: Self::Instantiator, runner: &mut ConductorRunner, ) -> Result { - let (modified_req, proxy_components, agent_component) = - instantiator.instantiate_proxies_and_agent(request).await?; + let InstantiatedChain { + request: modified_req, + proxies: proxy_components, + agent: agent_component, + } = instantiator.instantiate_proxies_and_agent(request).await?; debug!(?agent_component, "spawning agent"); @@ -1570,6 +1647,7 @@ impl ConductorHostRole for Agent { async fn handle_dispatch( &self, + _: Token, message: Dispatch, client_connection: ConnectionTo, conductor_tx: &mut mpsc::Sender, @@ -1593,10 +1671,10 @@ impl ConductorHostRole for Agent { } /// Conductor acting as a proxy -impl ConductorHostRole for Proxy { +impl HostRole for Proxy { type Instantiator = Box; - fn initialize_method(&self) -> &'static str { + fn initialize_method(&self, _: Token) -> &'static str { METHOD_INITIALIZE_PROXY } @@ -1605,6 +1683,7 @@ impl ConductorHostRole for Proxy { /// requested one, keeping v1's fallback-to-latest rule below v1. fn select_protocol_version( &self, + _: Token, requested: ProtocolVersion, ) -> Result { Ok(select_like_an_agent(requested)) @@ -1612,12 +1691,16 @@ impl ConductorHostRole for Proxy { async fn instantiate( &self, + _: Token, request: SelectedInitializeRequest, client_connection: ConnectionTo, instantiator: Self::Instantiator, runner: &mut ConductorRunner, ) -> Result { - let (modified_req, proxy_components) = instantiator.instantiate_proxies(request).await?; + let InstantiatedProxies { + request: modified_req, + proxies: proxy_components, + } = instantiator.instantiate_proxies(request).await?; runner.successor = Arc::new(GrandSuccessor); runner.spawn_proxies(client_connection, proxy_components)?; @@ -1627,6 +1710,7 @@ impl ConductorHostRole for Proxy { async fn handle_dispatch( &self, + _: Token, message: Dispatch, client_connection: ConnectionTo, conductor_tx: &mut mpsc::Sender, @@ -1676,7 +1760,8 @@ impl ConductorHostRole for Proxy { } } -pub trait ConductorSuccessor: Send + Sync + 'static { +/// Where the conductor's last proxy sends messages for its successor. +pub(crate) trait ConductorSuccessor: Send + Sync + 'static { /// Send a message to the successor. fn send_message<'a>( &self, @@ -1686,7 +1771,7 @@ pub trait ConductorSuccessor: Send + Sync + 'static { ) -> BoxFuture<'a, Result<(), agent_client_protocol::Error>>; } -impl ConductorSuccessor for agent_client_protocol::Error { +impl ConductorSuccessor for agent_client_protocol::Error { fn send_message<'a>( &self, _message: Dispatch, diff --git a/src/agent-client-protocol-conductor/src/lib.rs b/src/agent-client-protocol-conductor/src/lib.rs index 79c0e1b7..e38b1ef5 100644 --- a/src/agent-client-protocol-conductor/src/lib.rs +++ b/src/agent-client-protocol-conductor/src/lib.rs @@ -74,11 +74,14 @@ mod snoop; /// Trace event types for sequence diagram viewer pub mod trace; -pub use self::conductor::*; +pub use self::conductor::{ + AgentOnly, ConductorHostRole, ConductorImpl, InstantiateProxies, InstantiateProxiesAndAgent, + InstantiatedChain, InstantiatedProxies, ProxiesAndAgent, SelectedInitializeRequest, +}; use clap::{Parser, Subcommand}; -use agent_client_protocol::{AcpAgent, Stdio}; +use agent_client_protocol::{AcpAgent, BoxFuture, Stdio}; use agent_client_protocol::{Conductor, DynConnectTo}; use tracing::Instrument; use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt}; @@ -96,16 +99,10 @@ impl InstantiateProxies for CommandLineComponents { fn instantiate_proxies( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result< - (SelectedInitializeRequest, Vec>), - agent_client_protocol::Error, - >, - > { + ) -> BoxFuture<'static, Result> { Box::pin(async move { let proxies = self.0.into_iter().map(DynConnectTo::new).collect(); - Ok((req, proxies)) + Ok(InstantiatedProxies::new(req, proxies)) }) } } @@ -114,10 +111,7 @@ impl InstantiateProxiesAndAgent for CommandLineComponents { fn instantiate_proxies_and_agent( self: Box, req: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result, - > { + ) -> BoxFuture<'static, Result> { Box::pin(async move { let mut iter = self.0.into_iter().peekable(); let mut proxies: Vec> = Vec::new(); @@ -129,7 +123,7 @@ impl InstantiateProxiesAndAgent for CommandLineComponents { } else { // Last element is the agent let agent = DynConnectTo::new(component); - return Ok((req, proxies, agent)); + return Ok(InstantiatedChain::new(req, proxies, agent)); } } @@ -149,8 +143,13 @@ impl trace::WriteEvent for TraceHandleWriter { } } +/// Command-line arguments of the `agent-client-protocol-conductor` binary. +/// +/// New flags may be added in later releases, so create one by parsing +/// arguments with [`clap::Parser`]. #[derive(Parser, Debug)] #[command(author, version, about, long_about = None)] +#[non_exhaustive] pub struct ConductorArgs { /// Enable debug logging of all stdin/stdout/stderr from components #[arg(long)] @@ -179,9 +178,12 @@ pub struct ConductorArgs { pub command: ConductorCommand, } +/// The conductor's subcommand: the role it plays toward its client. #[derive(Subcommand, Debug)] +#[non_exhaustive] pub enum ConductorCommand { /// Run as agent orchestrator managing a proxy chain + #[non_exhaustive] Agent { /// Name of the agent #[arg(short, long, default_value = "conductor")] @@ -192,6 +194,7 @@ pub enum ConductorCommand { }, /// Run as a proxy orchestrating a proxy chain + #[non_exhaustive] Proxy { /// Name of the proxy #[arg(short, long, default_value = "conductor")] diff --git a/src/agent-client-protocol-conductor/src/trace.rs b/src/agent-client-protocol-conductor/src/trace.rs index c8646659..d4c1a043 100644 --- a/src/agent-client-protocol-conductor/src/trace.rs +++ b/src/agent-client-protocol-conductor/src/trace.rs @@ -20,7 +20,7 @@ use agent_client_protocol::{ use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; -use crate::ComponentIndex; +use crate::conductor::ComponentIndex; use crate::snoop::SnooperComponent; /// A trace event representing message flow between components. @@ -38,11 +38,11 @@ pub enum TraceEvent { Notification(NotificationEvent), } -/// Protocol type for messages. +/// The protocol a traced message belongs to. #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] #[non_exhaustive] -pub enum Protocol { +pub enum MessageProtocol { /// Standard ACP protocol messages. Acp, /// MCP messages carried over ACP. @@ -57,7 +57,7 @@ pub struct RequestEvent { pub ts: f64, /// Protocol: ACP or MCP. - pub protocol: Protocol, + pub protocol: MessageProtocol, /// Source component (e.g., "client", "proxy:0", "proxy:1", "agent"). pub from: String, @@ -101,7 +101,7 @@ pub struct ResponseEvent { /// Whether an error belongs to the ACP binding or the inner MCP peer. /// Older trace files omit this provenance. #[serde(default, skip_serializing_if = "Option::is_none")] - pub error_domain: Option, + pub error_domain: Option, /// Response result or error object. pub payload: serde_json::Value, @@ -115,7 +115,7 @@ pub struct NotificationEvent { pub ts: f64, /// Protocol: ACP or MCP. - pub protocol: Protocol, + pub protocol: MessageProtocol, /// Source component. pub from: String, @@ -186,7 +186,7 @@ impl std::fmt::Debug for TraceWriter { } struct RequestDetails { - protocol: Protocol, + protocol: MessageProtocol, request_from: ComponentIndex, request_to: ComponentIndex, } @@ -226,7 +226,7 @@ impl TraceWriter { #[expect(clippy::too_many_arguments)] fn request( &mut self, - protocol: Protocol, + protocol: MessageProtocol, from: ComponentIndex, to: ComponentIndex, id: serde_json::Value, @@ -261,7 +261,7 @@ impl TraceWriter { from: ComponentIndex, to: ComponentIndex, id: serde_json::Value, - error_domain: Option, + error_domain: Option, mut payload: serde_json::Value, ) { redact_credentials(&mut payload); @@ -279,7 +279,7 @@ impl TraceWriter { /// Write a notification event. fn notification( &mut self, - protocol: Protocol, + protocol: MessageProtocol, from: ComponentIndex, to: ComponentIndex, method: impl Into, @@ -390,7 +390,7 @@ impl TraceWriter { successor_index: ComponentIndex, successor: Successor, id: Option, - protocol: Protocol, + protocol: MessageProtocol, method: String, params: serde_json::Value, ) { @@ -531,19 +531,19 @@ fn params_from_transport(params: Option) -> serde_json::Value /// Keep an MCP outcome separate from a failure of its ACP binding. fn response_outcome( - protocol: Protocol, + protocol: MessageProtocol, outer_error: bool, payload: serde_json::Value, -) -> (Option, serde_json::Value) { +) -> (Option, serde_json::Value) { if outer_error { - return (Some(Protocol::Acp), payload); + return (Some(MessageProtocol::Acp), payload); } - if protocol == Protocol::Mcp { + if protocol == MessageProtocol::Mcp { match serde_json::from_value::(payload.clone()) { Ok(MessageMcpResponse::Result { result, .. }) => return (None, result), Ok(MessageMcpResponse::Error { error, .. }) => { return ( - Some(Protocol::Mcp), + Some(MessageProtocol::Mcp), serde_json::to_value(error).expect("MCP errors contain only JSON values"), ); } @@ -706,7 +706,7 @@ struct TracedMessage { struct MessageInfo { successor: Successor, id: Option, - protocol: Protocol, + protocol: MessageProtocol, method: String, params: serde_json::Value, } @@ -729,7 +729,12 @@ impl MessageInfo { let untyped = UntypedMessage::parse_message(&req.method, ¶ms_from_transport(req.params)) .expect("untyped message is infallible"); - Self::from_untyped_request(Successor(false), Some(req.id), Protocol::Acp, untyped) + Self::from_untyped_request( + Successor(false), + Some(req.id), + MessageProtocol::Acp, + untyped, + ) } fn from_notification(notification: RawJsonRpcNotification) -> Self { @@ -738,13 +743,13 @@ impl MessageInfo { ¶ms_from_transport(notification.params), ) .expect("untyped message is infallible"); - Self::from_untyped_notification(Successor(false), Protocol::Acp, untyped) + Self::from_untyped_notification(Successor(false), MessageProtocol::Acp, untyped) } fn from_untyped_request( successor: Successor, id: Option, - protocol: Protocol, + protocol: MessageProtocol, untyped: UntypedMessage, ) -> Self { if let Ok(m) = SuccessorMessage::parse_message(&untyped.method, &untyped.params) { @@ -758,7 +763,7 @@ impl MessageInfo { return Self::from_untyped_request( successor, id, - Protocol::Mcp, + MessageProtocol::Mcp, UntypedMessage { method: m.method, params, @@ -771,7 +776,7 @@ impl MessageInfo { fn from_untyped_notification( successor: Successor, - protocol: Protocol, + protocol: MessageProtocol, untyped: UntypedMessage, ) -> Self { if let Ok(m) = SuccessorMessage::parse_message(&untyped.method, &untyped.params) { @@ -784,7 +789,7 @@ impl MessageInfo { .map_or(serde_json::Value::Null, serde_json::Value::Object); return Self::from_untyped_notification( successor, - Protocol::Mcp, + MessageProtocol::Mcp, UntypedMessage { method: m.method, params, @@ -798,7 +803,7 @@ impl MessageInfo { fn new( successor: Successor, id: Option, - protocol: Protocol, + protocol: MessageProtocol, untyped: UntypedMessage, ) -> Self { Self { @@ -816,7 +821,9 @@ mod tests { use agent_client_protocol::RawJsonRpcMessage; use serde_json::json; - use super::{MessageInfo, Protocol, ResponseEvent, redact_credentials, response_outcome}; + use super::{ + MessageInfo, MessageProtocol, ResponseEvent, redact_credentials, response_outcome, + }; #[test] fn http_url_redaction_handles_encoded_keys_duplicates_and_userinfo() { @@ -967,20 +974,20 @@ mod tests { fn mcp_and_binding_errors_keep_their_domains() { let inner = json!({"code":-32000,"message":"peer","data":null,"extension":true}); assert_eq!( - response_outcome(Protocol::Mcp, false, json!({"error":inner})), - (Some(Protocol::Mcp), inner) + response_outcome(MessageProtocol::Mcp, false, json!({"error":inner})), + (Some(MessageProtocol::Mcp), inner) ); let outer = json!({"code":-33002,"message":"binding failure"}); assert_eq!( - response_outcome(Protocol::Mcp, true, outer.clone()), - (Some(Protocol::Acp), outer) + response_outcome(MessageProtocol::Mcp, true, outer.clone()), + (Some(MessageProtocol::Acp), outer) ); for result in [ json!(null), json!({"resultType":"input_required","requestState":"opaque"}), ] { assert_eq!( - response_outcome(Protocol::Mcp, false, json!({"result":result})), + response_outcome(MessageProtocol::Mcp, false, json!({"result":result})), (None, result) ); } @@ -1077,7 +1084,7 @@ mod tests { // Schema 1.10.1 requires named inner parameters. Do not pretend this // malformed envelope was a successfully decoded MCP notification. - assert_eq!(info.protocol, Protocol::Acp); + assert_eq!(info.protocol, MessageProtocol::Acp); assert_eq!(info.method, "mcp/message"); assert_eq!(info.params, params); } @@ -1096,7 +1103,7 @@ mod tests { unreachable!("notification constructor") }; let info = MessageInfo::from_notification(notification); - assert_eq!(info.protocol, Protocol::Mcp); + assert_eq!(info.protocol, MessageProtocol::Mcp); assert_eq!(info.method, "notifications/progress"); assert_eq!(info.params, params); } diff --git a/src/agent-client-protocol-conductor/tests/initialization_v2.rs b/src/agent-client-protocol-conductor/tests/initialization_v2.rs index d1218505..bbc49b6f 100644 --- a/src/agent-client-protocol-conductor/tests/initialization_v2.rs +++ b/src/agent-client-protocol-conductor/tests/initialization_v2.rs @@ -11,8 +11,8 @@ use agent_client_protocol::{ JsonRpcRequest, JsonRpcResponse, Proxy, UntypedMessage, }; use agent_client_protocol_conductor::{ - ConductorImpl, InstantiateProxies, InstantiateProxiesAndAgent, ProxiesAndAgent, - ProxiesAndAgentComponents, SelectedInitializeRequest, + ConductorImpl, InstantiateProxies, InstantiateProxiesAndAgent, InstantiatedChain, + InstantiatedProxies, ProxiesAndAgent, SelectedInitializeRequest, }; use futures::{StreamExt as _, channel::mpsc}; use serde::{Deserialize, Serialize}; @@ -117,7 +117,7 @@ impl InstantiateProxiesAndAgent for MutatingInstantiator { fn instantiate_proxies_and_agent( self: Box, request: SelectedInitializeRequest, - ) -> futures::future::BoxFuture<'static, Result> { + ) -> futures::future::BoxFuture<'static, Result> { let SelectedInitializeRequest::V2(mut request) = request else { return Box::pin(async { Err(Error::internal_error() @@ -133,7 +133,7 @@ impl InstantiateProxiesAndAgent for MutatingInstantiator { } } let agent = self.agent; - Box::pin(async move { Ok((request.into(), Vec::new(), agent)) }) + Box::pin(async move { Ok(InstantiatedChain::new(request, Vec::new(), agent)) }) } } @@ -143,10 +143,7 @@ impl InstantiateProxies for VersionMutatingProxyInstantiator { fn instantiate_proxies( self: Box, request: SelectedInitializeRequest, - ) -> futures::future::BoxFuture< - 'static, - Result<(SelectedInitializeRequest, Vec>), Error>, - > { + ) -> futures::future::BoxFuture<'static, Result> { let SelectedInitializeRequest::V2(mut request) = request else { return Box::pin(async { Err(Error::internal_error() @@ -154,7 +151,7 @@ impl InstantiateProxies for VersionMutatingProxyInstantiator { }); }; request.protocol_version = ProtocolVersion::from(3_u16); - Box::pin(async move { Ok((request.into(), Vec::new())) }) + Box::pin(async move { Ok(InstantiatedProxies::new(request, Vec::new())) }) } } @@ -230,7 +227,7 @@ async fn run_with_conductor( } async fn run_raw_client_with_conductor( - components: ProxiesAndAgent, + components: impl InstantiateProxiesAndAgent + 'static, editor_task: impl AsyncFnOnce(ConnectionTo) -> Result<(), Error>, ) -> Result<(), Error> { let (editor_out, conductor_in) = duplex(4096); @@ -533,12 +530,14 @@ async fn v2_nested_proxy_instantiator_cannot_change_selected_version() -> Result } #[tokio::test] -async fn v2_nested_proxy_rejection_is_flushed_and_remains_failed() -> Result<(), Error> { - let v1_only_proxy = |request: SelectedInitializeRequest| async move { - let request = request.into_v1()?; - Ok::<_, Error>((request.into(), Vec::>::new())) +async fn v2_nested_proxy_instantiation_failure_is_flushed_and_remains_failed() -> Result<(), Error> +{ + let failing_proxy = |_request: SelectedInitializeRequest| async move { + Err::( + Error::invalid_request().data("nested instantiator refused the chain"), + ) }; - let nested_conductor = ConductorImpl::new_proxy("v1-only-proxy", v1_only_proxy); + let nested_conductor = ConductorImpl::new_proxy("failing-proxy", failing_proxy); let request = initialize_request(); run_with_conductor( @@ -551,11 +550,11 @@ async fn v2_nested_proxy_rejection_is_flushed_and_remains_failed() -> Result<(), ) .await .expect("nested proxy rejection should not hang") - .expect_err("v1-only nested proxy must reject v2 initialize"); + .expect_err("failing nested instantiator must reject initialize"); assert!( error .to_string() - .contains("instantiator does not support ACP protocol version 2"), + .contains("nested instantiator refused the chain"), "unexpected nested proxy rejection on attempt {attempt}: {error:?}" ); } @@ -1345,28 +1344,25 @@ async fn v2_nested_conductor_canonicalizes_future_version_fields() -> Result<(), } #[tokio::test] -async fn v2_initialize_rejects_v1_only_instantiator_with_response() -> Result<(), Error> { +async fn v2_instantiator_failure_is_answered_and_remains_failed() -> Result<(), Error> { let request = initialize_request(); - let v1_only_instantiator = |request: SelectedInitializeRequest| async move { - let _request = request.into_v1()?; - Err::( - Error::internal_error().data("v1 instantiator unexpectedly called for v2"), + let failing_instantiator = |_request: SelectedInitializeRequest| async move { + Err::( + Error::invalid_request().data("instantiator refused the chain"), ) }; - run_with_conductor(v1_only_instantiator, async move |cx| { + run_with_conductor(failing_instantiator, async move |cx| { for attempt in 1..=2 { let error = tokio::time::timeout( std::time::Duration::from_secs(2), cx.send_request(request.clone()).block_task(), ) .await - .expect("v1-only instantiator rejection should not hang") - .expect_err("v1-only instantiator must reject v2 initialize"); + .expect("instantiator rejection should not hang") + .expect_err("failing instantiator must reject initialize"); assert!( - error - .to_string() - .contains("does not support ACP protocol version 2"), - "unexpected v2 rejection on attempt {attempt}: {error:?}" + error.to_string().contains("instantiator refused the chain"), + "unexpected rejection on attempt {attempt}: {error:?}" ); } Ok(()) @@ -1374,6 +1370,82 @@ async fn v2_initialize_rejects_v1_only_instantiator_with_response() -> Result<() .await } +/// A v1-only instantiator negotiates a v2 client down to v1 with +/// [`SelectedInitializeRequest::into_v1`], whether the conductor acts as an +/// agent or as a nested proxy: the chain initializes with v1 and the client +/// receives a v1 `initialize` response. +#[tokio::test] +async fn v1_only_instantiator_negotiates_a_v2_chain_down_to_v1() -> Result<(), Error> { + for nested in [false, true] { + let v1_hits = Arc::new(AtomicUsize::new(0)); + let v2_hits = Arc::new(AtomicUsize::new(0)); + let agent = counting_agent_router(Arc::clone(&v1_hits), Arc::clone(&v2_hits)); + let selected = Arc::new(Mutex::new(None)); + let run = { + let selected = Arc::clone(&selected); + async move |cx: ConnectionTo| { + let response = cx + .send_request(UntypedMessage::new( + "initialize", + serde_json::to_value(initialize_request()) + .map_err(Error::into_internal_error)?, + )?) + .block_task() + .await?; + assert_eq!(response["protocolVersion"], 1, "{response}"); + assert_eq!( + *selected.lock().expect("selected lock"), + Some(ProtocolVersion::V2), + "the conductor selects v2 before the instantiator negotiates down" + ); + + let response = cx + .send_request(UntypedMessage::new( + "session/new", + serde_json::json!({ "cwd": "/tmp", "mcpServers": [] }), + )?) + .block_task() + .await?; + assert_eq!(response["sessionId"], "v1-session", "{response}"); + Ok(()) + } + }; + if nested { + let v1_only_proxy = move |request: SelectedInitializeRequest| async move { + *selected.lock().expect("selected lock") = Some(request.protocol_version()); + let request = request.into_v1()?; + assert_eq!(request.protocol_version, ProtocolVersion::V1); + Ok(InstantiatedProxies::new( + request, + vec![DynConnectTo::::new(Proxy.v1())], + )) + }; + run_raw_client_with_conductor( + ProxiesAndAgent::new(agent) + .proxy(ConductorImpl::new_proxy("v1-only-proxy", v1_only_proxy)), + run, + ) + .await?; + } else { + let v1_only_instantiator = move |request: SelectedInitializeRequest| async move { + *selected.lock().expect("selected lock") = Some(request.protocol_version()); + let request = request.into_v1()?; + assert_eq!(request.protocol_version, ProtocolVersion::V1); + Ok(InstantiatedChain::new( + request, + Vec::new(), + DynConnectTo::new(agent), + )) + }; + run_raw_client_with_conductor(v1_only_instantiator, run).await?; + } + + assert_eq!(v1_hits.load(Ordering::SeqCst), 1, "nested: {nested}"); + assert_eq!(v2_hits.load(Ordering::SeqCst), 0, "nested: {nested}"); + } + Ok(()) +} + /// A v1 agent and a v2 agent behind one router, counting how often each /// implementation initializes. fn counting_agent_router( diff --git a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs index 20af85af..5f241303 100644 --- a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs +++ b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill.rs @@ -464,6 +464,94 @@ async fn http_downstream_receives_stable_transformed_declarations_for_all_setup_ Ok(()) } +/// A repeated `initialize` is forwarded through the polyfill as another +/// `proxy/initialize` instead of closing its connection. +#[tokio::test] +async fn repeated_initialize_keeps_the_polyfill_connection_open() +-> Result<(), agent_client_protocol::Error> { + let observed = Arc::new(ObservedRequests::default()); + let agent = RecordingAgent { + capabilities: agent_capabilities(McpCapabilities::new().http(true)), + observed: observed.clone(), + }; + + run_with_polyfill(agent, Arc::new(AtomicUsize::new(0)), async |connection| { + for _ in 0..2 { + let initialize = + recv(connection.send_request(InitializeRequest::new(ProtocolVersion::V1))).await?; + assert!(initialize.agent_capabilities.mcp_capabilities.acp); + } + recv(connection.send_request( + NewSessionRequest::new(PathBuf::from("/tmp")).mcp_servers(vec![native_server()]), + )) + .await?; + Ok(()) + }) + .await?; + + let setup = observed.setup.lock().unwrap(); + assert_eq!(setup.len(), 1); + assert!(matches!( + setup[0].mcp_servers.as_slice(), + [McpServer::Http(_)] + )); + Ok(()) +} + +/// A malformed repeated `initialize` is answered, and the polyfill's +/// connection stays open for later requests. +#[tokio::test] +async fn malformed_repeated_initialize_keeps_the_polyfill_connection_open() +-> Result<(), agent_client_protocol::Error> { + let observed = Arc::new(ObservedRequests::default()); + let agent = RecordingAgent { + capabilities: agent_capabilities(McpCapabilities::new().http(true)), + observed: observed.clone(), + }; + + let (editor_out, conductor_in) = duplex(4096); + let (conductor_out, editor_in) = duplex(4096); + let transport = + agent_client_protocol::ByteStreams::new(editor_out.compat_write(), editor_in.compat()); + + // The polyfill is the only proxy, so the conductor forwards the repeated + // `initialize` to it unchanged as `proxy/initialize`. + Client + .v1() + .name("polyfill-test-client") + .with_spawned(|_cx| async move { + ConductorImpl::new_agent( + "polyfill-test-conductor".to_string(), + ProxiesAndAgent::new(agent).proxy(McpOverAcpPolyfill::http()), + ) + .run(agent_client_protocol::ByteStreams::new( + conductor_out.compat_write(), + conductor_in.compat(), + )) + .await + }) + .connect_with(transport, async |connection| { + recv(connection.send_request(InitializeRequest::new(ProtocolVersion::V1))).await?; + // The polyfill's v1 connection negotiates `proxy/initialize` + // before its handler sees it, so a malformed version is answered + // with v1 rather than closing the connection. + let response = recv(connection.send_request( + agent_client_protocol::UntypedMessage::new( + "initialize", + serde_json::json!({ "protocolVersion": true }), + )?, + )) + .await?; + assert_eq!(response["protocolVersion"], 1, "{response}"); + recv(connection.send_request(NewSessionRequest::new(PathBuf::from("/tmp")))).await?; + Ok(()) + }) + .await?; + + assert_eq!(observed.setup.lock().unwrap().len(), 1); + Ok(()) +} + #[tokio::test] async fn native_downstream_keeps_capability_and_declaration_unchanged() -> Result<(), agent_client_protocol::Error> { diff --git a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs index 62222c06..04be9b62 100644 --- a/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs +++ b/src/agent-client-protocol-conductor/tests/mcp_over_acp_polyfill_v2.rs @@ -1,6 +1,7 @@ -#![cfg(feature = "unstable_protocol_v2")] - //! End-to-end v2 coverage for the request-scoped MCP HTTP adapter. +//! +//! Also needs the polyfill crate's `unstable_protocol_v2` feature. The test's +//! `required-features` in `Cargo.toml` skip it otherwise. use std::{ path::PathBuf, diff --git a/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs b/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs index c9add183..0fc8dd38 100644 --- a/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs +++ b/src/agent-client-protocol-conductor/tests/trace_client_mcp_server.rs @@ -361,6 +361,73 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:1"), + method: "initialize", + session: None, + params: Object { + "protocolVersion": Number(1), + "clientCapabilities": Object { + "fs": Object { + "readTextFile": Bool(false), + "writeTextFile": Bool(false), + }, + "terminal": Bool(false), + "auth": Object { + "terminal": Bool(false), + }, + }, + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:1"), + is_error: false, + error_domain: None, + payload: Object { + "protocolVersion": Number(1), + "agentCapabilities": Object { + "loadSession": Bool(true), + "promptCapabilities": Object { + "image": Bool(true), + "audio": Bool(true), + "embeddedContext": Bool(true), + }, + "mcpCapabilities": Object { + "http": Bool(true), + "sse": Bool(false), + "acp": Bool(false), + }, + "sessionCapabilities": Object { + "list": Object {}, + "delete": Object {}, + "additionalDirectories": Object {}, + "resume": Object {}, + "close": Object {}, + }, + "auth": Object { + "logout": Object {}, + }, + }, + "authMethods": Array [ + Object { + "id": String("testy-agent-auth"), + "name": String("Testy agent auth"), + "description": String("Deterministic no-op authentication for ACP client testing"), + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, @@ -410,7 +477,7 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:1"), + id: String("id:2"), method: "session/new", session: None, params: Object { @@ -425,12 +492,90 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:3"), + method: "session/new", + session: None, + params: Object { + "cwd": String("."), + "mcpServers": Array [ + Object { + "type": String("http"), + "name": String("echo-server"), + "url": String("http:endpoint:0"), + "headers": Array [ + Object { + "name": String("Authorization"), + "value": String("[REDACTED]"), + }, + ], + }, + ], + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:3"), + is_error: false, + error_domain: None, + payload: Object { + "sessionId": String("session:0"), + "modes": Object { + "currentModeId": String("chat"), + "availableModes": Array [ + Object { + "id": String("chat"), + "name": String("Chat"), + "description": String("Default deterministic chat mode"), + }, + Object { + "id": String("plan"), + "name": String("Plan"), + "description": String("Planning-focused test mode"), + }, + ], + }, + "configOptions": Array [ + Object { + "id": String("verbosity"), + "name": String("Verbosity"), + "description": String("Controls how much text Testy includes in summaries"), + "type": String("select"), + "currentValue": String("normal"), + "options": Array [ + Object { + "value": String("brief"), + "name": String("Brief"), + }, + Object { + "value": String("normal"), + "name": String("Normal"), + }, + Object { + "value": String("verbose"), + "name": String("Verbose"), + }, + ], + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:1"), + id: String("id:2"), is_error: false, error_domain: None, payload: Object { @@ -482,7 +627,27 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:2"), + id: String("id:4"), + method: "session/prompt", + session: None, + params: Object { + "sessionId": String("session:0"), + "prompt": Array [ + Object { + "type": String("text"), + "text": String("{\"command\":\"call_tool\",\"server\":\"echo-server\",\"tool\":\"echo\",\"params\":{\"message\":\"Hello from client test!\"}}"), + }, + ], + }, + }, + ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:5"), method: "session/prompt", session: None, params: Object { @@ -500,7 +665,7 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err NotificationEvent { ts: 0.0, protocol: Acp, - from: "Proxy(1)", + from: "Agent", to: "Proxy(0)", method: "session/update", session: None, @@ -517,12 +682,25 @@ async fn test_trace_client_mcp_server() -> Result<(), agent_client_protocol::Err }, }, ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:5"), + is_error: false, + error_domain: None, + payload: Object { + "stopReason": String("end_turn"), + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:2"), + id: String("id:4"), is_error: false, error_domain: None, payload: Object { diff --git a/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs b/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs index 424f3b41..5dbe3b96 100644 --- a/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs +++ b/src/agent-client-protocol-conductor/tests/trace_mcp_tool_call.rs @@ -402,6 +402,73 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(1)", + to: "Agent", + id: String("id:2"), + method: "initialize", + session: None, + params: Object { + "protocolVersion": Number(1), + "clientCapabilities": Object { + "fs": Object { + "readTextFile": Bool(false), + "writeTextFile": Bool(false), + }, + "terminal": Bool(false), + "auth": Object { + "terminal": Bool(false), + }, + }, + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(1)", + id: String("id:2"), + is_error: false, + error_domain: None, + payload: Object { + "protocolVersion": Number(1), + "agentCapabilities": Object { + "loadSession": Bool(true), + "promptCapabilities": Object { + "image": Bool(true), + "audio": Bool(true), + "embeddedContext": Bool(true), + }, + "mcpCapabilities": Object { + "http": Bool(true), + "sse": Bool(false), + "acp": Bool(false), + }, + "sessionCapabilities": Object { + "list": Object {}, + "delete": Object {}, + "additionalDirectories": Object {}, + "resume": Object {}, + "close": Object {}, + }, + "auth": Object { + "logout": Object {}, + }, + }, + "authMethods": Array [ + Object { + "id": String("testy-agent-auth"), + "name": String("Testy agent auth"), + "description": String("Deterministic no-op authentication for ACP client testing"), + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, @@ -494,7 +561,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:2"), + id: String("id:3"), method: "session/new", session: None, params: Object { @@ -509,7 +576,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Acp, from: "Proxy(0)", to: "Proxy(1)", - id: String("id:3"), + id: String("id:4"), method: "session/new", session: None, params: Object { @@ -524,12 +591,90 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(1)", + to: "Agent", + id: String("id:5"), + method: "session/new", + session: None, + params: Object { + "cwd": String("/"), + "mcpServers": Array [ + Object { + "type": String("http"), + "name": String("test"), + "url": String("http:endpoint:0"), + "headers": Array [ + Object { + "name": String("Authorization"), + "value": String("[REDACTED]"), + }, + ], + }, + ], + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(1)", + id: String("id:5"), + is_error: false, + error_domain: None, + payload: Object { + "sessionId": String("session:0"), + "modes": Object { + "currentModeId": String("chat"), + "availableModes": Array [ + Object { + "id": String("chat"), + "name": String("Chat"), + "description": String("Default deterministic chat mode"), + }, + Object { + "id": String("plan"), + "name": String("Plan"), + "description": String("Planning-focused test mode"), + }, + ], + }, + "configOptions": Array [ + Object { + "id": String("verbosity"), + "name": String("Verbosity"), + "description": String("Controls how much text Testy includes in summaries"), + "type": String("select"), + "currentValue": String("normal"), + "options": Array [ + Object { + "value": String("brief"), + "name": String("Brief"), + }, + Object { + "value": String("normal"), + "name": String("Normal"), + }, + Object { + "value": String("verbose"), + "name": String("Verbose"), + }, + ], + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(1)", to: "Proxy(0)", - id: String("id:3"), + id: String("id:4"), is_error: false, error_domain: None, payload: Object { @@ -580,7 +725,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:2"), + id: String("id:3"), is_error: false, error_domain: None, payload: Object { @@ -632,7 +777,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:4"), + id: String("id:6"), method: "session/prompt", session: None, params: Object { @@ -652,7 +797,27 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Acp, from: "Proxy(0)", to: "Proxy(1)", - id: String("id:5"), + id: String("id:7"), + method: "session/prompt", + session: None, + params: Object { + "sessionId": String("session:0"), + "prompt": Array [ + Object { + "type": String("text"), + "text": String("{\"command\":\"call_tool\",\"server\":\"test\",\"tool\":\"echo\",\"params\":{\"message\":\"Hello from trace test!\"}}"), + }, + ], + }, + }, + ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(1)", + to: "Agent", + id: String("id:8"), method: "session/prompt", session: None, params: Object { @@ -672,7 +837,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Mcp, from: "Proxy(1)", to: "Proxy(0)", - id: String("id:6"), + id: String("id:9"), method: "server/discover", session: None, params: Object { @@ -692,7 +857,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> ts: 0.0, from: "Proxy(0)", to: "Proxy(1)", - id: String("id:6"), + id: String("id:9"), is_error: false, error_domain: None, payload: Object { @@ -721,7 +886,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> protocol: Mcp, from: "Proxy(1)", to: "Proxy(0)", - id: String("id:7"), + id: String("id:10"), method: "tools/call", session: None, params: Object { @@ -746,7 +911,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> ts: 0.0, from: "Proxy(0)", to: "Proxy(1)", - id: String("id:7"), + id: String("id:10"), is_error: false, error_domain: None, payload: Object { @@ -768,7 +933,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> NotificationEvent { ts: 0.0, protocol: Acp, - from: "Proxy(2)", + from: "Agent", to: "Proxy(1)", method: "session/update", session: None, @@ -785,12 +950,25 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> }, }, ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(1)", + id: String("id:8"), + is_error: false, + error_domain: None, + payload: Object { + "stopReason": String("end_turn"), + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(1)", to: "Proxy(0)", - id: String("id:5"), + id: String("id:7"), is_error: false, error_domain: None, payload: Object { @@ -824,7 +1002,7 @@ async fn test_trace_mcp_tool_call() -> Result<(), agent_client_protocol::Error> ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:4"), + id: String("id:6"), is_error: false, error_domain: None, payload: Object { diff --git a/src/agent-client-protocol-conductor/tests/trace_snapshot.rs b/src/agent-client-protocol-conductor/tests/trace_snapshot.rs index c22e18eb..568a18ee 100644 --- a/src/agent-client-protocol-conductor/tests/trace_snapshot.rs +++ b/src/agent-client-protocol-conductor/tests/trace_snapshot.rs @@ -209,6 +209,70 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:1"), + method: "initialize", + session: None, + params: Object { + "protocolVersion": Number(1), + "clientCapabilities": Object { + "fs": Object { + "readTextFile": Bool(false), + "writeTextFile": Bool(false), + }, + "terminal": Bool(false), + }, + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:1"), + is_error: false, + error_domain: None, + payload: Object { + "protocolVersion": Number(1), + "agentCapabilities": Object { + "loadSession": Bool(true), + "promptCapabilities": Object { + "image": Bool(true), + "audio": Bool(true), + "embeddedContext": Bool(true), + }, + "mcpCapabilities": Object { + "http": Bool(true), + "sse": Bool(false), + "acp": Bool(false), + }, + "sessionCapabilities": Object { + "list": Object {}, + "delete": Object {}, + "additionalDirectories": Object {}, + "resume": Object {}, + "close": Object {}, + }, + "auth": Object { + "logout": Object {}, + }, + }, + "authMethods": Array [ + Object { + "id": String("testy-agent-auth"), + "name": String("Testy agent auth"), + "description": String("Deterministic no-op authentication for ACP client testing"), + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, @@ -258,7 +322,7 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:1"), + id: String("id:2"), method: "session/new", session: None, params: Object { @@ -267,12 +331,78 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { }, }, ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:3"), + method: "session/new", + session: None, + params: Object { + "cwd": String("."), + "mcpServers": Array [], + }, + }, + ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:3"), + is_error: false, + error_domain: None, + payload: Object { + "sessionId": String("session:0"), + "modes": Object { + "currentModeId": String("chat"), + "availableModes": Array [ + Object { + "id": String("chat"), + "name": String("Chat"), + "description": String("Default deterministic chat mode"), + }, + Object { + "id": String("plan"), + "name": String("Plan"), + "description": String("Planning-focused test mode"), + }, + ], + }, + "configOptions": Array [ + Object { + "id": String("verbosity"), + "name": String("Verbosity"), + "description": String("Controls how much text Testy includes in summaries"), + "type": String("select"), + "currentValue": String("normal"), + "options": Array [ + Object { + "value": String("brief"), + "name": String("Brief"), + }, + Object { + "value": String("normal"), + "name": String("Normal"), + }, + Object { + "value": String("verbose"), + "name": String("Verbose"), + }, + ], + }, + ], + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:1"), + id: String("id:2"), is_error: false, error_domain: None, payload: Object { @@ -324,7 +454,27 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { protocol: Acp, from: "Client", to: "Proxy(0)", - id: String("id:2"), + id: String("id:4"), + method: "session/prompt", + session: None, + params: Object { + "sessionId": String("session:0"), + "prompt": Array [ + Object { + "type": String("text"), + "text": String("{\"command\":\"greet\"}"), + }, + ], + }, + }, + ), + Request( + RequestEvent { + ts: 0.0, + protocol: Acp, + from: "Proxy(0)", + to: "Agent", + id: String("id:5"), method: "session/prompt", session: None, params: Object { @@ -342,7 +492,7 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { NotificationEvent { ts: 0.0, protocol: Acp, - from: "Proxy(1)", + from: "Agent", to: "Proxy(0)", method: "session/update", session: None, @@ -359,12 +509,25 @@ async fn test_trace_snapshot() -> Result<(), agent_client_protocol::Error> { }, }, ), + Response( + ResponseEvent { + ts: 0.0, + from: "Agent", + to: "Proxy(0)", + id: String("id:5"), + is_error: false, + error_domain: None, + payload: Object { + "stopReason": String("end_turn"), + }, + }, + ), Response( ResponseEvent { ts: 0.0, from: "Proxy(0)", to: "Client", - id: String("id:2"), + id: String("id:4"), is_error: false, error_domain: None, payload: Object { diff --git a/src/agent-client-protocol-conductor/tests/v1_chain_equivalence.rs b/src/agent-client-protocol-conductor/tests/v1_chain_equivalence.rs index e60e18b7..f5e57445 100644 --- a/src/agent-client-protocol-conductor/tests/v1_chain_equivalence.rs +++ b/src/agent-client-protocol-conductor/tests/v1_chain_equivalence.rs @@ -18,7 +18,9 @@ use agent_client_protocol::{ UntypedMessage, schema::{ProtocolVersion, v1}, }; -use agent_client_protocol_conductor::{ConductorImpl, SelectedInitializeRequest}; +use agent_client_protocol_conductor::{ + ConductorImpl, InstantiatedChain, SelectedInitializeRequest, +}; use agent_client_protocol_rmcp::rmcp_3::McpServerExt as _; use futures::{SinkExt as _, StreamExt as _, channel::mpsc}; use serde_json::{Value, json}; @@ -106,7 +108,7 @@ async fn run_v1_chain_with( let request = request.into_v1()?; let proxies = vec![proxy()]; let agent = DynConnectTo::new(recording_v1_agent(received)); - Ok((request.into(), proxies, agent)) + Ok(InstantiatedChain::new(request, proxies, agent)) } }); diff --git a/src/agent-client-protocol-polyfill/Cargo.toml b/src/agent-client-protocol-polyfill/Cargo.toml index e09c2e6d..a7327b78 100644 --- a/src/agent-client-protocol-polyfill/Cargo.toml +++ b/src/agent-client-protocol-polyfill/Cargo.toml @@ -10,6 +10,10 @@ description = "Polyfill proxies for Agent Client Protocol backward compatibility keywords = ["acp", "agent", "mcp", "polyfill"] categories = ["development-tools"] +[package.metadata.docs.rs] +all-features = true +rustdoc-args = ["--cfg", "docsrs"] + [features] default = [] unstable_protocol_v2 = ["agent-client-protocol/unstable_protocol_v2"] diff --git a/src/agent-client-protocol-polyfill/README.md b/src/agent-client-protocol-polyfill/README.md index d3d0492b..2f07cdc5 100644 --- a/src/agent-client-protocol-polyfill/README.md +++ b/src/agent-client-protocol-polyfill/README.md @@ -16,8 +16,11 @@ agent-client-protocol-polyfill = "3" The polyfill supports stable ACP v1 by default and explicitly enables the core's `unstable_mcp_over_acp` feature. For a draft-v2 chain, enable -`unstable_protocol_v2` on both the polyfill and conductor. The independent -`unstable_session_fork` feature forwards the schema's fork support. +`unstable_protocol_v2` on both the polyfill and conductor: the conductor's +feature does not enable the polyfill's, and a polyfill built without it +negotiates the chain down to v1. The polyfill adapts `session/fork` in every +build. The independent `unstable_session_fork` feature also validates fork +requests against the schema's fork type. - [API reference](https://docs.rs/agent-client-protocol-polyfill) - [Placement, lifecycle, and compatibility guide](https://agentclientprotocol.github.io/rust-sdk/mcp-bridge.html) 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 b54996b4..7faaa9d6 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 @@ -113,7 +113,11 @@ enum BridgeMessage { } /// Adapts native MCP-over-ACP servers to loopback Streamable HTTP for HTTP-only agents. +/// +/// Create one with [`McpOverAcpPolyfill::http`]. Configuration may be added +/// in later releases. #[derive(Debug, Default)] +#[non_exhaustive] pub struct McpOverAcpPolyfill; impl McpOverAcpPolyfill { @@ -229,11 +233,27 @@ impl PolyfillHandler { match message { Dispatch::Request(mut request, responder) => { if request.method() == agent_client_protocol::schema::METHOD_INITIALIZE_PROXY { - if self.protocol.is_some() { - return Err(agent_client_protocol::Error::invalid_request() - .data("MCP-over-ACP polyfill was already initialized")); - } - let protocol = PolyfillProtocol::from_initialize_request(&request)?; + // A repeated `proxy/initialize` is forwarded like the + // first one, as long as it stays on the connection's + // protocol version. Invalid requests are answered with an + // error instead of closing the connection. + let protocol = match PolyfillProtocol::from_initialize_request(&request) { + Ok(protocol) => match self.protocol { + Some(initialized) if initialized != protocol => { + responder.respond_with_error( + agent_client_protocol::Error::invalid_request().data(format!( + "MCP-over-ACP polyfill was initialized with {initialized}, not {protocol}", + )), + )?; + return Ok(Handled::Yes); + } + _ => protocol, + }, + Err(error) => { + responder.respond_with_error(error)?; + return Ok(Handled::Yes); + } + }; self.protocol = Some(protocol); request.method = "initialize".into(); let sent = cx diff --git a/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs b/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs index a10d0a53..2ac8f7da 100644 --- a/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs +++ b/src/agent-client-protocol-polyfill/src/mcp_over_acp/protocol.rs @@ -24,6 +24,16 @@ pub(super) enum NativeMcpOutcome { Error(Value), } +impl std::fmt::Display for PolyfillProtocol { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::V1 => "v1", + #[cfg(feature = "unstable_protocol_v2")] + Self::V2 => "v2", + }) + } +} + impl PolyfillProtocol { /// Validate against the negotiated ACP version before projecting onto HTTP. pub(super) fn message_response(self, value: Value) -> Result { @@ -63,7 +73,7 @@ impl PolyfillProtocol { Error::invalid_params().data("missing initialize.protocolVersion") })?, ) - .map_err(Error::into_internal_error)?; + .map_err(|_| Error::invalid_params().data("invalid initialize.protocolVersion"))?; let protocol = if requested == ProtocolVersion::V1 { Self::V1 } else { @@ -152,17 +162,21 @@ impl PolyfillProtocol { Ok(mode) } + /// Whether `method` creates a session whose `mcpServers` the polyfill + /// rewrites. + /// + /// `session/fork` counts in every build: without the + /// `unstable_session_fork` feature the polyfill cannot validate a fork + /// request, but forwarding its `McpServer::Acp` entries unchanged would + /// hand the agent servers it cannot reach. pub(crate) fn is_session_setup_method(self, method: &str) -> bool { match self { - Self::V1 => { - matches!(method, "session/new" | "session/load" | "session/resume") - || cfg!(feature = "unstable_session_fork") && method == "session/fork" - } + Self::V1 => matches!( + method, + "session/new" | "session/load" | "session/resume" | "session/fork" + ), #[cfg(feature = "unstable_protocol_v2")] - Self::V2 => { - matches!(method, "session/new" | "session/resume") - || cfg!(feature = "unstable_session_fork") && method == "session/fork" - } + Self::V2 => matches!(method, "session/new" | "session/resume" | "session/fork"), } } @@ -185,6 +199,9 @@ impl PolyfillProtocol { "session/fork" => { ForkSessionRequest::parse_message(request.method(), request.params())?; } + // Without the typed request, the agent validates the fork. + #[cfg(not(feature = "unstable_session_fork"))] + "session/fork" => {} _ => return Err(Error::invalid_request().data("not a session setup method")), }, #[cfg(feature = "unstable_protocol_v2")] @@ -199,6 +216,9 @@ impl PolyfillProtocol { "session/fork" => { v2::ForkSessionRequest::parse_message(request.method(), request.params())?; } + // Without the typed request, the agent validates the fork. + #[cfg(not(feature = "unstable_session_fork"))] + "session/fork" => {} _ => return Err(Error::invalid_request().data("not a session setup method")), }, } @@ -359,6 +379,23 @@ pub(crate) struct NativeMcpNotification { mod tests { use super::*; + #[test] + fn session_fork_is_a_setup_method_in_every_build() { + let fork = UntypedMessage::new( + "session/fork", + serde_json::json!({ "sessionId": "s", "cwd": "/tmp", "mcpServers": [] }), + ) + .unwrap(); + for protocol in [ + PolyfillProtocol::V1, + #[cfg(feature = "unstable_protocol_v2")] + PolyfillProtocol::V2, + ] { + assert!(protocol.is_session_setup_method("session/fork")); + protocol.validate_session_setup_request(&fork).unwrap(); + } + } + #[test] fn request_wrappers_keep_request_metadata_in_inner_params() { for protocol in [ diff --git a/src/agent-client-protocol-trace-viewer/src/lib.rs b/src/agent-client-protocol-trace-viewer/src/lib.rs index 90de5bbd..32660aba 100644 --- a/src/agent-client-protocol-trace-viewer/src/lib.rs +++ b/src/agent-client-protocol-trace-viewer/src/lib.rs @@ -20,7 +20,7 @@ pub const VIEWER_HTML: &str = include_str!("viewer.html"); /// Source of trace events for the viewer. #[derive(Debug, Clone)] -pub enum TraceSource { +enum TraceSource { /// Read events from a file (re-reads on each request for live updates). File(PathBuf), /// Read events from shared memory. @@ -63,7 +63,18 @@ struct AppState { } /// Configuration for the trace viewer server. -#[derive(Debug)] +/// +/// New options may be added in later releases, so start from +/// [`TraceViewerConfig::default`] and use the setters. +/// +/// ``` +/// let config = agent_client_protocol_trace_viewer::TraceViewerConfig::default() +/// .port(8080) +/// .open_browser(false); +/// assert_eq!(config.port, 8080); +/// ``` +#[derive(Debug, Clone)] +#[non_exhaustive] pub struct TraceViewerConfig { /// Port to serve on (0 = auto-select). pub port: u16, @@ -71,6 +82,22 @@ pub struct TraceViewerConfig { pub open_browser: bool, } +impl TraceViewerConfig { + /// Serve on `port`, or on a free port when `port` is 0. + #[must_use] + pub fn port(mut self, port: u16) -> Self { + self.port = port; + self + } + + /// Whether to open the viewer in a browser once the server is listening. + #[must_use] + pub fn open_browser(mut self, open_browser: bool) -> Self { + self.open_browser = open_browser; + self + } +} + impl Default for TraceViewerConfig { fn default() -> Self { Self { diff --git a/src/agent-client-protocol-trace-viewer/src/main.rs b/src/agent-client-protocol-trace-viewer/src/main.rs index 00d67b3a..b3201532 100644 --- a/src/agent-client-protocol-trace-viewer/src/main.rs +++ b/src/agent-client-protocol-trace-viewer/src/main.rs @@ -42,10 +42,9 @@ async fn main() -> anyhow::Result<()> { anyhow::bail!("Trace file not found: {}", args.trace_file.display()); } - let config = agent_client_protocol_trace_viewer::TraceViewerConfig { - port: args.port.unwrap_or(0), - open_browser: !args.no_open, - }; + let config = agent_client_protocol_trace_viewer::TraceViewerConfig::default() + .port(args.port.unwrap_or(0)) + .open_browser(!args.no_open); agent_client_protocol_trace_viewer::serve_file(args.trace_file, config).await }