feat: stdio server spawn-test (issue #2)
CI / Lint (ruff) (pull_request) Successful in 8s
CI / Tests (py3.10) (pull_request) Successful in 8s
CI / Tests (py3.12) (pull_request) Successful in 7s

Add spawn_test() to bcc_core — spawns a stdio server for up to 3 s,
captures stderr, and reports ok/exited/crashed/not_found. Key design
decisions driven by real MCP server behaviour:

- stdin=PIPE (never written): servers block on JSON-RPC input and stay
  alive, so "still running after timeout" reliably signals a healthy
  start. stdin=DEVNULL would send EOF, causing well-behaved servers to
  exit 0 and be misclassified as "exited".
- Command resolved via shutil.which(augmented_path()) before Popen so
  subprocess PATH resolution is unambiguous across platforms.
- start_new_session=True on POSIX + os.killpg on timeout: kills the
  whole process group, not just the launcher (npx, uvx), which would
  otherwise orphan the actual node/python grandchild process.
- stdout=DEVNULL: draining a PIPE we don't read would deadlock at ~64 KB.
- stderr drained in a daemon thread, capped at 4 KB.

GUI: SpawnTester(QThread) wraps spawn_test; "Test launch" button in
ServerEditor dep row (stdio only, visible when command resolves ok/warn).
Result colours match the existing dep-status palette (green/amber/red).
Stderr appended to the diagnostics panel if it is open.

7 new unit tests cover all outcomes and the stdin-open regression guard.
Ran python bcc.py locally: button appears for stdio servers whose command
resolves, is hidden for remote servers and missing-command servers.

Closes #2

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
AJ
2026-07-02 00:57:46 -04:00
parent e0eb3c2d1b
commit e98cb7abd7
3 changed files with 247 additions and 1 deletions
+55 -1
View File
@@ -132,6 +132,19 @@ class ConnTester(QThread):
self.done.emit(ok, detail)
class SpawnTester(QThread):
done = Signal(dict)
def __init__(self, data, timeout=3.0):
super().__init__()
self._data = data
self._timeout = timeout
def run(self):
result = core.spawn_test(self._data, self._timeout)
self.done.emit(result)
# --------------------------------------------------------------------------- #
# Small reusable: key/value editor (for env and headers)
# --------------------------------------------------------------------------- #
@@ -354,6 +367,10 @@ class ServerEditor(QFrame):
self.test_btn = QPushButton("Test connection")
self.test_btn.clicked.connect(self._test_remote)
self.test_btn.setVisible(False)
self.spawn_btn = QPushButton("Test launch")
self.spawn_btn.setToolTip("Spawn the server for 3 s and report whether it starts cleanly")
self.spawn_btn.clicked.connect(self._test_spawn)
self.spawn_btn.setVisible(False)
self.details_btn = QPushButton("Details ▸")
self.details_btn.setCheckable(True)
self.details_btn.toggled.connect(self._toggle_diag)
@@ -363,6 +380,7 @@ class ServerEditor(QFrame):
dep.addWidget(self.dep_label, 1)
dep.addWidget(self.fix_btn)
dep.addWidget(self.test_btn)
dep.addWidget(self.spawn_btn)
dep.addWidget(self.details_btn)
dep.addWidget(recheck)
outer.addLayout(dep)
@@ -585,7 +603,9 @@ class ServerEditor(QFrame):
res = core.check_dependency(data)
self._set_dep(res)
self.fix_btn.setVisible(res["status"] == "warn")
self.test_btn.setVisible(self.type.currentIndex() == 1)
is_remote = self.type.currentIndex() == 1
self.test_btn.setVisible(is_remote)
self.spawn_btn.setVisible(not is_remote and res["status"] in ("ok", "warn"))
problem = res["status"] in ("missing", "warn", "unknown")
if auto_open and problem and not self.details_btn.isChecked():
self.details_btn.setChecked(True) # opens panel (fills text via _toggle_diag)
@@ -650,6 +670,40 @@ class ServerEditor(QFrame):
self.dep_label.setText(f"reachable · {detail}" if ok else f"unreachable · {detail}")
self.dep_label.setStyleSheet(f"color: {color};")
def _test_spawn(self):
self.spawn_btn.setEnabled(False)
self.spawn_btn.setText("Launching…")
self.dep_dot.setText("")
self.dep_dot.setStyleSheet(f"color: {MUTED}; font-size: 14px;")
self.dep_label.setText("spawning server…")
self.dep_label.setStyleSheet(f"color: {MUTED};")
self._spawner = SpawnTester(self.dump_data(), timeout=3.0)
self._spawner.done.connect(self._on_spawn_done)
self._spawner.start()
def _on_spawn_done(self, result: dict):
self.spawn_btn.setEnabled(True)
self.spawn_btn.setText("Test launch")
outcome = result.get("outcome", "")
detail = result.get("detail", "")
stderr = result.get("stderr", "").strip()
if outcome == "ok":
color = GOOD
label = f"started · {detail}"
elif outcome == "exited":
color = WARN
label = f"exited cleanly · {detail}"
else:
color = BAD
label = f"{outcome} · {detail}"
self.dep_dot.setText("")
self.dep_dot.setStyleSheet(f"color: {color}; font-size: 14px;")
self.dep_label.setText(label)
self.dep_label.setStyleSheet(f"color: {color};")
if stderr and self.diag_card.isVisible():
current = self.diag_text.toPlainText()
self.diag_text.setPlainText(current + f"\n\n── stderr from Test launch ──\n{stderr}")
# --------------------------------------------------------------------------- #
# Arguments editor: one line = one argument, with a numbered gutter so that
+127
View File
@@ -20,8 +20,11 @@ import json
import os
import re
import shutil
import signal as _signal
import subprocess
import sys
import tempfile
import threading
import time
from dataclasses import dataclass
from pathlib import Path
@@ -1057,6 +1060,130 @@ def diagnostics_text(name: str, data: dict) -> str:
return "\n".join(L)
_STDERR_CAP = 4096 # bytes
def spawn_test(data: dict, timeout: float = 3.0) -> dict:
"""
Attempt to start a stdio server and observe it for `timeout` seconds.
Returns a dict:
outcome: "ok" — still running after timeout; server started correctly
"exited" — exited with code 0 before timeout; unusual for a server
"crashed" — exited with a non-zero code before timeout
"not_found" — command could not be resolved to an executable
"not_applicable" — remote server or no command; nothing to spawn
returncode: int | None
stderr: str (first ~4 KB)
detail: str
Run this off the UI thread — it blocks for up to `timeout` seconds.
"""
if "url" in data and "command" not in data:
return {
"outcome": "not_applicable",
"returncode": None,
"stderr": "",
"detail": "remote server",
}
cmd = (data.get("command") or "").strip()
if not cmd:
return {
"outcome": "not_applicable",
"returncode": None,
"stderr": "",
"detail": "no command set",
}
# Resolve to absolute path so subprocess doesn't fight with PATH in env.
resolved_cmd = shutil.which(cmd, path=augmented_path())
if resolved_cmd is None:
return {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": f"command not found: {cmd}",
}
args_list = [resolved_cmd] + [str(a) for a in (data.get("args") or [])]
merged_env = {**os.environ, "PATH": augmented_path()}
merged_env.update(data.get("env") or {})
stderr_chunks: list[bytes] = []
def _drain(pipe) -> None:
try:
while sum(len(c) for c in stderr_chunks) < _STDERR_CAP:
chunk = pipe.read(1024)
if not chunk:
break
stderr_chunks.append(chunk)
except Exception:
pass
popen_kwargs: dict = dict(
stdin=subprocess.PIPE, # keep open so servers block on read rather than seeing EOF
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
env=merged_env,
)
if os.name != "nt":
popen_kwargs["start_new_session"] = True # own process group → clean kill
try:
proc = subprocess.Popen(args_list, **popen_kwargs)
except (FileNotFoundError, OSError) as e:
return {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": str(e),
}
drain_thread = threading.Thread(target=_drain, args=(proc.stderr,), daemon=True)
drain_thread.start()
try:
proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
try:
if os.name != "nt":
os.killpg(os.getpgid(proc.pid), _signal.SIGKILL)
else:
proc.kill() # best-effort on Windows
except OSError:
pass
with contextlib.suppress(subprocess.TimeoutExpired):
proc.wait(timeout=1.0)
drain_thread.join(0.5)
stderr = b"".join(stderr_chunks).decode(errors="replace")
return {
"outcome": "ok",
"returncode": None,
"stderr": stderr,
"detail": f"still running after {timeout:.0f}s — server started successfully",
}
drain_thread.join(0.5)
stderr = b"".join(stderr_chunks).decode(errors="replace")
rc = proc.returncode
if rc == 0:
return {
"outcome": "exited",
"returncode": rc,
"stderr": stderr,
"detail": "process exited cleanly (code 0) — unusual; a healthy server should keep running",
}
return {
"outcome": "crashed",
"returncode": rc,
"stderr": stderr,
"detail": f"process exited with code {rc}",
}
def test_remote(url: str, timeout: float = 5.0) -> tuple[bool, str]:
"""
Reachability check for a url-based MCP server. ANY HTTP response (even 4xx/5xx)
+65
View File
@@ -2,6 +2,7 @@
proper test functions with tmp_path/monkeypatch fixtures)."""
import json
import sys
import pytest
@@ -238,3 +239,67 @@ def test_dep_check_marks_remote():
def test_dep_check_sees_through_shell_wrapper():
r = c.check_dependency({"command": "cmd", "args": ["/c", "definitely-not-real-xyz", "foo"]})
assert "definitely-not-real-xyz" in r["label"]
# --------------------------------------------------------------------------- #
# 7. Spawn test
# --------------------------------------------------------------------------- #
def test_spawn_test_ok_still_running():
# A process that sleeps longer than the timeout → "ok" (server started)
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "import time; time.sleep(60)"]}, timeout=0.3
)
assert r["outcome"] == "ok"
assert r["returncode"] is None
def test_spawn_test_crashed():
# A process that exits with a non-zero code immediately
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "raise SystemExit(1)"]}, timeout=0.3
)
assert r["outcome"] == "crashed"
assert r["returncode"] == 1
def test_spawn_test_exited_cleanly():
# A process that exits 0 before timeout (unusual for a server)
r = c.spawn_test({"command": sys.executable, "args": ["-c", "pass"]}, timeout=0.5)
assert r["outcome"] == "exited"
assert r["returncode"] == 0
def test_spawn_test_not_found():
r = c.spawn_test({"command": "definitely-not-a-real-binary-xyz"}, timeout=0.3)
assert r["outcome"] == "not_found"
def test_spawn_test_remote_not_applicable():
r = c.spawn_test({"url": "https://example.com/mcp"}, timeout=0.3)
assert r["outcome"] == "not_applicable"
def test_spawn_test_stderr_captured():
# A process that writes to stderr before crashing
r = c.spawn_test(
{
"command": sys.executable,
"args": ["-c", "import sys; sys.stderr.write('test-err\\n'); sys.exit(2)"],
},
timeout=0.5,
)
assert r["outcome"] == "crashed"
assert "test-err" in r["stderr"]
def test_spawn_test_stdin_not_closed():
# A process that blocks reading stdin should stay alive (outcome=ok), not exit 0.
# This guards against the regression where stdin=DEVNULL sends EOF and a server
# that reads stdin exits cleanly, being misclassified as "exited" instead of "ok".
r = c.spawn_test(
{"command": sys.executable, "args": ["-c", "import sys; sys.stdin.read()"]},
timeout=0.3,
)
assert r["outcome"] == "ok", f"expected ok (blocking on stdin), got {r['outcome']}"