test(windows): guard clean MCP stdio startup

Adds a native Windows guard for #654: a successful MCP initialize +
tools/list exchange, run against an isolated cache and runtime directory,
must leave nothing on stderr but the server's own structured log lines.

- tests/windows/test_mcp_stdio.py: the guard. The English and Spanish
  path-not-found messages reported in #654 are named cases; any other
  non-empty stderr line that is not a structured "level=<name> msg=<event>"
  record fails as well, so the same leak on a machine in another language
  is caught without listing its translation. The log format is pinned to
  text so an inherited CBM_LOG_FORMAT=json cannot change the verdict. Cache
  and runtime live under one temp root, so both are removed with it, also
  on failure, after the daemon is stopped.
- tests/windows/mcp_stdio.py: close() joins the stderr reader, and the new
  stderr_complete() reports whether the reader reached EOF. The guard reads
  stderr only after it has; a reader still running after the server exited
  is a SETUP failure (exit 2), never a verdict on a partial capture.
- scripts/test-windows.ps1: registers the guard in the green guard list
  and documents it in the header.

#2213 was an earlier duplicate of the same check.

Refs #654.

Distilled from #2217.

Co-authored-by: lorenzozanee <wyz0707@proton.me>
Co-authored-by: Shinken009 <182390694+Shinken009@users.noreply.github.com>
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
This commit is contained in:
Martin Vogel
2026-11-11 00:57:46 +01:00
co-authored by lorenzozanee Shinken009
parent 6d34ca15bd
commit c05de88247
3 changed files with 147 additions and 0 deletions
+2
View File
@@ -18,6 +18,7 @@
* test_hook_augment.py guards #618 (fixed by #619)
* test_ui_drive_listing.py guards #548 (roots field)
* test_cli_non_ascii_arg.py guards #423/#20 (wide-argv main())
* test_mcp_stdio.py guards #654 (clean stdio stderr)
* test_daemon_stability.py guards the daemon parameter
surface, crash recovery, busy-stop refusal, and churn
* test_windows_update_handoff.py guards that `update`
@@ -166,6 +167,7 @@ $guards = @(
"tests\windows\test_hook_augment.py",
"tests\windows\test_ui_drive_listing.py",
"tests\windows\test_cli_non_ascii_arg.py",
"tests\windows\test_mcp_stdio.py",
"tests\windows\test_windows_update_handoff.py"
)
+18
View File
@@ -25,6 +25,8 @@ class McpServer:
self._id = 0
self.proc = None
self._stderr = []
self._stderr_thread = None
self._stderr_eof = False
env = dict(os.environ)
if cache_dir:
env["CBM_CACHE_DIR"] = cache_dir # isolate the graph DB location
@@ -52,12 +54,27 @@ class McpServer:
try:
for line in self.proc.stderr:
self._stderr.append(line.decode("utf-8", "replace"))
self._stderr_eof = True
except Exception:
pass
def stderr_text(self):
return "".join(self._stderr)
def stderr_complete(self, timeout):
"""Wait for the stderr reader to reach EOF; True once it has.
Only then is stderr_text() the whole capture. False means the reader
is still running (some process still holds the pipe) or stopped on a
read error, so the text may be partial: a caller that asserts on
stderr must treat False as a setup failure, never as a verdict. The
timeout is a liveness backstop, not part of any verdict.
"""
if self._stderr_thread is None:
return False
self._stderr_thread.join(timeout)
return not self._stderr_thread.is_alive() and self._stderr_eof
def _send(self, obj):
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
self.proc.stdin.write(data + b"\n")
@@ -148,6 +165,7 @@ class McpServer:
self.proc.kill()
except Exception:
pass
self._stderr_thread.join(timeout=10)
def wait_projects_with_stats(server, timeout=90.0, poll=1.0):
+127
View File
@@ -0,0 +1,127 @@
"""Regression guard for clean Windows MCP stdio startup (#654).
A successful initialize + tools/list exchange must leave nothing on stderr but
the server's own structured log lines. #654 reported the OS "path not found"
message leaking to stderr on startup, in English and on a Spanish Windows. Those
two messages are named cases; any other non-empty line that is not a structured
`level=<name> msg=<event>` record fails as well, so the same leak on a machine
in any other language is caught without listing its translation.
The verdict is read only after the stderr reader has reached EOF. A reader that
is still running after the server exited means the capture may be partial, and
that is a setup failure, never a pass.
Exit codes follow the guard contract of scripts/test-windows.ps1:
0 = GREEN, 1 = RED (regression), 2 = SETUP / precondition.
"""
import os
import re
import subprocess
import sys
import tempfile
from mcp_stdio import McpServer
# The OS path-not-found message as reported in #654 (English and Spanish
# Windows). Named so a RED says exactly which known leak came back.
KNOWN_PATH_ERRORS = (
"The system cannot find the path specified.",
"El sistema no puede encontrar la ruta especificada.",
)
# The text-format record cbm_log writes (src/foundation/log.c):
# "level=<debug|info|warn|error> msg=<event> key=value ...".
STRUCTURED_LOG_LINE = re.compile(r"level=[a-z]+ msg=\S")
# Liveness backstop for the stderr reader. The server has already exited when
# this is waited on, so the reader is at EOF unless another process still holds
# the pipe. Expiry is a SETUP failure; it never decides PASS or FAIL.
STDERR_EOF_BACKSTOP_S = 60
def classify_stderr(text):
"""Return (known, raw): the named #654 messages found, and every other
non-empty stderr line that is not a structured log record."""
known = [message for message in KNOWN_PATH_ERRORS if message in text]
raw = []
for line in text.splitlines():
line = line.strip()
if not line or STRUCTURED_LOG_LINE.match(line):
continue
if any(message in line for message in KNOWN_PATH_ERRORS):
continue
raw.append(line)
return known, raw
def main():
if os.name != "nt":
print("SKIP: Windows-only MCP stdio guard")
return 2
binary = sys.argv[1] if len(sys.argv) > 1 else os.environ.get("CBM_TEST_BINARY")
if not binary:
print("SETUP FAIL: CBM_TEST_BINARY is required", file=sys.stderr)
return 2
with tempfile.TemporaryDirectory(prefix="cbm-mcp-stdio-") as work:
cache = os.path.join(work, "cache")
runtime = os.path.join(work, "runtime")
os.makedirs(cache)
os.makedirs(runtime)
# The verdict reads the text log shape, so select it explicitly: an
# inherited CBM_LOG_FORMAT=json must not turn every record into a raw
# line.
server = McpServer(
binary,
cache_dir=cache,
extra_env={"CBM_RUNTIME_DIR": runtime, "CBM_LOG_FORMAT": "text"},
)
try:
with server:
server.initialize(timeout=30)
server.tools_list(timeout=30)
finally:
stop = subprocess.run(
[binary, "daemon", "stop"],
env=server.env,
capture_output=True,
timeout=30,
)
if stop.returncode != 0:
detail = (stop.stdout + stop.stderr).decode("utf-8", "replace")
raise RuntimeError(
"daemon cleanup failed (%d): %s" % (stop.returncode, detail.strip())
)
if not server.stderr_complete(STDERR_EOF_BACKSTOP_S):
print(
"SETUP FAIL: the stderr reader did not reach EOF after the server "
"exited; refusing to judge a possibly partial capture",
file=sys.stderr,
)
return 2
known, raw = classify_stderr(server.stderr_text())
if known or raw:
if known:
print(
"FAIL: startup leaked an OS path error to stderr: " + ", ".join(known),
file=sys.stderr,
)
if raw:
print(
"FAIL: startup wrote %d stderr line(s) that are not structured "
"level= log records:" % len(raw),
file=sys.stderr,
)
for line in raw:
print(" " + line, file=sys.stderr)
return 1
print("PASS: successful MCP startup emitted only structured log lines on stderr")
return 0
if __name__ == "__main__":
raise SystemExit(main())