From afc01d699521627509f753f822c30234c2e9da03 Mon Sep 17 00:00:00 2001 From: Hongming Wang Date: Fri, 1 May 2026 18:26:24 -0700 Subject: [PATCH 1/2] fix(mcp): friendly fail-fast when stdio isn't pipe-compatible MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When molecule-mcp is launched with stdin or stdout redirected to a regular file (molecule-mcp > out.txt, ad-hoc CI smoke-tests, local debugging), asyncio.connect_read_pipe / connect_write_pipe later raise ValueError: Pipe transport is only for pipes, sockets and character devices — surfaced to the operator as a confusing traceback with no hint about what to do. Add _assert_stdio_is_pipe_compatible() to detect the same constraint synchronously before the event loop starts, exit cleanly with code 2, and print a stderr message that names: - which stream failed (stdin vs stdout) - the asyncio transport requirement - the two common causes (>file, &1 | tee out.txt) Wired into cli_main() (the synchronous wrapper around asyncio.run(main())) so wheel-smoke + the production launch path both go through the guard without changing the async stdio loop body. Closed/stale-fd case also handled — os.fstat OSError exits 2 with the same guidance instead of escaping. Tests: 4 new in TestStdioPipeAssertion — pipe-pair happy path, regular-file stdout (the bug condition), regular-file stdin (symmetric case), and closed-fd. Mutation-verified — all 4 fail without the prod helper. 37/37 in test_a2a_mcp_server.py. Closes Molecule-AI/molecule-ai-workspace-runtime#61. Co-Authored-By: Claude Opus 4.7 (1M context) --- workspace/a2a_mcp_server.py | 48 +++++++++++++ workspace/tests/test_a2a_mcp_server.py | 99 ++++++++++++++++++++++++++ 2 files changed, 147 insertions(+) diff --git a/workspace/a2a_mcp_server.py b/workspace/a2a_mcp_server.py index fc7e4862..3971c472 100644 --- a/workspace/a2a_mcp_server.py +++ b/workspace/a2a_mcp_server.py @@ -16,6 +16,7 @@ import asyncio import json import logging import os +import stat import sys from typing import Callable @@ -393,6 +394,52 @@ def _build_channel_notification(msg: dict) -> dict: # --- MCP Server (JSON-RPC over stdio) --- + +def _assert_stdio_is_pipe_compatible( + stdin_fd: int = 0, stdout_fd: int = 1 +) -> None: + """Fail fast with a friendly message when stdio isn't pipe-compatible. + + asyncio.connect_read_pipe / connect_write_pipe accept only pipes, + sockets, and character devices. When molecule-mcp is launched with + stdout redirected to a regular file (CI smoke tests, ad-hoc local + debugging that captures output), the asyncio call later raises + ``ValueError: Pipe transport is only for pipes, sockets and character + devices`` from inside the event loop — surfaced to the operator as a + confusing traceback. Detect early and exit cleanly with guidance + instead. See molecule-ai-workspace-runtime#61. + """ + for name, fd in (("stdin", stdin_fd), ("stdout", stdout_fd)): + try: + mode = os.fstat(fd).st_mode + except OSError as exc: + print( + f"molecule-mcp: cannot stat {name} (fd={fd}): {exc}.\n" + f" This MCP server expects bidirectional pipe stdio. Launch it from\n" + f" an MCP-aware client (Claude Code, Cursor, etc.) — not detached\n" + f" from a terminal or with stdio closed.", + file=sys.stderr, + ) + sys.exit(2) + if not ( + stat.S_ISFIFO(mode) or stat.S_ISSOCK(mode) or stat.S_ISCHR(mode) + ): + print( + f"molecule-mcp: {name} (fd={fd}) is a regular file, not a pipe,\n" + f" socket, or character device — asyncio's stdio transport rejects\n" + f" it with `ValueError: Pipe transport is only for pipes, sockets\n" + f" and character devices`. Common causes:\n" + f" molecule-mcp > out.txt # stdout → regular file (fails)\n" + f" molecule-mcp < input.json # stdin → regular file (fails)\n" + f" Launch molecule-mcp from an MCP-aware client (Claude Code, Cursor,\n" + f" hermes, OpenCode, etc.) so stdio is wired to a pipe pair, or use\n" + f" `tee`/process substitution if you need to capture output:\n" + f" molecule-mcp 2>&1 | tee out.txt # stdout stays a pipe", + file=sys.stderr, + ) + sys.exit(2) + + async def main(): # pragma: no cover """Run MCP server on stdio — reads JSON-RPC requests, writes responses.""" reader = asyncio.StreamReader() @@ -496,6 +543,7 @@ def cli_main() -> None: # pragma: no cover break every external-runtime operator's MCP install — the 0.1.16 ``main_sync`` rename incident is the cautionary precedent. """ + _assert_stdio_is_pipe_compatible() asyncio.run(main()) diff --git a/workspace/tests/test_a2a_mcp_server.py b/workspace/tests/test_a2a_mcp_server.py index 18d038c2..175558fb 100644 --- a/workspace/tests/test_a2a_mcp_server.py +++ b/workspace/tests/test_a2a_mcp_server.py @@ -2,6 +2,7 @@ import asyncio import json +import os from unittest.mock import AsyncMock, patch @@ -718,6 +719,104 @@ def test_inbox_bridge_swallows_closed_loop_runtime_error(): }) +class TestStdioPipeAssertion: + """Pin _assert_stdio_is_pipe_compatible — the friendly fail-fast guard + that turns asyncio's `ValueError: Pipe transport is only for pipes, + sockets and character devices` into a clear operator message + exit 2. + See molecule-ai-workspace-runtime#61. + """ + + def test_pipe_pair_passes_silently(self): + """Happy path — both fds are pipes (the production launch shape + from any MCP client). Should return None without printing or + exiting.""" + import a2a_mcp_server + + r, w = os.pipe() + try: + # No exit, no stderr noise. We don't capture stderr here + # because pipe path should produce zero output. + a2a_mcp_server._assert_stdio_is_pipe_compatible(stdin_fd=r, stdout_fd=w) + finally: + os.close(r) + os.close(w) + + def test_regular_file_stdout_exits_with_friendly_message( + self, tmp_path, capsys + ): + """Reproducer for runtime#61: stdout redirected to a regular file. + Pre-fix this would surface upstream as + `ValueError: Pipe transport is only for pipes...`. Post-fix we + exit with code 2 and a stderr message that names the symptom + + fix.""" + import a2a_mcp_server + + # stdin = pipe (so we isolate the stdout failure path); + # stdout = regular file (the bug condition). + r, _w = os.pipe() + regular = tmp_path / "captured.log" + f = open(regular, "wb") + try: + with pytest.raises(SystemExit) as excinfo: + a2a_mcp_server._assert_stdio_is_pipe_compatible( + stdin_fd=r, stdout_fd=f.fileno() + ) + assert excinfo.value.code == 2 + err = capsys.readouterr().err + # Names the failing stream + the asyncio constraint that + # would otherwise crash. Don't pin the exact wording — the + # asserts pin the operator-recoverable signal only. + assert "stdout" in err + assert "regular file" in err + assert "pipe" in err + finally: + f.close() + os.close(r) + + def test_regular_file_stdin_exits_with_friendly_message( + self, tmp_path, capsys + ): + """Symmetric case — stdin redirected from a regular file. Same + asyncio constraint applies via connect_read_pipe.""" + import a2a_mcp_server + + regular = tmp_path / "input.json" + regular.write_bytes(b'{"jsonrpc":"2.0","id":1,"method":"initialize"}\n') + f = open(regular, "rb") + _r, w = os.pipe() + try: + with pytest.raises(SystemExit) as excinfo: + a2a_mcp_server._assert_stdio_is_pipe_compatible( + stdin_fd=f.fileno(), stdout_fd=w + ) + assert excinfo.value.code == 2 + err = capsys.readouterr().err + assert "stdin" in err + assert "regular file" in err + finally: + f.close() + os.close(w) + + def test_closed_fd_exits_with_stat_error(self, capsys): + """If stdio is closed (rare but seen in detached daemonized + contexts), os.fstat raises OSError. We catch it and exit 2 with + a guidance message instead of letting the traceback escape.""" + import a2a_mcp_server + + r, w = os.pipe() + os.close(w) # Now `w` is a stale fd — fstat will fail. + try: + with pytest.raises(SystemExit) as excinfo: + a2a_mcp_server._assert_stdio_is_pipe_compatible( + stdin_fd=r, stdout_fd=w + ) + assert excinfo.value.code == 2 + err = capsys.readouterr().err + assert "cannot stat stdout" in err + finally: + os.close(r) + + def _readable(fd: int) -> bool: """True iff ``fd`` has bytes available without blocking. Lets us poll the pipe in a loop without the test hanging when the From f6a48d593e483cb684d19cbd9fcd8d080ec12e6f Mon Sep 17 00:00:00 2001 From: Hongming Wang Date: Fri, 1 May 2026 19:17:55 -0700 Subject: [PATCH 2/2] test: standardise on `from a2a_mcp_server import ...` in TestStdioPipeAssertion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit github-code-quality bot flagged 4 instances of `import a2a_mcp_server` in the new TestStdioPipeAssertion class — every other test in the file uses the `from a2a_mcp_server import ...` per-test pattern, so this is a real inconsistency. Switching the new tests to match. No behavior change; resolves the 4 unresolved review threads blocking the merge queue. Co-Authored-By: Claude Opus 4.7 (1M context) --- workspace/tests/test_a2a_mcp_server.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/workspace/tests/test_a2a_mcp_server.py b/workspace/tests/test_a2a_mcp_server.py index 175558fb..ac549d07 100644 --- a/workspace/tests/test_a2a_mcp_server.py +++ b/workspace/tests/test_a2a_mcp_server.py @@ -730,13 +730,13 @@ class TestStdioPipeAssertion: """Happy path — both fds are pipes (the production launch shape from any MCP client). Should return None without printing or exiting.""" - import a2a_mcp_server + from a2a_mcp_server import _assert_stdio_is_pipe_compatible r, w = os.pipe() try: # No exit, no stderr noise. We don't capture stderr here # because pipe path should produce zero output. - a2a_mcp_server._assert_stdio_is_pipe_compatible(stdin_fd=r, stdout_fd=w) + _assert_stdio_is_pipe_compatible(stdin_fd=r, stdout_fd=w) finally: os.close(r) os.close(w) @@ -749,7 +749,7 @@ class TestStdioPipeAssertion: `ValueError: Pipe transport is only for pipes...`. Post-fix we exit with code 2 and a stderr message that names the symptom + fix.""" - import a2a_mcp_server + from a2a_mcp_server import _assert_stdio_is_pipe_compatible # stdin = pipe (so we isolate the stdout failure path); # stdout = regular file (the bug condition). @@ -758,7 +758,7 @@ class TestStdioPipeAssertion: f = open(regular, "wb") try: with pytest.raises(SystemExit) as excinfo: - a2a_mcp_server._assert_stdio_is_pipe_compatible( + _assert_stdio_is_pipe_compatible( stdin_fd=r, stdout_fd=f.fileno() ) assert excinfo.value.code == 2 @@ -778,7 +778,7 @@ class TestStdioPipeAssertion: ): """Symmetric case — stdin redirected from a regular file. Same asyncio constraint applies via connect_read_pipe.""" - import a2a_mcp_server + from a2a_mcp_server import _assert_stdio_is_pipe_compatible regular = tmp_path / "input.json" regular.write_bytes(b'{"jsonrpc":"2.0","id":1,"method":"initialize"}\n') @@ -786,7 +786,7 @@ class TestStdioPipeAssertion: _r, w = os.pipe() try: with pytest.raises(SystemExit) as excinfo: - a2a_mcp_server._assert_stdio_is_pipe_compatible( + _assert_stdio_is_pipe_compatible( stdin_fd=f.fileno(), stdout_fd=w ) assert excinfo.value.code == 2 @@ -801,13 +801,13 @@ class TestStdioPipeAssertion: """If stdio is closed (rare but seen in detached daemonized contexts), os.fstat raises OSError. We catch it and exit 2 with a guidance message instead of letting the traceback escape.""" - import a2a_mcp_server + from a2a_mcp_server import _assert_stdio_is_pipe_compatible r, w = os.pipe() os.close(w) # Now `w` is a stale fd — fstat will fail. try: with pytest.raises(SystemExit) as excinfo: - a2a_mcp_server._assert_stdio_is_pipe_compatible( + _assert_stdio_is_pipe_compatible( stdin_fd=r, stdout_fd=w ) assert excinfo.value.code == 2