python-client: client-initiated turn cancellation (cancel)

.changeset/python-client-cancel.md

@@ -0,0 +1,12 @@
+---
+"@smooai/smooth-operator": minor
+---
+
+Add client-initiated turn cancellation (the "Stop button") to the Python async client SDK, mirroring the TypeScript client.
+
+- `SmoothAgentClient.cancel(request_id=..., session_id=None)` sends a `cancel` frame for an in-flight `send_message` turn.
+- `MessageTurn.cancel()` is the ergonomic "stop THIS turn" convenience — it cancels using the turn's own `request_id` + originating `session_id`.
+- The terminal `cancelled` event now settles the matching `MessageTurn` as a **user-stop**: the turn *resolves* (never raises), `await turn` yields the `Cancelled` event, the async iterator ends cleanly, and `turn.cancelled` is `True` so callers can tell a user-stop apart from an error.
+- `CancelRequest` / `Cancelled` are now first-class members of the `ClientAction` / `ServerEvent` unions (and the validator's schema maps).
+
+Idempotent: cancelling with no active turn — or receiving a `cancelled` with no matching in-flight turn — is a harmless no-op.

python/README.md

@@ -75,6 +75,23 @@ final = await turn                                    # the terminal eventual_re
 print("\nmessageId:", final.data.payload.message_id)
 ```
 
+### Stop button
+
+Call `turn.cancel()` (or `client.cancel(request_id=..., session_id=...)`) to stop an
+in-flight turn. The server aborts its LLM + tool work and replies with a terminal
+`cancelled` event: the turn **resolves** as a user-stop — `await turn` yields the
+`Cancelled` (never raises), the `async for` ends cleanly, and `turn.cancelled` is
+`True` so you can tell a stop apart from an error. Idempotent — cancelling with
+nothing in flight is a harmless no-op.
+
+```python
+turn = client.send_message(session_id=session.session_id, message="write a novel")
+turn.cancel()                       # the user hit Stop
+final = await turn                  # resolves; does NOT raise
+if turn.cancelled:                  # final.type == "cancelled", final.status == 499
+    print("stopped by user")
+```
+
 ```mermaid
 %%{init: {'theme':'base','themeVariables':{'background':'#020618','primaryColor':'#0b1426','primaryTextColor':'#e6edf6','primaryBorderColor':'#2b3a52','lineColor':'#7c8aa0','actorBkg':'#0b1426','actorBorder':'#2b3a52','actorTextColor':'#e6edf6','signalColor':'#7c8aa0','signalTextColor':'#e6edf6','noteBkgColor':'#f49f0a','noteTextColor':'#1a0f00','noteBorderColor':'#ff6b6c','fontFamily':'ui-sans-serif, system-ui, sans-serif'}}}%%
 sequenceDiagram

python/src/smooth_operator/__init__.py

@@ -22,6 +22,8 @@
     EVENT_TYPES,
     ActionType,
     AuthContext,
+    Cancelled,
+    CancelRequest,
     Checkpoint,
     ClientAction,
     Conversation,
@@ -95,6 +97,7 @@
     "SendMessageRequest",
     "GetSessionRequest",
     "GetMessagesRequest",
+    "CancelRequest",
     "PingRequest",
     "AuthContext",
     # response payloads
@@ -116,6 +119,7 @@
     "OtpSent",
     "OtpVerified",
     "OtpInvalid",
+    "Cancelled",
     "ErrorEvent",
     "Pong",
     "ErrorObject",

python/src/smooth_operator/client.py

@@ -28,6 +28,7 @@
 from . import _generated as _g
 from .transport import Transport, WebSocketTransport
 from .types import (
+    Cancelled,
     CreateConversationSessionResponse,
     EventualResponse,
     GetMessagesResponse,
@@ -89,10 +90,13 @@ def __init__(
         request_id: str,
         on_close: Callable[[], None],
         turn_timeout: float = 0.0,
+        on_cancel: Callable[[], None] | None = None,
     ) -> None:
         self.request_id = request_id
         self._on_close = on_close
         self._turn_timeout = turn_timeout
+        self._on_cancel = on_cancel
+        self._cancelled = False
         # asyncio.Queue / asyncio.Event bind to the running loop lazily on first use,
         # so they need no explicit loop capture. The settled future, however, is bound
         # at construction — use get_running_loop() so it attaches to the loop that is
@@ -102,10 +106,10 @@ def __init__(
         # to hang silently.
         self._queue: asyncio.Queue[ServerEvent] = asyncio.Queue()
         self._done = asyncio.Event()
-        self._final: EventualResponse | None = None
+        self._final: EventualResponse | Cancelled | None = None
         self._error: BaseException | None = None
         loop = asyncio.get_running_loop()
-        self._settled: asyncio.Future[EventualResponse] = loop.create_future()
+        self._settled: asyncio.Future[EventualResponse | Cancelled] = loop.create_future()
         # Avoid "Future exception was never retrieved" noise when the caller only
         # iterates (and surfaces the error via __aiter__) and never awaits the turn.
         self._settled.add_done_callback(lambda f: f.cancelled() or f.exception())
@@ -115,6 +119,15 @@ def __init__(
         if turn_timeout > 0:
             self._timeout_handle = loop.call_later(turn_timeout, self._on_timeout)
 
+    @property
+    def cancelled(self) -> bool:
+        """True if this turn ended because the user stopped it — a terminal
+        ``cancelled`` event settled it, as opposed to completing
+        (``eventual_response``) or erroring. This is the UI's signal to tell a
+        user-stop apart from a failure: on a user-stop the turn *resolves* (``await
+        turn`` yields the terminal :class:`Cancelled` event, never raises)."""
+        return self._cancelled
+
     # ── feed (called by the client dispatcher) ─────────────────────────────────
     def push(self, event: ServerEvent) -> None:
         if self._done.is_set():
@@ -130,6 +143,24 @@ def push(self, event: ServerEvent) -> None:
 
         if event.type == "eventual_response":
             self._finish(event, None)
+        elif event.type == "cancelled":
+            # Terminal user-stop: settle by *resolving* (never erroring) so the async
+            # iterator ends cleanly and ``await turn`` yields the Cancelled event.
+            # ``cancelled`` is set for the UI to tell a user-stop from a completed
+            # or errored turn.
+            self._cancelled = True
+            self._finish(event, None)
+
+    def cancel(self) -> None:
+        """Request cancellation of THIS turn — the ergonomic "Stop" button. Sends a
+        ``cancel`` frame carrying the turn's own ``requestId`` (and originating
+        ``sessionId``) via the client. Idempotent: a no-op once the turn has settled.
+        The turn itself settles when the server's terminal ``cancelled`` event arrives
+        (see :attr:`cancelled`); this only sends the request."""
+        if self._done.is_set():
+            return
+        if self._on_cancel is not None:
+            self._on_cancel()
 
     def abort(self, err: BaseException) -> None:
         """Force-close the turn (e.g. on disconnect)."""
@@ -143,7 +174,7 @@ def _on_timeout(self) -> None:
             return
         self._finish(None, TurnTimeoutError(self.request_id, self._turn_timeout))
 
-    def _finish(self, final: EventualResponse | None, err: BaseException | None) -> None:
+    def _finish(self, final: EventualResponse | Cancelled | None, err: BaseException | None) -> None:
         if self._done.is_set():
             return
         if self._timeout_handle is not None:
@@ -192,8 +223,11 @@ async def _iterate(self) -> AsyncIterator[ServerEvent]:
     def __await__(self):
         return self._settled.__await__()
 
-    async def result(self) -> EventualResponse:
-        """Await the terminal :class:`EventualResponse` (or raise ProtocolError)."""
+    async def result(self) -> EventualResponse | Cancelled:
+        """Await the terminal event: an :class:`EventualResponse` on completion, or a
+        :class:`Cancelled` on a user-stop (never raises for a cancel — check
+        :attr:`cancelled` or narrow on ``.type``). Raises :class:`ProtocolError` on an
+        ``error`` event."""
         return await self._settled
 
 
@@ -316,6 +350,7 @@ def send_message(self, *, session_id: str, message: str, stream: bool = True) ->
             request_id,
             lambda: self._turns.pop(request_id, None),
             turn_timeout=self._turn_timeout,
+            on_cancel=lambda: self.cancel(request_id=request_id, session_id=session_id),
         )
         self._turns[request_id] = turn
         try:
@@ -335,6 +370,23 @@ def send_message(self, *, session_id: str, message: str, stream: bool = True) ->
             turn.abort(err)
         return turn
 
+    def cancel(self, *, request_id: str, session_id: str | None = None) -> None:
+        """Client-initiated turn cancellation — the "Stop" button. Sends a ``cancel``
+        frame for the in-flight ``send_message`` turn identified by ``request_id``. The
+        server aborts the turn's LLM + tool work, frees the turn slot, and emits a
+        terminal ``cancelled`` event (in place of ``eventual_response``) echoing that
+        ``request_id``; the matching :class:`MessageTurn` then settles as a user-stop —
+        it *resolves* (never raises), ``await turn`` yields the :class:`Cancelled`
+        event, and ``turn.cancelled`` is ``True``.
+
+        Idempotent: a cancel with no active turn is a silent server no-op, and this
+        never raises on that account. For the common "stop THIS turn" case, prefer
+        :meth:`MessageTurn.cancel`."""
+        frame: dict = {"action": "cancel", "requestId": request_id}
+        if session_id is not None:
+            frame["sessionId"] = session_id
+        self._transport.send(json.dumps(frame))
+
     def confirm_tool_action(self, *, session_id: str, request_id: str, approved: bool) -> None:
         """Approve/reject a pending tool write, resuming the paused turn for
         ``request_id``. Resumed events flow back into the original :class:`MessageTurn`."""

python/src/smooth_operator/types.py

@@ -47,6 +47,7 @@
 GetMessagesRequest = _g.GetMessagesRequest
 ConfirmToolActionRequest = _g.ConfirmToolActionRequest
 VerifyOtpRequest = _g.VerifyOtpRequest
+CancelRequest = _g.CancelRequest
 PingRequest = _g.PingRequest
 AuthContext = _g.AuthContext
 
@@ -70,6 +71,7 @@
 OtpSent = _g.OtpSent
 OtpVerified = _g.OtpVerified
 OtpInvalid = _g.OtpInvalid
+Cancelled = _g.Cancelled
 Pong = _g.Pong
 
 # The generated `error` event model is named ``Error`` — which shadows the builtin.
@@ -107,6 +109,7 @@ class ActionType(StrEnum):
     get_conversation_messages = "get_conversation_messages"
     confirm_tool_action = "confirm_tool_action"
     verify_otp = "verify_otp"
+    cancel = "cancel"
     ping = "ping"
 
 
@@ -123,6 +126,7 @@ class EventType(StrEnum):
     otp_sent = "otp_sent"
     otp_verified = "otp_verified"
     otp_invalid = "otp_invalid"
+    cancelled = "cancelled"
     error = "error"
     pong = "pong"
 
@@ -147,6 +151,7 @@ class EventType(StrEnum):
         OtpSent,
         OtpVerified,
         OtpInvalid,
+        Cancelled,
         ErrorEvent,
         Pong,
     ],
@@ -166,6 +171,7 @@ class EventType(StrEnum):
         GetMessagesRequest,
         ConfirmToolActionRequest,
         VerifyOtpRequest,
+        CancelRequest,
         PingRequest,
     ],
     Field(discriminator="action"),
@@ -225,6 +231,7 @@ def is_client_action(frame: object) -> bool:
     "otp_sent",
     "otp_verified",
     "otp_invalid",
+    "cancelled",
     "error",
     "pong",
 ]

python/src/smooth_operator/validate.py

@@ -44,6 +44,7 @@
     "otp_sent": "events/otp-sent.schema.json",
     "otp_verified": "events/otp-verified.schema.json",
     "otp_invalid": "events/otp-invalid.schema.json",
+    "cancelled": "events/cancelled.schema.json",
     "error": "events/error.schema.json",
     "pong": "events/pong.schema.json",
 }
@@ -56,6 +57,7 @@
     "get_conversation_messages": "actions/get-messages.schema.json#/$defs/Request",
     "confirm_tool_action": "actions/confirm-tool-action.schema.json#/$defs/Request",
     "verify_otp": "actions/verify-otp.schema.json#/$defs/Request",
+    "cancel": "actions/cancel.schema.json#/$defs/Request",
     "ping": "actions/ping.schema.json#/$defs/Request",
 }
 

python/tests/test_client.py

@@ -271,6 +271,114 @@ async def iterate() -> None:
     assert seen == ["write_confirmation_required", "eventual_response"]
 
 
+# ─────────────────────────── cancellation ─────────────────────────────────────
+async def test_turn_cancel_emits_a_cancel_frame_for_the_turns_request_id() -> None:
+    client, transport = make_client()
+    await client.connect()
+
+    turn = client.send_message(session_id="sess-9", message="long one", stream=True)
+    req_id = transport.last_sent()["requestId"]
+
+    turn.cancel()
+
+    sent = transport.last_sent()
+    assert sent == {"action": "cancel", "requestId": req_id, "sessionId": "sess-9"}
+
+
+async def test_client_cancel_emits_a_cancel_frame_omitting_absent_session() -> None:
+    client, transport = make_client()
+    await client.connect()
+
+    client.cancel(request_id="req-abc")
+    assert transport.last_sent() == {"action": "cancel", "requestId": "req-abc"}
+
+
+async def test_cancelled_event_settles_the_turn_as_user_stop_not_error() -> None:
+    client, transport = make_client()
+    await client.connect()
+
+    turn = client.send_message(session_id="s", message="stop me")
+    req_id = transport.last_sent()["requestId"]
+
+    seen: list[str] = []
+
+    async def iterate() -> None:
+        # A user-stop must end the iterator cleanly — no exception raised here.
+        async for ev in turn:
+            seen.append(ev.type)
+
+    task = asyncio.create_task(iterate())
+    await asyncio.sleep(0)
+
+    transport.emit(
+        {
+            "type": "stream_token",
+            "requestId": req_id,
+            "token": "Hel",
+            "data": {"requestId": req_id, "token": "Hel"},
+        }
+    )
+    transport.emit(
+        {
+            "type": "cancelled",
+            "requestId": req_id,
+            "status": 499,
+            "data": {"requestId": req_id, "status": 499},
+        }
+    )
+
+    # await turn RESOLVES with the terminal Cancelled event (never raises).
+    final = await turn
+    await task
+
+    assert final.type == "cancelled"
+    assert final.status == 499
+    assert turn.cancelled is True
+    # The iterator yielded the token then the terminal cancelled, then ended cleanly.
+    assert seen == ["stream_token", "cancelled"]
+
+
+async def test_cancel_with_no_active_turn_is_a_harmless_no_op() -> None:
+    client, transport = make_client()
+    await client.connect()
+
+    # Cancel with nothing in flight — must not raise, just sends the frame.
+    client.cancel(request_id="never-started")
+    assert transport.last_sent()["action"] == "cancel"
+
+    # A cancelled event that matches no turn is dropped as an uncorrelated no-op.
+    transport.emit({"type": "cancelled", "requestId": "unknown", "status": 499})
+
+
+async def test_cancelled_is_idempotent_after_the_turn_already_settled() -> None:
+    client, transport = make_client()
+    await client.connect()
+
+    turn = client.send_message(session_id="s", message="q")
+    req_id = transport.last_sent()["requestId"]
+
+    transport.emit(
+        {
+            "type": "eventual_response",
+            "requestId": req_id,
+            "status": 200,
+            "data": {
+                "requestId": req_id,
+                "status": 200,
+                "data": {"messageId": "00000000-0000-0000-0000-000000000009", "response": None},
+            },
+        }
+    )
+    final = await turn
+    assert final.type == "eventual_response"
+    assert turn.cancelled is False
+
+    # A late cancel / cancelled after completion must not flip state or raise.
+    turn.cancel()
+    transport.emit({"type": "cancelled", "requestId": req_id, "status": 499})
+    assert turn.cancelled is False
+
+
 # ─────────────────────────── correlation ──────────────────────────────────────
 async def test_create_session_resolves_with_immediate_response_data() -> None:
     client, transport = make_client()