Pin the abandonment-cancellation contract in the interaction suite

maxisbey · maxisbey · commit 79c3ac3a8b84 · 2026-07-02T16:59:17.000Z
diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py
@@ -474,6 +474,20 @@ def __post_init__(self) -> None:
             "never reused within the session."
         ),
     ),
+    "protocol:request-id:caller-supplied": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/patterns/subscriptions#receiving-notifications",
+        behavior=(
+            "A caller can supply the id of a request it sends, so the id is known before any response "
+            "arrives; subscriptions/listen streams are demultiplexed by exactly that id."
+        ),
+        added_in="2026-07-28",
+        deferred=(
+            "No public API surface yet: the capability exists at the dispatcher seam "
+            "(CallOptions['request_id'], unit-tested there), but ClientSession.send_request does not "
+            "expose it. The public consumer arrives with the client-side listen driver (Client.listen), "
+            "whose interaction tests will exercise it end to end."
+        ),
+    ),
     "protocol:notifications:no-response": Requirement(
         source=f"{SPEC_BASE_URL}/basic#notifications",
         behavior=(
@@ -484,14 +498,29 @@ def __post_init__(self) -> None:
     "protocol:cancel:abort-signal": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#cancellation-flow",
         behavior=(
-            "Cancelling an in-flight request through the client API sends notifications/cancelled with "
-            "the request id and fails the local call."
+            "Abandoning an in-flight request client-side (cancelling the task awaiting it) cancels the "
+            "request itself: the server-side handler stops and the session serves later requests "
+            "normally. Each transport carries the signal in its own spelling - a cancelled frame on "
+            "stream wires, closing the request's own response stream at 2026-07-28 streamable HTTP."
         ),
-        deferred=(
-            "Not implemented in the SDK: there is no public client-side API to cancel an in-flight "
-            "request; cancellation requires hand-constructing the notification (which is how "
-            "protocol:cancel:in-flight exercises the receiving side)."
+        arm_exclusions=(
+            ArmExclusion(
+                reason="requires-session",
+                transport="streamable-http-stateless",
+                note=(
+                    "The 2025-era cancel frame POSTs on a fresh per-request transport that shares no "
+                    "in-flight state with the blocked request, so the handler is never interrupted."
+                ),
+            ),
+        ),
+    ),
+    "protocol:cancel:abort-scoped": Requirement(
+        source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#behavior-requirements",
+        behavior=(
+            "Abandoning one in-flight request cancels only that request: a concurrent request on the "
+            "same connection keeps running and returns its result."
         ),
+        arm_exclusions=(ArmExclusion(reason="requires-session", transport="streamable-http-stateless"),),
     ),
     "protocol:cancel:handler-abort-propagates": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#behavior-requirements",
@@ -550,6 +579,16 @@ def __post_init__(self) -> None:
             ArmExclusion(reason="server-initiated-request", spec_version="2026-07-28"),
         ),
     ),
+    "protocol:cancel:stream-frame": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/utilities/cancellation#transport-specific-cancellation",
+        behavior=(
+            "On stream (stdio-shaped) wires at 2026-07-28, abandoning an in-flight request sends exactly "
+            "one notifications/cancelled naming its request id - streams keep the frame spelling of "
+            "cancellation that streamable HTTP dropped."
+        ),
+        added_in="2026-07-28",
+        note="Exercised over the in-memory stream pair, the same dual-era wire stdio serves.",
+    ),
     "protocol:cancel:unknown-id-ignored": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#error-handling",
         behavior=(
@@ -3256,6 +3295,30 @@ def __post_init__(self) -> None:
         transports=("streamable-http",),
         note="Only observable over HTTP: Accept is an HTTP request header.",
     ),
+    "client-transport:http:cancel-closes-stream": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/transports/streamable-http#cancellation",
+        behavior=(
+            "At 2026-07-28, abandoning an in-flight request closes that request's own POST stream and "
+            "posts nothing further: no notifications/cancelled reaches the server (the revision defines "
+            "no client-to-server notifications), and the server treats the disconnect as cancellation "
+            "of exactly that request."
+        ),
+        transports=("streamable-http",),
+        added_in="2026-07-28",
+        supersedes=("client-transport:http:cancel-posts-frame",),
+        note="HTTP-only by nature: the response stream that closing constitutes the signal is an HTTP exchange.",
+    ),
+    "client-transport:http:cancel-posts-frame": Requirement(
+        source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#cancellation-flow",
+        behavior=(
+            "At 2025-era revisions, abandoning an in-flight request POSTs exactly one "
+            "notifications/cancelled naming its request id."
+        ),
+        transports=("streamable-http",),
+        removed_in="2026-07-28",
+        superseded_by="client-transport:http:cancel-closes-stream",
+        note="HTTP-only by nature: pins that the frame travels as its own POST on the legacy HTTP wire.",
+    ),
     "client-transport:http:concurrent-streams": Requirement(
         source="sdk",
         behavior="Multiple concurrent POST-initiated SSE streams each deliver their response to the right caller.",
diff --git a/tests/interaction/lowlevel/test_cancellation.py b/tests/interaction/lowlevel/test_cancellation.py
@@ -1,9 +1,10 @@
 """Cancellation interactions against the low-level Server, driven through the public Client API.
 
-There is no client-side cancellation API: cancelling means sending a CancelledNotification
-carrying the request id, which only the server-side handler can observe (`ctx.request_id`), so
-these tests capture the id from inside the blocked handler before cancelling. The handler blocks
-on an Event rather than a sleep, and every wait is bounded by `anyio.fail_after`.
+Client-side, cancelling means abandoning: cancelling the task that awaits a call makes the SDK
+carry the signal in the transport's own spelling (a cancelled frame on stream wires, closing the
+request's own response stream at 2026-07-28 streamable HTTP). The receiving-side tests instead
+script a CancelledNotification by hand, capturing the request id from inside the blocked handler.
+Handlers block on an Event rather than a sleep, and every wait is bounded by `anyio.fail_after`.
 """
 
 import anyio
@@ -20,9 +21,11 @@
     JSONRPCNotification,
     JSONRPCRequest,
     JSONRPCResponse,
+    ListToolsResult,
     PingRequest,
     ServerCapabilities,
     TextContent,
+    Tool,
 )
 
 from mcp import MCPError
@@ -344,3 +347,121 @@ async def scripted_server(streams: MessageStream) -> None:
         assert pong == snapshot(EmptyResult())
         # The stream is ordered, so a courtesy cancel would have arrived ahead of the ping.
         assert received_methods == snapshot(["initialize", "ping"])
+
+
+@requirement("protocol:cancel:abort-signal")
+async def test_abandoning_a_call_stops_the_server_handler(connect: Connect) -> None:
+    """Cancelling the task that awaits a call cancels the request itself, not just the local wait:
+    the server-side handler is interrupted, and the session serves later requests normally.
+
+    Spec-mandated (cancellation flow): the sender cancels requests it abandons; the wire spelling
+    is per-transport (frame on stream wires, response-stream close at 2026 streamable HTTP).
+    """
+    handler_started = anyio.Event()
+    handler_cancelled = anyio.Event()
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        if params.name == "block":
+            handler_started.set()
+            try:
+                await anyio.Event().wait()  # parked until the client's abandonment cancels it
+            except anyio.get_cancelled_exc_class():
+                handler_cancelled.set()
+                raise
+        assert params.name == "echo"
+        return CallToolResult(content=[TextContent(text="ok")])
+
+    async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
+        return ListToolsResult(tools=[Tool(name=name, input_schema={"type": "object"}) for name in ("block", "echo")])
+
+    server = Server("blocker", on_list_tools=list_tools, on_call_tool=call_tool)
+
+    async with connect(server) as client:
+        abandon = anyio.CancelScope()
+
+        async def call_and_abandon() -> None:
+            with abandon:
+                await client.call_tool("block", {})
+                raise NotImplementedError  # unreachable: the call never resolves
+            assert abandon.cancelled_caught
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(call_and_abandon)
+            with anyio.fail_after(5):
+                await handler_started.wait()
+            abandon.cancel()
+            with anyio.fail_after(5):
+                await handler_cancelled.wait()
+
+        # Let the abandoned call's late error response (sent on the legacy arms) arrive and be
+        # dropped while the client is still open, so teardown never races its delivery.
+        await anyio.wait_all_tasks_blocked()
+        result = await client.call_tool("echo", {})
+        assert result.content == [TextContent(text="ok")]
+
+
+@requirement("protocol:cancel:abort-scoped")
+async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Connect) -> None:
+    """Cancellation is scoped to the request it names: with two calls genuinely in flight,
+    abandoning the first interrupts only its handler and the second returns its result.
+
+    Steps:
+        1. `doomed` and `survivor` are both mid-flight (each handler has started).
+        2. The client abandons `doomed`; its handler observes cancellation.
+        3. `survivor` is released and completes normally.
+    """
+    doomed_started = anyio.Event()
+    doomed_cancelled = anyio.Event()
+    survivor_started = anyio.Event()
+    release_survivor = anyio.Event()
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        if params.name == "doomed":
+            doomed_started.set()
+            try:
+                await anyio.Event().wait()  # parked until the client's abandonment cancels it
+            except anyio.get_cancelled_exc_class():
+                doomed_cancelled.set()
+                raise
+        assert params.name == "survivor"
+        survivor_started.set()
+        with anyio.fail_after(5):
+            await release_survivor.wait()
+        return CallToolResult(content=[TextContent(text="survived")])
+
+    async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
+        return ListToolsResult(
+            tools=[Tool(name=name, input_schema={"type": "object"}) for name in ("doomed", "survivor")]
+        )
+
+    server = Server("pair", on_list_tools=list_tools, on_call_tool=call_tool)
+
+    async with connect(server) as client:
+        abandon = anyio.CancelScope()
+        results: list[CallToolResult] = []
+
+        async def doomed_call() -> None:
+            with abandon:
+                await client.call_tool("doomed", {})
+                raise NotImplementedError  # unreachable: the call never resolves
+
+        async def survivor_call() -> None:
+            results.append(await client.call_tool("survivor", {}))
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(doomed_call)
+            with anyio.fail_after(5):
+                await doomed_started.wait()
+            tg.start_soon(survivor_call)
+            with anyio.fail_after(5):
+                await survivor_started.wait()
+            abandon.cancel()
+            with anyio.fail_after(5):
+                await doomed_cancelled.wait()
+            release_survivor.set()
+
+        # Let the abandoned call's late error response (sent on the legacy arms) arrive and be
+        # dropped while the client is still open, so teardown never races its delivery.
+        await anyio.wait_all_tasks_blocked()
+
+    assert results == snapshot([CallToolResult(content=[TextContent(text="survived")])])
diff --git a/tests/interaction/lowlevel/test_wire.py b/tests/interaction/lowlevel/test_wire.py
@@ -308,3 +308,58 @@ async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelReq
 
     assert len(errors) == 1
     assert errors[0].code == INVALID_PARAMS
+
+
+@requirement("protocol:cancel:stream-frame")
+async def test_abandoning_a_call_on_a_modern_stream_wire_sends_one_cancelled_frame() -> None:
+    """At 2026-07-28 over a stream (stdio-shaped) wire, abandoning an in-flight call puts exactly
+    one notifications/cancelled naming that request on the wire, and the frame interrupts the
+    server-side handler - stream wires keep the frame spelling that 2026 streamable HTTP dropped.
+    """
+    handler_started = anyio.Event()
+    handler_cancelled = anyio.Event()
+
+    async def list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        raise NotImplementedError  # registered so tools/call is served; the stream wire never lists
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        assert params.name == "block"
+        handler_started.set()
+        try:
+            await anyio.Event().wait()  # parked until the client's abandonment cancels it
+        except anyio.get_cancelled_exc_class():
+            handler_cancelled.set()
+            raise
+        raise NotImplementedError  # unreachable
+
+    server = Server("blocker", on_list_tools=list_tools, on_call_tool=call_tool)
+    recording = RecordingTransport(InMemoryTransport(server))
+
+    async with Client(recording, mode="2026-07-28") as client:
+        abandon = anyio.CancelScope()
+
+        async def call_and_abandon() -> None:
+            with abandon:
+                await client.call_tool("block", {})
+                raise NotImplementedError  # unreachable: the call never resolves
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(call_and_abandon)
+            with anyio.fail_after(5):
+                await handler_started.wait()
+            abandon.cancel()
+            with anyio.fail_after(5):
+                await handler_cancelled.wait()
+
+        # Let the cancelled call's late error response arrive and be dropped while the client
+        # is still open, so teardown never races its delivery.
+        await anyio.wait_all_tasks_blocked()
+
+    call, cancel = [message.message for message in recording.sent]
+    assert isinstance(call, JSONRPCRequest)
+    assert call.method == "tools/call"
+    assert isinstance(cancel, JSONRPCNotification)
+    assert cancel.method == "notifications/cancelled"
+    assert cancel.params == {"requestId": call.id, "reason": "caller cancelled"}
diff --git a/tests/interaction/transports/test_client_transport_http.py b/tests/interaction/transports/test_client_transport_http.py
