fix(connection): support $/cancel_request request cancellation

docs/experimental-v2.md

@@ -150,6 +150,8 @@ 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.
+transport. Protocol-level request cancellation (`$/cancel_request`) is handled
+by the shared connection layer, as in v1: cancelling the task awaiting a request
+sends a best-effort cancellation to the peer, and an incoming cancellation
+cancels the handler task, which answers with its result or `-32800`.
+`session/cancel` remains available for cancelling active session work.

docs/quickstart.md

@@ -228,6 +228,21 @@ MCP requests return the inner JSON result unchanged, including `null`. Use
 and optional `params`. These methods share the same connections and routers
 across stdio, HTTP, and WebSocket transports.
 
+## Request cancellation
+
+Connections handle the protocol's `$/cancel_request` notification. When the
+peer cancels one of its requests, the SDK cancels the task running your handler.
+The handler may catch `asyncio.CancelledError` and return a (partial) result;
+otherwise the peer receives a `-32800` "Request cancelled" error. Cancellations
+for unknown or already finished requests are ignored; a cancelled request still
+gets exactly one response unless the connection is closed first.
+
+Cancelling the task that awaits an outgoing request (for example through
+`asyncio.wait_for`) still raises `CancelledError` locally without waiting for the
+peer, and additionally sends a best-effort `$/cancel_request` for that request.
+Cancellation support is optional for peers. Use `session/cancel` to stop a
+prompt turn.
+
 ## Maintaining protocol routes
 
 The `Agent` and `Client` protocols in `src/acp/interfaces.py` are the source of

src/acp/connection.py

@@ -14,6 +14,7 @@
 
 from ._transport import NdjsonTransport, Transport
 from .exceptions import RequestError
+from .meta import PROTOCOL_METHODS
 from .task import MessageSender, TaskSupervisor
 from .telemetry import span_context
 
@@ -37,6 +38,13 @@ class StreamEvent:
 
 StreamObserver = Callable[[StreamEvent], Awaitable[None] | None]
 
+_CANCEL_REQUEST_METHOD = PROTOCOL_METHODS["cancel_request"]
+
+
+def _is_request_id(value: Any) -> bool:
+    """Whether ``value`` is a JSON-RPC ``RequestId``: ``null``, an integer or a string."""
+    return value is None or isinstance(value, str) or (isinstance(value, int) and not isinstance(value, bool))
+
 
 class Connection:
     """Minimal JSON-RPC 2.0 connection over newline-delimited JSON frames."""
@@ -54,6 +62,7 @@ def __init__(
         self._handler = handler
         self._next_request_id = 0
         self._pending: dict[int, asyncio.Future[Any]] = {}
+        self._incoming: dict[Any, asyncio.Task[Any]] = {}
         self._tasks = TaskSupervisor(source="acp.Connection")
         self._tasks.add_error_handler(self._on_task_error)
         self._closed = False
@@ -117,16 +126,14 @@ async def send_request(self, method: str, params: JsonValue | None = None) -> An
         payload = {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}
         try:
             await self._transport.send(payload)
-        except BaseException:
-            self._pending.pop(request_id, None)
-            future.cancel()
-            raise
-        self._notify_observers(StreamDirection.OUTGOING, payload)
-        try:
+            self._notify_observers(StreamDirection.OUTGOING, payload)
             return await future
-        except asyncio.CancelledError:
-            self._pending.pop(request_id, None)
+        except BaseException as exc:
+            still_pending = self._pending.pop(request_id, None) is not None
             future.cancel()
+            if isinstance(exc, asyncio.CancelledError) and still_pending:
+                # The request may already be on the wire; ask the peer to stop working on it.
+                self._send_cancel_request(request_id)
             raise
 
     async def send_notification(self, method: str, params: JsonValue | None = None) -> None:
@@ -152,13 +159,20 @@ async def _receive_loop(self) -> None:
     def _process_message(self, message: dict[str, Any]) -> None:
         method = message.get("method")
         has_id = "id" in message
+        if method == _CANCEL_REQUEST_METHOD and not has_id:
+            self._cancel_incoming(message.get("params"))
+            return
         if method is not None:  # this is a request or notification
             # {"jsonrpc": "2.0", "id": 1, "method": "foo", "params": {...}}  # request
             # {"jsonrpc": "2.0", "method": "foo", "params: {...}}  # notification
-            self._tasks.create(
-                self._run_request(message) if has_id else self._run_notification(message),
-                name="acp.Connection.request" if has_id else "acp.Connection.notification",
-            )
+            if not has_id:
+                self._tasks.create(self._run_notification(message), name="acp.Connection.notification")
+                return
+            # The handler gets its own task so ``$/cancel_request`` can cancel it, even before it
+            # starts, without also cancelling delivery of the response the peer still expects.
+            handler = self._tasks.create(self._execute_request(message), name="acp.Connection.request")
+            self._track_incoming(message["id"], handler)
+            self._tasks.create(self._run_request(message, handler), name="acp.Connection.response")
             return
         if has_id:  # this is a response, {"id", "result" | "error"}
             self._handle_response(message)
@@ -184,8 +198,56 @@ def _notify_observers(self, direction: StreamDirection, message: dict[str, Any])
     def _on_observer_error(self, task: asyncio.Task[Any], exc: BaseException) -> None:
         logging.exception("Stream observer coroutine failed", exc_info=exc)
 
-    async def _run_request(self, message: dict[str, Any]) -> None:
-        payload = await self._execute_request(message)
+    def _track_incoming(self, request_id: Any, task: asyncio.Task[Any]) -> None:
+        # A ``$/cancel_request`` can only name a valid request id, and Python would alias an invalid
+        # one such as ``true`` or ``1.0`` with the integer ``1``.
+        if not _is_request_id(request_id):
+            return
+        self._incoming[request_id] = task
+
+        def _forget(done: asyncio.Task[Any]) -> None:
+            if self._incoming.get(request_id) is done:
+                del self._incoming[request_id]
+
+        task.add_done_callback(_forget)
+
+    def _cancel_incoming(self, params: Any) -> None:
+        # ``requestId`` is required and may be ``null``, so a missing one must not match a ``null`` id.
+        if not isinstance(params, dict) or "requestId" not in params:
+            return
+        request_id = params["requestId"]
+        if not _is_request_id(request_id):
+            return
+        # Unknown or already finished requests are ignored, as the protocol allows.
+        task = self._incoming.get(request_id)
+        if task is not None:
+            task.cancel()
+
+    def _send_cancel_request(self, request_id: int) -> None:
+        if self._closed or self._disconnected:
+            return
+        self._tasks.create(
+            self.send_notification(_CANCEL_REQUEST_METHOD, {"requestId": request_id}),
+            name="acp.Connection.cancel_request",
+            on_error=self._on_cancel_request_error,
+        )
+
+    def _on_cancel_request_error(self, task: asyncio.Task[Any], exc: BaseException) -> None:
+        logging.debug("Failed to send %s", _CANCEL_REQUEST_METHOD, exc_info=exc)
+
+    async def _run_request(
+        self, message: dict[str, Any], handler: asyncio.Future[dict[str, Any]] | None = None
+    ) -> None:
+        if handler is None:
+            handler = self._tasks.create(self._execute_request(message), name="acp.Connection.request")
+        # Unlike ``await handler``, ``asyncio.wait`` does not re-raise the handler's cancellation, and
+        # ``$/cancel_request`` only targets the handler, so only ``close()`` can abandon the response.
+        await asyncio.wait({handler})
+        if handler.cancelled():
+            # Cancelled by ``$/cancel_request`` or from inside the handler.
+            payload = {"jsonrpc": "2.0", "id": message["id"], "error": RequestError.request_cancelled().to_error_obj()}
+        else:
+            payload = handler.result()
         await self._transport.send(payload)
         self._notify_observers(StreamDirection.OUTGOING, payload)
 

src/acp/exceptions.py

@@ -42,5 +42,9 @@ def resource_not_found(cls, uri: str | None = None) -> RequestError:
         data = {"uri": uri} if uri is not None else None
         return cls(-32002, "Resource not found", data)
 
+    @classmethod
+    def request_cancelled(cls, data: dict[str, Any] | None = None) -> RequestError:
+        return cls(-32800, "Request cancelled", data)
+
     def to_error_obj(self) -> dict[str, Any]:
         return {"code": self.code, "message": str(self), "data": self.data}