THI-450: build_state forks less than half as many processes per refresh

CLAUDE.md

@@ -116,6 +116,8 @@ would double-load every global.
 | `config.py` | Cross-cutting paths + constants; `instance_file()` keeps prod/dev stores apart |
 | `log.py` | Logging + `_bg` / `_task` crash wrappers — every thread/task goes through these so crashes surface |
 | `pidfile.py` | Single-instance reclaim (reloader child counts as the same instance) |
+| `forks.py` | Audit-hook fork counter + each poll's cost (`/api/state` `build_stats`, the resource popover's poll block; the full map on `/api/healthz`) |
+| `gh.py` | Every `gh` call: token read every 10 minutes, telemetry and update check off |
 | `tmux.py` / `tmux_input.py` / `tmux_mirror.py` | Subprocess wrappers / persistent control-mode input client / control-mode pane mirror with reconcile frames |
 | `store.py` | `state.json` layer (`_STATE`, load/write, migrations) |
 | `channels.py` | In-process MCP server + tool implementations |

docs/invariants.md

@@ -8,6 +8,10 @@ These are the non-obvious behaviors worth preserving:
    active window in its session, or when the user acts on it via the
    dashboard (focus/send). See `update_focus_from_windows`.
 
+   The window rows come from one `list-panes -a` (`tmux.list_panes`), one
+   row per window off its active pane (`#{pane_active}`); `active` still
+   reads `#{window_active}`.
+
 2. **Claude detection is anchored to the bottom of the pane.** A pane is
    Claude when its composer (`❯` between two full-width rules) has no more
    than a footer under it, or the status line is in the last 4 non-empty
@@ -189,6 +193,14 @@ These are the non-obvious behaviors worth preserving:
     cross-poll residue; `grep "re-minting\|duplicate @periscope_id"`
     catches both.
 
+    `list_panes` carries `#{window_id}` and the window option
+    `#{@periscope_id}` on every pane row, so a stamp still targets the
+    window id. `list_windows` always forks a fresh `list-panes` — a stamp or
+    a new window must read back at once — and only `all_pane_ids` and
+    session_status's pane → pid map read the shared snapshot
+    (`tmux.panes_snapshot`: the poll's own rows inside `build_state`,
+    else under 1 s old).
+
 12. **Rebind eligibility (`_REBIND_TTL_S`, 15 min) is NOT GC retention
     (`_PID_TTL_S`, 30 days).** Rebind exists so persisted state reattaches
     when the tmux server restarts — window options are lost, so every

docs/testing.md

@@ -27,3 +27,14 @@ fixture seeds the cache so no plan-usage thread ever spawns, and
 `fresh_activity_db` teardown holds `activity._LOCK` before closing `_CONN`.
 Patching one call site (e.g. `periscope.app.cached_plan_usage`) is NOT enough —
 `routes.state` holds its own binding, so the fix lives at the cache layer.
+
+**The shared snapshots are seeded, so their readers don't fork.** conftest
+seeds `tmux._panes_snapshot`, `agent_processes._snapshot_cache` and
+`gh._token_cache` with unexpiring empty entries, and clears git_pr's per-path
+caches (a stale entry would start a real `_bg` git refresh). So no test runs
+`gh auth token`, and `all_pane_ids`, session_status and the process readers
+fork no `list-panes` or `ps` — a test that calls the real `list_windows`
+still forks one `list-panes`. A test of one of those paths resets the seed
+itself (`tests/test_gh.py` overrides `_no_gh_token` by name). The live-tmux
+oracle in `test_panes.py` and the live-`ps` tests in
+`test_agent_processes.py` and `test_session_status.py` fork on purpose.

periscope/agent_processes.py

@@ -1,107 +1,180 @@
 """Process-tree evidence for interactive coding agents.
 
-This module is deliberately stdlib-only.  A failed ``ps`` invocation means
-"no opinion" rather than "the agent exited"; callers use ``None`` to preserve
-that distinction.
+This module is deliberately stdlib-only.  A failed ``ps`` invocation serves
+the last table while it is recent, then means "no opinion" rather than "the
+agent exited"; callers use ``None`` to preserve that distinction.
 """
 
 from __future__ import annotations
 
+import contextlib
+import logging
 import os
 import re
 import subprocess
 import threading
 import time
+from collections import deque
+from contextvars import ContextVar
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
+
+log = logging.getLogger(__name__)
 
 
 @dataclass(frozen=True)
 class ProcessInfo:
     pid: int
     ppid: int
     started_at: int
-    comm: str
+    # argv[0]: the executable as launched. Not ps's comm, which macOS cuts to
+    # 16 characters unless it is the last column (`/Users/tom/.loca`).
+    exe: str
     argv: str
     state: str
 
 
 _PS_ROW_RE = re.compile(
     r"^\s*(\d+)\s+(\d+)\s+"
     r"(\w{3}\s+\w{3}\s+\d+\s+\d\d:\d\d:\d\d\s+\d{4})\s+"
-    r"(\S+)\s+(\S+)(?:\s+(.*))?$"
+    r"(\S+)(?:\s+(.*))?$"
 )
 _PS_TTL_S = 1.0
+_PS_TIMEOUT_S = 2.0
+# A failed ps serves the last table while it is at most this old, instead of
+# blanking session state, the pane -> claude map, codex detection and resources
+# for a poll. A loaded poll runs 3-13 s, so the last table is up to ~14 s old
+# when the next poll's ps times out.
+_PS_KEEP_S = 15.0
+_PS_FAILURE_LOG_EVERY_S = 60.0
 _lock = threading.Lock()
 _snapshot_cache: tuple[float, dict[int, ProcessInfo] | None] | None = None
+# (monotonic time, table) from the last ps that succeeded.
+_last_table: tuple[float, dict[int, ProcessInfo]] | None = None
+_failure_logged_at: float | None = None
+# Inside pinned_snapshot(): the table the block took, in a list its context
+# copies share.
+_pinned: ContextVar[list[dict[int, ProcessInfo] | None]] = ContextVar("pinned_process_snapshot")
 # pane_id -> (pane pid, pane start, agent pid, agent start)
 _pane_cache: dict[str, tuple[int, int, int, int]] = {}
 
 
 def parse_ps_snapshot(output: str) -> dict[int, ProcessInfo]:
-    """Parse macOS ``ps`` output emitted by :func:`_take_snapshot`."""
+    """Parse the `pid ppid lstart state args` rows :func:`_take_snapshot` reads."""
     result: dict[int, ProcessInfo] = {}
     for line in output.splitlines():
         match = _PS_ROW_RE.match(line)
         if not match:
             continue
-        pid, ppid, started, state, comm, argv = match.groups()
+        pid, ppid, started, state, argv = match.groups()
         try:
-            started_at = int(datetime.strptime(started, "%a %b %d %H:%M:%S %Y").timestamp())
+            started_at = int(datetime.strptime(started, "%a %b %d %H:%M:%S %Y").replace(tzinfo=UTC).timestamp())
         except ValueError:
             continue
+        argv = (argv or "").strip()
         info = ProcessInfo(
             pid=int(pid),
             ppid=int(ppid),
             started_at=started_at,
-            comm=comm,
-            argv=argv or comm,
+            exe=argv.split(maxsplit=1)[0] if argv else "",
+            argv=argv,
             state=state,
         )
         result[info.pid] = info
     return result
 
 
+def ps_env() -> dict[str, str]:
+    """LC_ALL=C so lstart parses and argv bytes encode the same in both ps
+    calls `_env_tail` compares; TZ=UTC so lstart is unambiguous across DST."""
+    return {**os.environ, "LC_ALL": "C", "TZ": "UTC"}
+
+
 def _take_snapshot() -> dict[int, ProcessInfo] | None:
     try:
         proc = subprocess.run(
-            ["ps", "-axo", "pid=,ppid=,lstart=,state=,comm=,args="],
+            ["ps", "-axo", "pid=,ppid=,lstart=,state=,args="],
             check=True,
             capture_output=True,
             text=True,
-            timeout=2,
+            timeout=_PS_TIMEOUT_S,
+            env=ps_env(),
         )
     except (OSError, subprocess.SubprocessError):
         return None
     return parse_ps_snapshot(proc.stdout)
 
 
+@contextlib.contextmanager
+def pinned_snapshot():
+    """Inside the block, and in copies of its context, process_snapshot()
+    serves the table it took first, however long the block runs.
+    build_state's poll is the block: a loaded one runs 3–13 s, and every
+    reader in it gets one table."""
+    token = _pinned.set([])
+    try:
+        yield
+    finally:
+        _pinned.reset(token)
+
+
 def process_snapshot() -> dict[int, ProcessInfo] | None:
-    """Return a briefly cached system process snapshot."""
-    global _snapshot_cache
+    """The system process table: the pinned one inside pinned_snapshot(),
+    else the last one while under _PS_TTL_S old, else a fresh `ps`."""
+    global _snapshot_cache, _last_table, _failure_logged_at
+    pinned = _pinned.get(None)
+    if pinned:
+        return pinned[-1]
     now = time.monotonic()
     with _lock:
         if _snapshot_cache and now - _snapshot_cache[0] < _PS_TTL_S:
-            return _snapshot_cache[1]
-        snapshot = _take_snapshot()
-        _snapshot_cache = (now, snapshot)
-        return snapshot
+            snapshot = _snapshot_cache[1]
+        else:
+            snapshot = _take_snapshot()
+            # Stamped when ps returns, not when it started: a ps slower than the
+            # TTL would otherwise be expired on arrival. Callers queued on the lock
+            # read `now` before it, so they reuse this table.
+            returned_at = time.monotonic()
+            if snapshot is not None:
+                _last_table = (returned_at, snapshot)
+            else:
+                kept = _last_table if _last_table and returned_at - _last_table[0] <= _PS_KEEP_S else None
+                snapshot = kept[1] if kept else None
+                if _failure_logged_at is None or returned_at - _failure_logged_at >= _PS_FAILURE_LOG_EVERY_S:
+                    _failure_logged_at = returned_at
+                    log.warning(
+                        "ps failed or ran past %.0f s; %s",
+                        _PS_TIMEOUT_S,
+                        f"serving the table from {returned_at - kept[0]:.1f} s ago" if kept else "no process table",
+                    )
+            _snapshot_cache = (returned_at, snapshot)
+    if pinned is not None:
+        pinned.append(snapshot)
+    return snapshot
 
 
 def _is_codex_executable(proc: ProcessInfo) -> bool:
     # Match the executable itself, never an arbitrary mention in argv.
-    return os.path.basename(proc.comm).lower() in {"codex", "codex.exe"}
+    return os.path.basename(proc.exe).lower() in {"codex", "codex.exe"}
+
+
+def is_claude_executable(proc: ProcessInfo) -> bool:
+    # The launcher (`claude`, `.../bin/claude`) and the versioned binary
+    # (`.../share/claude/versions/<v>`) both say so; a `node` whose arguments
+    # merely mention ~/.claude doesn't.
+    return "claude" in proc.exe.lower()
 
 
 def descendants(root_pid: int, snapshot: dict[int, ProcessInfo]):
-    """`root_pid`'s process (when in the snapshot) and every process below it."""
+    """`root_pid`'s process (when in the snapshot) and every process below it,
+    nearest first."""
     children: dict[int, list[ProcessInfo]] = {}
     for proc in snapshot.values():
         children.setdefault(proc.ppid, []).append(proc)
-    pending = [root_pid]
+    pending = deque([root_pid])
     seen: set[int] = set()
     while pending:
-        parent = pending.pop()
+        parent = pending.popleft()
         if parent in seen:
             continue
         seen.add(parent)
@@ -158,7 +231,9 @@ def codex_process_for_pane(pane_id: str, pane_pid: int | str | None) -> bool | N
 
 def clear_caches() -> None:
     """Test/support hook; runtime callers normally rely on TTL/invalidation."""
-    global _snapshot_cache
+    global _snapshot_cache, _last_table, _failure_logged_at
     with _lock:
         _snapshot_cache = None
+        _last_table = None
+        _failure_logged_at = None
         _pane_cache.clear()

periscope/app.py

@@ -52,8 +52,9 @@
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
-    from periscope import config
+    from periscope import config, forks
 
+    forks.install()
     log.info("periscope starting (pid=%d, port=%d)", os.getpid(), config.PORT)
     from periscope import store
 

periscope/channels.py

@@ -155,13 +155,21 @@ def _unknown_account_error(arguments: dict) -> list | None:
     return _tool_result({"ok": False, "error": f"unknown account {account!r}; registered: {', '.join(ids)}"})
 
 
-def _channel_gc(known_pane_ids: set[str]) -> None:
-    """Drop alert state for panes that no longer exist. Session registry is
-    GC'd by the connection handler on disconnect, not here."""
+def _channel_gc(known_pane_ids: set[str], since: float) -> None:
+    """Drop alert state for panes that no longer exist, except a pane whose
+    newest alert landed at or after `since` — the poll's pane set can be as
+    old as the poll itself (its pinned list-panes rows), so a pane
+    spawned mid-poll must not look dead before its own first alert gets here.
+    Session registry is GC'd by the connection handler on disconnect, not
+    here."""
     with _CHANNELS_LOCK:
-        for d in (_CHANNEL_ALERTS, _CHANNEL_UNREAD):
-            for stale in [k for k in d if k not in known_pane_ids]:
-                d.pop(stale, None)
+        stale = {k for k in (*_CHANNEL_ALERTS, *_CHANNEL_UNREAD) if k not in known_pane_ids}
+        for pane_id in stale:
+            newest = max((a.get("ts") or 0 for a in _CHANNEL_ALERTS.get(pane_id, [])), default=0)
+            if newest >= since:
+                continue
+            _CHANNEL_ALERTS.pop(pane_id, None)
+            _CHANNEL_UNREAD.pop(pane_id, None)
 
 
 def _record_alert(pane: str, message: str, kind: str = "info", severity: str = "info") -> dict:
@@ -1146,8 +1154,7 @@ def pane_channel_ready(pane_id: str) -> bool | None:
     pid = session_status.pane_claude_pids().get(pane_id)
     if pid is None:
         return None
-    _kids, cmds = session_status.proc_table()
-    cmd = cmds.get(pid)
+    cmd = session_status.command_lines().get(pid)
     if cmd is None:
         return None
     return config.CHANNEL_FLAG in cmd

periscope/cleanup.py

@@ -24,7 +24,7 @@
 import time
 from typing import TypedDict
 
-from periscope import worktrees
+from periscope import gh, worktrees
 from periscope.gitutil import detect_default_branch, worktree_root
 from periscope.log import log
 from periscope.projects import MAIN_KEY, Project, all_projects
@@ -70,11 +70,7 @@ def _pr_state(repo: str, pr_number: int) -> str | None:
         cached = _pr_state_cache.get(key)
         if cached and _now() - cached[0] < _PR_STATE_TTL:
             return cached[1]
-    code, out = _run(
-        ["gh", "pr", "view", str(pr_number), "--json", "state"],
-        cwd=repo,
-        timeout=10.0,
-    )
+    code, out = gh.run("pr", "view", str(pr_number), "--json", "state", cwd=repo, timeout=10.0)
     state: str | None = None
     if code == 0 and out:
         try: