refactor(v2): simplify experimental runtime

Author: PsiACE

docs/experimental-v2.md

@@ -4,62 +4,43 @@
 > 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:
+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.open_session(
+session = await connection.new_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(
+await connection.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")
-```
+`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`:
 
@@ -74,3 +55,8 @@ 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.

src/acp/experimental/__init__.py

@@ -4,24 +4,10 @@
 from .negotiation import (
     AgentProtocolConnection,
     AgentProtocolRouter,
-    ClientNegotiator,
-    NegotiatedClient,
-    NegotiatedV1,
-    NegotiatedV2,
-    UnsupportedProtocolVersionError,
-    V1ClientConfig,
-    V2ClientConfig,
 )
 
 __all__ = [
     "AgentProtocolConnection",
     "AgentProtocolRouter",
-    "ClientNegotiator",
-    "NegotiatedClient",
-    "NegotiatedV1",
-    "NegotiatedV2",
-    "UnsupportedProtocolVersionError",
-    "V1ClientConfig",
-    "V2ClientConfig",
     "v2",
 ]

src/acp/experimental/negotiation.py

@@ -2,15 +2,13 @@
 
 import asyncio
 from collections.abc import Callable
-from dataclasses import dataclass
 from typing import Any, cast
 
 from pydantic import BaseModel
 
 from acp import meta as v1_meta
 from acp import schema as v1_schema
 from acp.agent.connection import AgentSideConnection as V1AgentSideConnection
-from acp.client.connection import ClientSideConnection as V1ClientSideConnection
 from acp.connection import Connection, MethodHandler
 from acp.exceptions import RequestError
 from acp.interfaces import Agent as V1Agent
@@ -20,23 +18,13 @@
 from .v2._connection import open_connection
 from .v2.agent import AgentFactory as V2AgentFactory
 from .v2.agent import AgentSideConnection as V2AgentSideConnection
-from .v2.client import ClientFactory as V2ClientFactory
-from .v2.client import ClientSideConnection as V2ClientSideConnection
 
 __all__ = [
     "AgentProtocolConnection",
     "AgentProtocolRouter",
-    "ClientNegotiator",
-    "NegotiatedClient",
-    "NegotiatedV1",
-    "NegotiatedV2",
-    "UnsupportedProtocolVersionError",
-    "V1ClientConfig",
-    "V2ClientConfig",
 ]
 
 V1AgentFactory = Callable[[V1Client], V1Agent]
-V1ClientFactory = Callable[[V1Agent], V1Client]
 
 
 def _dump(model: BaseModel) -> dict[str, Any]:
@@ -80,33 +68,6 @@ def _normalize_initialize(params: Any, selected_version: int) -> dict[str, Any]:
     return _dump(request)
 
 
-class _SwitchingHandler:
-    def __init__(self) -> None:
-        self._handler: MethodHandler | None = None
-        self._failure: BaseException | None = None
-        self._ready = asyncio.Event()
-
-    def bind(self, handler: MethodHandler) -> None:
-        if self._handler is not None or self._failure is not None:
-            raise RuntimeError("Protocol handler has already been resolved")
-        self._handler = handler
-        self._ready.set()
-
-    def fail(self, error: BaseException) -> None:
-        if self._handler is not None:
-            return
-        self._failure = error
-        self._ready.set()
-
-    async def __call__(self, method: str, params: Any | None, is_notification: bool) -> Any:
-        await self._ready.wait()
-        if self._failure is not None:
-            raise self._failure
-        if self._handler is None:
-            raise RuntimeError("Protocol handler was not resolved")
-        return await self._handler(method, params, is_notification)
-
-
 class _AgentNegotiationHandler:
     def __init__(
         self,
@@ -246,130 +207,3 @@ async def run(
             await connection.listen()
         finally:
             await asyncio.shield(connection.close())
-
-
-@dataclass(frozen=True, slots=True)
-class V1ClientConfig:
-    client: V1ClientFactory | V1Client
-    initialize: v1_schema.InitializeRequest
-
-    def __post_init__(self) -> None:
-        if self.initialize.protocol_version != v1_meta.PROTOCOL_VERSION:
-            raise ValueError(f"V1ClientConfig requires protocol version {v1_meta.PROTOCOL_VERSION}")
-
-
-@dataclass(frozen=True, slots=True)
-class V2ClientConfig:
-    client: V2ClientFactory | v2.Client
-    initialize: v2.schema.InitializeRequest
-
-    def __post_init__(self) -> None:
-        if self.initialize.protocol_version != v2.PROTOCOL_VERSION:
-            raise ValueError(f"V2ClientConfig requires protocol version {v2.PROTOCOL_VERSION}")
-
-
-@dataclass(frozen=True, slots=True)
-class NegotiatedV1:
-    connection: V1ClientSideConnection
-    initialize: v1_schema.InitializeResponse
-    protocol_version: int = v1_meta.PROTOCOL_VERSION
-
-
-@dataclass(frozen=True, slots=True)
-class NegotiatedV2:
-    connection: V2ClientSideConnection
-    initialize: v2.schema.InitializeResponse
-    protocol_version: int = v2.PROTOCOL_VERSION
-
-
-NegotiatedClient = NegotiatedV1 | NegotiatedV2
-
-
-class UnsupportedProtocolVersionError(ValueError):
-    def __init__(self, requested: int, offered: int, supported: frozenset[int]) -> None:
-        self.requested = requested
-        self.offered = offered
-        self.supported = supported
-        super().__init__(f"Agent selected ACP protocol {offered}; requested {requested}, supported {sorted(supported)}")
-
-
-class ClientNegotiator:
-    """Send one initialize request and return the selected typed client."""
-
-    def __init__(
-        self,
-        input_stream: Any,
-        output_stream: Any = None,
-        *,
-        v1: V1ClientConfig | None = None,
-        v2: V2ClientConfig | None = None,
-        **connection_kwargs: Any,
-    ) -> None:
-        if v1 is None and v2 is None:
-            raise ValueError("Configure at least one ACP client version")
-        self._v1 = v1
-        self._v2 = v2
-        self._handler = _SwitchingHandler()
-        self._connection = open_connection(
-            self._handler,
-            input_stream,
-            output_stream,
-            **connection_kwargs,
-        )
-        self._lock = asyncio.Lock()
-        self._resolved: NegotiatedClient | None = None
-        self._failure: BaseException | None = None
-
-    async def negotiate(self) -> NegotiatedClient:
-        async with self._lock:
-            if self._resolved is not None:
-                return self._resolved
-            if self._failure is not None:
-                raise self._failure
-            try:
-                self._resolved = await self._negotiate_once()
-            except BaseException as error:
-                self._failure = error
-                self._handler.fail(error)
-                await self._connection.close()
-                raise
-            return self._resolved
-
-    async def _negotiate_once(self) -> NegotiatedClient:
-        offered_request: BaseModel = (
-            self._v2.initialize if self._v2 is not None else cast(V1ClientConfig, self._v1).initialize
-        )
-
-        response = await self._connection.send_request(v2.AGENT_METHODS["initialize"], _dump(offered_request))
-        offered = _read_protocol_version(response)
-        requested = offered_request.protocol_version
-
-        if offered == v2.PROTOCOL_VERSION and self._v2 is not None:
-            initialize = v2.schema.InitializeResponse.model_validate(response)
-            connection, handler = V2ClientSideConnection._attach(self._v2.client, self._connection)
-            connection._complete_initialization(self._v2.initialize, initialize)
-            self._handler.bind(handler)
-            return NegotiatedV2(connection, initialize)
-        if offered == v1_meta.PROTOCOL_VERSION and self._v1 is not None:
-            initialize = v1_schema.InitializeResponse.model_validate(response)
-            connection, handler = V1ClientSideConnection._attach(self._v1.client, self._connection)
-            self._handler.bind(handler)
-            return NegotiatedV1(connection, initialize)
-        supported = frozenset(
-            version
-            for version, config in (
-                (v1_meta.PROTOCOL_VERSION, self._v1),
-                (v2.PROTOCOL_VERSION, self._v2),
-            )
-            if config is not None
-        )
-        raise UnsupportedProtocolVersionError(requested, offered, supported)
-
-    async def close(self) -> None:
-        await self._connection.close()
-
-    async def __aenter__(self) -> ClientNegotiator:
-        return self
-
-    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
-        await self.close()

src/acp/experimental/v2/__init__.py

@@ -5,21 +5,16 @@
 from .client import ClientSideConnection, connect_to_agent
 from .interfaces import Agent, Client
 from .meta import AGENT_METHODS, CLIENT_METHODS, PROTOCOL_METHODS, PROTOCOL_VERSION
-from .session import ActiveSession, SessionMessage, SessionStop, SessionUpdate
 
 __all__ = [
     "AGENT_METHODS",
     "CLIENT_METHODS",
     "PROTOCOL_METHODS",
     "PROTOCOL_VERSION",
-    "ActiveSession",
     "Agent",
     "AgentSideConnection",
     "Client",
     "ClientSideConnection",
-    "SessionMessage",
-    "SessionStop",
-    "SessionUpdate",
     "connect_to_agent",
     "run_agent",
     "schema",

src/acp/experimental/v2/agent.py

@@ -20,7 +20,7 @@
 )
 from ._router import MethodRouter
 from .interfaces import Agent, Client
-from .meta import CLIENT_METHODS, PROTOCOL_METHODS
+from .meta import CLIENT_METHODS
 
 __all__ = ["AgentSideConnection", "run_agent"]
 
@@ -37,11 +37,6 @@ def __init__(self, agent: Agent, state: InitializationState) -> None:
         self._state = state
 
     async def __call__(self, method: str, params: Any | None, is_notification: bool) -> Any:
-        if is_notification and method == PROTOCOL_METHODS["cancel_request"]:
-            if self._state.phase not in {"initializing", "initialized"}:
-                await self._state.require(method)
-            return None
-
         initialize = self._router.request_spec("initialize")
         if not is_notification and initialize is not None and method == initialize.method:
             request = cast(schema.InitializeRequest, initialize.request.validate_python(params))