feat(v2): add experimental runtime and negotiation (#139)

* feat(v2): add strict experimental runtime

* feat(v2): track active session lifecycle

* feat(protocol): negotiate experimental protocol versions

* docs(v2): document experimental runtime

* refactor(v2): simplify experimental runtime

* refactor(v2): minimize experimental runtime API

* refactor(v2): remove redundant type casts

* refactor(v2): remove redundant agent protocol

Author: PsiACE

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. Install update handlers on the
+client before opening a session because updates are independent connection
+traffic:
+
+```python
+from acp.experimental import v2
+
+class MyClient:
+    async def session_update(
+        self,
+        notification: v2.schema.UpdateSessionNotification,
+    ) -> None:
+        handle_update(notification)
+
+
+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.new_session(
+    v2.schema.NewSessionRequest(cwd="/workspace")
+)
+await connection.prompt(
+    v2.schema.PromptRequest(
+        session_id=session.session_id,
+        prompt=[v2.schema.TextContentBlock(text="Hello")],
+    )
+)
+```
+
+`session/prompt` returns when the agent accepts the prompt. It does not define a
+boundary for session updates: they may arrive before, during, or after that
+request, and they do not carry a prompt identifier. Applications decide how to
+buffer or present them.
+
+Agents that serve both versions use `AgentProtocolRouter`:
+
+```python
+from acp.experimental import AgentProtocolRouter
+
+router = AgentProtocolRouter(
+    v1=lambda connection: V1Agent(connection),
+    v2=lambda connection: V2Agent(connection),
+)
+await router.run()
+```
+
+The selected factory is called once per connection. Return a fresh agent from
+each call to avoid sharing connection state.
+
+Extension method names are explicit and must include the protocol-required `_`
+prefix:
+
+```python
+result = await connection.send_extension_request("_vendor/method", {"value": 1})
+await connection.send_extension_notification("_vendor/event", {"value": 1})
+```
+
+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.
+
+Client-side fallback is application controlled and may require opening a new
+transport. Protocol-level request cancellation is not yet exposed by the
+experimental runtime; `session/cancel` remains available for cancelling active
+session work.

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,13 @@
 """Experimental ACP APIs."""
+
+from . import v2
+from .negotiation import (
+    AgentProtocolConnection,
+    AgentProtocolRouter,
+)
+
+__all__ = [
+    "AgentProtocolConnection",
+    "AgentProtocolRouter",
+    "v2",
+]