feat(v2): add experimental runtime and negotiation

docs/experimental-v2.md

@@ -0,0 +1,76 @@
+# Experimental Protocol v2
+
+> **Experimental.** Protocol v2 is a draft. Import it from `acp.experimental` and
+> expect its API and generated models to change with the upstream schema.
+
+The v2 runtime is separate from the stable v1 API. Its methods accept and return
+generated request and response models directly:
+
+```python
+from acp.experimental import v2
+
+connection = v2.connect_to_agent(MyClient(), transport)
+initialized = await connection.initialize(
+    v2.schema.InitializeRequest(
+        protocol_version=v2.PROTOCOL_VERSION,
+        info=v2.schema.Implementation(name="my-client", version="1.0.0"),
+    )
+)
+session = await connection.open_session(
+    v2.schema.NewSessionRequest(cwd="/workspace")
+)
+```
+
+`open_session()` returns an `ActiveSession`. A v2 prompt is accepted before the
+agent finishes it, so consume session updates until `SessionStop` marks the
+`running` to `idle` transition:
+
+```python
+await session.prompt(
+    v2.schema.PromptRequest(
+        session_id=session.session_id,
+        prompt=[v2.schema.TextContentBlock(text="Hello")],
+    )
+)
+stopped = await session.wait_for_idle()
+```
+
+## Negotiate v1 or v2
+
+Use `ClientNegotiator` when the same client can speak both versions. It sends
+exactly one `initialize` request and returns a version-tagged connection:
+
+```python
+from acp.experimental import (
+    ClientNegotiator,
+    NegotiatedV2,
+    V1ClientConfig,
+    V2ClientConfig,
+)
+
+negotiator = ClientNegotiator(
+    transport,
+    v1=V1ClientConfig(client=v1_client, initialize=v1_initialize),
+    v2=V2ClientConfig(client=v2_client, initialize=v2_initialize),
+)
+negotiated = await negotiator.negotiate()
+
+if isinstance(negotiated, NegotiatedV2):
+    session = await negotiated.connection.open_session(v2_new_session)
+else:
+    session = await negotiated.connection.new_session(cwd="/workspace")
+```
+
+Agents that serve both versions use `AgentProtocolRouter`:
+
+```python
+from acp.experimental import AgentProtocolRouter
+
+router = AgentProtocolRouter(v1=v1_agent, v2=v2_agent)
+await router.run()
+```
+
+The selected runtime remains strict after initialization: v1 messages are not
+accepted by a v2 connection, and v2 messages are not translated into v1 calls.
+Only the initial v2 request is reduced to the common v1 initialization fields
+when an agent selects v1.

mkdocs.yml

@@ -12,6 +12,7 @@ nav:
   - Quick Start: quickstart.md
   - Use Cases: use-cases.md
   - Web Transport (HTTP/WS): web-transport.md
+  - Experimental Protocol v2: experimental-v2.md
   - Experimental Contrib: contrib.md
   - Releasing: releasing.md
   - 0.11 Migration Guide: migration-guide-0.11.md

src/acp/agent/connection.py

@@ -7,7 +7,7 @@
 from pydantic import TypeAdapter
 
 from .._transport import Transport
-from ..connection import Connection
+from ..connection import Connection, MethodHandler
 from ..interfaces import Agent, Client
 from ..meta import CLIENT_METHODS
 from ..schema import (
@@ -88,8 +88,7 @@ def __init__(
         use_unstable_protocol: bool = False,
         **connection_kwargs: Any,
     ) -> None:
-        agent = to_agent(self) if callable(to_agent) else to_agent
-        handler = build_agent_router(cast(Agent, agent), use_unstable_protocol=use_unstable_protocol)
+        agent, handler = self._prepare(to_agent, use_unstable_protocol=use_unstable_protocol)
         if isinstance(input_stream, Transport):
             if output_stream is not None:
                 raise TypeError(_AGENT_CONNECTION_ERROR)
@@ -100,6 +99,32 @@ def __init__(
             ):
                 raise TypeError(_AGENT_CONNECTION_ERROR)
             self._conn = Connection(handler, input_stream, output_stream, listening=listening, **connection_kwargs)
+        self._notify_connected(agent)
+
+    @classmethod
+    def _attach(
+        cls,
+        to_agent: Callable[[Client], Agent] | Agent,
+        connection: Connection,
+        *,
+        use_unstable_protocol: bool = False,
+    ) -> tuple[AgentSideConnection, MethodHandler]:
+        self = cls.__new__(cls)
+        agent, handler = self._prepare(to_agent, use_unstable_protocol=use_unstable_protocol)
+        self._conn = connection
+        self._notify_connected(agent)
+        return self, handler
+
+    def _prepare(
+        self,
+        to_agent: Callable[[Client], Agent] | Agent,
+        *,
+        use_unstable_protocol: bool,
+    ) -> tuple[Agent, MethodHandler]:
+        agent = cast(Agent, to_agent(self) if callable(to_agent) else to_agent)
+        return agent, build_agent_router(agent, use_unstable_protocol=use_unstable_protocol)
+
+    def _notify_connected(self, agent: Agent) -> None:
         if on_connect := getattr(agent, "on_connect", None):
             on_connect(self)
 

src/acp/client/connection.py

@@ -6,7 +6,7 @@
 from typing import Any, cast, final
 
 from .._transport import Transport
-from ..connection import Connection
+from ..connection import Connection, MethodHandler
 from ..exceptions import RequestError
 from ..interfaces import Agent, Client
 from ..meta import AGENT_METHODS, CLIENT_METHODS
@@ -122,9 +122,7 @@ def __init__(
         use_unstable_protocol: bool = False,
         **connection_kwargs: Any,
     ) -> None:
-        client = to_client(self) if callable(to_client) else to_client
-        self._session_updates = _SessionUpdateTracker(cast(Client, client))
-        handler = build_client_router(cast(Client, self._session_updates), use_unstable_protocol=use_unstable_protocol)
+        client, handler = self._prepare(to_client, use_unstable_protocol=use_unstable_protocol)
 
         if isinstance(input_stream, Transport):
             if output_stream is not None:
@@ -136,6 +134,34 @@ def __init__(
             ):
                 raise TypeError(_CLIENT_CONNECTION_ERROR)
             self._conn = Connection(handler, input_stream, output_stream, **connection_kwargs)
+        self._notify_connected(client)
+
+    @classmethod
+    def _attach(
+        cls,
+        to_client: Callable[[Agent], Client] | Client,
+        connection: Connection,
+        *,
+        use_unstable_protocol: bool = False,
+    ) -> tuple[ClientSideConnection, MethodHandler]:
+        self = cls.__new__(cls)
+        client, handler = self._prepare(to_client, use_unstable_protocol=use_unstable_protocol)
+        self._conn = connection
+        self._notify_connected(client)
+        return self, handler
+
+    def _prepare(
+        self,
+        to_client: Callable[[Agent], Client] | Client,
+        *,
+        use_unstable_protocol: bool,
+    ) -> tuple[Client, MethodHandler]:
+        client = cast(Client, to_client(self) if callable(to_client) else to_client)
+        self._session_updates = _SessionUpdateTracker(client)
+        handler = build_client_router(cast(Client, self._session_updates), use_unstable_protocol=use_unstable_protocol)
+        return client, handler
+
+    def _notify_connected(self, client: Client) -> None:
         if on_connect := getattr(client, "on_connect", None):
             on_connect(self)
 

src/acp/experimental/__init__.py

@@ -1 +1,27 @@
 """Experimental ACP APIs."""
+
+from . import v2
+from .negotiation import (
+    AgentProtocolConnection,
+    AgentProtocolRouter,
+    ClientNegotiator,
+    NegotiatedClient,
+    NegotiatedV1,
+    NegotiatedV2,
+    UnsupportedProtocolVersionError,
+    V1ClientConfig,
+    V2ClientConfig,
+)
+
+__all__ = [
+    "AgentProtocolConnection",
+    "AgentProtocolRouter",
+    "ClientNegotiator",
+    "NegotiatedClient",
+    "NegotiatedV1",
+    "NegotiatedV2",
+    "UnsupportedProtocolVersionError",
+    "V1ClientConfig",
+    "V2ClientConfig",
+    "v2",
+]