Merge lab into fix/procfs-pid-file-liveness

src/agent_tools/web_tools.py

@@ -18,7 +18,7 @@ import urllib.request
 from pathlib import Path
 from typing import Dict, Any
 
-from core.platform_compat import pid_alive
+from core import platform_compat
 from src.constants import MAX_OUTPUT_CHARS
 
 PDF_EXTRACT_MAX_BYTES = 80_000_000
@@ -124,7 +124,6 @@ def _browser_pid_file_candidates(
 # Linux exposes one command line per pid under /proc; macOS and Windows do not.
 # Kept as a module attribute so the procfs-dependent paths stay testable on a
 # host that has no procfs, and on one that does.
-_PROC_ROOT = Path("/proc")
 
 
 def _process_command_line(pid: int) -> str | None:
@@ -136,7 +135,7 @@ def _process_command_line(pid: int) -> str | None:
     """
 
     try:
-        return (_PROC_ROOT / str(pid) / "cmdline").read_bytes().replace(
+        return (platform_compat.PROC_ROOT / str(pid) / "cmdline").read_bytes().replace(
             b"\0", b" "
         ).decode("utf-8", errors="replace")
     except (OSError, UnicodeError):
@@ -160,7 +159,7 @@ def _process_is_alive(pid: int) -> bool:
     reading in both.
     """
 
-    return pid_alive(pid)
+    return platform_compat.pid_alive(pid)
 
 _SCHOLARLY_METADATA_CUE_RE = re.compile(
     r"\b(?:accept(?:ed|ance)?|publish(?:ed|ing|cation)?|venue|conference|"
@@ -2359,14 +2358,14 @@ class PrivateBrowserTool:
         except OSError:
             return
         profile_prefix = str(tmpdir / "agent-browser-chrome-")
-        if not _PROC_ROOT.is_dir():
+        if not platform_compat.has_procfs():
             # Without procfs there is no way to match a reparented Chrome by
             # its command line, and the sweep is an optimisation rather than a
             # correctness requirement.  Leave those trees to the daemon's own
             # lifecycle instead of failing the whole shutdown path.
             return
         pids: list[int] = []
-        for entry in _PROC_ROOT.iterdir():
+        for entry in platform_compat.PROC_ROOT.iterdir():
             if not entry.name.isdigit():
                 continue
             try:

src/model_discovery.py

@@ -38,7 +38,10 @@ def discover_tailscale_hosts() -> List[str]:
     global _hosts_cache, _hosts_cache_time
 
     now = time.time()
-    if _hosts_cache and (now - _hosts_cache_time) < _HOSTS_CACHE_TTL:
+    # Gate on the timestamp, not the list: a successful query that found no
+    # eligible peers is a real answer, and testing the list's truthiness made
+    # that case re-run `tailscale status` (up to a 5s timeout) on every call.
+    if _hosts_cache_time and (now - _hosts_cache_time) < _HOSTS_CACHE_TTL:
         return list(_hosts_cache)
 
     hosts = []

src/task_scheduler.py

@@ -97,19 +97,30 @@ async def _cached(key: Tuple, ttl: float, fetch: Callable[[], Awaitable[Any]]) -
             pending = fut
             owner = True
     if not owner:
-        return await pending
+        # A cancelled waiter must not cancel the shared Future for the owner
+        # and every other waiter.
+        return await asyncio.shield(pending)
     try:
         val = await fetch()
         async with _shared_cache_lock:
             _shared_cache[key] = (time.monotonic() + ttl, val)
-            _shared_cache_pending.pop(key, None)
         pending.set_result(val)
         return val
+    except asyncio.CancelledError:
+        # Cancellation is a BaseException on supported Python versions, so it
+        # bypasses the Exception handler below. Wake all current waiters while
+        # allowing a later caller to retry the fetch.
+        pending.cancel()
+        raise
     except Exception as e:
-        async with _shared_cache_lock:
-            _shared_cache_pending.pop(key, None)
         pending.set_exception(e)
         raise
+    finally:
+        # Keep this cleanup synchronous so a second cancellation cannot
+        # interrupt it and leave a permanently pending Future behind. All
+        # access runs on the scheduler's event-loop thread.
+        if _shared_cache_pending.get(key) is pending:
+            _shared_cache_pending.pop(key, None)
 
 
 def compute_next_run(schedule: str, scheduled_time: str,

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 = f.read()
+                raw = (proc_root / pid_dir / "cmdline").read_bytes()
             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:
-                for pid_name in os.listdir("/proc"):
+            # No procfs means no way to match a survivor by its command line.
+            # 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