review(#2484): restore exception propagation from background

reader/writer and close streams on crash

Self-review of the first commit caught a regression: the asyncio
branch spawned stdout_reader / stdin_writer via asyncio.create_task
and did `try: await task except BaseException: pass` during cleanup,
which silently swallowed any exception the reader or writer raised.
An anyio task group would have re-raised that through the `async
with` block — the asyncio path has to reproduce that contract to
keep the fix observably equivalent.

Two changes:

1. `add_done_callback(_on_background_task_done)` — if a background
   task crashes while the caller is still inside the `yield`, close
   the memory streams immediately so any in-flight user read wakes
   up with `ClosedResourceError` instead of hanging forever. Mirrors
   the scope-cancellation side effect of a task group. The callback
   logs the exception at debug so operators can trace crashes that
   we only surface on the way out.

2. On `__aexit__` we now collect task results and re-raise the first
   non-cancellation, non-closed-resource exception. CancelledError
   and anyio.ClosedResourceError are expected during the teardown we
   just forced in `_cleanup_process_and_streams`, so they are
   swallowed; anything else (a real crash) propagates.

New regression test:
`test_stdio_client_reader_crash_propagates_on_asyncio` — injects a
TextReceiveStream that raises on the first `__anext__`, asserts the
exception is visible at the async-with exit. Confirmed it FAILS on
the first-commit version ("Failed: DID NOT RAISE"), so it is real
regression coverage, not a tautology.

All 14 stdio tests pass (10 pre-existing + 4 new, 1 Unix-only skip).
Full `tests/client/` pass (203 passed, 1 pre-existing xfail).

Author: sakenuGOD

src/mcp/client/stdio.py

@@ -215,26 +215,67 @@ async def _cleanup_process_and_streams() -> None:
     # design, and cross-task cleanup is fundamentally incompatible with
     # that model, so callers on trio still have to clean up LIFO.
     if sniffio.current_async_library() == "asyncio":
+
+        def _on_background_task_done(task: asyncio.Task[None]) -> None:
+            """
+            If a background reader/writer crashes while the caller is
+            still using the transport, close the memory streams so that
+            any in-flight user read wakes up with ``ClosedResourceError``
+            instead of hanging forever. An anyio task group would have
+            produced the same effect via scope cancellation — this
+            restores that observability on the asyncio path.
+            """
+            if task.cancelled():
+                return
+            exc = task.exception()
+            if exc is None:
+                return
+            logger.debug(
+                "stdio_client background task raised %s — closing streams to wake up caller",
+                type(exc).__name__,
+                exc_info=exc,
+            )
+            for stream in (read_stream_writer, write_stream):
+                try:
+                    stream.close()
+                except Exception:  # pragma: no cover
+                    pass
+
         async with process:
-            stdout_task = asyncio.create_task(stdout_reader())
-            stdin_task = asyncio.create_task(stdin_writer())
+            stdout_task: asyncio.Task[None] = asyncio.create_task(stdout_reader())
+            stdin_task: asyncio.Task[None] = asyncio.create_task(stdin_writer())
+            stdout_task.add_done_callback(_on_background_task_done)
+            stdin_task.add_done_callback(_on_background_task_done)
+            background_tasks = (stdout_task, stdin_task)
             try:
                 yield read_stream, write_stream
             finally:
                 try:
                     await _cleanup_process_and_streams()
                 finally:
-                    for task in (stdout_task, stdin_task):
+                    for task in background_tasks:
                         if not task.done():
                             task.cancel()
-                    for task in (stdout_task, stdin_task):
+                    # Collect results; swallow CancelledError (expected for
+                    # teardown) and anyio.ClosedResourceError (surfaced when
+                    # we closed the streams out from under the reader/writer
+                    # during cleanup). Re-raise anything else so a crash in
+                    # the background does not go unnoticed — matching the
+                    # exception propagation we'd get from an anyio task
+                    # group on the trio path.
+                    pending_exc: BaseException | None = None
+                    for task in background_tasks:
                         try:
                             await task
-                        except BaseException:  # noqa: BLE001, S110
-                            # Reader/writer cancellation or late I/O errors
-                            # surfaced after the streams were closed: those
-                            # are expected during teardown.
+                        except asyncio.CancelledError:
+                            pass
+                        except anyio.ClosedResourceError:
                             pass
+                        except BaseException as exc:  # noqa: BLE001
+                            if pending_exc is None:
+                                pending_exc = exc
+                    if pending_exc is not None:
+                        raise pending_exc
     else:
         async with anyio.create_task_group() as tg, process:
             tg.start_soon(stdout_reader)

tests/client/test_stdio.py

@@ -647,3 +647,39 @@ async def test_stdio_client_shared_exit_stack_fifo_on_asyncio(anyio_backend):
     async with AsyncExitStack() as stack:
         await stack.enter_async_context(stdio_client(params))
         await stack.enter_async_context(stdio_client(params))
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("anyio_backend", ["asyncio"])
+async def test_stdio_client_reader_crash_propagates_on_asyncio(anyio_backend, monkeypatch):
+    """
+    Guardrail for the asyncio branch of #577: moving the reader/writer
+    out of an anyio task group must NOT silently swallow exceptions
+    they raise. An anyio task group would have re-raised those through
+    the async ``with`` block; the asyncio path has to reproduce that
+    contract by collecting the task results in ``__aexit__`` and
+    re-raising anything that isn't cancellation / closed-resource.
+    """
+    from mcp.client import stdio as stdio_mod
+
+    class _BoomTextStream:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            raise RuntimeError("deliberate reader crash for #577 regression test")
+
+    monkeypatch.setattr(stdio_mod, "TextReceiveStream", _BoomTextStream)
+
+    params = StdioServerParameters(command=sys.executable, args=["-c", _QUIET_STDIN_STUB])
+
+    with pytest.raises(RuntimeError, match="deliberate reader crash"):
+        async with stdio_client(params):
+            # Give the reader a chance to raise. The crash should close
+            # the streams out from under us, so we just wait a moment
+            # and then exit the context — the exception is surfaced on
+            # the way out.
+            await anyio.sleep(0.2)