Merge pull request 'feat: stdio server spawn-test (#2)' (#11) from feat/2-stdio-spawn-test into main

Merge feat/2-stdio-spawn-test into main

bcc.py

@@ -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)
 # --------------------------------------------------------------------------- #
@@ -308,6 +321,7 @@ class ServerEditor(QFrame):
         self._before_change = before_change  # called before any row add/remove in tables
         self._extra = {}  # preserve unknown server fields verbatim
         self._loading = False
+        self._last_spawn: dict | None = None  # most recent spawn_test result
 
         outer = QVBoxLayout(self)
         outer.setContentsMargins(16, 16, 16, 16)
@@ -354,6 +368,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 +381,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)
@@ -470,6 +489,7 @@ class ServerEditor(QFrame):
             self.headers.load({})
             self.details_btn.setChecked(False)
             self.diag_text.clear()
+            self._last_spawn = None
             self._set_dep({"status": "unknown", "label": "—"})
             self.args_warn.hide()
             self.args_fix_btn.hide()
@@ -585,12 +605,14 @@ 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)
         if self.diag_card.isVisible():
-            self.diag_text.setPlainText(core.diagnostics_text(self.current_name(), data))
+            self.diag_text.setPlainText(self._full_diag_text())
 
     def _set_dep(self, res: dict):
         status = res.get("status", "unknown")
@@ -600,13 +622,17 @@ class ServerEditor(QFrame):
         color = STATUS_COLORS.get(status, MUTED) if status in ("missing", "warn") else MUTED
         self.dep_label.setStyleSheet(f"color: {color};")
 
+    def _full_diag_text(self) -> str:
+        text = core.diagnostics_text(self.current_name(), self.dump_data())
+        if self._last_spawn and self._last_spawn.get("stderr", "").strip():
+            text += "\n\n── stderr from last Test launch ──\n" + self._last_spawn["stderr"].strip()
+        return text
+
     def _toggle_diag(self, on):
         self.diag_card.setVisible(on)
         self.details_btn.setText("Details ▾" if on else "Details ▸")
         if on and self.isEnabled():
-            self.diag_text.setPlainText(
+            self.diag_text.setPlainText(self._full_diag_text())
-                core.diagnostics_text(self.current_name(), self.dump_data())
-            )
 
     def _recheck(self):
         core.refresh_path_cache()  # re-scan PATH, e.g. after installing a runtime
@@ -650,6 +676,42 @@ 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")
+        self._last_spawn = result
+        outcome = result.get("outcome", "")
+        detail = result.get("detail", "")
+        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};")
+        # Auto-open the details panel on non-ok outcomes so stderr is visible.
+        if outcome != "ok" and not self.details_btn.isChecked():
+            self.details_btn.setChecked(True)  # triggers _toggle_diag → fills diag_text
+        elif self.diag_card.isVisible():
+            self.diag_text.setPlainText(self._full_diag_text())
+
 
 # --------------------------------------------------------------------------- #
 # Arguments editor: one line = one argument, with a numbered gutter so that

bcc_core.py

@@ -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,136 @@ 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:
+        # Read until EOF so the pipe buffer never fills and blocks the subprocess.
+        # Only keep the first _STDERR_CAP bytes; the rest is discarded.
+        captured = 0
+        try:
+            while True:
+                chunk = pipe.read(1024)
+                if not chunk:
+                    break
+                if captured < _STDERR_CAP:
+                    keep = min(len(chunk), _STDERR_CAP - captured)
+                    stderr_chunks.append(chunk[:keep])
+                    captured += keep
+        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)

tests/test_core.py

@@ -2,6 +2,7 @@
 proper test functions with tmp_path/monkeypatch fixtures)."""
 
 import json
+import sys
 
 import pytest
 
@@ -238,3 +239,82 @@ 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']}"
+
+
+def test_spawn_test_large_stderr_does_not_deadlock():
+    # A process that writes >4 KB (pipe buffer territory) to stderr then exits non-zero.
+    # Without EOF-draining, the subprocess blocks on a full pipe and is wrongly
+    # classified as "ok" (still running) instead of "crashed".
+    script = "import sys; sys.stderr.write('x' * 65536); sys.stderr.flush(); sys.exit(3)"
+    r = c.spawn_test(
+        {"command": sys.executable, "args": ["-c", script]},
+        timeout=2.0,
+    )
+    assert r["outcome"] == "crashed", f"expected crashed, got {r['outcome']}"
+    assert r["returncode"] == 3
+    # stderr is capped, not the full 64 KB
+    assert len(r["stderr"]) <= c._STDERR_CAP