From 9a002cbf96c02affce54b4c724bb22ba97519f5f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 1 Nov 2026 20:02:30 +0000 Subject: [PATCH] Relay stdio stderr into errlog when it is not a real file Jupyter's stderr reports a descriptor that points at the launch terminal, and streams such as io.StringIO have none, so handing errlog to the child either hid startup crashes or failed the spawn. Inherit a real file; otherwise pipe stderr and copy it into errlog.write(), including the lines written while the server exits. Co-authored-by: Evan Liu --- docs/client/transports.md | 2 +- src/mcp/client/stdio.py | 93 ++++++- src/mcp/os/win32/utilities.py | 6 +- tests/client/test_stdio.py | 300 ++++++++++++++++++++++- tests/transports/stdio/conftest.py | 2 +- tests/transports/stdio/test_lifecycle.py | 139 +++++++++++ 6 files changed, 527 insertions(+), 15 deletions(-) diff --git a/docs/client/transports.md b/docs/client/transports.md index b41fe51348..4b8d9d641a 100644 --- a/docs/client/transports.md +++ b/docs/client/transports.md @@ -114,7 +114,7 @@ Describe the process with `StdioServerParameters` and hand it to `Client`: Entering the block spawns the process. Leaving it shuts the subprocess down: close stdin, wait, kill if it lingers. You never clean it up yourself. -The child's stderr goes to yours. To send it somewhere else, build the transport yourself with `stdio_client` (from `mcp`) and pass that instead: `Client(stdio_client(server, errlog=log_file))`. +The child's stderr goes to yours; in a Jupyter notebook, that is the cell's output. To send it somewhere else, build the transport yourself with `stdio_client` (from `mcp`) and pass that instead: `Client(stdio_client(server, errlog=log_file))`. A file is handed to the child as its stderr. Any other text stream, such as an `io.StringIO`, gets the child's output through its `write()` while the client is connected. !!! warning The child does **not** inherit your environment. It gets a minimal allow-list (`HOME`, `LOGNAME`, diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py index 6a3ad12111..85817dbf6c 100644 --- a/src/mcp/client/stdio.py +++ b/src/mcp/client/stdio.py @@ -8,8 +8,10 @@ process nor hang on one. """ +import io import logging import os +import subprocess import sys from collections.abc import AsyncGenerator from contextlib import asynccontextmanager, suppress @@ -69,6 +71,9 @@ # Time for the writer to flush accepted messages before stdin closes. _WRITER_FLUSH_TIMEOUT = 0.5 +# Time for the stderr relay to copy what an exited server left in the pipe. +_STDERR_DRAIN_TIMEOUT = 0.5 + # How often to poll returncode while waiting for the process to die. _EXIT_POLL_INTERVAL = 0.01 @@ -105,7 +110,7 @@ class StdioServerParameters(BaseModel): """The working directory to use when spawning the process.""" encoding: str = "utf-8" - """Text encoding for messages to and from the server.""" + """Text encoding for messages to and from the server, and for its relayed stderr.""" encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict" """Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers.""" @@ -117,17 +122,24 @@ async def stdio_client( ) -> AsyncGenerator[TransportStreams, None]: """Spawns an MCP server subprocess and connects to it over stdin/stdout. + The server's stderr goes to `errlog`. A real file is handed to the server + as its stderr. Any other text stream (an `io.StringIO`, Jupyter's + `sys.stderr`, a logging shim) receives the server's output through + `write()`, decoded with `server.encoding` (undecodable bytes are replaced), + until shutdown. + Raises: OSError: If the server process cannot be spawned. ValueError: If the spawn parameters are invalid (embedded NUL bytes). """ command = await _get_executable_command(server.command) + relay_stderr = _needs_stderr_relay(errlog) process = await _create_platform_compatible_process( command=command, args=server.args, env=get_default_environment() | (server.env or {}), - errlog=errlog, + errlog=subprocess.PIPE if relay_stderr else errlog, cwd=server.cwd, ) @@ -138,6 +150,7 @@ async def stdio_client( shutting_down = False writer_done = anyio.Event() + stderr_relayed = anyio.Event() if relay_stderr else None async def stdout_reader() -> None: assert process.stdout, "Opened process is missing stdout" @@ -193,7 +206,7 @@ async def shutdown() -> None: await writer_done.wait() if flush_scope.cancelled_caught: await anyio.lowlevel.cancel_shielded_checkpoint() # resync coverage on 3.11 (gh-106749) - await _stop_server_process(process) + await _stop_server_process(process, stderr_relayed) await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader) # One pass so unblocked tasks exit via their except paths before the cancel. await anyio.lowlevel.checkpoint() @@ -201,6 +214,8 @@ async def shutdown() -> None: async with anyio.create_task_group() as tg: tg.start_soon(stdout_reader) tg.start_soon(stdin_writer) + if stderr_relayed is not None: + tg.start_soon(_relay_stderr, process, errlog, server.encoding, stderr_relayed) try: yield read_stream, write_stream finally: @@ -216,6 +231,63 @@ async def shutdown() -> None: await anyio.lowlevel.cancel_shielded_checkpoint() +def _needs_stderr_relay(errlog: TextIO) -> bool: + """Whether the server's stderr has to reach `errlog` through `write()`. + + A file's descriptor is handed to the server, as before. Any other text + stream only shows what goes through `write()`, even when `fileno()` + succeeds: Jupyter's leads to the terminal behind the notebook. + """ + if isinstance(errlog, io.TextIOBase) and not isinstance(errlog, io.TextIOWrapper): + return True + try: + errlog.fileno() + except (AttributeError, io.UnsupportedOperation): + # No descriptor (pytest's capsys, a logging shim). An int such as + # `subprocess.DEVNULL`, or a binary stream such as `io.BytesIO`, still + # goes to the spawn, which accepts or rejects it up front. + return hasattr(errlog, "write") and not isinstance(errlog, io.RawIOBase | io.BufferedIOBase) + return False + + +async def _relay_stderr(process: ServerProcess, errlog: TextIO, encoding: str, done: anyio.Event) -> None: + """Copies the server's stderr into `errlog` until the pipe ends or shutdown closes it. + + A write `errlog` rejects loses only that chunk, and only the first is logged. + The pipe is read to the end regardless, or a chatty server would block on it. + `errlog` is flushed once at the end: Jupyter flushes on its own timer, and + an explicit flush there waits on another thread. + """ + assert process.stderr, "Opened process is missing stderr" + + write_failed = False + try: + # Shielded like the stdout drain: a cancelled caller must not leave the + # server blocked on a full pipe. Shutdown ends the read by closing it. + with anyio.CancelScope(shield=True): + # Diagnostics must not take the transport down, so bad bytes are replaced. + async for text in TextReceiveStream(process.stderr, encoding=encoding, errors="replace"): + try: + errlog.write(text) + except (OSError, ValueError): + if not write_failed: + write_failed = True + # A broken errlog is often sys.stderr, where logging reports its own failures. + with suppress(OSError, ValueError): + logger.exception( + "Writing the MCP server's stderr to errlog failed; later failures are not logged" + ) + except (anyio.ClosedResourceError, anyio.BrokenResourceError, OSError): + pass # shutdown closed the pipe under the read, or the pipe broke + finally: + # Logging shims installed as sys.stderr may only implement write(). + flush = getattr(errlog, "flush", None) + if flush is not None: + with suppress(OSError, ValueError): + flush() + done.set() + + def _parse_line(line: str) -> SessionMessage | Exception: """Parses one stdout line, returning parse errors as values for the session to surface.""" try: @@ -246,7 +318,7 @@ async def _drain_stdout(process: ServerProcess) -> None: await process.stdout.receive() -async def _stop_server_process(process: ServerProcess) -> None: +async def _stop_server_process(process: ServerProcess, stderr_relayed: anyio.Event | None = None) -> None: """Closes stdin, waits out the grace period, then kills the whole tree. The escalation order is spec text; timeouts and tree-wide scope are SDK policy: @@ -263,8 +335,17 @@ async def _stop_server_process(process: ServerProcess) -> None: # Reaps surviving Windows job members now, not at GC; no-op on POSIX. close_process_job(process) - # A kill survivor can hold the stdout pipe open; poison the reader anyway. + if stderr_relayed is not None: + # Wait before closing the pipe. Closing it discards unread output, such + # as the server's last words. + with anyio.move_on_after(_STDERR_DRAIN_TIMEOUT) as drain_scope: + await stderr_relayed.wait() + if drain_scope.cancelled_caught: + await anyio.lowlevel.cancel_shielded_checkpoint() # resync coverage on 3.11 (gh-106749) + # A kill survivor can hold a pipe open; poison the readers anyway. await _close_pipe(process.stdout) + if process.stderr is not None: + await _close_pipe(process.stderr) _close_subprocess_transport(process) @@ -330,7 +411,7 @@ async def _create_platform_compatible_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, + errlog: TextIO | int = sys.stderr, cwd: Path | str | None = None, ) -> ServerProcess: """Spawns the server in its own kill scope. diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py index 321fda8a66..625965b10d 100644 --- a/src/mcp/os/win32/utilities.py +++ b/src/mcp/os/win32/utilities.py @@ -92,9 +92,11 @@ def __init__(self, popen_obj: subprocess.Popen[bytes]) -> None: self.popen: subprocess.Popen[bytes] = popen_obj stdin = popen_obj.stdin stdout = popen_obj.stdout + stderr = popen_obj.stderr self.stdin = FileWriteStream(cast(BinaryIO, stdin)) if stdin else None self.stdout = FileReadStream(cast(BinaryIO, stdout)) if stdout else None + self.stderr = FileReadStream(cast(BinaryIO, stderr)) if stderr else None async def wait(self) -> int: """Waits for exit by polling the Popen. @@ -137,7 +139,7 @@ async def create_windows_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, + errlog: TextIO | int | None = sys.stderr, cwd: Path | str | None = None, ) -> Process | FallbackProcess: """Creates a subprocess with Job Object support for tree termination. @@ -177,7 +179,7 @@ async def _create_windows_fallback_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, + errlog: TextIO | int | None = sys.stderr, cwd: Path | str | None = None, ) -> FallbackProcess: """Spawns via subprocess.Popen and wraps it in FallbackProcess.""" diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index b64eb01902..242afac0d8 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -9,10 +9,12 @@ import errno import gc +import io import logging import math import os import signal +import subprocess import sys import threading from collections.abc import Callable @@ -132,6 +134,38 @@ async def aclose(self) -> None: await anyio.lowlevel.checkpoint() +class _FakeStderr: + """The fake process's stderr pipe, present only when the spawn asked for one. + + Closing it fails a pending read with `ClosedResourceError`, as anyio's pipe + wrapper does, instead of reporting EOF. + """ + + def __init__(self) -> None: + self._send, self._receive = anyio.create_memory_object_stream[bytes | anyio.ClosedResourceError](math.inf) + + async def feed(self, data: bytes) -> None: + await self._send.send(data) + + def feed_nowait(self, data: bytes) -> None: + self._send.send_nowait(data) + + def close_write_end(self) -> None: + self._send.close() + + async def receive(self) -> bytes: + item = await self._receive.receive() + if isinstance(item, anyio.ClosedResourceError): + raise item + return item + + async def aclose(self) -> None: + with suppress(anyio.ClosedResourceError): # the write end is already closed + self._send.send_nowait(anyio.ClosedResourceError()) + self._receive.close() + await anyio.lowlevel.checkpoint() + + class FakeProcess: """In-memory stand-in for the spawned server process. @@ -168,6 +202,7 @@ def __init__( self.stdin_send_gate = stdin_send_gate self.on_stdout_receive = on_stdout_receive self.stdin = _FakeStdin(self) + self.stderr: _FakeStderr | None = None def _dispatch_stdout_receive(self) -> None: # Late-bound so a test can assign `on_stdout_receive` after construction. @@ -182,10 +217,21 @@ def close_stdout(self) -> None: """End the fake process's stdout, as the kernel does when it dies.""" self._stdout_send.close() + async def feed_stderr(self, data: bytes) -> None: + """Make `data` readable on the fake process's stderr pipe.""" + assert self.stderr is not None, "the spawn did not ask for a stderr pipe" + await self.stderr.feed(data) + + def close_stderr(self) -> None: + """End the stderr pipe, as the kernel does once nothing holds its write end.""" + if self.stderr is not None: + self.stderr.close_write_end() + def exit(self, code: int = 0) -> None: - """Die: set the exit code and EOF stdout, as the kernel does.""" + """Die: set the exit code and EOF stdout and stderr, as the kernel does.""" self.returncode = code self.close_stdout() + self.close_stderr() def pending_stdout_chunks(self) -> int: """How many fed chunks the client has not yet pulled off the fake stdout.""" @@ -210,9 +256,11 @@ async def fake_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, + errlog: TextIO | int = sys.stderr, cwd: Path | str | None = None, ) -> FakeProcess: + if errlog == subprocess.PIPE: + process.stderr = _FakeStderr() return process async def fake_terminate_tree(proc: FakeProcess) -> None: @@ -1027,6 +1075,243 @@ async def stubborn_terminate(proc: FakeProcess) -> None: process.close_stdout() +# --------------------------------------------------------------------------- +# Server stderr relayed to an errlog that is not a file +# --------------------------------------------------------------------------- +# +# The real-process half (the relay reaching the stream, shutdown keeping the server's +# last words, files still inherited) is in tests/transports/stdio/test_lifecycle.py. + + +@pytest.mark.anyio +async def test_server_stderr_reaches_the_sys_stderr_that_capsys_installs( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Server stderr reaches pytest's capsys stream, a text wrapper with no file descriptor.""" + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=sys.stderr): + await process.feed_stderr(b"warming up\n") + + assert capsys.readouterr().err == "warming up\n" + + +class _StreamToLogger: + """Stands in for a logging shim installed as sys.stderr: only `write()`.""" + + def __init__(self) -> None: + self.written: list[str] = [] + + def write(self, text: str) -> int: + self.written.append(text) + return len(text) + + +@pytest.mark.anyio +async def test_server_stderr_reaches_a_logging_shim_that_only_has_write( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Server stderr reaches a stream with `write()` and neither `fileno()` nor `flush()`. + + Handing such a stream to the spawn used to fail with `AttributeError`. + """ + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + errlog = _StreamToLogger() + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=cast(TextIO, errlog)): + await process.feed_stderr(b"warming up\n") + + assert errlog.written == ["warming up\n"] + + +class _FlakyStream(io.StringIO): + """Rejects every write that contains `bad`, as a stream with a transient fault can.""" + + def write(self, s: str) -> int: + if "bad" in s: + raise ValueError("rejected") + return super().write(s) + + +@pytest.mark.anyio +async def test_a_write_errlog_rejects_loses_only_that_chunk_and_is_logged_once( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A rejected write loses only its own chunk; later output still arrives, and one failure is logged. + + A relay that stopped reading would leave a chatty server blocked on a full pipe. + """ + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + errlog = _FlakyStream() + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=errlog): + for chunk in (b"one\n", b"bad two\n", b"bad three\n", b"four\n"): + await process.feed_stderr(chunk) + + assert errlog.getvalue() == "one\nfour\n" + assert [record.getMessage() for record in caplog.records if record.name == stdio.__name__] == [ + "Writing the MCP server's stderr to errlog failed; later failures are not logged" + ] + + +@pytest.mark.anyio +async def test_shutdown_stops_waiting_for_a_stderr_pipe_that_a_survivor_holds_open( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Shutdown completes, bounded, when a descendant keeps the server's stderr open past its exit. + + What the server wrote before exiting still reaches `errlog`. The shortened drain + bound is the timeout under test. + """ + + def exit_leaving_stderr_to_a_survivor() -> None: + process.returncode = 0 + process.close_stdout() + + process = FakeProcess(on_stdin_close=exit_leaving_stderr_to_a_survivor) + install_fake_process(monkeypatch, process) + monkeypatch.setattr(stdio, "_STDERR_DRAIN_TIMEOUT", 0.05) + errlog = io.StringIO() + + try: + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=errlog): + await process.feed_stderr(b"before exit\n") + + assert errlog.getvalue() == "before exit\n" + finally: + # The survivor finally exits; without this the pipe's GC-time ResourceWarning + # would fail a later, unrelated test. + process.close_stderr() + + +@pytest.mark.anyio +async def test_cancelling_the_client_still_relays_the_stderr_a_server_writes_while_exiting( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cancelled caller keeps the stderr relay reading through shutdown, as the stdout drain does. + + SDK policy: otherwise the server's last words are lost, and a server writing more + than a pipe buffer of them blocks and is killed instead of exiting. + """ + + def say_bye_and_exit() -> None: + assert process.stderr is not None + process.stderr.feed_nowait(b"bye\n") + process.exit(0) + + process = FakeProcess(on_stdin_close=say_bye_and_exit) + terminated = install_fake_process(monkeypatch, process) + errlog = io.StringIO() + entered = anyio.Event() + # Cancelled from the test's task, for the coverage reason given in + # test_cancelling_the_client_still_runs_the_full_shutdown. + cancel_scope = anyio.CancelScope() + + async def run_client_until_cancelled() -> None: + with cancel_scope: + async with stdio_client(FAKE_PARAMS, errlog=errlog): + entered.set() + await anyio.sleep_forever() + + with anyio.fail_after(5): + async with anyio.create_task_group() as tg: + tg.start_soon(run_client_until_cancelled) + await entered.wait() + cancel_scope.cancel() + + assert errlog.getvalue() == "bye\n" + assert terminated == [] + + +@pytest.mark.anyio +async def test_undecodable_server_stderr_is_replaced_and_split_characters_are_rejoined( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Bytes that are not valid in the server encoding become U+FFFD instead of failing the relay. + + A character split across two pipe reads still arrives whole. + """ + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + errlog = io.StringIO() + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=errlog): + for chunk in (b"caf\xe9\n", b"na\xc3", b"\xafve\n"): + await process.feed_stderr(chunk) + + assert errlog.getvalue() == "caf\ufffd\nna\u00efve\n" + + +@pytest.mark.anyio +@pytest.mark.parametrize("errlog", [None, subprocess.DEVNULL], ids=["none-as-under-pythonw", "devnull"]) +async def test_an_errlog_the_spawn_takes_as_is_opens_no_stderr_pipe( + monkeypatch: pytest.MonkeyPatch, + errlog: int | None, +) -> None: + """`None` and `subprocess.DEVNULL` still go straight to the spawn. + + `sys.stderr` is `None` under pythonw, which makes it the default `errlog` there; + `DEVNULL` is what callers pass when they want the server's stderr discarded. + """ + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=cast(TextIO, errlog)): + pass + + assert process.stderr is None + + +@pytest.mark.anyio +async def test_a_binary_errlog_is_still_rejected_at_spawn() -> None: + """A binary in-memory stream cannot take text, so the spawn rejects it before anything starts.""" + params = StdioServerParameters(command=sys.executable, args=["-c", "pass"]) + + with pytest.raises(io.UnsupportedOperation): + async with stdio_client(params, errlog=cast(TextIO, io.BytesIO())): + raise NotImplementedError + + +class _BrokenStream(io.StringIO): + """A stream whose every write fails, as a closed one's does.""" + + def write(self, s: str) -> int: + raise ValueError("I/O operation on closed file") + + +@pytest.mark.anyio +async def test_an_errlog_that_is_a_broken_sys_stderr_does_not_end_the_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A closed `errlog` does not end the session, even when it is `sys.stderr`. + + Reporting the failed write goes through logging, whose last-resort handler + writes to `sys.stderr` and lets that `ValueError` escape. + """ + broken = _BrokenStream() + monkeypatch.setattr(sys, "stderr", broken) + monkeypatch.setattr(stdio.logger, "propagate", False) + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=broken): + await process.feed_stderr(b"lost\n") + + assert process.returncode == 0 + + # --------------------------------------------------------------------------- # POSIX tree-termination policy, tested through the sanctioned killpg seam # --------------------------------------------------------------------------- @@ -1251,7 +1536,7 @@ async def recording_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, + errlog: TextIO | int = sys.stderr, cwd: Path | str | None = None, ) -> anyio.abc.Process | FallbackProcess: process = await _create_platform_compatible_process(command, args, env, errlog, cwd) @@ -1415,9 +1700,11 @@ async def test_escalation_kills_a_process_that_ignores_sigterm( # pragma: lax n @pytest.mark.anyio @pytest.mark.skipif(not Path("/proc/self/fd").is_dir(), reason="needs procfs to enumerate open file descriptors") +@pytest.mark.parametrize("make_errlog", [lambda: sys.stderr, io.StringIO], ids=["inherited-stderr", "relayed-stderr"]) # lax no cover: Windows CI jobs enforce 100% coverage per job, have no procfs, and skip this. async def test_a_graceful_exit_with_a_surviving_child_leaks_no_pipe_fds( # pragma: lax no cover monkeypatch: pytest.MonkeyPatch, + make_errlog: Callable[[], TextIO], ) -> None: """A graceful exit with a surviving child must not leak the client's pipe fds. @@ -1425,8 +1712,11 @@ async def test_a_graceful_exit_with_a_surviving_child_leaks_no_pipe_fds( # prag inherited pipe ends (the POSIX policy: survivors are the server's business). The client must still release its own pipe fds and subprocess transport at shutdown (on asyncio nothing else ever closes them while the orphan holds the pipe) instead - of leaking them for the orphan's lifetime. + of leaking them for the orphan's lifetime. That includes the stderr pipe read for + an `errlog` that is not a file, which the orphan keeps open past the drain window. """ + # The relayed case waits out this bound: the orphan never closes stderr. + monkeypatch.setattr(stdio, "_STDERR_DRAIN_TIMEOUT", 0.05) spawned = _record_spawned_processes(monkeypatch) async with AsyncExitStack() as stack: @@ -1445,7 +1735,7 @@ async def test_a_graceful_exit_with_a_surviving_child_leaks_no_pipe_fds( # prag # Two interpreter cold starts on a loaded runner; healthy runs take ~0.3s. with anyio.fail_after(15.0): - async with stdio_client(server_params): + async with stdio_client(server_params, errlog=make_errlog()): stream = await _accept_alive(sock) await stream.aclose() diff --git a/tests/transports/stdio/conftest.py b/tests/transports/stdio/conftest.py index e9601ac6d5..43562d43a0 100644 --- a/tests/transports/stdio/conftest.py +++ b/tests/transports/stdio/conftest.py @@ -37,7 +37,7 @@ async def recording_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, + errlog: TextIO | int = sys.stderr, cwd: Path | str | None = None, ) -> anyio.abc.Process | FallbackProcess: process = await _create_platform_compatible_process(command, args, env, errlog, cwd) diff --git a/tests/transports/stdio/test_lifecycle.py b/tests/transports/stdio/test_lifecycle.py index 157d93df5f..f7ea72e542 100644 --- a/tests/transports/stdio/test_lifecycle.py +++ b/tests/transports/stdio/test_lifecycle.py @@ -1,5 +1,6 @@ """Cross-platform stdio lifecycle tests using real subprocesses.""" +import io import os import subprocess import sys @@ -7,6 +8,7 @@ from contextlib import AsyncExitStack from pathlib import Path from textwrap import dedent +from typing import TextIO import anyio import anyio.abc @@ -168,6 +170,143 @@ async def test_server_stderr_output_reaches_the_errlog_file( assert spawned_processes[0].returncode == 0 +@pytest.mark.anyio +async def test_a_file_errlog_is_handed_to_the_server_as_its_own_stderr( + tmp_path: Path, + spawned_processes: list[anyio.abc.Process | FallbackProcess], +) -> None: + """A file `errlog` becomes the server's stderr itself, not a pipe the client copies from. + + The server reports the identity of its fd 2, which must be the file's. + """ + async with AsyncExitStack() as stack: + sock, port = await open_liveness_listener() + stack.push_async_callback(sock.aclose) + + server = ( + f"import os, socket, sys\n" + f"st = os.fstat(2)\n" + f"sys.stderr.write(f'{{st.st_dev}} {{st.st_ino}}\\n')\n" + f"sys.stderr.flush()\n" + f"s = socket.create_connection(('127.0.0.1', {port}))\n" + f"s.sendall(b'alive')\n" + f"sys.stdin.read()\n" + ) + params = StdioServerParameters(command=sys.executable, args=["-c", server]) + + with (tmp_path / "errlog.txt").open("w+", encoding="utf-8") as errlog: + # Allow one cold interpreter start on loaded CI. + with anyio.fail_after(10.0): + async with stdio_client(params, errlog=errlog): + stream = await accept_alive(sock) + stack.push_async_callback(stream.aclose) + + errlog_stat = os.fstat(errlog.fileno()) + errlog.seek(0) + content = errlog.read() + + assert content == f"{errlog_stat.st_dev} {errlog_stat.st_ino}\n" + assert spawned_processes[0].returncode == 0 + + +@pytest.mark.anyio +async def test_server_stderr_written_while_exiting_reaches_an_in_memory_errlog( + spawned_processes: list[anyio.abc.Process | FallbackProcess], +) -> None: + """Server stderr reaches an `errlog` with no file descriptor, including what it writes while exiting. + + The last line is written after stdin closes, so it is still in the pipe when + shutdown sees the process gone. SDK-defined: a fileless stream used to raise + `UnsupportedOperation` from the spawn instead (#156). + """ + async with AsyncExitStack() as stack: + sock, port = await open_liveness_listener() + stack.push_async_callback(sock.aclose) + + server = ( + f"import socket, sys\n" + f"sys.stderr.write('starting up\\n')\n" + f"sys.stderr.flush()\n" + f"s = socket.create_connection(('127.0.0.1', {port}))\n" + f"s.sendall(b'alive')\n" + f"sys.stdin.read()\n" + f"sys.stderr.write('shutting down\\n')\n" + f"sys.stderr.flush()\n" + ) + params = StdioServerParameters(command=sys.executable, args=["-c", server]) + errlog = io.StringIO() + + # Allow one cold interpreter start on loaded CI. + with anyio.fail_after(10.0): + async with stdio_client(params, errlog=errlog): + stream = await accept_alive(sock) + stack.push_async_callback(stream.aclose) + + lines = errlog.getvalue().splitlines() + + assert lines == ["starting up", "shutting down"] + assert spawned_processes[0].returncode == 0 + + +class _NotebookStderr(io.StringIO): + """Stands in for Jupyter's `sys.stderr`. + + The notebook shows what goes through `write()`, while `fileno()` leads past it + to the terminal the notebook server runs in. + """ + + def __init__(self, terminal: TextIO) -> None: + super().__init__() + self._terminal = terminal + + def fileno(self) -> int: + return self._terminal.fileno() + + +@pytest.mark.anyio +async def test_server_stderr_reaches_a_notebook_stream_rather_than_the_descriptor_it_reports( + tmp_path: Path, + spawned_processes: list[anyio.abc.Process | FallbackProcess], +) -> None: + """A text stream's own `fileno()` is not handed to the server; its output goes through `write()`. + + Jupyter's `sys.stderr` reports the terminal behind the notebook, so inheriting that + descriptor hid server stderr from the notebook (#156). + """ + marker = "stdio-lifecycle notebook marker 5151" + + async with AsyncExitStack() as stack: + sock, port = await open_liveness_listener() + stack.push_async_callback(sock.aclose) + + server = ( + f"import socket, sys\n" + f"sys.stderr.write({marker!r} + '\\n')\n" + f"sys.stderr.flush()\n" + f"s = socket.create_connection(('127.0.0.1', {port}))\n" + f"s.sendall(b'alive')\n" + f"sys.stdin.read()\n" + ) + params = StdioServerParameters(command=sys.executable, args=["-c", server]) + + with (tmp_path / "terminal.txt").open("w+", encoding="utf-8") as terminal: + notebook = _NotebookStderr(terminal) + # Like Jupyter's, the descriptor works; it just isn't where the notebook reads. + assert notebook.fileno() == terminal.fileno() + # Allow one cold interpreter start on loaded CI. + with anyio.fail_after(10.0): + async with stdio_client(params, errlog=notebook): + stream = await accept_alive(sock) + stack.push_async_callback(stream.aclose) + + terminal.seek(0) + terminal_output = terminal.read() + + assert notebook.getvalue().splitlines() == [marker] + assert terminal_output == "" + assert spawned_processes[0].returncode == 0 + + @pytest.mark.skipif( not hasattr(os, "waitid"), reason="needs os.waitid(WNOWAIT); absent on Windows and macOS before 3.13" )