reliquary/plugins/spire/skills/forks/forks.py

# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""spire:forks helper: trains one headless implementer per build and forks it per block.

Run as `uv run <abs path>/forks.py <command> [args]`. Every call prints exactly one JSON
object on stdout (newline-terminated); diagnostics go to stderr. Exit codes: 0 ok, 2 usage
error or refused, 3 `wait` condition not met at timeout, 4 `wait` condition met and a job
in it did not end `ok`.

Standard library only. State lives next to the run log: `<run log without .md>-forks/`.
"""
from __future__ import annotations

import argparse
import contextlib
import datetime as _dt
import hashlib
import json
import ntpath
import os
import platform
import posixpath
import re
import shutil
import signal
import string
import subprocess
import sys
import threading
import time
import traceback
import uuid

# --------------------------------------------------------------------------------------
# Constants
# --------------------------------------------------------------------------------------

SCHEMA = 1

EXIT_OK = 0
EXIT_USAGE = 2
EXIT_WAIT_TIMEOUT = 3
EXIT_JOB_FAILED = 4

IS_WINDOWS = os.name == "nt"

# How the detached supervisor is created (C6, P0.1 rung 3 hidden; POSIX: new session).
SPAWN_METHOD = "cim-hidden" if IS_WINDOWS else "setsid"
CREATE_NO_WINDOW = 0x08000000
_NOWIN = CREATE_NO_WINDOW if IS_WINDOWS else 0

# Registry file semantics (C3a). A lock is stale when its holder (pid + creation time) is dead,
# or when it is older than LOCK_ABANDONED_S. Every critical section is a read-modify-write of a few
# small JSON files: milliseconds, at worst a few seconds of PermissionError retries (40 x 50 ms
# per file). 300 s is two orders of magnitude beyond that, so only a holder that cannot be judged
# (garbage content, an unverifiable pid) and is really gone ever reaches it.
LOCK_ABANDONED_S = 300.0
LOCK_WAIT_S = 45.0          # a dead holder's lock is broken at once; a live holder is waited for
LOCK_POLL_S = 0.05
RETRY_TRIES = 40            # PermissionError retries for replace and read (>= 20)
RETRY_SLEEP_S = 0.05

MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$")
ONE_M_SUFFIX = "[1m]"
EFFORTS = ("low", "medium", "high", "xhigh", "max")
# `claude --help` 2.1.299 (checked by the Phase 1 review); `plan` is in the list but refused.
PERMISSION_MODES = ("acceptEdits", "auto", "bypassPermissions", "manual", "dontAsk", "plan")
REFUSED_PERMISSION_MODES = ("plan",)

DEFAULT_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,Skill"
DEFAULT_DISALLOWED = ["Bash(git push:*)", "Bash(git rebase:*)", "Bash(git reset --hard:*)",
                      "Bash(claude:*)"]

CONFIG_DEFAULTS = {
    "parallel_cap": 2,
    "orphan_after": 1200,     # N1, seconds; > 2 x 570
    "pings": "on",            # P0.6 passed; used in Phase 2
    "ttl_s": 3600,
    "max_minutes": 90,
    "stall_after_s": 900,
}
THRESHOLD_DEFAULTS = {
    "rebase_commits": 6,
    "rebase_lines": 1500,
    "rebase_delta_tokens": 12000,
    "max_base_tokens": 250000,
    "max_rebases": 4,
}

# Env pinned into every launch's fingerprint (design 3.3).
ENV_PINNED_LAUNCH = {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"}

# --- Env scrub: one constant (design 4 + plan 2 + P0.0) ---
SCRUB_DROP = frozenset({
    # the twelve Desktop vars (design 4, research E1)
    "CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_CODE_HOST_SESSION_ID",
    "CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_DESKTOP_APP_VERSION",
    "CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_TERMINAL_MCP_TOOLS",
    "CLAUDE_CODE_REPORT_FINDINGS", "CLAUDE_CODE_SESSION_ID", "CLAUDE_AGENT_SDK_VERSION",
    "CLAUDE_CODE_ENABLE_ASK_USER_QUESTION_TOOL", "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING",
    # fingerprint and auth path
    "CLAUDE_EFFORT", "ANTHROPIC_MODEL", "ANTHROPIC_SMALL_FAST_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL",
    "ENABLE_PROMPT_CACHING_1H", "FORCE_PROMPT_CACHING_5M", "CLAUDE_CODE_SUBAGENT_PROMPT_CACHE_TTL",
    "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX",
    # uv (plan 2, P0.0)
    "VIRTUAL_ENV",
})
SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r"^UV_"))
SCRUB_KEEP = frozenset({
    "CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH",
    "PATH", "HOME", "USERPROFILE",
    "ANTHROPIC_BASE_URL",      # kept (gateways need it); `where` warns when it is not the default
})
DEFAULT_BASE_URL = "https://api.anthropic.com"
SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_FORKS_CHILD": "1"}
SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH"   # set to the stored absolute binary path
CHILD_MARKER = "SPIRE_FORKS_CHILD"             # N3
SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_")   # `where` lists unknown vars with these

# C8: the user settings file. Module-level so tests can point it at a temp file.
USER_SETTINGS_PATH: str | None = None

PHASE2_COMMANDS = ("rebase", "usage")
# N3: a fork may only run where, list, usage and wait (`tick` launches pings).
CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init", "tick")

# Warmth and keep-warm (design 5, block 2.1). Module-level so tests can shorten them.
TICK_SUP_S = 60.0           # a supervisor runs one tick pass this often
TICK_WAIT_S = 30.0          # `wait` runs one tick pass at its start and then this often
ACTIVITY_WINDOW_S = 600.0   # ping only while a fork/continue ran or `wait` was called this recently
PING_PROMPT = "Reply OK"
PING_MAX_S = 300.0          # wall-time limit of one ping (a single turn)
TTL_1H, TTL_5M = 3600, 300
# Price ratios against the base input price, in hundredths (integers: floor(1.9 / 0.1) is 18 in floats).
PRICE_RATIOS = {"read": 10, "write_1h": 200, "write_5m": 125}
REAL_KINDS = ("train", "fork", "continue", "rebase")   # every kind but `ping`

# Supervision and jobs (1.2). Module-level so tests can shorten the in-process ones.
HEARTBEAT_S = 10.0          # the supervisor writes heartbeat_at at least this often (<= 30 s)
LOST_AFTER_S = 120.0        # heartbeat older than this and a dead supervisor -> `lost`
SUP_TICK_S = 0.25           # supervisor main-loop period
CLOCK_JUMP_S = 60.0         # a tick gap above this is a host suspend / clock jump (M3), not run time
ORPHAN_CHECK_S = 1.0        # how often the supervisor reads run.last_orchestrator_call_at (N1)
WAIT_POLL_S = 2.0           # `wait` poll period
WAIT_DEFAULT_S = 540
WAIT_MAX_S = 570
KILL_VERIFY_S = 10.0        # how long `kill` keeps sweeping before it reports survivors
READER_JOIN_S = 10.0        # after claude exits, how long the supervisor waits for the stream reader
CACHE_HIT_RATIO = 0.9       # C5
RUNNING = ("created", "running")
FINAL = ("ok", "failed", "timeout", "killed", "lost")
LAUNCH_FIELDS = ("model", "effort", "permission_mode", "tools", "allowed_tools", "disallowed_tools",
                 "agent", "cwd", "binary")   # fixed at init; fork/continue refuse them
MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens")
TOKEN_KEYS = ("input", "output", "cache_creation", "cache_creation_1h", "cache_creation_5m", "cache_read")
TEMPLATES_DIR = "templates"

# CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars.
POWERSHELL_CIM = (
    "$si = New-CimInstance -ClassName Win32_ProcessStartup -ClientOnly -Property @{ShowWindow=[uint16]0}; "
    "$r = Invoke-CimMethod -ClassName Win32_Process -MethodName Create -Arguments "
    "@{CommandLine=$env:SPIRE_FORKS_CMD; CurrentDirectory=$env:SPIRE_FORKS_CWD; ProcessStartupInformation=$si}; "
    "\"{0} {1}\" -f $r.ReturnValue, $r.ProcessId")

# --------------------------------------------------------------------------------------
# Errors, output, clock
# --------------------------------------------------------------------------------------


class ForksError(Exception):
    """A refusal or usage error: exit `code`, payload printed as the stdout JSON object."""

    def __init__(self, error: str, code: int = EXIT_USAGE, **fields):
        super().__init__(error)
        self.code = code
        self.payload = {"error": error, **fields}


class _HelpRequested(Exception):
    def __init__(self, text: str):
        super().__init__("help")
        self.text = text


def _emit(obj: dict) -> None:
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()


def _diag(msg: str) -> None:
    sys.stderr.write("forks: " + msg + "\n")
    sys.stderr.flush()


def _now() -> float:
    """The only clock (epoch seconds). Tests monkeypatch it."""
    return time.time()


def _iso(t: float | None = None) -> str:
    """ISO 8601 with offset, from `_now()` unless given."""
    if t is None:
        t = _now()
    return _dt.datetime.fromtimestamp(t).astimezone().isoformat(timespec="seconds")


def _ts(iso) -> float | None:
    """Epoch seconds of an ISO string written by `_iso`, or None."""
    if not iso or not isinstance(iso, str):
        return None
    try:
        return _dt.datetime.fromisoformat(iso).timestamp()
    except ValueError:
        return None


def _uniq(items) -> list:
    out = []
    for x in items:
        if x not in out:
            out.append(x)
    return out


def _sha256_bytes(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def _sha256_file(path: str) -> str:
    with open(path, "rb") as f:
        return _sha256_bytes(f.read())


def canonical_json(obj) -> str:
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def fingerprint_hash(launch: dict) -> str:
    return _sha256_bytes(canonical_json(launch).encode("utf-8"))


# --------------------------------------------------------------------------------------
# Paths
# --------------------------------------------------------------------------------------

_GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)")


def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str:
    """Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms.

    Windows: absolute path with backslashes and an upper-case drive letter.
    POSIX: `os.path.abspath` semantics (relative to `cwd`).
    """
    if windows is None:
        windows = IS_WINDOWS
    if not isinstance(p, str) or not p.strip():
        raise ForksError("empty path")
    p = p.strip()
    if not windows:
        base = cwd if cwd is not None else os.getcwd()
        return posixpath.normpath(posixpath.join(base, p))
    m = _GITBASH_DRIVE.match(p)
    if m:
        p = m.group(1).upper() + ":" + (p[2:] or "/")
    p = p.replace("/", "\\")
    drive, rest = ntpath.splitdrive(p)
    if not (drive and rest.startswith("\\")):
        base = norm_path(cwd if cwd is not None else os.getcwd(), windows=True)
        if not drive and rest.startswith("\\"):
            p = ntpath.splitdrive(base)[0] + rest          # rooted, no drive: cwd's drive
        else:
            p = ntpath.join(base, p)
    p = ntpath.normpath(p)
    drive, rest = ntpath.splitdrive(p)
    if not drive or not rest.startswith("\\"):
        raise ForksError("cannot normalise path", path=p)
    if len(drive) == 2 and drive[1] == ":":
        p = drive.upper() + rest
    return p


def state_dir_for(run: str) -> str:
    """`reports/<x>-run.md` -> `reports/<x>-run-forks` (a sibling of the run log)."""
    run_n = norm_path(run)
    if not run_n.lower().endswith(".md"):
        raise ForksError("--run must name the run log .md file", run=run_n)
    return run_n[:-3] + "-forks"


STATE_SUBDIRS = ("jobs", "briefs", "notes", "results", "logs", "raw")
STATE_GITIGNORE = "raw/\n*.lock\n"
MCP_EMPTY = {"mcpServers": {}}


def registry_path(state: str) -> str:
    return os.path.join(state, "registry.json")


def lock_path(state: str) -> str:
    return os.path.join(state, "registry.lock")


def ledger_path(state: str) -> str:
    return os.path.join(state, "ledger.jsonl")


def mcp_empty_path(state: str) -> str:
    return os.path.join(state, "mcp-empty.json")


def ensure_state(state: str) -> None:
    """Create the state dir and its fixed files. Idempotent."""
    os.makedirs(state, exist_ok=True)
    for d in STATE_SUBDIRS:
        os.makedirs(os.path.join(state, d), exist_ok=True)
    gi = os.path.join(state, ".gitignore")
    if not os.path.exists(gi):
        atomic_write_bytes(gi, STATE_GITIGNORE.encode("utf-8"))
    mcp = mcp_empty_path(state)
    if not os.path.exists(mcp):
        atomic_write_bytes(mcp, (json.dumps(MCP_EMPTY) + "\n").encode("utf-8"))


# --------------------------------------------------------------------------------------
# Atomic files, retries, the registry lock (C3)
# --------------------------------------------------------------------------------------


def _retry(fn, tries: int = RETRY_TRIES, sleep_s: float = RETRY_SLEEP_S):
    """Call fn(); retry PermissionError (Windows sharing violations). Returns (value, attempts)."""
    for attempt in range(1, tries + 1):
        try:
            return fn(), attempt
        except PermissionError:
            if attempt == tries:
                raise
            time.sleep(sleep_s)
    raise AssertionError("unreachable")


def atomic_write_bytes(path: str, data: bytes) -> int:
    """Write to a temp file in the same dir, then `os.replace` with retries. Returns attempts."""
    d = os.path.dirname(path) or "."
    tmp = os.path.join(d, ".%s.%d.%s.tmp" % (os.path.basename(path), os.getpid(), uuid.uuid4().hex[:8]))
    with open(tmp, "wb") as f:
        f.write(data)
        f.flush()
        os.fsync(f.fileno())
    try:
        _, attempts = _retry(lambda: os.replace(tmp, path))
    except BaseException:
        with contextlib.suppress(OSError):
            os.remove(tmp)
        raise
    return attempts


def write_json_atomic(path: str, obj) -> int:
    return atomic_write_bytes(path, (json.dumps(obj, indent=1, sort_keys=True) + "\n").encode("utf-8"))


def read_bytes_retry(path: str) -> bytes:
    def _read():
        with open(path, "rb") as f:
            return f.read()
    return _retry(_read)[0]


def read_json(path: str):
    """Read a JSON file with PermissionError retries. Returns None if it does not exist."""
    try:
        data = read_bytes_retry(path)
    except FileNotFoundError:
        return None
    try:
        return json.loads(data.decode("utf-8"))
    except ValueError as e:
        raise ForksError("corrupt state file", path=path, detail=str(e))


class RegistryLock:
    """`registry.lock`: create-exclusive. A waiter breaks a lock only when it is stale (its holder's
    pid + creation time is dead, or it is older than LOCK_ABANDONED_S), and only atomically: it
    renames the lock aside and keeps it only if the renamed content is the one it judged stale;
    otherwise it puts it back. So a waiter never deletes a fresh lock."""

    def __init__(self, state: str, timeout: float = LOCK_WAIT_S):
        self.path = lock_path(state)
        self.timeout = timeout
        self.token = uuid.uuid4().hex
        self.held = False
        self.broke_stale = False

    def _content(self) -> bytes:
        return json.dumps({"pid": os.getpid(), "ctime": proc_ctime(os.getpid()),
                           "at": _now(), "token": self.token}).encode("utf-8")

    def _try_create(self) -> bool:
        flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0)
        try:
            fd = os.open(self.path, flags)
        except FileExistsError:
            return False
        except PermissionError:      # Windows: the old lock is being deleted
            return False
        try:
            os.write(fd, self._content())
        finally:
            os.close(fd)
        return True

    def _read_lock(self):
        """(raw bytes, age in seconds, holder dict or None) of the current lock, or
        (None, None, None) if it is gone or unreadable right now."""
        try:
            raw = read_bytes_retry(self.path)
            mtime = os.path.getmtime(self.path)
        except (FileNotFoundError, PermissionError):
            return None, None, None
        holder, at = None, None
        try:
            holder = json.loads(raw.decode("utf-8"))
            at = float(holder.get("at"))
        except (ValueError, TypeError, AttributeError):
            pass
        if at is None:
            at = mtime
        return raw, _now() - at, holder if isinstance(holder, dict) else None

    @staticmethod
    def _is_stale(holder, age) -> bool:
        if age is not None and age > LOCK_ABANDONED_S:
            return True
        if holder and holder.get("pid"):
            try:
                return not pid_alive(int(holder["pid"]), holder.get("ctime"))
            except (TypeError, ValueError):
                return False
        return False

    def _restore(self, aside: str) -> bool:
        """Put a lock renamed aside back, but never over a lock that appeared meanwhile."""
        try:
            if IS_WINDOWS:
                _retry(lambda: os.rename(aside, self.path))   # Windows rename never overwrites
            else:
                os.link(aside, self.path)                     # fails if the path exists
                os.remove(aside)
            return True
        except OSError:
            return False

    def _break(self, raw: bytes) -> bool:
        """Remove the lock atomically if it still is `raw` (the content judged stale)."""
        aside = os.path.join(os.path.dirname(self.path),
                             "registry.%s.broken.lock" % uuid.uuid4().hex[:12])
        try:
            os.rename(self.path, aside)       # atomic: we now own whatever lock was there
        except OSError:                       # gone already, or busy: judge again later
            return False
        try:
            got = read_bytes_retry(aside)
        except OSError:
            got = None
        if got == raw:
            with contextlib.suppress(OSError):
                _retry(lambda: os.remove(aside))
            return True
        if not self._restore(aside):
            _diag("registry lock changed while being broken and could not be put back (left at %s)"
                  % aside)
        return False

    def acquire(self) -> "RegistryLock":
        deadline = time.monotonic() + self.timeout
        while True:
            if self._try_create():
                self.held = True
                return self
            raw, age, holder = self._read_lock()
            if raw is not None and self._is_stale(holder, age) and self._break(raw):
                self.broke_stale = True
                _diag("broke stale registry lock (age %.1f s, holder pid %s)"
                      % (age, (holder or {}).get("pid")))
                continue                      # the path is free: try to create at once
            if time.monotonic() >= deadline:
                raise ForksError("registry locked", lock=self.path, waited_s=self.timeout)
            time.sleep(LOCK_POLL_S)

    def release(self) -> None:
        if not self.held:
            return
        self.held = False
        try:
            raw = read_bytes_retry(self.path)
        except FileNotFoundError:
            return
        try:
            mine = json.loads(raw.decode("utf-8")).get("token") == self.token
        except (ValueError, AttributeError):
            mine = False
        if mine:
            with contextlib.suppress(FileNotFoundError):
                _retry(lambda: os.remove(self.path))
        else:
            _diag("registry lock was taken over while held; not removing it")

    def __enter__(self):
        return self.acquire()

    def __exit__(self, *exc):
        self.release()
        return False


def new_registry() -> dict:
    return {"schema": SCHEMA, "config": None, "bases": {}, "jobs": {}, "run": {"created_at": _iso()}}


def read_registry(state: str) -> dict | None:
    reg = read_json(registry_path(state))
    if reg is None:
        return None
    if not isinstance(reg, dict) or reg.get("schema") != SCHEMA:
        raise ForksError("unsupported registry schema", path=registry_path(state),
                         schema=(reg or {}).get("schema") if isinstance(reg, dict) else None,
                         expected=SCHEMA)
    return reg


def write_registry(state: str, reg: dict) -> int:
    """Caller must hold RegistryLock(state)."""
    reg["schema"] = SCHEMA
    return write_json_atomic(registry_path(state), reg)


def _ledger_jobs(state: str) -> tuple[set, bool]:
    """(job ids that already have a ledger line, whether the file ends without a newline)."""
    try:
        data = read_bytes_retry(ledger_path(state))
    except FileNotFoundError:
        return set(), False
    jobs = set()
    for ln in data.splitlines():
        try:
            obj = json.loads(ln.decode("utf-8", "replace"))
        except ValueError:
            continue
        if isinstance(obj, dict) and obj.get("job"):
            jobs.add(obj["job"])
    return jobs, bool(data) and not data.endswith(b"\n")


def ledger_append(state: str, line: dict) -> bool:
    """Append one JSON line, at most one per job (M1: skipped if the job already has one).
    Caller must hold RegistryLock(state) (C3c). Returns True if a line was written."""
    have, torn = _ledger_jobs(state)
    if line.get("job") and line["job"] in have:
        _diag("ledger already has a line for %s; not appending another" % line["job"])
        return False
    data = (("\n" if torn else "") + json.dumps(line, sort_keys=True) + "\n").encode("utf-8")

    def _append():
        with open(ledger_path(state), "ab") as f:
            f.write(data)
    _retry(_append)
    return True


# --------------------------------------------------------------------------------------
# Processes: pid + creation time (A19), subprocess helpers, git (C7)
# --------------------------------------------------------------------------------------

if IS_WINDOWS:
    import ctypes
    import ctypes.wintypes as _wt

    _k32 = ctypes.WinDLL("kernel32", use_last_error=True)
    _k32.OpenProcess.argtypes = [_wt.DWORD, _wt.BOOL, _wt.DWORD]
    _k32.OpenProcess.restype = _wt.HANDLE
    _k32.GetProcessTimes.argtypes = [_wt.HANDLE] + [ctypes.POINTER(_wt.FILETIME)] * 4
    _k32.GetProcessTimes.restype = _wt.BOOL
    _k32.GetExitCodeProcess.argtypes = [_wt.HANDLE, ctypes.POINTER(_wt.DWORD)]
    _k32.GetExitCodeProcess.restype = _wt.BOOL
    _k32.CloseHandle.argtypes = [_wt.HANDLE]
    _k32.CloseHandle.restype = _wt.BOOL
    _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
    _PROCESS_TERMINATE = 0x0001
    _STILL_ACTIVE = 259
    _ERROR_ACCESS_DENIED = 5
    _TH32CS_SNAPPROCESS = 0x00000002
    _INVALID_HANDLE = _wt.HANDLE(-1).value

    class _PE32(ctypes.Structure):
        _fields_ = [("dwSize", _wt.DWORD), ("cntUsage", _wt.DWORD), ("th32ProcessID", _wt.DWORD),
                    ("th32DefaultHeapID", ctypes.c_size_t), ("th32ModuleID", _wt.DWORD),
                    ("cntThreads", _wt.DWORD), ("th32ParentProcessID", _wt.DWORD),
                    ("pcPriClassBase", _wt.LONG), ("dwFlags", _wt.DWORD),
                    ("szExeFile", _wt.WCHAR * 260)]

    _k32.CreateToolhelp32Snapshot.argtypes = [_wt.DWORD, _wt.DWORD]
    _k32.CreateToolhelp32Snapshot.restype = _wt.HANDLE
    _k32.Process32FirstW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
    _k32.Process32FirstW.restype = _wt.BOOL
    _k32.Process32NextW.argtypes = [_wt.HANDLE, ctypes.POINTER(_PE32)]
    _k32.Process32NextW.restype = _wt.BOOL
    _k32.TerminateProcess.argtypes = [_wt.HANDLE, _wt.UINT]
    _k32.TerminateProcess.restype = _wt.BOOL


def proc_ctime(pid: int) -> int | None:
    """Process creation time: Windows FILETIME ticks; POSIX /proc start time; else None."""
    if IS_WINDOWS:
        h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, int(pid))
        if not h:
            return None
        try:
            c, e, k, u = _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME(), _wt.FILETIME()
            if not _k32.GetProcessTimes(h, ctypes.byref(c), ctypes.byref(e), ctypes.byref(k),
                                        ctypes.byref(u)):
                return None
            return (c.dwHighDateTime << 32) | c.dwLowDateTime
        finally:
            _k32.CloseHandle(h)
    try:
        with open("/proc/%d/stat" % int(pid), "rb") as f:
            stat = f.read().decode("ascii", "replace")
        return int(stat[stat.rindex(")") + 2:].split()[19])
    except (OSError, ValueError, IndexError):
        return None


_FILETIME_EPOCH = 116444736000000000      # 1970-01-01 in FILETIME ticks (100 ns since 1601)


def epoch_to_ctime(t: float | None) -> int | None:
    """Epoch seconds in the units of `proc_ctime` (Windows FILETIME ticks; POSIX clock ticks since
    boot), or None when the conversion is unknown here."""
    if t is None:
        return None
    if IS_WINDOWS:
        return int(float(t) * 10_000_000) + _FILETIME_EPOCH
    try:
        with open("/proc/stat", "rb") as f:
            btime = next(int(ln.split()[1]) for ln in f.read().splitlines() if ln.startswith(b"btime "))
        return int((float(t) - btime) * os.sysconf("SC_CLK_TCK"))
    except (OSError, ValueError, StopIteration, IndexError, AttributeError):
        return None


def pid_record(pid: int) -> dict:
    """{"pid", "ctime"}: the pair every stored pid is compared by."""
    return {"pid": int(pid), "ctime": proc_ctime(pid)}


def pid_alive(pid: int, ctime: int | None = None) -> bool:
    """True if `pid` runs and (when both are known) its creation time equals `ctime`."""
    if pid is None or int(pid) <= 0:
        return False
    pid = int(pid)
    if IS_WINDOWS:
        h = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not h:
            # Exists but not openable: alive, creation time unverifiable.
            return ctypes.get_last_error() == _ERROR_ACCESS_DENIED
        try:
            code = _wt.DWORD()
            if not _k32.GetExitCodeProcess(h, ctypes.byref(code)) or code.value != _STILL_ACTIVE:
                return False
        finally:
            _k32.CloseHandle(h)
    else:
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            return False
        except PermissionError:
            pass
    if ctime is not None:
        now_ct = proc_ctime(pid)
        if now_ct is not None and now_ct != ctime:
            return False
    return True


def record_alive(rec: dict | None) -> bool:
    return bool(rec) and pid_alive(rec.get("pid"), rec.get("ctime"))


def process_table() -> list[tuple[int, int, str]]:
    """[(pid, ppid, name)] of every process. Windows: one Toolhelp32 snapshot (fast, no PowerShell)."""
    out = []
    if IS_WINDOWS:
        snap = _k32.CreateToolhelp32Snapshot(_TH32CS_SNAPPROCESS, 0)
        if not snap or snap == _INVALID_HANDLE:
            return out
        try:
            e = _PE32()
            e.dwSize = ctypes.sizeof(_PE32)
            ok = _k32.Process32FirstW(snap, ctypes.byref(e))
            while ok:
                out.append((int(e.th32ProcessID), int(e.th32ParentProcessID), e.szExeFile))
                ok = _k32.Process32NextW(snap, ctypes.byref(e))
        finally:
            _k32.CloseHandle(snap)
        return out
    try:
        names = os.listdir("/proc")
    except OSError:
        return out
    for n in names:
        if not n.isdigit():
            continue
        try:
            with open("/proc/%s/stat" % n, "rb") as f:
                st = f.read().decode("ascii", "replace")
            comm = st[st.index("(") + 1: st.rindex(")")]
            ppid = int(st[st.rindex(")") + 2:].split()[1])
            out.append((int(n), ppid, comm))
        except (OSError, ValueError, IndexError):
            continue
    return out


ALIVE_UNTIL_SLACK_S = 1.0   # job-file times are ISO with whole seconds: allow the truncated part


def process_tree(rec: dict | None, alive_until: float | None = None) -> list[dict]:
    """`rec` (if alive) and all its descendants, each {pid, ctime, ppid, name}. A process counts as
    a child only if it was created at or after its parent (guards against pid reuse).

    Dead root (I2): its pid may have been reused by a later process whose own children now carry
    the same stale parent pid. Its direct children are therefore taken only if they were created
    after the root's recorded creation time AND before `alive_until` (epoch seconds: the last
    time the job was known alive, e.g. the job file's `heartbeat_at` / `ended_at`). Without a
    recorded creation time or `alive_until`, a dead root claims nothing. Below a claimed (live,
    verified) child the normal rule applies."""
    if not rec or not rec.get("pid"):
        return []
    kids, names = {}, {}
    for pid, ppid, name in process_table():
        names[pid] = name
        if pid != ppid:
            kids.setdefault(ppid, []).append(pid)
    root = int(rec["pid"])
    root_ct = rec.get("ctime")
    out = []
    if record_alive(rec):
        out.append({"pid": root, "ctime": root_ct, "ppid": None, "name": names.get(root)})
        dead_bounds = None
    else:
        upper = epoch_to_ctime(None if alive_until is None else alive_until + ALIVE_UNTIL_SLACK_S)
        if root_ct is None or upper is None:
            return []
        dead_bounds = (int(root_ct), upper)
    stack, seen = [(root, root_ct)], {root}
    while stack:
        ppid, pct = stack.pop()
        is_dead_root = ppid == root and dead_bounds is not None
        if not is_dead_root:
            now_ct = proc_ctime(ppid)
            if pct is not None and now_ct is not None and now_ct != pct:
                continue      # this pid now belongs to another process: its children are not ours
        for c in kids.get(ppid, []):
            if c in seen:
                continue
            ct = proc_ctime(c)
            if is_dead_root:
                lo, hi = dead_bounds
                if ct is None or not (lo < ct <= hi):
                    continue  # created before the job, or after it was last known alive: not ours
            elif pct is not None and ct is not None and ct < pct:
                continue
            seen.add(c)
            out.append({"pid": c, "ctime": ct, "ppid": ppid, "name": names.get(c)})
            stack.append((c, ct))
    return out


def terminate(rec: dict) -> bool:
    """Kill one process identified by pid + creation time. True if it is gone or was signalled."""
    if not record_alive(rec):
        return True
    pid = int(rec["pid"])
    if IS_WINDOWS:
        h = _k32.OpenProcess(_PROCESS_TERMINATE | _PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not h:
            return False
        try:
            return bool(_k32.TerminateProcess(h, 1))
        finally:
            _k32.CloseHandle(h)
    try:
        os.kill(pid, signal.SIGKILL)
        return True
    except OSError:
        return False


def _taskkill() -> str:
    p = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "taskkill.exe")
    return p if os.path.isfile(p) else "taskkill"


def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> dict:
    """Snapshot the tree of `rec` (plus the trees of `extra` records), kill it (`taskkill /T /F` on
    Windows, the process group on POSIX), sweep every snapshot pid still alive, and verify.
    `alive_until` bounds the descendants of a dead root (see `process_tree`).
    Returns {"snapshot", "taskkill_rc", "survivors"}; pids are always checked with their ctime."""
    roots = [r for r in [rec, *extra] if r and r.get("pid")]
    snap, seen = [], set()
    for r in roots:
        for p in process_tree(r, alive_until):
            key = (p["pid"], p["ctime"])
            if key not in seen:
                seen.add(key)
                snap.append(p)
    rc = None
    for r in roots:
        if not record_alive(r):
            continue
        if IS_WINDOWS:
            try:
                rc = _run([_taskkill(), "/T", "/F", "/PID", str(int(r["pid"]))], timeout=60).returncode
            except (OSError, subprocess.SubprocessError) as e:
                _diag("taskkill failed: %s" % e)
        else:
            with contextlib.suppress(OSError):
                os.killpg(int(r["pid"]), signal.SIGKILL)
        break   # one tree kill of the first live root; the sweep covers the rest
    deadline = time.monotonic() + KILL_VERIFY_S
    while True:
        survivors = [p for p in snap if record_alive(p)]
        if not survivors or time.monotonic() >= deadline:
            break
        for p in survivors:
            terminate(p)
        time.sleep(0.1)
    return {"snapshot": snap, "taskkill_rc": rc, "survivors": survivors}


def _run(args, cwd=None, timeout=60, env=None) -> subprocess.CompletedProcess:
    """Run a short helper process: stdout captured (never leaks to ours), no window."""
    return subprocess.run(args, cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True,
                          timeout=timeout, creationflags=_NOWIN)


def git(args, cwd=None, timeout=60) -> subprocess.CompletedProcess:
    """Every helper git call goes through here: `git --no-optional-locks …` (C7)."""
    return _run(["git", "--no-optional-locks", *args], cwd=cwd, timeout=timeout)


def git_toplevel(cwd: str | None = None) -> str | None:
    try:
        r = git(["rev-parse", "--show-toplevel"], cwd=cwd)
    except (OSError, subprocess.SubprocessError):
        return None
    if r.returncode != 0:
        return None
    out = r.stdout.decode("utf-8", "replace").strip()
    return norm_path(out) if out else None


# --------------------------------------------------------------------------------------
# Env scrub
# --------------------------------------------------------------------------------------


def scrub_class(name: str) -> str | None:
    """'drop', 'pinned', 'keep' or None (not on any list)."""
    n = name.upper()
    if n in SCRUB_DROP or any(p.match(n) for p in SCRUB_DROP_PATTERNS):
        return "drop"
    if n in SCRUB_PINNED or n == SCRUB_PINNED_BINARY:
        return "pinned"
    if n in SCRUB_KEEP:
        return "keep"
    return None


def _venv_prefixes(environ) -> list[str]:
    out = []
    venv = environ.get("VIRTUAL_ENV")
    if venv:
        out.append(venv)
    if os.path.normcase(os.path.abspath(sys.prefix)) != os.path.normcase(os.path.abspath(sys.base_prefix)):
        out.append(sys.prefix)
    return [os.path.normcase(os.path.normpath(os.path.abspath(x))) for x in out]


def _under(path: str, prefixes) -> bool:
    try:
        p = os.path.normcase(os.path.normpath(os.path.abspath(path)))
    except ValueError:
        return False
    return any(p == px or p.startswith(px.rstrip("\\/") + os.sep) for px in prefixes)


def scrubbed_env(binary: str, environ=None) -> dict:
    """The child env for `claude`: host env minus the drop list, venv PATH entries removed,
    pinned vars on top, `CLAUDE_CODE_EXECPATH` = the stored absolute binary path."""
    if environ is None:
        environ = os.environ
    prefixes = _venv_prefixes(environ)
    env = {}
    for k, v in environ.items():
        cls = scrub_class(k)
        if cls in ("drop", "pinned"):
            continue
        if k.upper() == "PATH":
            v = os.pathsep.join(e for e in v.split(os.pathsep) if e and not _under(e, prefixes))
        env[k] = v
    env.update(SCRUB_PINNED)
    env[SCRUB_PINNED_BINARY] = binary
    return env


def scrub_unknown(environ=None) -> list[str]:
    """Host CLAUDE_* / ANTHROPIC_* vars on neither list (kept, but worth a look)."""
    if environ is None:
        environ = os.environ
    return sorted(k for k in environ
                  if k.upper().startswith(SCRUB_WATCH_PREFIXES) and scrub_class(k) is None)


def scrub_report(environ=None) -> dict:
    return {
        "drop": sorted(SCRUB_DROP),
        "drop_patterns": [p.pattern for p in SCRUB_DROP_PATTERNS],
        "drop_path_entries": "under VIRTUAL_ENV, or under sys.prefix when it differs from sys.base_prefix",
        "keep": sorted(SCRUB_KEEP),
        "keep_rule": "everything not dropped is kept",
        "pinned": dict(SCRUB_PINNED, **{SCRUB_PINNED_BINARY: "<stored binary path>"}),
        "unknown": scrub_unknown(environ),
    }


# --------------------------------------------------------------------------------------
# Binary, uv, python
# --------------------------------------------------------------------------------------


def resolve_binary(explicit: str | None = None, environ=None) -> tuple[str | None, str | None]:
    """(absolute path, how): `--binary`, else $CLAUDE_CODE_EXECPATH, else PATH `claude`."""
    if environ is None:
        environ = os.environ
    if explicit:
        return norm_path(explicit), "--binary"
    ep = environ.get("CLAUDE_CODE_EXECPATH")
    if ep:
        return norm_path(ep), "CLAUDE_CODE_EXECPATH"
    w = shutil.which("claude")
    if w:
        return norm_path(w), "PATH"
    return None, None


def _binary_version(path: str) -> str | None:
    try:
        r = _run([path, "--version"], timeout=60)
    except (OSError, subprocess.SubprocessError):
        return None
    toks = r.stdout.decode("utf-8", "replace").split()
    return toks[0] if r.returncode == 0 and toks else None


def binary_info(path: str) -> dict:
    """{"path", "version", "size", "mtime"} of an existing binary; raises ForksError otherwise."""
    if not path or not os.path.isfile(path):
        raise ForksError("claude binary not found", binary=path)
    st = os.stat(path)
    version = _binary_version(path)
    if not version:
        raise ForksError("claude binary did not report a version", binary=path)
    return {"path": path, "version": version, "size": st.st_size, "mtime": st.st_mtime}


def _uv_report(environ=None) -> dict:
    if environ is None:
        environ = os.environ
    out = {"path": None, "version": None}
    w = shutil.which("uv")
    if w:
        out["path"] = norm_path(w)
        try:
            r = _run([w, "--version"], timeout=30)
            toks = r.stdout.decode("utf-8", "replace").split()
            out["version"] = toks[1] if len(toks) > 1 else (toks[0] if toks else None)
        except (OSError, subprocess.SubprocessError) as e:
            out["error"] = str(e)
    if environ.get("SCRIPTS_UV"):
        out["SCRIPTS_UV"] = environ["SCRIPTS_UV"]
    return out


def _git_report() -> dict:
    out = {"path": None, "version": None}
    w = shutil.which("git")
    if w:
        out["path"] = norm_path(w)
        try:
            r = git(["--version"], timeout=30)
            toks = r.stdout.decode("utf-8", "replace").split()
            out["version"] = toks[2] if len(toks) > 2 else None
        except (OSError, subprocess.SubprocessError) as e:
            out["error"] = str(e)
    return out


def script_path() -> str:
    return norm_path(os.path.abspath(__file__))


def preamble_path() -> str:
    return norm_path(os.path.join(os.path.dirname(os.path.abspath(__file__)), "preamble.md"))


def supervisor_interpreter() -> str:
    """C6: the base interpreter when `sys.executable` is a venv launcher."""
    base = getattr(sys, "_base_executable", None)
    return base if base and base != sys.executable else sys.executable


# --------------------------------------------------------------------------------------
# C8: availableModels admission
# --------------------------------------------------------------------------------------


def user_settings_path() -> str:
    if USER_SETTINGS_PATH:
        return USER_SETTINGS_PATH
    return os.path.join(os.path.expanduser("~"), ".claude", "settings.json")


def settings_files(cwd: str) -> list[str]:
    return [user_settings_path(),
            os.path.join(cwd, ".claude", "settings.json"),
            os.path.join(cwd, ".claude", "settings.local.json")]


def model_family(model: str) -> str | None:
    toks = model.replace(ONE_M_SUFFIX, "").split("-")
    return toks[1] if len(toks) > 1 else None


def model_admission(model: str, cwd: str) -> dict:
    """Apply the P0.5 rule. Returns {"admitted", "lists": [{"file", "availableModels"}], "via"}."""
    lists = []
    for f in settings_files(cwd):
        try:
            data = read_json(f)
        except ForksError as e:
            _diag("ignoring unreadable settings file %s (%s)" % (f, e.payload.get("detail")))
            continue
        except OSError:
            continue
        if isinstance(data, dict) and isinstance(data.get("availableModels"), list):
            lists.append({"file": f, "availableModels": [str(x) for x in data["availableModels"]]})
    if not lists:
        return {"admitted": True, "lists": [], "via": "no availableModels list"}
    candidates = [("exact id", model)]
    if model.endswith(ONE_M_SUFFIX):
        candidates.append(("id without [1m]", model[: -len(ONE_M_SUFFIX)]))
    fam = model_family(model)
    if fam:
        candidates.append(("family alias", fam))
    for via, cand in candidates:
        for entry in lists:
            if cand in entry["availableModels"]:
                return {"admitted": True, "lists": lists, "via": "%s %r in %s" % (via, cand, entry["file"])}
    return {"admitted": False, "lists": lists, "via": None}


# --------------------------------------------------------------------------------------
# Stream parsing, usage rule (P0.3), cache verdict (C5), warmth (C1)
# --------------------------------------------------------------------------------------


class StreamState:
    """What one `claude -p --output-format stream-json --verbose` stream told us so far."""

    def __init__(self):
        self.init = None
        self.session_id = None
        self.msg_ids: list = []
        self._ids: set = set()
        self.first_usage = None       # usage of the first assistant event (cache verdict)
        self.first_at = None          # its arrival time (last_hit)
        self.last_usage = None        # usage of the last assistant event (base context)
        self.last_assistant_at = None
        self.last_event_at = None
        self.result = None
        self.refused = None           # content of a model-restricted warning
        self.events = 0

    def feed(self, ev: dict, t: float | None) -> None:
        if not isinstance(ev, dict):
            return
        self.events += 1
        if t is not None:
            self.last_event_at = t
        ty = ev.get("type")
        if ty == "system":
            st = ev.get("subtype")
            if st == "init" and self.init is None:
                self.init = ev
                self.session_id = ev.get("session_id") or self.session_id
            elif (st == "informational" and ev.get("level") == "warning"
                  and "is restricted" in str(ev.get("content", ""))):
                self.refused = str(ev.get("content"))
        elif ty == "assistant":
            m = ev.get("message") or {}
            mid = m.get("id")
            if mid not in self._ids:
                self._ids.add(mid)
                self.msg_ids.append(mid)
            u = m.get("usage") or {}
            if self.first_usage is None:
                self.first_usage, self.first_at = u, t
            self.last_usage, self.last_assistant_at = u, t
        elif ty == "result":
            self.result = ev
            self.session_id = self.session_id or ev.get("session_id")

    @property
    def turns(self) -> int:
        return len(self.msg_ids)

    @classmethod
    def from_file(cls, path: str) -> "StreamState":
        """Re-parse a raw stream file (event times unknown)."""
        ss = cls()
        try:
            data = read_bytes_retry(path)
        except OSError:
            return ss
        for line in data.splitlines():
            try:
                ss.feed(json.loads(line.decode("utf-8", "replace")), None)
            except ValueError:
                continue
        return ss


def context_tokens(usage: dict | None) -> int:
    """Prompt-side tokens of one request: input + cache creation + cache read."""
    if not usage:
        return 0
    return int(usage.get("input_tokens") or 0) + int(usage.get("cache_creation_input_tokens") or 0) \
        + int(usage.get("cache_read_input_tokens") or 0)


def runtime_fingerprint(init: dict | None) -> dict | None:
    """The runtime fields of a `system/init` event that must match the base's."""
    if not init:
        return None
    tools = sorted(str(x) for x in (init.get("tools") or []))
    plugins = sorted("%s@%s" % (p.get("name"), p.get("version") or "")
                     for p in (init.get("plugins") or []) if isinstance(p, dict))
    agents = sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (init.get("agents") or []))
    return {"tools": _sha256_bytes(canonical_json(tools).encode("utf-8")),
            "permissionMode": init.get("permissionMode"), "plugins": plugins, "agents": agents,
            "apiKeySource": init.get("apiKeySource")}


def runtime_drift(rt: dict | None, ref: dict | None, launch: dict) -> list[str]:
    if not rt:
        return []
    out = []
    if ref:
        out += [k for k in sorted(rt) if rt.get(k) != ref.get(k)]
    elif rt.get("permissionMode") not in (None, launch.get("permission_mode")):
        out.append("permissionMode")
    return out


def usage_from_result(result: dict, before_mu: dict, cost_before: float) -> dict:
    """P0.3: the four columns and the 1h/5m split from `result.usage`; `models_used` = modelUsage
    keys that grew against the session total before the call; cost delta against the session's
    total before the call; `mismatch` when `result.usage` differs from the modelUsage delta."""
    usage = result.get("usage") or {}
    cc = usage.get("cache_creation") or {}
    tokens = {
        "input": int(usage.get("input_tokens") or 0),
        "output": int(usage.get("output_tokens") or 0),
        "cache_creation": int(usage.get("cache_creation_input_tokens") or 0),
        "cache_creation_1h": int(cc.get("ephemeral_1h_input_tokens") or 0),
        "cache_creation_5m": int(cc.get("ephemeral_5m_input_tokens") or 0),
        "cache_read": int(usage.get("cache_read_input_tokens") or 0),
    }
    after = result.get("modelUsage") or {}
    before_mu = before_mu or {}
    delta = {}
    for m, v in after.items():
        b = before_mu.get(m) or {}
        delta[m] = {f: int((v or {}).get(f) or 0) - int(b.get(f) or 0) for f in MU_FIELDS}
    models_used = sorted(m for m, d in delta.items() if any(x > 0 for x in d.values()))
    sums = {f: sum(d[f] for d in delta.values()) for f in MU_FIELDS}
    mismatch = (sums["inputTokens"], sums["outputTokens"], sums["cacheCreationInputTokens"],
                sums["cacheReadInputTokens"]) != (tokens["input"], tokens["output"],
                                                   tokens["cache_creation"], tokens["cache_read"])
    total = result.get("total_cost_usd")
    cost_delta = None if total is None else round(float(total) - float(cost_before or 0.0), 6)
    return {"tokens": tokens, "models_used": models_used, "mismatch": mismatch,
            "cost_delta": cost_delta, "total": total, "model_usage": after}


_API_ERROR_TEXT = (   # checked in this order, after the status code
    ("prompt_too_long", re.compile(r"prompt is too long|prompt too long|prompt_too_long", re.I)),
    ("usage_limit", re.compile(r"usage limit|rate[ _-]?limit|hit your limit|api error:?\s*429", re.I)),
    ("overloaded", re.compile(r"overloaded|api error:?\s*529", re.I)),
    ("api_error", re.compile(r"api error", re.I)),
)


def api_error_reason(res: dict | None) -> tuple[str, str] | None:
    """(reason, detail) for a result that carries an API error, else None (I1). `reason` is
    `usage_limit` (429 / rate or usage limit), `overloaded` (529), `prompt_too_long`, or
    `api_error` (any other API error status); `detail` holds the status and a short text."""
    if not res:
        return None
    status = res.get("api_error_status")
    try:
        status = int(status) if status not in (None, "") else None
    except (TypeError, ValueError):
        status = None
    if not res.get("is_error") and status is None:
        return None
    parts = [res.get("result"), res.get("error"), res.get("errors")]
    text = " | ".join(p if isinstance(p, str) else json.dumps(p) for p in parts if p)
    detail = "api_error_status %s: %s" % (status, text[:200] if text else "(no text)")
    if status == 429:
        return "usage_limit", detail
    if status == 529:
        return "overloaded", detail
    for reason, rx in _API_ERROR_TEXT:
        if rx.search(text):
            return reason, detail
    if status is not None:
        return "api_error", detail
    return None


def cache_verdict(first_usage: dict | None, base_context) -> str | None:
    """C5: first assistant event's cache_read against the base's context tokens."""
    if first_usage is None:
        return None
    cr = int(first_usage.get("cache_read_input_tokens") or 0)
    if base_context and cr >= CACHE_HIT_RATIO * float(base_context):
        return "hit"
    return "partial" if cr > 0 else "miss"


def infer_ttl(tokens: dict | None, prev, default=None):
    """Design 5: the TTL a base's cache entries live for, from its most recent call's tokens.
    1h written -> 3600; only 5m written -> 300; otherwise (read only, or nothing) the previous
    known TTL, and with none known `default` (the configured ttl_s)."""
    if not tokens:
        return prev if prev is not None else default
    if int(tokens.get("cache_creation_1h") or 0) > 0:
        return TTL_1H
    if int(tokens.get("cache_creation_5m") or 0) > 0:
        return TTL_5M
    return prev if prev is not None else default


def wrote_only_5m(tokens: dict | None) -> bool:
    """`ttl_downgrade`: the call wrote 5-minute cache entries and no 1-hour ones."""
    return bool(tokens) and int(tokens.get("cache_creation_5m") or 0) > 0 \
        and int(tokens.get("cache_creation_1h") or 0) == 0


def break_even(ttl) -> int:
    """floor((write - read) / read): pings pay while fewer are needed (19 at 1 h, 11 at 5 min)."""
    write = PRICE_RATIOS["write_1h"] if float(ttl) > TTL_5M else PRICE_RATIOS["write_5m"]
    return (write - PRICE_RATIOS["read"]) // PRICE_RATIOS["read"]


def base_warmth(base: dict, cfg: dict, now: float | None = None) -> dict:
    """Design 5: ttl = the base's inferred TTL, else the configured ttl_s; warm_until = last_hit +
    ttl; margin = init's warm_margin_s if set, else max(120, ttl/10)."""
    ttl = float((base or {}).get("ttl_s") or cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"])
    wm = cfg.get("warm_margin_s")
    margin = float(wm) if wm is not None else max(120.0, ttl / 10.0)
    out = {"ttl_s": int(ttl) if ttl.is_integer() else ttl,
           "margin_s": int(margin) if margin.is_integer() else margin}
    last = _ts((base or {}).get("last_hit"))
    if last is None:
        return {"base_state": "cold", "warm_until": None, **out}
    now = _now() if now is None else now
    wu = last + ttl
    state = "warm" if now < wu - margin else ("expiring" if now < wu else "cold")
    return {"base_state": state, "warm_until": _iso(wu), **out}


# --------------------------------------------------------------------------------------
# Jobs: paths, templates, spawning, finalising (C3b, C3c)
# --------------------------------------------------------------------------------------


def job_file(state: str, job: str) -> str:
    return os.path.join(state, "jobs", job + ".json")


def result_file(state: str, job: str) -> str:
    return os.path.join(state, "results", job + ".json")


def raw_file(state: str, job: str, ext: str = ".jsonl") -> str:
    return os.path.join(state, "raw", job + ext)


def _safe_name(s: str) -> str:
    return re.sub(r"[^A-Za-z0-9._-]", "_", str(s))


def note_rel(block) -> str | None:
    return None if block in (None, "") else "notes/%s.md" % _safe_name(block)


def _phase(v):
    if v is None:
        return None
    return int(v) if re.fullmatch(r"\d+", str(v)) else str(v)


def render_template(name: str, values: dict) -> str:
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), TEMPLATES_DIR, name)
    with open(path, "r", encoding="utf-8") as f:
        text = f.read()
    return string.Template(text).safe_substitute(values)


def _powershell() -> str:
    w = shutil.which("powershell")
    if w:
        return w
    return os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "WindowsPowerShell",
                        "v1.0", "powershell.exe")


def spawn_supervisor(state: str, run: str, job: str) -> dict:
    """Start `_supervise` detached (C6). Windows: CIM `Win32_Process.Create` with a hidden window,
    through a `powershell` that itself has CREATE_NO_WINDOW. POSIX: a new session.
    Returns {"ok", "pid", "ctime", "method"} or {"ok": False, "detail"}."""
    argv = [supervisor_interpreter(), "-I", script_path(), "_supervise", "--run", run, "--job", job]
    if IS_WINDOWS:
        env = dict(os.environ)
        env["SPIRE_FORKS_CMD"] = subprocess.list2cmdline(argv)
        env["SPIRE_FORKS_CWD"] = state
        try:
            r = _run([_powershell(), "-NoProfile", "-NonInteractive", "-Command", POWERSHELL_CIM],
                     env=env, timeout=120)
        except (OSError, subprocess.SubprocessError) as e:
            return {"ok": False, "method": SPAWN_METHOD, "detail": "powershell: %s" % e}
        toks = r.stdout.decode("utf-8", "replace").split()
        try:
            rv, pid = int(toks[-2]), int(toks[-1])
        except (ValueError, IndexError):
            return {"ok": False, "method": SPAWN_METHOD,
                    "detail": "unparsable CIM output: %r / %r" % (r.stdout[-200:], r.stderr[-400:])}
        if rv != 0:
            return {"ok": False, "method": SPAWN_METHOD, "detail": "Win32_Process.Create returned %d" % rv}
        return {"ok": True, "method": SPAWN_METHOD, "pid": pid, "ctime": proc_ctime(pid)}
    log = open(raw_file(state, job, ".sup.log"), "ab")
    try:
        p = subprocess.Popen(argv, cwd=state, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
                             start_new_session=True, close_fds=True)
    except OSError as e:
        return {"ok": False, "method": SPAWN_METHOD, "detail": str(e)}
    finally:
        log.close()
    return {"ok": True, "method": SPAWN_METHOD, "pid": p.pid, "ctime": proc_ctime(p.pid)}


def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: str, flags, reason,
              detail, exit_code, wall_s, base_commit, now: float):
    """Build the result, the ledger line, and apply the registry updates for one finished job.
    `hint`: "exited" (decide ok/failed from the stream), or a final status given by the caller."""
    cfg = reg["config"]
    launch = cfg["launch"]
    kind = entry["kind"]
    base = reg["bases"].get(entry.get("base")) if entry.get("base") else None
    res = ss.result if ss else None
    u = usage_from_result(res, entry.get("model_usage_before") or {}, entry.get("cost_before") or 0.0) \
        if res else None
    flags = list(flags or [])
    models_used = u["models_used"] if u else []
    if hint == "exited":
        requested = entry.get("model_requested")
        # Refused = the restriction warning, or a substitution: some other model billed and the
        # requested one did not. An empty delta (an error before the model answered) is not one.
        substituted = bool(models_used) and requested not in models_used
        api = api_error_reason(res)
        if ss and (ss.refused or (res and substituted)):
            status, reason = "failed", "model_refused"
            detail = ss.refused or "models_used %s, requested %s" % (models_used, requested)
        elif res is None:
            status = "failed"
            reason = "exit_code" if exit_code else "no_result"
            detail = "exit %s, no result event" % exit_code
        elif api:
            status, (reason, detail) = "failed", api
        elif res.get("is_error"):
            status, reason, detail = "failed", "is_error", str(res.get("subtype"))
        elif exit_code:
            status, reason, detail = "failed", "exit_code", "exit %s" % exit_code
        else:
            status, reason, detail = "ok", None, None
        kr = entry.get("kill_requested")
        if kr and status != "ok":
            status, reason, detail = "killed", kr.get("by"), "kill requested"
            if kr.get("by") == "expire":
                flags.append("expired")
    else:
        status = hint
    if u and u["mismatch"]:
        flags.append("usage_mismatch")
    denials = len((res or {}).get("permission_denials") or [])
    if denials:
        flags.append("needs_attention")
    cache = cache_verdict(ss.first_usage if ss else None, entry.get("base_context")) \
        if kind in ("fork", "continue", "ping") else None
    rt = runtime_fingerprint(ss.init) if ss else None
    drift = list(entry.get("launch_drift") or [])
    ref = (base or {}).get("runtime_fingerprint") if kind != "train" else None
    drift += runtime_drift(rt, ref, launch)
    drift = sorted(set(drift))
    fingerprint = "ok" if not drift else "drift:" + ",".join(drift)
    base_state = entry.get("base_state_at_launch")
    if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit")
                                                                   and fingerprint == "ok")):
        flags.append("cold_start")
    if u and wrote_only_5m(u["tokens"]):
        flags.append("ttl_downgrade")
    # A ping refreshes the base unless it missed (P0.6: a `partial` ping still kept the base warm;
    # it re-created about the shared system prefix). A miss or a failed ping leaves last_hit alone.
    ping_refreshed = (kind == "ping" and status == "ok" and cache in ("hit", "partial")
                      and bool(ss and ss.first_at))
    if kind == "ping" and not ping_refreshed:
        flags.append("ping_missed")
    flags = _uniq(flags)
    turns = int(res.get("num_turns") or 0) if res else (ss.turns if ss else 0)
    note = note_rel(entry.get("block")) if kind != "train" else None
    if note and not os.path.isfile(os.path.join(state, note)):
        note = None
    session = (ss.session_id if ss and ss.session_id else None) or entry.get("session")
    tokens = u["tokens"] if u else {k: None for k in TOKEN_KEYS}
    ended = _iso(now)
    result = {
        "job": entry["job"], "kind": kind, "base": entry.get("base"), "block": entry.get("block"),
        "phase": entry.get("phase"), "status": status, "flags": flags, "fingerprint": fingerprint,
        "session": session, "parent_session": entry.get("parent_session"), "models_used": models_used,
        "tokens": tokens, "cost_delta_usd": u["cost_delta"] if u else None, "cache": cache,
        "base_state": base_state, "wall_s": None if wall_s is None else round(float(wall_s), 1),
        "turns": turns, "denials": denials, "note": note, "raw": "raw/%s.jsonl" % entry["job"],
        "paths": entry.get("paths"), "reason": reason, "model_requested": entry.get("model_requested"),
    }
    ledger = {
        "job": entry["job"], "kind": kind, "phase": entry.get("phase"), "block": entry.get("block"),
        "base": entry.get("base"), "session": session, "parent_session": entry.get("parent_session"),
        "started": entry.get("started_at") or entry.get("launched_at") or entry.get("created_at"),
        "ended": ended, "wall_s": result["wall_s"], "binary": entry.get("binary_version"),
        "model_requested": entry.get("model_requested"), "models_used": models_used, "tokens": tokens,
        "cost_delta_usd": result["cost_delta_usd"], "cost_cumulative_usd": u["total"] if u else None,
        "turns": turns, "cache": cache, "base_state": base_state, "fingerprint": fingerprint,
        "denials": denials, "status": status, "flags": flags, "reason": reason, "note": note,
    }
    # registry updates
    entry.update({"status": status, "flags": flags, "reason": reason, "reason_detail": detail,
                  "ended_at": ended, "session": session, "turns": turns, "exit_code": exit_code,
                  "fingerprint": fingerprint, "cache": cache})
    if ss and ss.first_at:
        entry["first_event_at"] = _iso(ss.first_at)
    if u:
        entry["cost_cum_usd"], entry["model_usage_cum"] = u["total"], u["model_usage"]
        holder = entry if kind == "fork" else reg["jobs"].get(entry.get("continues") or "")
        if kind in ("fork", "continue") and holder is not None:
            holder["session_cost_cum_usd"] = u["total"]
            holder["session_model_usage_cum"] = u["model_usage"]
    if base is not None:
        # Design 5: only a request that reads exactly the base prefix refreshes it: a fork's first
        # request, a ping that did not miss (and the training's end, which writes it). A continue
        # reads the fork's own longer prefix, so it leaves the base's last_hit alone (I3).
        hit_t = None
        if ss and kind == "train":
            hit_t = ss.last_assistant_at
        elif ss and kind == "fork":
            hit_t = ss.first_at
        elif ping_refreshed:
            hit_t = ss.first_at
        if hit_t and (_ts(base.get("last_hit")) or 0) < hit_t:
            base["last_hit"] = _iso(hit_t)
        if u:   # design 5: TTL from the most recent call on this base (any kind)
            base["ttl_s"] = infer_ttl(u["tokens"], base.get("ttl_s"))
            base["ttl_from"] = entry["job"]
        if kind == "ping" and base.get("ping_in_flight") == entry["job"]:
            base["ping_in_flight"] = None      # C3(d): cleared when the ping finishes, any status
        if kind == "train" and base.get("train_job") == entry["job"]:
            if status == "ok" and base.get("status") == "training":
                base.update({"status": "active", "session": session, "base_commit": base_commit,
                             "context_tokens": context_tokens(ss.last_usage if ss else None),
                             "training_cost_usd": u["cost_delta"] if u else None,
                             "cost_cum_usd": u["total"] if u else None,
                             "model_usage_cum": u["model_usage"] if u else {},
                             "runtime_fingerprint": rt, "trained_at": ended})
            elif base.get("status") == "training":
                base["status"] = "failed"
    return result, ledger


def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None,
                 detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None:
    """Write `results/<job>.json`, the registry update and the ledger line, exactly once, under the
    lock (C3c). The registry status is written first: it is the once-only guard, so a finaliser
    killed after it leaves no second finaliser to run (M1); the ledger append also skips a job
    that already has a line. Returns the result, or None when the job was already final."""
    with RegistryLock(state):
        reg = read_registry(state)
        entry = (reg or {}).get("jobs", {}).get(job)
        if not entry or entry.get("status") not in RUNNING:
            return None
        result, ledger = _conclude(state, reg, entry, ss, hint, flags, reason, detail, exit_code,
                                   wall_s, base_commit, _now())
        write_json_atomic(result_file(state, job), result)
        write_registry(state, reg)
        ledger_append(state, ledger)
        return result


def bump_last_hit(state: str, base_name: str, t: float) -> None:
    """Record a hit on the base prefix (first assistant event of a fork) as it happens."""
    with RegistryLock(state):
        reg = read_registry(state)
        base = (reg or {}).get("bases", {}).get(base_name)
        if base is None or (_ts(base.get("last_hit")) or 0) >= t:
            return
        base["last_hit"] = _iso(t)
        write_registry(state, reg)


def in_child() -> bool:
    """N3: this call comes from inside a fork (`SPIRE_FORKS_CHILD` is set)."""
    return bool(os.environ.get(CHILD_MARKER))


def stamp_orchestrator(reg: dict, t: float | None = None, wait: bool = False) -> None:
    """N1: set `run.last_orchestrator_call_at` in `reg` (caller holds the lock and writes it);
    `wait=True` also sets `run.last_wait_at` (the ping rule's activity condition, design 5).
    A fork's own calls never stamp, so a fork cannot keep its own watchdog alive (M4)."""
    if in_child():
        return
    run = reg.setdefault("run", {})
    run["last_orchestrator_call_at"] = _iso(t)
    if wait:
        run["last_wait_at"] = run["last_orchestrator_call_at"]


def touch_orchestrator(state: str, wait: bool = False) -> None:
    """N1: record that the orchestrator called us (wait, fork, continue, train, list)."""
    if in_child():
        return
    with RegistryLock(state):
        reg = read_registry(state)
        if reg is None:
            return
        stamp_orchestrator(reg, wait=wait)
        write_registry(state, reg)


def _job_runtime(state: str, job: str) -> dict:
    try:
        return read_json(job_file(state, job)) or {}
    except ForksError:
        return {}


def last_alive_at(rt: dict) -> float | None:
    """The last time a job's supervisor was known alive: the later of its job file's
    `heartbeat_at` and `ended_at` (bounds a dead root's descendants, I2)."""
    ts = [t for t in (_ts(rt.get("heartbeat_at")), _ts(rt.get("ended_at"))) if t is not None]
    return max(ts) if ts else None


def reap_lost(state: str) -> list[str]:
    """Decision 7 / A19: a running job whose heartbeat is older than LOST_AFTER_S and whose
    supervisor (pid + creation time) is dead becomes `lost`. Its claude tree, if any is left,
    is killed with it."""
    reg = read_registry(state)
    lost = []
    now = _now()
    for job, e in sorted((reg or {}).get("jobs", {}).items()):
        if e.get("status") not in RUNNING:
            continue
        rt = _job_runtime(state, job)
        hb = _ts(rt.get("heartbeat_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at"))
        if hb is not None and now - hb <= LOST_AFTER_S:
            continue
        sup = rt.get("supervisor") or e.get("supervisor")
        if sup and record_alive(sup):
            continue
        if rt.get("claude"):
            kill_tree(rt["claude"], alive_until=last_alive_at(rt))
        start = _ts(rt.get("started_at")) or _ts(e.get("launched_at"))
        res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="lost",
                           flags=rt.get("flags") or [], reason="supervisor_dead",
                           detail="heartbeat %s, supervisor %s dead" % (rt.get("heartbeat_at"), sup),
                           wall_s=(hb - start) if (hb and start) else None)
        if res is not None:
            lost.append(job)
    return lost


def kill_job(state: str, job: str, by: str) -> dict:
    """`kill` (by="kill") and `expire` (by="expire"): snapshot the supervisor's tree, taskkill it,
    sweep, verify, finalise as `killed`."""
    with RegistryLock(state):
        reg = read_registry(state)
        entry = reg["jobs"].get(job)
        if entry is None:
            raise ForksError("unknown job", job=job)
        if entry.get("status") not in RUNNING:
            return {"job": job, "killed": False, "status": entry.get("status"), "ok": True}
        entry["kill_requested"] = {"by": by, "at": _iso()}
        write_registry(state, reg)
    rt = _job_runtime(state, job)
    sup = rt.get("supervisor") or entry.get("supervisor")
    cl = rt.get("claude")
    sup_alive = record_alive(sup)
    k = kill_tree(sup, extra=[cl] if cl else [], alive_until=last_alive_at(rt))
    flags = list(rt.get("flags") or [])
    if by == "expire":
        flags.append("expired")
    start = _ts(rt.get("started_at")) or _ts(entry.get("launched_at"))
    res = finalize_job(state, job, ss=StreamState.from_file(raw_file(state, job)), hint="killed",
                       flags=flags, reason=by, detail="killed by `%s`" % by,
                       wall_s=(_now() - start) if start else None)
    if res is not None and not sup_alive and rt:
        rt.update({"status": "killed", "ended_at": _iso()})
        with contextlib.suppress(OSError):
            write_json_atomic(job_file(state, job), rt)
    final = read_registry(state)["jobs"][job].get("status")
    return {"job": job, "killed": True, "status": final, "ok": not k["survivors"],
            "snapshot": k["snapshot"], "survivors": k["survivors"], "taskkill_rc": k["taskkill_rc"]}


# --------------------------------------------------------------------------------------
# Argument parsing (argparse that never prints to stdout or exits on its own)
# --------------------------------------------------------------------------------------


class _Parser(argparse.ArgumentParser):
    def error(self, message):
        raise ForksError("usage", detail=message, usage=self.format_usage().strip())

    def print_help(self, file=None):
        raise _HelpRequested(self.format_help())

    def exit(self, status=0, message=None):
        if message:
            _diag(message.strip())
        raise ForksError("usage", detail=(message or "").strip() or "argument parser exit")


def _split_top_level_commas(s: str) -> list[str]:
    """Split on commas outside parentheses: `Bash(a,b),Read` -> [`Bash(a,b)`, `Read`]."""
    out, depth, cur = [], 0, []
    for ch in s:
        if ch == "(":
            depth += 1
        elif ch == ")":
            depth = max(0, depth - 1)
        if ch == "," and depth == 0:
            out.append("".join(cur).strip())
            cur = []
        else:
            cur.append(ch)
    out.append("".join(cur).strip())
    return [x for x in out if x]


def _tool_list(values) -> list[str] | None:
    if values is None:
        return None
    out = []
    for v in values:
        out.extend(_split_top_level_commas(v))
    return out


def _pos_int(s: str) -> int:
    try:
        v = int(s)
    except ValueError:
        raise argparse.ArgumentTypeError("not an integer: %r" % s)
    if v <= 0:
        raise argparse.ArgumentTypeError("must be > 0: %r" % s)
    return v


def _pos_float(s: str) -> float:
    try:
        v = float(s)
    except ValueError:
        raise argparse.ArgumentTypeError("not a number: %r" % s)
    if not v > 0:
        raise argparse.ArgumentTypeError("must be > 0: %r" % s)
    return v


COMMANDS = ("where", "init", "train", "fork", "continue", "wait", "list", "expire", "kill",
            "rebase", "usage", "tick", "_supervise")


def build_parser() -> _Parser:
    p = _Parser(prog="forks.py", description="spire:forks helper (one JSON object on stdout).")
    sub = p.add_subparsers(dest="command", metavar="command", parser_class=_Parser)
    sub.required = True

    def cmd(name, help_text):
        sp = (sub.add_parser(name, help=help_text, description=help_text) if help_text
              else sub.add_parser(name))
        if name != "where":
            sp.add_argument("--run", required=True, help="path of the run log .md")
        return sp

    w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.")
    w.add_argument("--binary", help="claude binary to report instead of the default resolution")

    i = cmd("init", "Create the state dir and the launch config (idempotent).")
    i.add_argument("--model", required=True, help="full model id, e.g. claude-opus-5-5[1m]")
    i.add_argument("--effort", default="high", choices=EFFORTS)
    i.add_argument("--permission-mode", default="dontAsk")
    i.add_argument("--tools", default=DEFAULT_TOOLS, help="comma-separated tool names")
    i.add_argument("--allowed-tools", nargs="+", action="extend", default=None,
                   help="allow rules (default: the tools)")
    i.add_argument("--disallowed-tools", nargs="+", action="extend", default=None,
                   help="deny rules (default: git push/rebase/reset --hard, claude)")
    i.add_argument("--agent", help="project agent name (<cwd>/.claude/agents/<name>.md)")
    i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)")
    i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)")
    i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"])
    i.add_argument("--orphan-after", type=_pos_int, default=CONFIG_DEFAULTS["orphan_after"])
    i.add_argument("--pings", choices=("on", "off"), default=CONFIG_DEFAULTS["pings"])
    i.add_argument("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"],
                   help="TTL assumed until a base's calls show one (default 3600)")
    i.add_argument("--warm-margin-s", type=_pos_int, default=None,
                   help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10))")
    i.add_argument("--max-minutes", type=_pos_int, default=CONFIG_DEFAULTS["max_minutes"])
    i.add_argument("--stall-after-s", type=_pos_int, default=CONFIG_DEFAULTS["stall_after_s"])
    for k, v in THRESHOLD_DEFAULTS.items():
        i.add_argument("--" + k.replace("_", "-"), type=_pos_int, default=v)

    t = cmd("train", "Train a new base (headless session).")
    t.add_argument("--base", required=True)
    t.add_argument("--brief", required=True)
    t.add_argument("--block")
    t.add_argument("--phase")

    def launch_overrides(sp):
        # Accepted only so that they can be refused with a clear message (plan 4.3).
        g = sp.add_argument_group("refused launch fields (fixed at init; read from the base record)")
        for name in LAUNCH_FIELDS:
            flag = "--" + name.replace("_", "-")
            if name in ("allowed_tools", "disallowed_tools"):
                g.add_argument(flag, nargs="+", default=None, help=argparse.SUPPRESS)
            else:
                g.add_argument(flag, default=None, help=argparse.SUPPRESS)

    f = cmd("fork", "Fork a base for one block.")
    f.add_argument("--base", required=True)
    f.add_argument("--block", required=True)
    f.add_argument("--phase")
    f.add_argument("--brief", required=True)
    f.add_argument("--mode", choices=("sequential", "parallel"), default="sequential")
    f.add_argument("--paths", nargs="+", action="extend", default=None)
    f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
    f.add_argument("--max-budget-usd", type=_pos_float)
    f.add_argument("--force", action="store_true", help="fork a base that is not active")
    launch_overrides(f)

    c = cmd("continue", "One more turn in a fork (resume without --fork-session).")
    c.add_argument("--job", required=True)
    c.add_argument("--brief", required=True)
    c.add_argument("--max-minutes", type=_pos_float)
    launch_overrides(c)

    wt = cmd("wait", "Block until jobs finish or the timeout (default 540, max 570).")
    wt.add_argument("jobs", nargs="*")
    wt.add_argument("--all", action="store_true", help="every job not yet settled by a wait")
    wt.add_argument("--any", action="store_true", help="return when any one of them has finished")
    wt.add_argument("--timeout", type=_pos_int, default=WAIT_DEFAULT_S)

    cmd("list", "Show every base and job.")

    e = cmd("expire", "Retire a base (or all); running jobs of it are killed.")
    eg = e.add_mutually_exclusive_group(required=True)
    eg.add_argument("--base")
    eg.add_argument("--all", action="store_true")

    k = cmd("kill", "Kill a job's supervisor and claude process tree.")
    k.add_argument("--job", required=True)

    s = cmd("_supervise", None)   # internal
    s.add_argument("--job", required=True)

    cmd("tick", "One keep-warm pass: ping every base the rule picks (design 5).")

    for name, text in (("rebase", "Rebase a base onto the current state (Phase 2)."),
                       ("usage", "Aggregate the ledger (Phase 2).")):
        cmd(name, text)   # arguments arrive with Phase 2; until then extra args are tolerated
    return p


# --------------------------------------------------------------------------------------
# Commands
# --------------------------------------------------------------------------------------


def cmd_where(args) -> tuple[dict, int]:
    reasons = []
    py_ok = sys.version_info >= (3, 10)
    if not py_ok:
        reasons.append("python < 3.10")
    py = {"version": platform.python_version(), "executable": sys.executable,
          "base_executable": getattr(sys, "_base_executable", sys.executable),
          "supervisor_interpreter": supervisor_interpreter(), "ok": py_ok}
    path, how = resolve_binary(args.binary)
    binary = {"path": path, "how": how}
    if path is None:
        reasons.append("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
    else:
        try:
            binary.update(binary_info(path))
        except ForksError as e:
            binary["error"] = e.payload["error"]
            reasons.append("%s: %s" % (e.payload["error"], path))
    warnings = []
    base_url = os.environ.get("ANTHROPIC_BASE_URL")
    if base_url is not None and base_url.strip().rstrip("/").lower() != DEFAULT_BASE_URL:
        warnings.append("ANTHROPIC_BASE_URL is %r, not the default %s; it is passed to every "
                        "base and fork, so they authenticate and bill through that endpoint"
                        % (base_url, DEFAULT_BASE_URL))
    out = {
        "script": script_path(),
        "python": py,
        "uv": _uv_report(),
        "binary": binary,
        "git": _git_report(),
        "spawn": SPAWN_METHOD,
        "scrub": scrub_report(),
        "platform": sys.platform,
        "warnings": warnings,
        "ok": not reasons,
    }
    if reasons:
        out["reason"] = "; ".join(reasons)
        out["error"] = out["reason"]
        return out, EXIT_USAGE
    return out, EXIT_OK


def _resolve_cwd(arg: str | None) -> str:
    if arg:
        cwd = norm_path(arg)
    else:
        cwd = git_toplevel()
        if not cwd:
            raise ForksError("no --cwd given and the current directory is not inside a git repo")
    if not os.path.isdir(cwd):
        raise ForksError("--cwd is not a directory", cwd=cwd)
    return cwd


def build_init_config(args, state: str) -> dict:
    """Validate the launch fields and build the config block. Writes nothing."""
    model = args.model
    if not MODEL_RE.match(model or ""):
        raise ForksError("model must be a full id (aliases are refused)", model=model,
                         pattern=MODEL_RE.pattern)
    mode = args.permission_mode
    if mode in REFUSED_PERMISSION_MODES:
        raise ForksError("permission mode refused", permission_mode=mode,
                         detail="plan mode swaps models; use dontAsk")
    if mode not in PERMISSION_MODES:
        raise ForksError("unknown permission mode", permission_mode=mode,
                         allowed=[m for m in PERMISSION_MODES if m not in REFUSED_PERMISSION_MODES])
    tools = _tool_list([args.tools])
    if not tools:
        raise ForksError("--tools is empty")
    allowed = _tool_list(args.allowed_tools)
    if allowed is None:
        allowed = list(tools)
    disallowed = _tool_list(args.disallowed_tools)
    if disallowed is None:
        disallowed = list(DEFAULT_DISALLOWED)

    cwd = _resolve_cwd(args.cwd)

    adm = model_admission(model, cwd)
    if not adm["admitted"]:
        files = [x["file"] for x in adm["lists"]]
        raise ForksError(
            "model not admitted by availableModels", model=model, files=files,
            message=("Model %s is not admitted by the availableModels list(s) in: %s. The user must "
                     "add the id (or its family alias) to one of these files; forks never writes "
                     "settings." % (model, ", ".join(files))))

    bpath, how = resolve_binary(args.binary)
    if bpath is None:
        raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
    binfo = binary_info(bpath)

    pre = preamble_path()
    if not os.path.isfile(pre):
        raise ForksError("preamble.md missing next to forks.py", path=pre)

    agent = None
    if args.agent:
        name = args.agent
        if not re.match(r"^[A-Za-z0-9_.:-]+$", name) or name in (".", ".."):
            raise ForksError("bad agent name", agent=name)
        apath = os.path.join(cwd, ".claude", "agents", name + ".md")
        if not os.path.isfile(apath):
            raise ForksError("agent file not found", agent=name, path=apath)
        agent = {"name": name, "path": apath, "sha256": _sha256_file(apath)}

    launch = {
        "model": model,
        "effort": args.effort,
        "permission_mode": mode,
        "tools": tools,
        "allowed_tools": allowed,
        "disallowed_tools": disallowed,
        "mcp": {"strict": True, "config": mcp_empty_path(state)},
        "append_system_prompt": {"path": pre, "sha256": _sha256_file(pre)},
        "agent": agent,
        "cwd": cwd,
        "binary": binfo,
        "env": {"pinned": dict(ENV_PINNED_LAUNCH)},
    }
    cfg = {
        "launch": launch,
        "fingerprint": fingerprint_hash(launch),
        "binary_resolved_by": how,
        "parallel_cap": args.parallel_cap,
        "orphan_after": args.orphan_after,
        "pings": args.pings,
        "ttl_s": args.ttl_s,
        "warm_margin_s": args.warm_margin_s,
        "max_minutes": args.max_minutes,
        "stall_after_s": args.stall_after_s,
        "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS},
        "model_admission": adm["via"],
    }
    return cfg


def _diff_fields(a, b, prefix="") -> list[str]:
    if isinstance(a, dict) and isinstance(b, dict):
        out = []
        for k in sorted(set(a) | set(b)):
            out += _diff_fields(a.get(k), b.get(k), prefix + k + ".")
        return out
    return [] if a == b else [prefix.rstrip(".")]


def cmd_init(args) -> tuple[dict, int]:
    state = state_dir_for(args.run)
    cfg = build_init_config(args, state)          # all validation before any write
    ensure_state(state)
    with RegistryLock(state):
        reg = read_registry(state)
        created = reg is None
        if created:
            reg = new_registry()
        old = reg.get("config")
        changed_fields = [] if old is None else _diff_fields(old, cfg)
        # `model_admission` and `binary_resolved_by` describe how, not what: not a change.
        changed_fields = [f for f in changed_fields
                          if f not in ("model_admission", "binary_resolved_by")]
        if old is not None and changed_fields and reg.get("bases"):
            raise ForksError("init values differ and a base exists; train a new run instead",
                             fields=changed_fields, bases=sorted(reg["bases"]), state=state)
        if old is None or changed_fields:
            reg["config"] = cfg
            reg.setdefault("run", {})["init_at"] = _iso()
            write_registry(state, reg)
        else:
            cfg = old
    return {"ok": True, "state": state, "registry": registry_path(state), "created": created,
            "changed": bool(old is not None and changed_fields), "changed_fields": changed_fields,
            "config": cfg}, EXIT_OK


def _not_implemented(block=None, phase=None):
    def f(args):
        out = {"error": "not implemented", "command": args.command}
        if block:
            out["block"] = block
        if phase:
            out["phase"] = phase
        return out, EXIT_USAGE
    return f


# --- launching: train, fork, continue -------------------------------------------------


def _load_run(args) -> tuple[str, dict]:
    state = state_dir_for(args.run)
    reg = read_registry(state) if os.path.isdir(state) else None
    if not reg or not reg.get("config"):
        raise ForksError("no launch config for this run; call init first", state=state)
    return state, reg


def _refuse_overrides(args) -> None:
    given = [n for n in LAUNCH_FIELDS if getattr(args, n, None) is not None]
    if given:
        raise ForksError("launch fields are fixed at init; %s reads them from the base record and "
                         "cannot override them (a different value needs a new base)" % args.command,
                         fields=given, command=args.command)


def _read_brief(path: str) -> tuple[str, str]:
    p = norm_path(path)
    try:
        with open(p, "r", encoding="utf-8") as f:
            return p, f.read()
    except OSError as e:
        raise ForksError("cannot read brief", brief=p, detail=str(e))


def _check_binary(ref: dict, launch: dict) -> tuple[dict, list[str]]:
    """Launch-time binary re-check (version, size, mtime) against `ref`; also the hashed files."""
    path = launch["binary"]["path"]
    if not os.path.isfile(path):
        raise ForksError("claude binary missing (the stored path is gone); the base cannot be "
                         "launched with another binary", binary=path, reason="binary_missing")
    info = binary_info(path)
    drift = [] if all(info.get(k) == (ref or {}).get(k) for k in ("version", "size", "mtime")) else ["binary"]
    asp = launch["append_system_prompt"]
    try:
        if _sha256_file(asp["path"]) != asp["sha256"]:
            drift.append("append_system_prompt")
    except OSError:
        drift.append("append_system_prompt")
    ag = launch.get("agent")
    if ag:
        try:
            if _sha256_file(ag["path"]) != ag["sha256"]:
                drift.append("agent")
        except OSError:
            drift.append("agent")
    return info, drift


def claude_argv(launch: dict) -> list[str]:
    """The fingerprinted command line (passed in full on every call; flags are not inherited)."""
    a = [launch["binary"]["path"], "-p", "--output-format", "stream-json", "--verbose",
         "--model", launch["model"], "--effort", launch["effort"],
         "--permission-mode", launch["permission_mode"], "--tools", ",".join(launch["tools"])]
    if launch.get("allowed_tools"):
        a += ["--allowedTools", ",".join(launch["allowed_tools"])]
    if launch.get("disallowed_tools"):
        a += ["--disallowedTools", ",".join(launch["disallowed_tools"])]
    a += ["--strict-mcp-config", "--mcp-config", launch["mcp"]["config"],
          "--append-system-prompt-file", launch["append_system_prompt"]["path"]]
    if launch.get("agent"):
        a += ["--agent", launch["agent"]["name"]]
    return a


def _git_head(cwd: str) -> str | None:
    try:
        r = git(["rev-parse", "HEAD"], cwd=cwd)
    except (OSError, subprocess.SubprocessError):
        return None
    out = r.stdout.decode("utf-8", "replace").strip()
    return out if r.returncode == 0 and out else None


def delta_header(cwd: str, base_commit: str | None) -> str:
    """Commits since the base's training commit plus the working tree's shortstat against it."""
    if not base_commit:
        return ("No base commit is recorded (the base's cwd was not a git repository, or git failed "
                "at training). Treat everything on disk as possibly changed since training.")
    parts = []
    try:
        lg = git(["log", "--oneline", "%s..HEAD" % base_commit], cwd=cwd)
        st = git(["diff", "--shortstat", base_commit], cwd=cwd)
        log_txt = lg.stdout.decode("utf-8", "replace").strip() if lg.returncode == 0 else \
            "(git log failed: %s)" % lg.stderr.decode("utf-8", "replace").strip()
        st_txt = st.stdout.decode("utf-8", "replace").strip() if st.returncode == 0 else \
            "(git diff failed: %s)" % st.stderr.decode("utf-8", "replace").strip()
    except (OSError, subprocess.SubprocessError) as e:
        log_txt = st_txt = "(git failed: %s)" % e
    parts.append("Base commit: `%s`." % base_commit)
    parts.append("Commits since then (`git log --oneline %s..HEAD`):\n\n```\n%s\n```"
                 % (base_commit[:12], log_txt or "(none)"))
    parts.append("Working tree against the base commit (`git diff --shortstat %s`):\n\n```\n%s\n```"
                 % (base_commit[:12], st_txt or "(no changes)"))
    return "\n\n".join(parts)


def _next_job_id(reg: dict) -> str:
    n = 0
    for j in reg.get("jobs", {}):
        m = re.fullmatch(r"j(\d+)", j)
        if m:
            n = max(n, int(m.group(1)))
    return "j%04d" % (n + 1)


def _running_capped(reg: dict) -> list[str]:
    return sorted(j for j, e in reg.get("jobs", {}).items()
                  if e.get("kind") in ("fork", "continue") and e.get("status") in RUNNING)


def _new_entry(job: str, kind: str, **kw) -> dict:
    e = {"job": job, "kind": kind, "base": None, "block": None, "phase": None, "mode": None,
         "paths": None, "status": "created", "flags": [], "reason": None, "session": None,
         "parent_session": None, "model_requested": None, "created_at": _iso(), "launched_at": None,
         "ended_at": None, "supervisor": None, "cwd": None, "argv": None,
         "brief": "briefs/%s.md" % job, "raw": "raw/%s.jsonl" % job, "max_s": None,
         "cost_before": 0.0, "model_usage_before": {}, "base_context": None,
         "base_state_at_launch": None, "launch_drift": [], "binary_version": None,
         "head_at_launch": None, "continues": None, "delivered_at": None, "settled_at": None}
    e.update(kw)
    return e


def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dict:
    """Write the brief and the child env, spawn the supervisor, record it. A failed spawn ends the
    job as `failed`, reason `spawn_failed`."""
    atomic_write_bytes(os.path.join(state, "briefs", job + ".md"), prompt.encode("utf-8"))
    write_json_atomic(raw_file(state, job, ".env.json"), scrubbed_env(launch["binary"]["path"]))
    sp = spawn_supervisor(state, run, job)
    if not sp.get("ok"):
        finalize_job(state, job, ss=None, hint="failed", reason="spawn_failed", detail=sp.get("detail"))
        return sp
    with RegistryLock(state):
        reg = read_registry(state)
        e = reg["jobs"][job]
        e["supervisor"] = {"pid": sp["pid"], "ctime": sp.get("ctime")}
        e["launched_at"] = _iso()
        if e.get("status") == "created":
            e["status"] = "running"
        write_registry(state, reg)
    return sp


def cmd_train(args) -> tuple[dict, int]:
    state, reg = _load_run(args)
    name = args.base
    if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""):
        raise ForksError("bad base name", base=name)
    cfg = reg["config"]
    la = cfg["launch"]
    brief_path, brief = _read_brief(args.brief)
    binfo, drift = _check_binary(la["binary"], la)
    session = str(uuid.uuid4())
    argv = claude_argv(la) + ["--session-id", session]
    head = _git_head(la["cwd"])
    with RegistryLock(state):
        reg = read_registry(state)
        stamp_orchestrator(reg)
        old = reg["bases"].get(name)
        if old and old.get("status") != "failed":
            raise ForksError("base exists", base=name, status=old.get("status"))
        job = _next_job_id(reg)
        reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"],
                              "fingerprint": cfg["fingerprint"], "binary": binfo, "created_at": _iso(),
                              "train_job": job, "phase": _phase(args.phase), "last_hit": None,
                              "base_commit": None, "context_tokens": None, "forks": []}
        reg["jobs"][job] = _new_entry(
            job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session,
            model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60,
            launch_drift=drift, binary_version=binfo["version"], head_at_launch=head,
            brief_source=brief_path)
        write_registry(state, reg)
    prompt = render_template("train.md", {"base": name, "run": norm_path(args.run), "state_dir": state,
                                          "notes_dir": os.path.join(state, "notes"),
                                          "logs_dir": os.path.join(state, "logs"), "brief": brief})
    _start_job(state, norm_path(args.run), job, prompt, la)
    return {"job": job, "base": name, "session_id": session}, EXIT_OK


def cmd_fork(args) -> tuple[dict, int]:
    _refuse_overrides(args)
    if args.mode == "parallel" and not args.paths:
        raise ForksError("--mode parallel needs --paths (the files this fork may touch)", mode="parallel")
    state, reg = _load_run(args)
    cfg = reg["config"]
    la = cfg["launch"]
    base = reg["bases"].get(args.base)
    if base is None:
        raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"]))
    if base.get("status") != "active" and not args.force:
        raise ForksError("base is not active (use --force to fork it anyway)", base=args.base,
                         status=base.get("status"))
    if not base.get("session"):
        raise ForksError("base has no session", base=args.base, status=base.get("status"))
    brief_path, brief = _read_brief(args.brief)
    binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
    delta = delta_header(la["cwd"], base.get("base_commit"))
    head = _git_head(la["cwd"])
    session = str(uuid.uuid4())
    argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session]
    if args.max_budget_usd:
        argv += ["--max-budget-usd", "%g" % args.max_budget_usd]
    paths = list(args.paths) if args.paths else None
    reap_lost(state)
    with RegistryLock(state):
        reg = read_registry(state)
        stamp_orchestrator(reg)
        base = reg["bases"][args.base]
        if base.get("status") != "active" and not args.force:
            raise ForksError("base is not active (use --force to fork it anyway)", base=args.base,
                             status=base.get("status"))
        running = _running_capped(reg)
        if len(running) >= int(cfg["parallel_cap"]):
            raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running)
        job = _next_job_id(reg)
        warmth = base_warmth(base, cfg)
        reg["jobs"][job] = _new_entry(
            job, "fork", base=args.base, block=args.block, phase=_phase(args.phase), mode=args.mode,
            paths=paths, session=session, parent_session=base["session"], model_requested=la["model"],
            cwd=la["cwd"], argv=argv,
            max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60,
            cost_before=float(base.get("cost_cum_usd") or 0.0),
            model_usage_before=base.get("model_usage_cum") or {},
            base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
            warm_until_at_launch=warmth["warm_until"], launch_drift=drift,
            binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path,
            forced=bool(args.force and base.get("status") != "active"))
        base.setdefault("forks", []).append(job)
        write_registry(state, reg)
    if args.mode == "parallel":
        commit_rule = ("This fork runs in parallel with other forks: do NOT commit and do not stage. "
                       "Touch only these paths: %s. The orchestrator commits after its gate."
                       % ", ".join("`%s`" % p for p in paths))
    else:
        commit_rule = ("This fork commits: after your gate passes, stage and commit your block by "
                       "explicit paths only (never `git add -A` or `.`).")
        if paths:
            commit_rule += " Expected paths: %s." % ", ".join("`%s`" % p for p in paths)
    prompt = render_template("fork.md", {
        "base": args.base, "run": norm_path(args.run), "state_dir": state, "block": args.block,
        "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, "delta": delta,
        "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")),
        "logs_dir": os.path.join(state, "logs"), "brief": brief})
    _start_job(state, norm_path(args.run), job, prompt, la)
    return {"job": job, "base": args.base, "base_state": warmth["base_state"],
            "warm_until": warmth["warm_until"], "rebase_due": False, "retrain_due": False,
            "reasons": []}, EXIT_OK


def cmd_continue(args) -> tuple[dict, int]:
    _refuse_overrides(args)
    state, reg = _load_run(args)
    cfg = reg["config"]
    la = cfg["launch"]
    fj = reg["jobs"].get(args.job)
    if fj is None:
        raise ForksError("unknown job", job=args.job)
    if fj.get("kind") != "fork":
        raise ForksError("continue takes a fork job", job=args.job, kind=fj.get("kind"))

    def busy(r):
        return [j for j, e in r["jobs"].items() if e.get("status") in RUNNING
                and (j == args.job or (e.get("kind") == "continue" and e.get("continues") == args.job))]
    if busy(reg):
        raise ForksError("the fork is still running", job=args.job, running=busy(reg))
    if not fj.get("session"):
        raise ForksError("the fork has no session", job=args.job)
    base = reg["bases"].get(fj.get("base")) or {}
    brief_path, brief = _read_brief(args.brief)
    binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
    head = _git_head(la["cwd"])
    argv = claude_argv(la) + ["--resume", fj["session"]]
    reap_lost(state)
    with RegistryLock(state):
        reg = read_registry(state)
        stamp_orchestrator(reg)
        fj = reg["jobs"][args.job]
        if busy(reg):
            raise ForksError("the fork is still running", job=args.job, running=busy(reg))
        running = _running_capped(reg)
        if len(running) >= int(cfg["parallel_cap"]):
            raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running)
        base = reg["bases"].get(fj.get("base")) or {}
        cum = fj.get("session_cost_cum_usd")
        job = _next_job_id(reg)
        warmth = base_warmth(base, cfg)
        reg["jobs"][job] = _new_entry(
            job, "continue", base=fj.get("base"), block=fj.get("block"), phase=fj.get("phase"),
            mode=fj.get("mode"), paths=fj.get("paths"), session=fj["session"],
            parent_session=fj.get("parent_session"), model_requested=la["model"], cwd=la["cwd"],
            argv=argv, max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60,
            cost_before=float(cum if cum is not None else (base.get("cost_cum_usd") or 0.0)),
            model_usage_before=(fj.get("session_model_usage_cum") if cum is not None
                                else base.get("model_usage_cum")) or {},
            base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
            launch_drift=drift, binary_version=binfo["version"], head_at_launch=head,
            continues=args.job, brief_source=brief_path)
        write_registry(state, reg)
    _start_job(state, norm_path(args.run), job, brief, la)
    return {"job": job}, EXIT_OK


# --- keep warm: the ping rule and the tick (design 5, C3d) ----------------------------


def run_for_state(state: str) -> str:
    """The run log path of a state dir (inverse of `state_dir_for`)."""
    return state[: -len("-forks")] + ".md"


def launch_record(base: dict | None, cfg: dict) -> dict:
    """The launch fields a launch on `base` uses: the base's own record when it has one, else the
    run config (every base has the init config's fields until bases carry their own record)."""
    return (base or {}).get("launch") or cfg["launch"]


def _job_num(job) -> int:
    m = re.fullmatch(r"j(\d+)", str(job or ""))
    return int(m.group(1)) if m else -1


def pings_since_real_call(reg: dict, name: str) -> int:
    """Pings of base `name` launched after its last real (non-ping) call. Job ids are monotonic
    per run, so "after" is by job number (launch times have whole seconds only)."""
    jobs = reg.get("jobs", {})
    last_real = max((_job_num(j) for j, e in jobs.items()
                     if e.get("base") == name and e.get("kind") in REAL_KINDS), default=-1)
    return sum(1 for j, e in jobs.items()
               if e.get("base") == name and e.get("kind") == "ping" and _job_num(j) > last_real)


def ping_in_flight(reg: dict, base: dict | None) -> str | None:
    """The base's in-flight ping job (C3d), or None. A mark whose job is no longer running (or
    gone) does not count: it clears itself on the next marking."""
    j = (base or {}).get("ping_in_flight")
    if j and (reg.get("jobs", {}).get(j) or {}).get("status") in RUNNING:
        return j
    return None


def run_active(reg: dict, now: float) -> bool:
    """Design 5's activity condition: a fork/continue job ran, or `wait` was called, in the last
    ACTIVITY_WINDOW_S (the Athanor is in its turn)."""
    lw = _ts((reg.get("run") or {}).get("last_wait_at"))
    if lw is not None and now - lw <= ACTIVITY_WINDOW_S:
        return True
    for e in reg.get("jobs", {}).values():
        if e.get("kind") not in ("fork", "continue"):
            continue
        if e.get("status") in RUNNING:
            return True
        end = _ts(e.get("ended_at"))
        if end is not None and now - end <= ACTIVITY_WINDOW_S:
            return True
    return False


def ping_decision(reg: dict, name: str, now: float | None = None) -> dict:
    """The ping rule for one base: pings on, base active, state expiring, no ping in flight, the
    run active in the last 10 min, and fewer pings since its last real call than break_even."""
    now = _now() if now is None else now
    cfg = reg.get("config") or {}
    base = reg["bases"][name]
    w = base_warmth(base, cfg, now)
    n = pings_since_real_call(reg, name)
    be = break_even(w["ttl_s"])
    inflight = ping_in_flight(reg, base)
    if cfg.get("pings", CONFIG_DEFAULTS["pings"]) != "on":
        why = "pings_off"
    elif base.get("status") != "active" or not base.get("session"):
        why = "status:%s" % base.get("status")
    elif w["base_state"] != "expiring":
        why = "state:%s" % w["base_state"]
    elif inflight:
        why = "in_flight:%s" % inflight
    elif not run_active(reg, now):
        why = "no_activity"
    elif n >= be:
        why = "cap:%d/%d" % (n, be)
    else:
        why = None
    return {"base": name, "ping": why is None, "why": why, "base_state": w["base_state"],
            "warm_until": w["warm_until"], "ttl_s": w["ttl_s"], "pings_since_real_call": n,
            "break_even": be, "ping_in_flight": inflight}


def tick_pass(state: str) -> dict:
    """One pass of the ping rule over every base of the run. The decision and the `ping_in_flight`
    mark are made under the registry lock (C3d), so concurrent tickers launch one ping per base.
    The binary check runs before the lock for the bases that look eligible. Pings go through the
    same hidden spawn as any job. Returns {"pinged": [job...], "bases": [decision...]}."""
    reg = read_registry(state)
    if not reg or not reg.get("config"):
        return {"pinged": [], "bases": []}
    now = _now()
    decisions = [ping_decision(reg, n, now) for n in sorted(reg.get("bases", {}))]
    pre = [d["base"] for d in decisions if d["ping"]]
    if not pre:
        return {"pinged": [], "bases": decisions}
    checks = {}
    for n in pre:
        base = reg["bases"][n]
        la = launch_record(base, reg["config"])
        try:
            checks[n] = _check_binary(base.get("binary") or la["binary"], la)
        except ForksError as e:
            checks[n] = e
    launches = []
    with RegistryLock(state):
        reg = read_registry(state)
        cfg = reg["config"]
        now = _now()
        decisions = []
        for n in sorted(reg["bases"]):
            d = ping_decision(reg, n, now)
            if d["ping"] and n not in checks:
                d.update(ping=False, why="eligible_since_this_tick_began")   # the next tick takes it
            elif d["ping"] and isinstance(checks[n], ForksError):
                d.update(ping=False, why=checks[n].payload.get("reason") or checks[n].payload["error"])
            if d["ping"]:
                base = reg["bases"][n]
                la = launch_record(base, cfg)
                binfo, drift = checks[n]
                job = _next_job_id(reg)
                session = str(uuid.uuid4())
                argv = claude_argv(la) + ["--resume", base["session"], "--fork-session",
                                          "--session-id", session, "--max-turns", "1"]
                reg["jobs"][job] = _new_entry(
                    job, "ping", base=n, session=session, parent_session=base["session"],
                    model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=PING_MAX_S,
                    cost_before=float(base.get("cost_cum_usd") or 0.0),
                    model_usage_before=base.get("model_usage_cum") or {},
                    base_context=base.get("context_tokens"), base_state_at_launch=d["base_state"],
                    warm_until_at_launch=d["warm_until"], launch_drift=drift,
                    binary_version=binfo["version"])
                base["ping_in_flight"] = job
                base["ping_in_flight_at"] = _iso(now)
                d["job"] = job
                launches.append((job, la))
            decisions.append(d)
        if launches:
            write_registry(state, reg)
    run = run_for_state(state)
    for job, la in launches:
        _start_job(state, run, job, PING_PROMPT, la)
    return {"pinged": [j for j, _ in launches], "bases": decisions}


def safe_tick(state: str, where: str) -> dict | None:
    """A tick pass for `wait` and the supervisor: never raises (diagnostics on stderr)."""
    try:
        out = tick_pass(state)
    except Exception as ex:   # a keep-warm failure must not break `wait` or a supervisor
        _diag("%s tick failed: %s: %s" % (where, type(ex).__name__, ex))
        return None
    if out["pinged"]:
        _diag("%s tick: pinged %s" % (where, ", ".join("%s (%s)" % (d["job"], d["base"])
                                                         for d in out["bases"] if d.get("job"))))
    return out


def cmd_tick(args) -> tuple[dict, int]:
    """`tick --run R`: one pass by hand (N3-refused in a fork)."""
    state, _ = _load_run(args)
    reap_lost(state)
    out = tick_pass(state)
    return {"state": state, **out}, EXIT_OK


# --- the supervisor -------------------------------------------------------------------


class SupClock:
    """The supervisor's run time with host suspends cut out (M3). Each loop tick reports the
    wall clock; a gap above `jump_s` between ticks (the tick is ~SUP_TICK_S) is a suspend or a
    clock jump: the gap minus one tick is left out of `elapsed`, and the orphan watchdog is
    re-armed at the jump (one more `orphan_after` from now)."""

    def __init__(self, t0: float, tick_s: float | None = None, jump_s: float | None = None):
        self.t0 = self.prev = t0
        self.tick_s = SUP_TICK_S if tick_s is None else tick_s
        self.jump_s = CLOCK_JUMP_S if jump_s is None else jump_s
        self.excluded_s = 0.0
        self.rearmed_at: float | None = None
        self.jumps: list[float] = []

    def tick(self, now: float) -> float | None:
        """Record a tick; returns the excluded seconds if this tick follows a jump, else None."""
        gap, self.prev = now - self.prev, now
        if gap <= self.jump_s:
            return None
        skipped = gap - self.tick_s
        self.excluded_s += skipped
        self.rearmed_at = now
        self.jumps.append(skipped)
        return skipped

    def elapsed(self, now: float) -> float:
        """Seconds of run time since t0 (what `max_minutes` is measured against)."""
        return now - self.t0 - self.excluded_s

    def orphan_age(self, now: float, last_call: float) -> float:
        """Seconds the orphan watchdog counts: since the later of the last orchestrator call and
        the last re-arm."""
        ref = last_call if self.rearmed_at is None else max(last_call, self.rearmed_at)
        return now - ref


class Supervisor:
    """`_supervise`: runs one `claude` call, parses its stream, watches it, writes the result.
    The only writer of `jobs/<job>.json` (C3b)."""

    def __init__(self, state: str, job: str):
        self.state, self.job = state, job
        reg = read_registry(state)
        if not reg or job not in reg.get("jobs", {}):
            raise ForksError("unknown job", job=job, state=state)
        self.cfg = reg["config"]
        self.entry = reg["jobs"][job]
        self.env = read_json(raw_file(state, job, ".env.json"))
        self.ss = StreamState()
        self.mu = threading.Lock()
        self.flags: list[str] = []
        self.last_output = _now()
        # `messages` (M16): a progress counter of distinct assistant message ids; the result's and
        # the ledger's `turns` is `result.num_turns`.
        self.rt = {"job": job, "status": "running", "supervisor": pid_record(os.getpid()), "claude": None,
                   "started_at": _iso(), "heartbeat_at": _iso(), "messages": 0, "last_event_at": None,
                   "flags": []}
        self._last_write = 0.0

    def write_rt(self, **kw) -> None:
        with self.mu:
            self.rt.update(kw)
            self.rt["messages"] = self.ss.turns
            self.rt["last_event_at"] = _iso(self.ss.last_event_at) if self.ss.last_event_at else None
            self.rt["flags"] = list(self.flags)
            self.rt["heartbeat_at"] = _iso()
            snap = dict(self.rt)
        write_json_atomic(job_file(self.state, self.job), snap)
        self._last_write = _now()

    def _reader(self, stream, raw) -> None:
        for line in iter(stream.readline, b""):
            t = _now()
            raw.write(line)
            raw.flush()
            try:
                ev = json.loads(line.decode("utf-8", "replace"))
            except ValueError:
                ev = None
            with self.mu:
                self.last_output = t
                if ev is not None:
                    self.ss.feed(ev, t)

    def _feeder(self, stdin, data: bytes) -> None:
        with contextlib.suppress(OSError, ValueError):
            stdin.write(data)
        with contextlib.suppress(OSError, ValueError):
            stdin.close()

    def run(self) -> tuple[dict, int]:
        e = self.entry
        if e.get("status") not in RUNNING or e.get("kill_requested"):
            return {"job": self.job, "skipped": "job is %s" % e.get("status")}, EXIT_OK
        if self.env is None:
            finalize_job(self.state, self.job, ss=None, hint="failed", reason="supervisor_error",
                         detail="missing env file")
            return {"error": "missing env file"}, EXIT_USAGE
        os.environ.clear()                 # git and taskkill lookups use the child's PATH
        os.environ.update(self.env)
        self.write_rt()
        with open(os.path.join(self.state, "briefs", self.job + ".md"), "rb") as f:
            prompt = f.read()
        t0 = _now()
        raw = open(raw_file(self.state, self.job), "wb")
        err = open(raw_file(self.state, self.job, ".err"), "wb")
        try:
            p = subprocess.Popen(e["argv"], cwd=e["cwd"], env=self.env, stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE, stderr=err, creationflags=_NOWIN,
                                 start_new_session=not IS_WINDOWS)
        except OSError as ex:
            raw.close()
            err.close()
            finalize_job(self.state, self.job, ss=self.ss, hint="failed", reason="launch_error",
                         detail=str(ex), wall_s=0)
            self.write_rt(status="failed")
            return {"error": "launch_error", "detail": str(ex)}, EXIT_USAGE
        crec = pid_record(p.pid)
        self.write_rt(claude=crec)
        reader = threading.Thread(target=self._reader, args=(p.stdout, raw), daemon=True)
        reader.start()
        threading.Thread(target=self._feeder, args=(p.stdin, prompt), daemon=True).start()
        max_s = float(e.get("max_s") or float(self.cfg["max_minutes"]) * 60)
        stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"])
        orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"])
        hint, reason, detail, flags_extra = "exited", None, None, []
        hit_written = e["kind"] != "fork"            # only a fork's first request refreshes (I3)
        clock = SupClock(t0)
        last_orphan_check = 0.0
        last_messages = 0
        last_tick = t0
        while p.poll() is None:
            now = _now()
            skipped = clock.tick(now)
            if skipped is not None:
                _diag("clock jump: %.0f s between supervisor ticks at %s (host suspend?); not counted "
                      "toward max_minutes, orphan watchdog re-armed" % (skipped, _iso(now)))
                if "clock_jump" not in self.flags:
                    self.flags.append("clock_jump")
                self.write_rt()
            with self.mu:
                refused, first_at, messages = self.ss.refused, self.ss.first_at, self.ss.turns
                quiet = now - self.last_output
            if refused:
                kill_tree(crec, alive_until=time.time())
                break                                  # decided as model_refused from the stream
            if clock.elapsed(now) > max_s:
                kill_tree(crec, alive_until=time.time())
                hint, detail = "timeout", "run time over max_minutes (%.0f s)" % max_s
                break
            if quiet > stall_s and "stalled" not in self.flags:
                self.flags.append("stalled")
                self.write_rt()
            if not hit_written and first_at:
                hit_written = True
                with contextlib.suppress(ForksError, OSError):
                    bump_last_hit(self.state, e["base"], first_at)
            if now - last_orphan_check >= ORPHAN_CHECK_S:
                last_orphan_check = now
                try:
                    reg = read_registry(self.state)
                    last = _ts(((reg or {}).get("run") or {}).get("last_orchestrator_call_at"))
                except (ForksError, OSError):
                    last = None
                if last is not None and clock.orphan_age(now, last) > orphan_s:
                    kill_tree(crec, alive_until=time.time())
                    hint, reason, flags_extra = "killed", "orphaned", ["orphaned"]
                    detail = "no orchestrator call for %.0f s (orphan_after %.0f s)" % (
                        clock.orphan_age(now, last), orphan_s)
                    break
            if now - self._last_write >= HEARTBEAT_S or (messages != last_messages
                                                          and now - self._last_write >= 1.0):
                last_messages = messages
                self.write_rt()
            if now - last_tick >= TICK_SUP_S:          # keep warm (design 5): one pass a minute
                last_tick = now
                safe_tick(self.state, "supervisor %s" % self.job)
            time.sleep(SUP_TICK_S)
        exit_code = p.wait()
        # leftovers of claude (backgrounded commands); claude exited just now
        sweep = kill_tree(crec, alive_until=time.time())
        if sweep["snapshot"]:
            _diag("swept %d leftover process(es) of claude: %s" % (
                len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]]))
        reader.join(READER_JOIN_S)
        with contextlib.suppress(OSError):
            raw.close()
        with contextlib.suppress(OSError):
            err.close()
        wall = _now() - t0
        base_commit = None
        if e["kind"] == "train":
            base_commit = _git_head(e["cwd"]) or e.get("head_at_launch")
        res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra,
                           reason=reason, detail=detail, exit_code=exit_code, wall_s=wall,
                           base_commit=base_commit)
        final = res["status"] if res else (read_registry(self.state)["jobs"][self.job].get("status"))
        self.write_rt(status=final, ended_at=_iso(), exit_code=exit_code)
        return {"job": self.job, "status": final}, EXIT_OK


def cmd_supervise(args) -> tuple[dict, int]:
    state = state_dir_for(args.run)
    if not os.path.isdir(os.path.join(state, "raw")) or not re.fullmatch(r"j\d+", args.job):
        raise ForksError("no such job state", state=state, job=args.job)
    log = open(raw_file(state, args.job, ".sup.log"), "a", encoding="utf-8", buffering=1)
    sys.stdout = sys.stderr = log
    _diag("supervisor %d for %s starting (%s)" % (os.getpid(), args.job, _iso()))
    try:
        return Supervisor(state, args.job).run()
    except Exception as ex:
        traceback.print_exc(file=log)
        with contextlib.suppress(Exception):
            rt = _job_runtime(state, args.job)
            if rt.get("claude"):
                kill_tree(rt["claude"], alive_until=time.time())
            finalize_job(state, args.job, ss=StreamState.from_file(raw_file(state, args.job)),
                         hint="failed", reason="supervisor_error", detail="%s: %s" % (type(ex).__name__, ex))
        raise


# --- wait, list, kill, expire -----------------------------------------------------------


def _running_info(state: str, job: str, e: dict, now: float) -> dict:
    rt = _job_runtime(state, job)
    start = _ts(rt.get("started_at")) or _ts(e.get("launched_at")) or _ts(e.get("created_at"))
    last = _ts(rt.get("last_event_at"))
    return {"job": job, "block": e.get("block"),
            "elapsed_s": None if start is None else round(now - start, 1),
            "last_event_s_ago": None if last is None else round(now - last, 1),
            "messages": int(rt.get("messages") or 0)}


def cmd_wait(args) -> tuple[dict, int]:
    state, reg = _load_run(args)
    if args.timeout > WAIT_MAX_S:
        raise ForksError("--timeout over the maximum", timeout=args.timeout, max=WAIT_MAX_S)
    if args.jobs and args.all:
        raise ForksError("name jobs or --all, not both")
    if not args.jobs and not args.all:
        raise ForksError("name jobs or --all")
    unknown = [j for j in args.jobs if j not in reg["jobs"]]
    if unknown:
        raise ForksError("unknown job", jobs=unknown)
    pings = [j for j in args.jobs if reg["jobs"][j].get("kind") == "ping"]
    if pings:
        raise ForksError("pings are not waited on; their results arrive in `pings`", jobs=pings)
    touch_orchestrator(state, wait=True)
    deadline = time.monotonic() + args.timeout
    ticks = not in_child()          # N3: a fork's `wait` never launches pings
    next_tick = time.monotonic()    # one tick pass at the start, then every TICK_WAIT_S

    def evaluate(r):
        members = list(args.jobs) if args.jobs else sorted(
            j for j, x in r["jobs"].items() if not x.get("settled_at") and x.get("kind") != "ping")
        finished = [j for j in members if r["jobs"][j].get("status") in FINAL]
        running = [j for j in members if j not in finished]
        met = bool(finished) or not members if args.any else not running
        return members, finished, running, met

    while True:
        reap_lost(state)
        if ticks and time.monotonic() >= next_tick:
            next_tick = time.monotonic() + TICK_WAIT_S
            safe_tick(state, "wait")
        members, finished, running, met = evaluate(read_registry(state))
        left = deadline - time.monotonic()
        if met or left <= 0:
            break
        time.sleep(min(WAIT_POLL_S, left))
    with RegistryLock(state):
        reg = read_registry(state)
        members, finished, running, met = evaluate(reg)
        done, now = [], _now()
        for j in finished:
            e = reg["jobs"][j]
            if e.get("delivered_at"):
                continue
            res = read_json(result_file(state, j))
            if res is not None:
                done.append(res)
                e["delivered_at"] = _iso(now)
        if met:
            for j in finished:
                reg["jobs"][j]["settled_at"] = _iso(now)
        # Pings never come back through `done` and never count toward the exit code: each
        # finished ping result is returned once, in `pings`.
        ping_results = []
        for j, e in sorted(reg["jobs"].items()):
            if e.get("kind") != "ping" or e.get("status") not in FINAL or e.get("delivered_at"):
                continue
            res = read_json(result_file(state, j))
            if res is not None:
                ping_results.append(res)
                e["delivered_at"] = _iso(now)
        stamp_orchestrator(reg, now, wait=True)
        write_registry(state, reg)
        statuses = {j: reg["jobs"][j].get("status") for j in finished}
        entries = {j: dict(reg["jobs"][j]) for j in running}
    out = {"done": done, "running": [_running_info(state, j, entries[j], now) for j in running],
           "pings": ping_results}
    if not met:
        return out, EXIT_WAIT_TIMEOUT
    return out, (EXIT_OK if all(s == "ok" for s in statuses.values()) else EXIT_JOB_FAILED)


LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "status", "flags", "reason",
                 "reason_detail", "session", "parent_session", "continues", "created_at", "launched_at",
                 "ended_at", "fingerprint", "cache", "delivered_at", "settled_at")
LIST_BASE_KEYS = ("name", "status", "session", "cwd", "base_commit", "context_tokens",
                  "training_cost_usd", "cost_cum_usd", "last_hit", "train_job", "forks", "phase",
                  "created_at", "trained_at", "ttl_from", "ping_in_flight")


def cmd_list(args) -> tuple[dict, int]:
    state, _ = _load_run(args)
    touch_orchestrator(state)
    lost = reap_lost(state)
    reg = read_registry(state)
    now = _now()
    cfg = reg["config"]
    bases = []
    for name, b in sorted(reg["bases"].items()):
        row = {k: b.get(k) for k in LIST_BASE_KEYS}
        row["name"] = name
        row.update(base_warmth(b, cfg, now))
        row.update({"rebase_due": False, "retrain_due": False, "reasons": []})
        bases.append(row)
    jobs, running = [], []
    for j, e in sorted(reg["jobs"].items()):
        row = {k: e.get(k) for k in LIST_JOB_KEYS}
        if e.get("status") in RUNNING:
            info = _running_info(state, j, e, now)
            row.update({k: info[k] for k in ("elapsed_s", "last_event_s_ago", "messages")})
            running.append(j)
        else:
            row["turns"] = e.get("turns")
        jobs.append(row)
    return {"state": state, "bases": bases, "jobs": jobs, "running": running, "lost_now": lost}, EXIT_OK


def cmd_kill(args) -> tuple[dict, int]:
    state, reg = _load_run(args)
    if args.job not in reg["jobs"]:
        raise ForksError("unknown job", job=args.job)
    return kill_job(state, args.job, "kill"), EXIT_OK


def cmd_expire(args) -> tuple[dict, int]:
    state, reg = _load_run(args)
    names = sorted(reg["bases"]) if args.all else [args.base]
    unknown = [n for n in names if n not in reg["bases"]]
    if unknown:
        raise ForksError("unknown base", bases=unknown)
    retired = []
    with RegistryLock(state):
        reg = read_registry(state)
        for n in names:
            b = reg["bases"][n]
            if b.get("status") not in ("retired", "superseded"):
                b["status"] = "retired"
                b["retired_at"] = _iso()
                retired.append(n)
        write_registry(state, reg)
        targets = sorted(j for j, e in reg["jobs"].items()
                         if e.get("base") in names and e.get("status") in RUNNING)
    killed = [kill_job(state, j, "expire") for j in targets]
    return {"retired": retired, "bases": names, "killed": killed,
            "ok": all(k.get("ok", True) for k in killed)}, EXIT_OK


HANDLERS = {
    "where": cmd_where,
    "init": cmd_init,
    "train": cmd_train,
    "fork": cmd_fork,
    "continue": cmd_continue,
    "wait": cmd_wait,
    "list": cmd_list,
    "expire": cmd_expire,
    "kill": cmd_kill,
    "_supervise": cmd_supervise,
    "rebase": _not_implemented(phase=2),
    "usage": _not_implemented(phase=2),
    "tick": cmd_tick,
}


def run(argv) -> tuple[dict, int]:
    parser = build_parser()
    try:
        args, unknown = parser.parse_known_args(argv)
    except _HelpRequested as h:
        return {"help": h.text, "commands": [c for c in COMMANDS if not c.startswith("_")]}, EXIT_OK
    if unknown and args.command not in PHASE2_COMMANDS:
        parser.error("unrecognized arguments: %s" % " ".join(unknown))
    if args.command in CHILD_REFUSED and in_child():
        raise ForksError("`%s` is refused inside a fork (%s is set); a fork may only run where, list, "
                         "usage and wait" % (args.command, CHILD_MARKER), command=args.command)
    return HANDLERS[args.command](args)


def main(argv=None) -> int:
    if argv is None:
        argv = sys.argv[1:]
    try:
        out, code = run(argv)
    except ForksError as e:
        out, code = e.payload, e.code
    except Exception as e:  # never break the stdout invariant
        traceback.print_exc(file=sys.stderr)
        out, code = {"error": "internal error", "type": type(e).__name__, "detail": str(e)}, EXIT_USAGE
    _emit(out)
    return code


if __name__ == "__main__":
    sys.exit(main())