fix(cookbook): skip the pid sweep when the host has no procfs

`_cookbook_kill_session` kills the tmux session, then sweeps /proc for
model servers that survived the SIGHUP. The sweep had no guard, so on
macOS and Windows `os.listdir("/proc")` raised FileNotFoundError after
the kill had already succeeded. The function's outer except turned that
into `{"error": "...No such file or directory: '/proc'", "exit_code": 1}`
and skipped the state write that marks the session stopped — the agent
is told a stop failed that actually worked.

Guard the sweep with a procfs check, the way `_scan_running_model_processes`
already does a few hundred lines up. The root and the check now live in
`core/platform_compat`, which is where OS differences belong and which
makes both branches patchable from a test on either kind of host.

Adds a class-level guard test: the third instance of this defect, and two
of the three were found by reading source rather than by a test.

core/platform_compat.py

@@ -36,6 +36,19 @@ IS_APPLE_SILICON = (
 )
 
 
+# ── procfs ──────────────────────────────────────────────────────────────────
+# Linux exposes one directory per pid under /proc; macOS and Windows have no
+# procfs at all. Any code that walks it must skip the walk rather than raise.
+# Kept as a module attribute so both branches stay testable on either kind of
+# host.
+PROC_ROOT = Path("/proc")
+
+
+def has_procfs() -> bool:
+    """True when the host exposes a procfs pid tree that can be scanned."""
+    return PROC_ROOT.is_dir()
+
+
 # ── File permissions ────────────────────────────────────────────────────────
 def safe_chmod(path, mode: int) -> bool:
     """``os.chmod`` that is a harmless no-op on Windows.

src/tools/cookbook.py

@@ -17,6 +17,7 @@ import re
 from typing import Any, Dict, List, Optional
 
 from fastapi import HTTPException
+from core import platform_compat
 from routes._validators import validate_remote_host, validate_ssh_port
 
 from src.tools._common import _parse_tool_args
@@ -676,17 +677,17 @@ def _scan_running_model_processes() -> List[Dict[str, Any]]:
     a dict shaped like a cookbook task so the caller can merge cleanly.
     """
     import os
-    if not os.path.isdir("/proc"):
+    if not platform_compat.has_procfs():
         return []
+    proc_root = platform_compat.PROC_ROOT
     out: List[Dict[str, Any]] = []
     seen_keys = set()
     try:
-        for pid_dir in os.listdir("/proc"):
+        for pid_dir in os.listdir(proc_root):
             if not pid_dir.isdigit():
                 continue
             try:
-                with open(f"/proc/{pid_dir}/cmdline", "rb") as f:
+                raw = (proc_root / pid_dir / "cmdline").read_bytes()
-                    raw = f.read()
             except (OSError, PermissionError):
                 continue
             if not raw:
@@ -1124,12 +1125,16 @@ async def _cookbook_kill_session(session_id: str, *, remote_host: str = "",
             import signal
             tracked_cmd = str((matched.get("payload") or {}).get("_cmd") or "").strip()
             matched_pids: list[int] = []
-            if tracked_cmd:
+            # No procfs means no way to match a survivor by its command line.
-                for pid_name in os.listdir("/proc"):
+            # The tmux kill above already stopped the session, so skip the
+            # sweep instead of failing a stop that worked.
+            if tracked_cmd and platform_compat.has_procfs():
+                proc_root = platform_compat.PROC_ROOT
+                for pid_name in os.listdir(proc_root):
                     if not pid_name.isdigit() or int(pid_name) == os.getpid():
                         continue
                     try:
-                        raw = open(f"/proc/{pid_name}/cmdline", "rb").read()
+                        raw = (proc_root / pid_name / "cmdline").read_bytes()
                         process_cmd = raw.replace(b"\x00", b" ").decode("utf-8", errors="replace").strip()
                     except (OSError, PermissionError):
                         continue

tests/test_cookbook_stop_without_procfs.py

@@ -0,0 +1,166 @@
+"""Stopping a Cookbook server must succeed on a host with no procfs.
+
+The tmux kill is what actually stops the server; the pid sweep that follows
+it only catches model servers that survive the session's SIGHUP. On macOS and
+Windows there is no ``/proc`` to sweep, and letting that raise turned a
+successful stop into a reported failure *and* skipped the state write that
+marks the session stopped for the Cookbook UI.
+"""
+import asyncio
+import json
+import signal
+
+import pytest
+
+from core import platform_compat
+from src import tool_implementations as tools
+
+
+class FakeResponse:
+    def __init__(self, data=None, status_code=200):
+        self._data = data or {}
+        self.status_code = status_code
+        self.text = json.dumps(self._data)
+
+    def json(self):
+        return self._data
+
+
+def _tracked_state(session_id="serve-abc123", cmd="python -m vllm.entrypoints.openai.api_server"):
+    return {
+        "tasks": [
+            {
+                "sessionId": session_id,
+                "model": "org/model",
+                "type": "serve",
+                "status": "running",
+                "payload": {"_cmd": cmd},
+            }
+        ]
+    }
+
+
+def _install_httpx_client(monkeypatch, state):
+    """Serve cookbook state over a fake httpx and record every POST body."""
+    import httpx
+
+    posts = []
+
+    class FakeAsyncClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def get(self, url, **kwargs):
+            return FakeResponse(state)
+
+        async def post(self, url, json=None, **kwargs):
+            posts.append((url, json))
+            return FakeResponse({"ok": True})
+
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+    return posts
+
+
+def _install_successful_tmux_kill(monkeypatch):
+    """Replace the real ``tmux kill-session`` with a process that succeeds."""
+
+    class FakeProc:
+        returncode = 0
+
+        async def communicate(self):
+            return b"", b""
+
+    async def fake_exec(*argv, **kwargs):
+        assert argv[:2] == ("tmux", "kill-session")
+        return FakeProc()
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec)
+
+
+def _stopped_statuses(posts, session_id):
+    out = []
+    for _url, body in posts:
+        for task in (body or {}).get("tasks") or []:
+            if task.get("sessionId") == session_id:
+                out.append(task.get("status"))
+    return out
+
+
+@pytest.mark.asyncio
+async def test_stop_marks_session_stopped_when_the_host_has_no_procfs(
+    monkeypatch, tmp_path
+):
+    state = _tracked_state()
+    posts = _install_httpx_client(monkeypatch, state)
+    _install_successful_tmux_kill(monkeypatch)
+    monkeypatch.setattr(platform_compat, "PROC_ROOT", tmp_path / "no-procfs")
+
+    import os
+
+    def _unexpected_listdir(*args, **kwargs):
+        raise AssertionError("the pid sweep must not run without procfs")
+
+    monkeypatch.setattr(os, "listdir", _unexpected_listdir)
+
+    result = await tools.do_stop_served_model(
+        json.dumps({"session_id": "serve-abc123"})
+    )
+
+    assert result == {"output": "Stopped server serve-abc123", "exit_code": 0}
+    assert _stopped_statuses(posts, "serve-abc123") == ["stopped"]
+
+
+@pytest.mark.asyncio
+async def test_stop_sweeps_surviving_pids_when_procfs_is_present(
+    monkeypatch, tmp_path
+):
+    tracked_cmd = "python -m vllm.entrypoints.openai.api_server --model org/model"
+    state = _tracked_state(cmd=tracked_cmd)
+    posts = _install_httpx_client(monkeypatch, state)
+    _install_successful_tmux_kill(monkeypatch)
+
+    proc = tmp_path / "proc"
+
+    def _write_pid(pid, cmdline):
+        entry = proc / pid
+        entry.mkdir(parents=True)
+        (entry / "cmdline").write_bytes(cmdline.replace(" ", "\0").encode())
+
+    _write_pid("101", tracked_cmd)
+    _write_pid("202", "python -m http.server")
+    (proc / "self").mkdir()
+    monkeypatch.setattr(platform_compat, "PROC_ROOT", proc)
+
+    signalled = []
+    import os
+
+    monkeypatch.setattr(os, "kill", lambda pid, sig: signalled.append((pid, sig)))
+
+    result = await tools.do_stop_served_model(
+        json.dumps({"session_id": "serve-abc123"})
+    )
+
+    assert result["exit_code"] == 0
+    assert (101, signal.SIGTERM) in signalled
+    assert not any(pid == 202 for pid, _sig in signalled)
+    assert _stopped_statuses(posts, "serve-abc123") == ["stopped"]
+
+
+def test_model_process_scan_returns_empty_without_procfs(monkeypatch, tmp_path):
+    """The other procfs scan in the same module already guards; pin it."""
+    monkeypatch.setattr(platform_compat, "PROC_ROOT", tmp_path / "no-procfs")
+
+    import os
+
+    def _unexpected_listdir(*args, **kwargs):
+        raise AssertionError("the model-process scan must not run without procfs")
+
+    monkeypatch.setattr(os, "listdir", _unexpected_listdir)
+
+    assert tools._scan_running_model_processes() == []

tests/test_procfs_scan_guard.py

@@ -0,0 +1,178 @@
+"""Every procfs pid scan in the app tree must be guarded by an existence check.
+
+This is the third instance of the same defect: code walks ``/proc`` on a host
+that has no procfs, and the resulting ``FileNotFoundError`` breaks a path that
+had otherwise succeeded. Two of the three were found by reading source, so
+this pins the class rather than the instances.
+
+Deliberate AST assertion under the narrow exception in
+``tests/TESTING_STANDARD.md``: the invariant is "no *other* module grows an
+unguarded scan", which cannot be driven at runtime without importing and
+exercising every procfs-touching code path on both a Linux and a non-Linux
+host. ``test_cookbook_stop_without_procfs.py`` covers the behaviour itself.
+"""
+import ast
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+# Trees that ship as the application. Tests and vendored code are excluded.
+APP_TREES = (
+    "app.py",
+    "core",
+    "routes",
+    "src",
+    "services",
+    "scripts",
+    "mcp_servers",
+    "integrations",
+    "companion",
+)
+
+# Calls that enumerate a directory's entries. Reading one known file under
+# /proc is a different shape — it fails per-file and callers already handle
+# that — so only the enumerating calls are in scope here.
+_SCAN_FUNCS = {"listdir", "scandir"}
+_SCAN_METHODS = {"iterdir", "glob", "rglob"}
+
+# Calls that prove the scan is conditional on procfs being present.
+_GUARD_FUNCS = {"has_procfs", "isdir", "is_dir", "exists"}
+
+PROCFS_ROOT = "/proc"
+
+
+def _is_procfs_root(node: ast.AST) -> bool:
+    """True if ``node`` evaluates to the procfs root directory."""
+    if isinstance(node, ast.Constant) and node.value == PROCFS_ROOT:
+        return True
+    # Path("/proc")
+    if isinstance(node, ast.Call):
+        return any(_is_procfs_root(arg) for arg in node.args)
+    # PROC_ROOT / _PROC_ROOT / proc_root / platform_compat.PROC_ROOT
+    name = None
+    if isinstance(node, ast.Name):
+        name = node.id
+    elif isinstance(node, ast.Attribute):
+        name = node.attr
+    return bool(name) and name.lower().lstrip("_") == "proc_root"
+
+
+def _scan_sites(tree: ast.AST) -> list[ast.Call]:
+    sites = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if isinstance(func, ast.Attribute) and func.attr in _SCAN_FUNCS:
+            if node.args and _is_procfs_root(node.args[0]):
+                sites.append(node)
+        elif isinstance(func, ast.Name) and func.id in _SCAN_FUNCS:
+            if node.args and _is_procfs_root(node.args[0]):
+                sites.append(node)
+        elif isinstance(func, ast.Attribute) and func.attr in _SCAN_METHODS:
+            if _is_procfs_root(func.value):
+                sites.append(node)
+    return sites
+
+
+def _guard_lines(tree: ast.AST) -> list[int]:
+    """Line numbers of calls that test whether procfs is present."""
+    lines = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
+        if name not in _GUARD_FUNCS:
+            continue
+        subject_is_procfs = (
+            name in {"has_procfs", "is_wsl"}
+            or (isinstance(func, ast.Attribute) and _is_procfs_root(func.value))
+            or any(_is_procfs_root(arg) for arg in node.args)
+        )
+        if subject_is_procfs:
+            lines.append(node.lineno)
+    return lines
+
+
+def _enclosing_scope(tree: ast.AST, node: ast.AST) -> ast.AST:
+    """Smallest function/module scope containing ``node``."""
+    best = tree
+    for candidate in ast.walk(tree):
+        if not isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        end = getattr(candidate, "end_lineno", None) or candidate.lineno
+        if candidate.lineno <= node.lineno <= end:
+            if best is tree or candidate.lineno > best.lineno:
+                best = candidate
+    return best
+
+
+def _app_python_files() -> list[Path]:
+    files = []
+    for entry in APP_TREES:
+        target = REPO_ROOT / entry
+        if target.is_file():
+            files.append(target)
+        elif target.is_dir():
+            files.extend(
+                p for p in target.rglob("*.py") if "__pycache__" not in p.parts
+            )
+    return sorted(files)
+
+
+def _collect_sites() -> tuple[list[str], list[str]]:
+    """Return (guarded, unguarded) ``path:line`` labels for procfs scans."""
+    guarded, unguarded = [], []
+    for path in _app_python_files():
+        try:
+            source = path.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            continue
+        # Cheap pre-filter: a scan has to name the root somehow.
+        if PROCFS_ROOT not in source and "proc_root" not in source.lower():
+            continue
+        try:
+            tree = ast.parse(source)
+        except SyntaxError:
+            continue
+        sites = _scan_sites(tree)
+        if not sites:
+            continue
+        guards = _guard_lines(tree)
+        for site in sites:
+            scope = _enclosing_scope(tree, site)
+            start = getattr(scope, "lineno", 0)
+            label = f"{path.relative_to(REPO_ROOT)}:{site.lineno}"
+            if any(start <= g < site.lineno for g in guards):
+                guarded.append(label)
+            else:
+                unguarded.append(label)
+    return guarded, unguarded
+
+
+def test_every_procfs_scan_is_guarded_by_an_existence_check():
+    _guarded, unguarded = _collect_sites()
+    assert not unguarded, (
+        "procfs pid scans with no existence check in the enclosing function — "
+        "these raise FileNotFoundError on macOS and Windows: "
+        + ", ".join(unguarded)
+    )
+
+
+def test_the_guard_detector_still_sees_the_known_scans():
+    """A rename must not silently turn the assertion above into a no-op.
+
+    Lower bound, not an exact count: a new *guarded* scan is fine and should
+    not fail this. What must not happen is the detector going blind, which
+    shows up as sites disappearing.
+    """
+    guarded, unguarded = _collect_sites()
+    found = set(guarded) | set(unguarded)
+    files = {label.rsplit(":", 1)[0] for label in found}
+    known = {"src/agent_tools/web_tools.py", "src/tools/cookbook.py"}
+    assert known <= files, (
+        "the detector no longer sees a known procfs scan — check whether the "
+        f"root was renamed. Found: {sorted(found)}"
+    )
+    assert len(found) >= 3, f"expected at least 3 procfs scans, found {sorted(found)}"