feat(acp): Add unstable (very experimental!) protocol v2 support (#170)

* feat(acp): Add unstable protocol v2 support

Round 2. This time only do the magic for the agent side.

* Clippy

* Clean up clones

* fix(acp): drain failed initialize negotiation state

* fix(acp): Use pending version during initialize

* fix(acp): clear pending version on init failure

* fix(acp): Default builders to v1 protocol mode

* fix(acp): Handle outgoing conversion failures locally

Author: benbrandt

.github/workflows/ci.yml

@@ -40,7 +40,9 @@ jobs:
           config: ./typos.toml
 
       - name: Lint
-        run: cargo clippy
+        run: |
+          cargo clippy
+          cargo clippy --all-targets --all-features
 
       - name: Build
         run: cargo build --all-targets --all-features

md/SUMMARY.md

@@ -6,6 +6,7 @@
 
 - [Design Overview](./design.md)
 - [Protocol Reference](./protocol.md)
+- [Protocol V2](./protocol-v2.md)
 
 # Conductor (agent-client-protocol-conductor)
 

md/protocol-v2.md

@@ -0,0 +1,75 @@
+# Protocol V2
+
+The core SDK can opt into the draft ACP protocol v2 surface with the
+`unstable_protocol_v2` crate feature:
+
+```toml
+agent-client-protocol = { version = "...", features = ["unstable_protocol_v2"] }
+```
+
+This feature is separate from the broad `unstable` feature because protocol v2
+is a versioning experiment, not just an unstable method family.
+
+By default, `Client.builder()` and `Agent.builder()` continue to expose the
+stable v1 API and advertise protocol v1. To use the v2 API for a connection,
+construct the builder with `Client.v2()` or `Agent.v2()`:
+
+```rust
+use agent_client_protocol::schema::{ProtocolVersion, v2};
+use agent_client_protocol::{Agent, Client};
+
+# async fn run(agent_transport: impl agent_client_protocol::ConnectTo<agent_client_protocol::Client>) -> agent_client_protocol::Result<()> {
+Client
+    .v2()
+    .connect_with(agent_transport, async |cx| {
+        let initialize = cx
+            .send_request(v2::InitializeRequest::new(ProtocolVersion::V1))
+            .block_task()
+            .await?;
+
+        assert_eq!(initialize.protocol_version, ProtocolVersion::V2);
+        Ok(())
+    })
+    .await?;
+# Ok(())
+# }
+
+# async fn serve(client_transport: impl agent_client_protocol::ConnectTo<agent_client_protocol::Agent>) -> agent_client_protocol::Result<()> {
+Agent
+    .v2()
+    .on_receive_request(
+        async |initialize: v2::InitializeRequest, responder, _cx| {
+            responder.respond(v2::InitializeResponse::new(initialize.protocol_version))
+        },
+        agent_client_protocol::on_receive_request!(),
+    )
+    .connect_to(client_transport)
+    .await?;
+# Ok(())
+# }
+```
+
+When v2 mode is enabled, application code should use types from
+`agent_client_protocol::schema::v2`. The flat `agent_client_protocol::schema::*`
+exports remain the stable v1 schema. This will likely change as v2 gets closer
+to release.
+
+The SDK handles the `initialize` negotiation at the JSON-RPC boundary:
+
+- A v2 client advertises protocol v2 as its latest supported version.
+- A v2 client requires a v2 agent. If the agent responds with v1, the
+  `initialize` request resolves with an error and the caller must explicitly
+  fall back to a v1 client implementation if that is acceptable.
+- A v2 agent responds with v2 when the client supports it, or v1 when the client
+  only supports v1. Agent handlers still receive v2 schema types; the SDK tracks
+  the negotiated wire version separately and adapts supported behavior at the
+  transport boundary.
+- If the agent responds with any other unsupported version, the request resolves
+  with an error so the client can close the connection.
+- After initialization, the SDK converts supported messages and responses between
+  the local API version and the negotiated wire version.
+
+That means an agent can be implemented against v2 request and response types
+while still serving v1 clients. The goal is for agent-side v1 compatibility to
+live in the SDK wherever it can be represented as protocol adaptation. Clients
+should opt into v2 separately and should not assume v2 behavior from v1 agents.

src/agent-client-protocol/Cargo.toml

@@ -37,6 +37,7 @@ unstable_session_delete = ["agent-client-protocol-schema/unstable_session_delete
 unstable_session_fork = ["agent-client-protocol-schema/unstable_session_fork"]
 unstable_session_model = ["agent-client-protocol-schema/unstable_session_model"]
 unstable_session_usage = ["agent-client-protocol-schema/unstable_session_usage"]
+unstable_protocol_v2 = ["agent-client-protocol-schema/unstable_protocol_v2"]
 
 [dependencies]
 agent-client-protocol-schema.workspace = true

src/agent-client-protocol/src/jsonrpc.rs

@@ -6,6 +6,7 @@ pub use jsonrpcmsg;
 
 // Types re-exported from crate root
 use serde::{Deserialize, Serialize};
+use std::any::TypeId;
 use std::fmt::Debug;
 use std::panic::Location;
 use std::pin::pin;
@@ -19,6 +20,7 @@ mod dynamic_handler;
 pub(crate) mod handlers;
 mod incoming_actor;
 mod outgoing_actor;
+mod protocol_compat;
 pub(crate) mod run;
 mod task_actor;
 mod transport_actor;
@@ -28,6 +30,7 @@ pub use crate::jsonrpc::handlers::NullHandler;
 use crate::jsonrpc::handlers::{ChainedHandler, NamedHandler};
 use crate::jsonrpc::handlers::{MessageHandler, NotificationHandler, RequestHandler};
 use crate::jsonrpc::outgoing_actor::{OutgoingMessageTx, send_raw_message};
+use crate::jsonrpc::protocol_compat::{ProtocolCompat, ProtocolMode};
 use crate::jsonrpc::run::SpawnedRun;
 use crate::jsonrpc::run::{ChainRun, NullRun, RunWithConnectionTo};
 use crate::jsonrpc::task_actor::{Task, TaskTx};
@@ -554,6 +557,21 @@ where
 
     /// Responder for background tasks.
     responder: Runner,
+
+    /// Protocol version mode for the public API and wire compatibility layer.
+    protocol_mode: ProtocolMode,
+}
+
+fn default_protocol_mode<Host: Role>() -> ProtocolMode {
+    let role = TypeId::of::<Host>();
+
+    if role == TypeId::of::<Agent>() {
+        ProtocolMode::v1_agent()
+    } else if role == TypeId::of::<Client>() {
+        ProtocolMode::v1_client()
+    } else {
+        ProtocolMode::disabled()
+    }
 }
 
 impl<Host: Role> Builder<Host, NullHandler, NullRun> {
@@ -566,6 +584,7 @@ impl<Host: Role> Builder<Host, NullHandler, NullRun> {
             name: None,
             handler: NullHandler,
             responder: NullRun,
+            protocol_mode: default_protocol_mode::<Host>(),
         }
     }
 }
@@ -581,6 +600,7 @@ where
             name: None,
             handler,
             responder: NullRun,
+            protocol_mode: default_protocol_mode::<Host>(),
         }
     }
 }
@@ -597,6 +617,28 @@ impl<
         self
     }
 
+    pub(crate) fn v1_agent(mut self) -> Self {
+        self.protocol_mode = ProtocolMode::v1_agent();
+        self
+    }
+
+    pub(crate) fn v1_client(mut self) -> Self {
+        self.protocol_mode = ProtocolMode::v1_client();
+        self
+    }
+
+    #[cfg(feature = "unstable_protocol_v2")]
+    pub(crate) fn v2_agent(mut self) -> Self {
+        self.protocol_mode = ProtocolMode::v2_agent();
+        self
+    }
+
+    #[cfg(feature = "unstable_protocol_v2")]
+    pub(crate) fn v2_client(mut self) -> Self {
+        self.protocol_mode = ProtocolMode::v2_client();
+        self
+    }
+
     /// Merge another [`Builder`] into this one.
     ///
     /// Prefer [`Self::on_receive_request`] or [`Self::on_receive_notification`].
@@ -613,14 +655,22 @@ impl<
         impl HandleDispatchFrom<Host::Counterpart>,
         impl RunWithConnectionTo<Host::Counterpart>,
     > {
+        let Builder {
+            name: other_name,
+            handler: other_handler,
+            responder: other_responder,
+            protocol_mode: other_protocol_mode,
+            host: _,
+        } = other;
         Builder {
             host: self.host,
             name: self.name,
             handler: ChainedHandler::new(
                 self.handler,
-                NamedHandler::new(other.name, other.handler),
+                NamedHandler::new(other_name, other_handler),
             ),
-            responder: ChainRun::new(self.responder, other.responder),
+            responder: ChainRun::new(self.responder, other_responder),
+            protocol_mode: self.protocol_mode.merge(other_protocol_mode),
         }
     }
 
@@ -637,6 +687,7 @@ impl<
             name: self.name,
             handler: ChainedHandler::new(self.handler, handler),
             responder: self.responder,
+            protocol_mode: self.protocol_mode,
         }
     }
 
@@ -653,6 +704,7 @@ impl<
             name: self.name,
             handler: self.handler,
             responder: ChainRun::new(self.responder, responder),
+            protocol_mode: self.protocol_mode,
         }
     }
 
@@ -1173,6 +1225,7 @@ impl<
             handler,
             responder,
             host: me,
+            protocol_mode,
         } = self;
 
         let (outgoing_tx, outgoing_rx) = mpsc::unbounded();
@@ -1198,6 +1251,7 @@ impl<
         } = transport_channel;
 
         let (reply_tx, reply_rx) = mpsc::unbounded();
+        let protocol_compat = ProtocolCompat::new(protocol_mode);
 
         let future = crate::util::instrument_with_connection_name(name, {
             let connection = connection.clone();
@@ -1211,6 +1265,7 @@ impl<
                             outgoing_rx,
                             reply_tx.clone(),
                             transport_outgoing_tx,
+                            protocol_compat.clone(),
                         ),
                         // Protocol layer: jsonrpcmsg::Message → handler/reply routing
                         incoming_actor::incoming_protocol_actor(
@@ -1220,6 +1275,7 @@ impl<
                             dynamic_handler_rx,
                             reply_rx,
                             handler,
+                            protocol_compat,
                         ),
                         task_actor::task_actor(new_task_rx, &connection),
                         responder.run_with_connection_to(connection.clone()),
@@ -1341,6 +1397,9 @@ enum OutgoingMessage {
     Response {
         id: jsonrpcmsg::Id,
 
+        /// Method of the incoming request this response completes.
+        method: String,
+
         response: Result<serde_json::Value, crate::Error>,
     },
 
@@ -1907,6 +1966,7 @@ impl Responder<serde_json::Value> {
     /// The response will be serialized to JSON and sent over the wire.
     fn new(message_tx: OutgoingMessageTx, method: String, id: jsonrpcmsg::Id) -> Self {
         let id_clone = id.clone();
+        let method_clone = method.clone();
         Self {
             method,
             id,
@@ -1915,6 +1975,7 @@ impl Responder<serde_json::Value> {
                     &message_tx,
                     OutgoingMessage::Response {
                         id: id_clone,
+                        method: method_clone,
                         response,
                     },
                 )