fix(connection): support $/cancel_request request cancellation (#158)

* fix(connection): honor $/cancel_request and reply -32800 to cancelled requests

ACP v1 defines the `$/cancel_request` notification (part of the stable v1
schema vendored in schema/, 1.23.0; see
https://agentclientprotocol.com/protocol/v1/cancellation), and the SDK
already ships `PROTOCOL_METHODS["cancel_request"]`, but `Connection` never
handled it:

- An incoming `$/cancel_request` fell through to the router, which logged
  an ERROR traceback (method not found) while the targeted handler kept
  running and the peer never received the required response.
- A handler that ended with `CancelledError` (cancelled from inside the
  agent) produced no response at all, leaving the peer waiting forever.
- Cancelling the task awaiting `send_request` only dropped the local
  future; the peer kept working on the request.
- `close()` hung forever when a handler answered its own cancellation with
  a result, because the response was queued on an already-closed sender.

`Connection` now tracks in-flight incoming requests by id, cancels the
handler task on `$/cancel_request`, answers cancelled requests with
`-32800 Request cancelled` (a handler may still return a partial result),
sends `$/cancel_request` when a still-pending outgoing request is
cancelled locally, and skips replies once the connection is closed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* fix(connection): keep request responses out of the handler's cancellation

Two races could still lose the response to a cancelled request:

- A `$/cancel_request` read in the same buffer as its request cancelled the
  request task before it first ran, so the coroutine that would have sent
  `-32800` never executed.
- A late or repeated `$/cancel_request` arriving while the response was being
  sent cancelled that send, so the peer got no response at all.

Each incoming request now runs its handler in its own task, which is the only
thing `$/cancel_request` targets. A separate supervised task waits for the
handler with `asyncio.wait` and sends its result, or `-32800` if the handler
was cancelled. `close()` still cancels both, so shutdown never waits on a
blocked send, and the `Task.uncancel()` / closed-send special cases are no
longer needed.

Also document request cancellation in the quickstart and state that a
cancelled handler answers with its result or `-32800`.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* fix(connection): only cancel requests named by a valid requestId

`$/cancel_request` requires a `requestId`, and a request id is `null`, an
integer or a string. The handler did not check either, so a malformed
cancellation could abort an unrelated request:

- A notification without `params`, with non-object `params` or without
  `requestId` was treated as `requestId: null` and cancelled a pending
  request whose id is `null`.
- `requestId: true` / `false` cancelled pending requests `1` / `0`, and
  `1.0` cancelled `1`, because Python equates these dict keys. Likewise a
  request sent with a non-integer id such as `true` or `1.0` could be
  cancelled by a cancellation for `1`.

Cancellations without a `requestId`, or with one that is not a valid request
id, are now ignored, and only requests with a valid id are tracked for
cancellation. An explicit `requestId: null` still cancels a `null` request.

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: monody0007

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}