Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ assigned to their author are closed automatically.

## Code Quality

- Keep comments brief. Explain only non-obvious reasons or constraints; do not
narrate the code or restate names, types, or assertions.
- Type hints required for all code
- Public APIs must have docstrings. When a public API raises exceptions a
caller would reasonably catch, document them in a `Raises:` section. Don't
Expand Down
27 changes: 4 additions & 23 deletions tests/interaction/transports/_stdio_server.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,4 @@
"""A real low-level Server over the stdio transport, for the suite's one subprocess test.

Runnable as `python -m tests.interaction.transports._stdio_server` from the repo root; the test
launches it that way via `stdio_client`. Kept separate from the test module so the server lives in
its own importable file (subprocess coverage applies) while the test file follows the suite's
test-only-functions convention.
"""
"""Low-level stdio server for the interaction suite's subprocess test."""

import sys
import warnings
Expand Down Expand Up @@ -63,26 +57,13 @@ async def set_logging_level(ctx: ServerRequestContext, params: SetLevelRequestPa
async def main() -> None:
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
# Flush this process's coverage data before the clean-exit line below. Without this, the
# data is only written by coverage's atexit hook during interpreter teardown -- and on a
# slow Windows runner that can overrun the transport's termination grace, so the kill
# silently destroys the data file and the 100% gate trips on this module's subprocess-only
# lines. Saving here puts the write before the line the test synchronizes on: once the
# parent has seen "clean exit", the data is durably on disk and the escalation is harmless.
# Nothing measured may execute after the save (it would be unrecordable by construction),
# hence the excluded lines below. The branch is pragma'd because under coverage the
# instance always exists, and without coverage nothing is measured anyway.
# Save subprocess coverage before the marker so forced teardown cannot lose it.
cov = getattr(coverage.process_startup, "coverage", None)
if cov is not None: # pragma: no branch
# stop() is load-bearing twice over: it ends tracing, making itself the last
# recordable line, and it leaves nothing new for coverage's atexit re-save to flush --
# so a kill landing during interpreter teardown cannot corrupt the file save() wrote
# (coverage opens it with sqlite journaling off; a torn rewrite would not roll back).
# Leave nothing for coverage's atexit hook to rewrite if teardown is interrupted.
cov.stop()
cov.save() # pragma: lax no cover - untraced: stop() above already ended measurement
# Reached only when the run loop exits because stdin closed; if the process were terminated
# the test's stderr capture would not see this line. lax no cover: runs after the coverage
# save by design, so it can never appear covered.
# The test uses this marker to distinguish clean exit from termination.
print("stdio-echo: clean exit", file=sys.stderr, flush=True) # pragma: lax no cover


Expand Down
48 changes: 8 additions & 40 deletions tests/interaction/transports/test_stdio.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,6 @@
"""The stdio transport: one subprocess end-to-end test and one in-process framing test.
"""Stdio subprocess round-trip and in-process framing tests.

The subprocess test proves the client-server round trip over the transport's real process
boundary; its server lives in `_stdio_server.py` and is launched via `python -m` so subprocess
coverage measurement applies. The framing test drives `stdio_server` over injected in-process
streams instead.

stdio is deliberately not a leg of the `connect`-fixture matrix: a subprocess per test would be
slow, and the matrix already proves transport-agnosticism in-process. Process-lifecycle edge
cases (terminate/kill escalation, parse errors) stay in `tests/client/test_stdio.py`.
Lifecycle edge cases remain in `tests/client/test_stdio.py`.
"""

import io
Expand Down Expand Up @@ -50,20 +43,8 @@
async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A stdio-subprocess Client round-trips a tool call, a notification, and a clean exit.

The Client initializes, calls a tool with arguments, and receives the server's log
notification before the call returns; the server exits when the transport closes its
stdin.
"""
# After stdin closes, the child must unwind, flush its subprocess coverage data, and write
# the clean-exit line before escalation (the server saves coverage *before* printing, so a
# post-print kill can no longer silently lose the data file -- see _stdio_server.main). The
# production 2s default is too tight for the unwind+save tail on loaded Windows runners
# (measured in-situ p99 of the whole test is ~7s); a kill before the print fails the stderr
# assertion below loudly rather than tripping the coverage gate. The 20s grace covers even a
# badly starved runner (a >10s stall has been seen once in CI) and costs nothing when the
# child exits promptly. Not under test.
"""A stdio client round-trips a tool call and notification before clean exit."""
# Allow slow Windows runners to flush subprocess coverage before escalation.
monkeypatch.setattr(stdio, "PROCESS_TERMINATION_TIMEOUT", 20.0)

received: list[LoggingMessageNotificationParams] = []
Expand All @@ -77,10 +58,7 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
command=sys.executable,
args=["-m", _stdio_server.__name__],
cwd=str(_REPO_ROOT),
# stdio_client filters the inherited environment, dropping the variables
# coverage.py's subprocess support uses; pass them through so the server module is
# measured. PYTHONWARNINGS: the child recompiles anyio (pytest's pyc tag differs),
# and on 3.14 anyio's return-in-finally SyntaxWarning would land on the snapshot stderr.
# Preserve subprocess coverage and suppress anyio's `SyntaxWarning` on Python 3.14.
env={key: value for key, value in os.environ.items() if key.startswith("COVERAGE_")}
| {"PYTHONWARNINGS": "ignore::SyntaxWarning"},
),
Expand All @@ -98,26 +76,18 @@ async def collect(params: LoggingMessageNotificationParams) -> None:
captured_stderr = errlog.read()

assert result == snapshot(CallToolResult(content=[TextContent(text="across\nprocesses")]))
# stdio carries one ordered server-to-client stream, so the same notification-before-response
# guarantee holds here as for the in-memory transport.
# Stdio preserves notification-before-response ordering.
assert received == snapshot(
[LoggingMessageNotificationParams(level="info", logger="echo", data="echoing across\nprocesses")]
)
# The server writes this line only after its run loop returns on stdin close: seeing it proves
# a self-exit, not the terminate escalation. The capture itself proves stderr passthrough.
# The marker distinguishes clean exit from termination.
assert captured_stderr == snapshot("stdio-echo: clean exit\n")


@requirement("transport:stdio:stream-purity")
@requirement("transport:stdio:no-embedded-newlines")
async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None:
"""Every `stdio_server` write is one valid JSON-RPC message on its own line.

Each line is newline-terminated with payload newlines JSON-escaped. This proves the
transport's own framing over injected streams; the descriptor-level guard that keeps
handler code off the wire is pinned by tests/server/test_stdio.py (see the narrowed
divergence on `transport:stdio:stream-purity`).
"""
"""Each `stdio_server` write is one newline-terminated JSON-RPC message."""
captured = io.StringIO()
sent_line = json.dumps(initialize_body(request_id=1)) + "\n"

Expand Down Expand Up @@ -148,7 +118,5 @@ async def test_stdio_server_writes_one_jsonrpc_message_per_line() -> None:
assert len(lines) == 2
messages = [jsonrpc_message_adapter.validate_json(line) for line in lines]
assert [type(message).__name__ for message in messages] == snapshot(["JSONRPCResponse", "JSONRPCNotification"])
# The newline inside the payload is JSON-escaped on the wire, not a literal newline that would
# break the one-message-per-line framing.
assert r"line\nbreak" in lines[0]
assert r"two\nlines" in lines[1]
Loading
Loading