#!/usr/bin/env python3
"""Auto-resume watcher: move a stuck interactive session to another pooled account.

An interactive ``claude`` or ``codex`` that runs into a usage limit does NOT exit. Claude
idles on "continuing automatically at <reset>" (which can be days away) and codex sits on
"You've hit your usage limit ... try again at <date>", both waiting for a human. The shim
(bin/claude, bin/codex) spawns this watcher, detached, right before its byte-identical
``exec`` of an interactive TUI inside tmux. The watcher tails that session's own
transcript (claude) or rollout (codex) and, on a verdict it can act on:

1. asks the SHIM which account a relaunch would get (probe mode: the real candidate loop,
   ``pick=<acct> tier=<t>``, never an exec) — so it only stops a session there is
   somewhere to move to;
2. writes a single-use relaunch file (verdict + argv + chain state);
3. stops the client with SIGTERM (SIGKILL after a grace);
4. waits for the pane to be back at its SHELL and types
   `` CLAUDE_MULTIACC_AR=<token>:<sid> [CLAUDE_ACCOUNTS_ROOT=<root>] <shim path>`` there.

It never types into a TUI: Claude's limit menu can "Add funds" or spend the one-shot
``/limit-reset`` on a stray Enter. The relaunch goes back through the shim, which parks
the old account with the existing writers and runs ordinary selection.

Everything fails OPEN: an unreadable state file, a missing transcript, an ambiguous
rollout, a pane that is not the launching shell — the watcher simply exits and the
session stays exactly as it would be without it. docs/AUTORESUME.md is the reference.

CLI (``python3 -I lib/autoresume.py ...``)::

    watch --state FILE                                   the daemon (spawned by the shim)
    classify --provider claude|codex FILE [--since EPOCH] JSON verdicts (debug, tests)
    relaunch-argv --provider P --state FILE --sid ID --class C   JSON argv (tests)
    trust --acct-dir DIR --cwd PATH                      mark PATH trusted in DIR/.claude.json

Python 3.9 compatible, stdlib only. The pure parts (classifiers, argv allowlist and
sanitizer, relaunch-file writer, budget math, tmux command builders, the codex "try again
at" parser) are plain functions; every process side effect (ps, lsof, kill, tmux, the
probe) goes through ``Runner`` so the tests can inject a fake one.
"""

from __future__ import annotations

import calendar
import glob
import json
import os
import re
import secrets
import shutil
import signal
import stat
import string
import subprocess
import sys
import tempfile
import time
import traceback

# ---- vocabulary ---------------------------------------------------------------------

PROVIDERS = ("claude", "codex")

# The fixed prefix of the auto-resume prompt. A user record that starts with it is the
# watcher's own relaunch, never a human typing, so it never cancels a verdict.
PROMPT_PREFIX = "(claude-multiacc auto-resume)"

PROMPT_TEMPLATE = (
    PROMPT_PREFIX + " This session was restarted automatically{where} because {reason}. "
    "Continue the task from where you left off; the user has not sent a new message. "
    "Anything that was running in the background before the restart was stopped, so "
    "re-check it before relying on it, and do not repeat work that is already done."
)

REASONS = {
    "quota": "the previous account hit its usage limit",
    "model": "the previous account hit this model's usage limit",
    "auth": "the previous account's login failed",
    "blocked": "the previous account was refused",
    "transient": "an API error ended the last turn",
    "crash": "the previous process exited unexpectedly",
}

# Classes that MOVE the session to a different account (the probe must offer an eligible
# one that is not the current account). transient/crash may land on the same account.
ROTATE = ("quota", "auth", "blocked", "model")

# How long the chain keeps away from the account it is leaving. quota uses the server's
# own reset instead (QUOTA_FALLBACK when it named none that is still ahead).
AVOID_SECONDS = {"auth": 3600, "blocked": 6 * 3600, "model": 5 * 3600}
QUOTA_FALLBACK = 3600

CLAUDE_QUOTA_TYPES = ("five_hour", "seven_day")
CLAUDE_TRANSIENT = ("overloaded", "server_error", "unknown")
CLAUDE_BLOCKED = ("oauth_org_not_allowed", "account_on_hold", "billing_error",
                  "verification_required")
# "You've reached your Fable limit. Switch to another model ..." — a MODEL-scoped limit:
# the account still serves other models, so it is avoided for a while, never parked.
MODEL_TEXT = re.compile(r"reached your .{0,40} limit|switch to another model", re.I)
# User records that are machinery, not a human at the keyboard.
NON_HUMAN_PREFIXES = ("<task-notification", "<local-command", "<system-reminder")

CODEX_QUOTA = ("usage_limit_exceeded", "rate_limit_exceeded")
CODEX_AUTH = ("unauthorized",)
CODEX_TRANSIENT = ("server_overloaded", "internal_server_error", "http_connection_failed",
                   "response_stream_connection_failed", "response_stream_disconnected",
                   "response_too_many_failed_attempts")
CODEX_CANCEL = ("user_message", "task_started", "turn_aborted")

# Shells the typed ``VAR=x /path/cmd`` line works in. fish only learned that syntax in
# 3.1; nu, tcsh, xonsh, pwsh... never: a pane running one of them is never stopped.
SHELLS = ("zsh", "bash", "sh", "dash", "ksh")

TOKEN_ALPHABET = string.ascii_letters + string.digits
ACCT_RE = re.compile(r"^acct-[A-Za-z0-9._-]{1,64}$")
PANE_RE = re.compile(r"^%[0-9]{1,9}$")
CHAIN_RE = re.compile(r"^[A-Za-z0-9]{1,64}$")
TOKEN_RE = re.compile(r"^[A-Za-z0-9]{8,64}$")
CLASS_RE = re.compile(r"^[a-z_]{1,16}$")
# What the relaunch line may carry unquoted (the shims' gate refuses anything else).
TYPEABLE_RE = re.compile(r"^/[A-Za-z0-9/._+-]*$")
ISO_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(\.\d+)?"
                    r"(Z|[+-]\d{2}:?\d{2})?$")
TRY_AGAIN_RE = re.compile(
    r"try again at\s+([A-Za-z]{3})[A-Za-z]*\.?\s+(\d{1,2})(?:st|nd|rd|th)?,?\s+(\d{4})"
    r"\s+(\d{1,2}):(\d{2})\s*([AaPp])\.?[Mm]\.?")
MONTHS = {m: i for i, m in enumerate(
    ("jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"), 1)}

STATE_KEYS = ("v", "provider", "pid", "ppid", "acct", "acct_dir", "cwd", "launched", "tmux",
              "pane", "self", "acc_root", "depth", "avoid", "hist", "chain")
RELAUNCH_KEYS = ("v", "provider", "class", "acct", "reset", "rtype", "marked", "sid", "cwd",
                 "depth", "avoid", "hist", "chain")
HIST_KEEP = 32

# ---- knobs --------------------------------------------------------------------------
# Every timing constant is overridable as <P>_MULTIACC_AR_<NAME> (P = CLAUDE or CODEX) so
# the suites can run the whole stop/relaunch sequence in a second. The last five are not
# in the spec's list; they bound waits the spec fixes (probe 30 s, rollout discovery
# 5 min, subagent quiet 20 s capped at 15 min, the typed relaunch taking effect 10 s) and
# exist for the same reason.
KNOBS = {
    "POLL": 1.0,
    "GRACE": 5.0,
    "HOLD_REPROBE": 60.0,
    "MAX_DEPTH": 20,
    "ROTATE_PER_HOUR": 8,
    "TRANSIENT_PER_HOUR": 3,
    "CRASH_PER_10MIN": 2,
    "CRASH_MIN_RUNTIME": 60.0,
    "TERM_GRACE": 10.0,
    "PANE_WAIT": 15.0,
    "PROBE_TIMEOUT": 30.0,
    "DISCOVER_TIMEOUT": 300.0,
    "SUBAGENT_QUIET": 20.0,
    "SUBAGENT_CAP": 900.0,
    "RELAUNCH_WAIT": 10.0,
}
LADDER_DEFAULT = (30.0, 60.0, 120.0)
# After DISCOVER_TIMEOUT a codex watcher still looks for its rollout (a session first
# used an hour after it was opened is still a session), just this rarely.
DISCOVER_SLOW = 15.0
# The pane counts as back at its prompt only when two readings this far apart agree: one
# reading can land in the instant between two programs a shell loop runs.
SHELL_SETTLE = 0.5
MAX_TREE = 64          # more "descendants" than this is not one TUI's tree: kill none of them
# The cwd fallback is for a machine that cannot list open files; a lookup that is there
# but has failed this long counts as not there.
LOOKUP_FAILED = 30.0


def _knob(raw, default):
    try:
        val = float(raw)
    except (TypeError, ValueError):
        return default
    if val != val or val < 0 or val > 10 ** 7:  # NaN, negative, absurd
        return default
    return int(val) if isinstance(default, int) else val


def _ladder(raw):
    if not raw:
        return LADDER_DEFAULT
    steps = []
    for part in str(raw).split(","):
        val = _knob(part.strip(), -1.0)
        if val < 0:
            return LADDER_DEFAULT
        steps.append(float(val))
    return tuple(steps) or LADDER_DEFAULT


class Config:
    """The watcher's constants for one provider, read from the environment."""

    def __init__(self, provider, environ=None):
        env = os.environ if environ is None else environ
        up = provider.upper()
        pfx = up + "_MULTIACC_AR_"
        for name, default in KNOBS.items():
            setattr(self, name.lower(), _knob(env.get(pfx + name), default))
        # A zero poll would spin a core; 20 ms is already far below anything useful.
        self.poll = max(self.poll, 0.02)
        self.transient_ladder = _ladder(env.get(pfx + "TRANSIENT_LADDER"))
        self.anypane = env.get(pfx + "TEST_TMUX_ANYPANE") == "1"
        # The shim's own no-terminal test knob: a client with no terminal cannot be in its
        # foreground, so only then is the foreground requirement waived.
        self.test_tty = env.get(pfx + "TEST_TTY") == "1"
        self.debug = env.get(up + "_MULTIACC_AUTORESUME_DEBUG") == "1"
        override = env.get(up + "_MULTIACC_AUTORESUME_PROMPT")
        self.prompt = override if prompt_override_ok(override) else None


# ---- small parsers ------------------------------------------------------------------

def num_ok(val):
    """The shims' num_ok: digits only, at most 18 of them."""
    return isinstance(val, str) and val.isdigit() and val.isascii() and len(val) <= 18


def sess_id_ok(val):
    """The shims' sess_id_ok: hex and dashes, never a leading dash."""
    return (isinstance(val, str) and val != "" and not val.startswith("-")
            and all(c in "0123456789abcdefABCDEF-" for c in val))


def uuid_ok(val):
    """A session id a relaunch may name (the shims' ar_uuid_ok): dashed, bounded."""
    return sess_id_ok(val) and "-" in val and len(val) <= 64


def epoch_of(val):
    """A positive epoch (seconds) from a JSON number or digit string; None otherwise."""
    if isinstance(val, bool):
        return None
    if isinstance(val, (int, float)):
        if val != val or val <= 0 or val >= 10 ** 11:
            return None
        return int(val)
    if isinstance(val, str) and num_ok(val) and 0 < int(val) < 10 ** 11:
        return int(val)
    return None


def iso_to_epoch(val):
    """``2026-09-22T21:36:58.009Z`` -> float epoch; None when it is not that shape."""
    if not isinstance(val, str):
        return None
    m = ISO_RE.match(val.strip())
    if not m:
        return None
    try:
        base = calendar.timegm((int(m.group(1)), int(m.group(2)), int(m.group(3)),
                                int(m.group(4)), int(m.group(5)), int(m.group(6)), 0, 0, 0))
    except (ValueError, OverflowError):
        return None
    frac = float(m.group(7)) if m.group(7) else 0.0
    tz = m.group(8)
    off = 0
    if tz and tz != "Z":
        sign = -1 if tz[0] == "-" else 1
        digits = tz[1:].replace(":", "")
        off = sign * (int(digits[:2]) * 3600 + int(digits[2:]) * 60)
    return base + frac - off


def utc_iso(epoch):
    """Epoch -> ``YYYY-MM-DDTHH:MM:SSZ`` (the shims' marked_at / sel_log shape)."""
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(int(epoch)))


def parse_kv(text, keys):
    """``key=value`` lines -> dict of the known keys (last one wins, unknown ignored)."""
    out = {}
    for line in text.split("\n"):
        if "=" not in line:
            continue
        key, val = line.split("=", 1)
        if key in keys:
            out[key] = val
    return out


def avoid_parse(raw, now):
    """``acct-NN:until,...`` -> {acct: until} with only the entries still in force."""
    out = {}
    for entry in (raw or "").split(",")[:256]:
        acct, _, until = entry.partition(":")
        if not ACCT_RE.match(acct) or not num_ok(until) or int(until) <= now:
            continue
        out[acct] = max(out.get(acct, 0), int(until))
    return out


def avoid_format(entries):
    return ",".join("%s:%d" % (a, entries[a]) for a in sorted(entries))


def avoid_merge(raw, now, acct=None, until=None):
    """The chain's AVOID list with expired entries dropped, plus ``acct`` until ``until``
    (never shortening an entry already there)."""
    entries = avoid_parse(raw, now)
    if acct and until and int(until) > now:
        entries[acct] = max(entries.get(acct, 0), int(until))
    return avoid_format(entries)


def hist_parse(raw):
    """``class:epoch,...`` -> [(class, epoch)] — malformed entries dropped."""
    out = []
    for entry in (raw or "").split(",")[:1024]:
        cls, _, when = entry.partition(":")
        if CLASS_RE.match(cls) and num_ok(when):
            out.append((cls, int(when)))
    return out


def hist_append(raw, cls, now):
    """The chain history with this verdict appended; the newest HIST_KEEP entries."""
    entries = hist_parse(raw) + [(cls, int(now))]
    return ",".join("%s:%d" % e for e in entries[-HIST_KEEP:])


def make_token(length=16):
    return "".join(secrets.choice(TOKEN_ALPHABET) for _ in range(length))


# ---- budgets ------------------------------------------------------------------------

def budget_check(cls, depth, hist, now, cfg):
    """(ok, reason). A chain stops moving at MAX_DEPTH relaunches for good; the hourly
    windows only pause it (they slide, so a held verdict may proceed later)."""
    if depth >= cfg.max_depth:
        return False, "depth"
    entries = hist_parse(hist) if isinstance(hist, str) else list(hist)
    if cls in ROTATE:
        used = sum(1 for c, t in entries if c in ROTATE and now - t < 3600)
        if used >= cfg.rotate_per_hour:
            return False, "rotate-budget"
    elif cls == "transient":
        used = sum(1 for c, t in entries if c == "transient" and now - t < 3600)
        if used >= cfg.transient_per_hour:
            return False, "transient-budget"
    elif cls == "crash":
        used = sum(1 for c, t in entries if c == "crash" and now - t < 600)
        if used >= cfg.crash_per_10min:
            return False, "crash-budget"
    return True, ""


def ladder_wait(hist, now, ladder):
    """How long a transient verdict must stand before the watcher acts: the next step of
    the ladder after this chain's transient relaunches in the last hour."""
    entries = hist_parse(hist) if isinstance(hist, str) else list(hist)
    used = sum(1 for c, t in entries if c == "transient" and now - t < 3600)
    return ladder[min(used, len(ladder) - 1)] if ladder else 0.0


def avoid_until(verdict, now):
    """Until when the chain keeps away from the account a verdict is leaving."""
    cls = verdict["class"]
    if cls == "quota":
        reset = verdict.get("reset")
        return int(reset) if reset and reset > now else int(now) + QUOTA_FALLBACK
    return int(now) + AVOID_SECONDS.get(cls, QUOTA_FALLBACK)


# ---- claude classification ----------------------------------------------------------

def _message_text(rec):
    msg = rec.get("message")
    content = msg.get("content") if isinstance(msg, dict) else None
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return " ".join(b.get("text") for b in content
                        if isinstance(b, dict) and isinstance(b.get("text"), str))
    return ""


def claude_human_input(rec, prefixes=(PROMPT_PREFIX,)):
    """True for a main-thread user record a HUMAN produced (typed text, a slash command,
    a pasted image) — never a tool result, meta/compaction record, a harness
    notification, or the watcher's own auto-resume prompt."""
    if rec.get("isMeta") is True or rec.get("isCompactSummary") is True \
            or rec.get("isVisibleInTranscriptOnly") is True or "toolUseResult" in rec:
        return False
    msg = rec.get("message")
    content = msg.get("content") if isinstance(msg, dict) else None
    if isinstance(content, list):
        blocks = [b for b in content if isinstance(b, dict)]
        if not blocks or any(b.get("type") == "tool_result" for b in blocks):
            return False
        text = " ".join(b.get("text") for b in blocks
                        if b.get("type") == "text" and isinstance(b.get("text"), str))
    elif isinstance(content, str):
        text = content
    else:
        return False
    head = text.lstrip()
    if head.startswith(NON_HUMAN_PREFIXES):
        return False
    return not any(p and head.startswith(p) for p in prefixes)


def _verdict(cls, code, at, reset=None, rtype=""):
    return {"class": cls, "code": code or "?", "at": at, "reset": reset, "rtype": rtype,
            "marked": utc_iso(at) if at else ""}


def classify_claude(rec, prefixes=(PROMPT_PREFIX,)):
    """One transcript record -> None | ("error", verdict) | ("never", code) |
    ("cancel", why). No time filtering here — the tracker applies it."""
    if not isinstance(rec, dict) or rec.get("isSidechain") is True:
        return None
    typ = rec.get("type")
    if typ == "user":
        return ("cancel", "user") if claude_human_input(rec, prefixes) else None
    if typ != "assistant":
        return None  # system records ("continuing automatically at ...") never count
    if rec.get("isApiErrorMessage") is not True:
        return ("cancel", "assistant")  # a real reply: the user or auto-continue moved on
    code = rec.get("error") if isinstance(rec.get("error"), str) else ""
    at = iso_to_epoch(rec.get("timestamp"))
    if rec.get("apiError") == "model_requires_usage_credits":
        return ("error", _verdict("model", code or "model_requires_usage_credits", at))
    if code == "rate_limit":
        quota = rec.get("quotaLimits")
        if isinstance(quota, dict) and quota.get("status") == "rejected":
            rtype = quota.get("rateLimitType")
            if rtype in CLAUDE_QUOTA_TYPES:
                return ("error", _verdict("quota", code, at, epoch_of(quota.get("resetsAt")),
                                          rtype))
            # seven_day_opus, seven_day_overage_included, ...: one model's bucket.
            return ("error", _verdict("model", code, at))
        if MODEL_TEXT.search(_message_text(rec)):
            return ("error", _verdict("model", code, at))
        return ("error", _verdict("transient", code, at))
    if code in CLAUDE_TRANSIENT:
        return ("error", _verdict("transient", code, at))
    if code == "authentication_failed":
        return ("error", _verdict("auth", code, at))
    if code in CLAUDE_BLOCKED:
        return ("error", _verdict("blocked", code, at))
    # invalid_request, max_output_tokens, model_not_found, ...: another account would
    # fail the same way, so it is logged once and left alone.
    return ("never", code or "?")


class ClaudeTracker:
    """Folds a claude transcript's records into at most one pending verdict."""

    def __init__(self, since, prefixes=(PROMPT_PREFIX,)):
        self.since = since
        self.prefixes = prefixes
        self.pending = None
        self.pending_since = None

    def clear(self):
        self.pending = None
        self.pending_since = None

    def wants(self, raw):
        """Cheap byte prefilter: with nothing pending only an API error can matter (a
        cancel needs something to cancel), so a resumed 100 MB transcript is scanned for
        one substring instead of parsed line by line."""
        return self.pending is not None or b'"isApiErrorMessage"' in raw

    def feed(self, rec, now):
        ev = classify_claude(rec, self.prefixes)
        if ev is None:
            return None
        kind, val = ev
        stamp = iso_to_epoch(rec.get("timestamp"))
        if kind == "cancel":
            # A cancel only ever stops an action, so one without a timestamp still counts.
            if self.pending is None or (stamp is not None and stamp < self.since):
                return None
            self.clear()
            return ev
        # An error must PROVE it is from this launch: a resumed transcript carries every
        # rejection the session ever met, and an old one must never fire.
        if stamp is None or stamp < self.since:
            return None
        if kind == "never":
            return ev
        if self.pending is None or self.pending["class"] != val["class"]:
            self.pending_since = now
        self.pending = val
        return ev


# ---- codex classification -----------------------------------------------------------

def codex_error_code(info):
    """``codex_error_info`` is a plain string or a one-key object (a Rust enum variant
    with data, e.g. {"http_connection_failed": {"http_status_code": null}})."""
    if isinstance(info, str):
        return info
    if isinstance(info, dict) and info:
        key = next(iter(info))
        return key if isinstance(key, str) else ""
    return ""


def codex_rate_snapshot(rate_limits):
    """The usable window of a token_count's ``rate_limits``: only ``limit_id == "codex"``
    with a non-null window counts (the all-null ``premium`` line that precedes most
    errors says nothing), and of its windows the most-used one."""
    if not isinstance(rate_limits, dict) or rate_limits.get("limit_id") != "codex":
        return None
    best = None
    for key in ("primary", "secondary"):
        win = rate_limits.get(key)
        if not isinstance(win, dict):
            continue
        used = win.get("used_percent")
        used = float(used) if isinstance(used, (int, float)) and not isinstance(used, bool) \
            else -1.0
        cand = {"used_percent": used, "resets_at": epoch_of(win.get("resets_at")),
                "window_minutes": win.get("window_minutes")}
        if best is None or cand["used_percent"] > best["used_percent"]:
            best = cand
    return best


def parse_try_again(message):
    """"... try again at Sep 26th, 2026 11:22 AM." -> epoch, read as LOCAL time (that is
    how the codex CLI renders it); None when the message carries no such date."""
    if not isinstance(message, str):
        return None
    m = TRY_AGAIN_RE.search(message)
    if not m:
        return None
    month = MONTHS.get(m.group(1).lower())
    day, year, hour, minute = int(m.group(2)), int(m.group(3)), int(m.group(4)), \
        int(m.group(5))
    if not month or not 1 <= day <= 31 or not 1 <= hour <= 12 or minute > 59:
        return None
    hour = hour % 12 + (12 if m.group(6).lower() == "p" else 0)
    try:
        return int(time.mktime((year, month, day, hour, minute, 0, 0, 0, -1)))
    except (OverflowError, ValueError):
        return None


def codex_quota_reset(snapshot, message, now):
    """(reset epoch, window label) for a codex quota verdict — used for AVOID only (the
    shim writes a 10-minute error-cooldown, never a sticky weekly marker): the last codex
    token_count's most-used window, else the message's "try again at", else an hour."""
    if snapshot and snapshot.get("resets_at") and snapshot["resets_at"] > now:
        mins = snapshot.get("window_minutes")
        label = ""
        if isinstance(mins, (int, float)) and not isinstance(mins, bool):
            label = "7d" if mins >= 1440 else "5h"
        return int(snapshot["resets_at"]), label
    when = parse_try_again(message)
    if when and when > now:
        return when, ""
    return int(now) + QUOTA_FALLBACK, ""


def classify_codex(rec):
    """One rollout record -> None | ("meta", payload) | ("rate", snapshot) |
    ("error", verdict) | ("never", code) | ("cancel", type). Untimed."""
    if not isinstance(rec, dict):
        return None
    payload = rec.get("payload")
    if not isinstance(payload, dict):
        return None
    if rec.get("type") == "session_meta":
        return ("meta", payload)
    if rec.get("type") != "event_msg":
        return None
    ptype = payload.get("type")
    if ptype == "token_count":
        snap = codex_rate_snapshot(payload.get("rate_limits"))
        return ("rate", snap) if snap else None
    if ptype in CODEX_CANCEL:
        return ("cancel", ptype)
    if ptype != "task_complete" or not isinstance(payload.get("error"), dict):
        return None
    err = payload["error"]
    code = codex_error_code(err.get("codex_error_info"))
    at = epoch_of(payload.get("completed_at"))
    if at is None:
        at = iso_to_epoch(rec.get("timestamp"))
    if code in CODEX_QUOTA:
        cls = "quota"
    elif code in CODEX_AUTH:
        cls = "auth"
    elif code in CODEX_TRANSIENT:
        cls = "transient"
    else:
        return ("never", code or "?")
    verdict = _verdict(cls, code, at)
    # Kept in memory for the "try again at" parse only; never logged or printed.
    msg = err.get("message")
    verdict["_message"] = msg if isinstance(msg, str) else ""
    return ("error", verdict)


class CodexTracker:
    """Folds a codex rollout's records into at most one pending verdict."""

    def __init__(self, since):
        self.since = since
        self.pending = None
        self.pending_since = None
        self.snapshot = None

    def clear(self):
        self.pending = None
        self.pending_since = None

    def wants(self, raw):
        if b'"token_count"' in raw or b'"task_complete"' in raw:
            return True
        return self.pending is not None and (
            b'"user_message"' in raw or b'"task_started"' in raw or b'"turn_aborted"' in raw)

    def feed(self, rec, now):
        ev = classify_codex(rec)
        if ev is None or ev[0] == "meta":
            return None
        kind, val = ev
        if kind == "error":
            stamp = val["at"]
        else:
            stamp = iso_to_epoch(rec.get("timestamp"))
        if kind == "rate":
            if stamp is not None and stamp >= self.since:
                self.snapshot = val
            return None
        if kind == "cancel":
            if self.pending is None or (stamp is not None and stamp < self.since):
                return None
            self.clear()
            return ev
        if stamp is None or stamp < self.since:
            return None
        if kind == "never":
            return ev
        if val["class"] == "quota":
            val["reset"], val["rtype_window"] = codex_quota_reset(
                self.snapshot, val.get("_message"), now)
        if self.pending is None or self.pending["class"] != val["class"]:
            self.pending_since = now
        self.pending = val
        return ev


def make_tracker(provider, since, prefixes=(PROMPT_PREFIX,)):
    return ClaudeTracker(since, prefixes) if provider == "claude" else CodexTracker(since)


def verdict_public(verdict):
    """A verdict without its in-memory-only fields (the codex error message)."""
    if verdict is None:
        return None
    return {k: v for k, v in verdict.items() if not k.startswith("_")}


# ---- argv: allowlist and relaunch ---------------------------------------------------

def _value_ok(val):
    return isinstance(val, str) and val != "" and not val.startswith("-")


def claude_argv_parse(argv):
    """The argv shapes a claude relaunch can rebuild (mirrors bin/claude's ar_argv_ok) ->
    the kept options, with -c/--continue/--resume and the prompt dropped, or None for
    anything else."""
    kept, prompt, i = [], False, 0
    while i < len(argv):
        arg = argv[i]
        if arg in ("--dangerously-skip-permissions", "--allow-dangerously-skip-permissions"):
            kept.append(arg)
        elif arg in ("-c", "--continue"):
            pass
        elif arg in ("-r", "--resume"):
            if i + 1 >= len(argv) or not uuid_ok(argv[i + 1]):
                return None
            i += 1
        elif arg.startswith("--resume="):
            if not uuid_ok(arg[len("--resume="):]):
                return None
        elif arg in ("--model", "--effort", "--permission-mode"):
            if i + 1 >= len(argv) or not _value_ok(argv[i + 1]):
                return None
            kept += [arg, argv[i + 1]]
            i += 1
        elif arg.startswith("--model=") and len(arg) > len("--model="):
            kept.append(arg)
        elif arg.startswith("-"):
            return None
        else:
            # One positional, and it must look like a prompt: a single word may be a
            # subcommand (`claude doctor`), which is nothing to resume.
            if prompt or " " not in arg:
                return None
            prompt = True
        i += 1
    return kept


def codex_argv_parse(argv):
    """The argv shapes a codex relaunch can rebuild (mirrors bin/codex's ar_argv_ok) ->
    (kept options, resume target or None), or None for anything else."""
    kept, resume, target, prompt, i = [], False, None, False, 0
    while i < len(argv):
        arg = argv[i]
        if arg in ("--dangerously-bypass-approvals-and-sandbox", "--yolo"):
            kept.append(arg)
        elif arg in ("-m", "--model"):
            if i + 1 >= len(argv) or not _value_ok(argv[i + 1]):
                return None
            kept += [arg, argv[i + 1]]
            i += 1
        elif arg.startswith("--model=") and len(arg) > len("--model="):
            kept.append(arg)
        elif arg == "--last":
            if not resume or target is not None:
                return None
            target = "--last"
        elif arg == "resume" and not resume and not prompt:
            resume = True
        elif arg.startswith("-"):
            return None
        elif resume and target is None:
            if not uuid_ok(arg):
                return None
            target = arg
        else:
            if prompt or " " not in arg:
                return None
            prompt = True
        i += 1
    if resume and target is None:
        return None  # a bare `resume` is the picker: no session to follow
    return kept, target


def argv_ok(provider, argv):
    parse = claude_argv_parse if provider == "claude" else codex_argv_parse
    return parse(argv) is not None


def codex_resume_id(argv):
    """The session id a codex launch resumes by name (None for --last / a new session)."""
    parsed = codex_argv_parse(argv)
    if parsed and parsed[1] and parsed[1] != "--last":
        return parsed[1]
    return None


def prompt_override_ok(text):
    """An operator prompt must survive as ONE positional the next shim still accepts: not
    empty, not option-shaped, containing a space, and without NUL/newlines."""
    return (isinstance(text, str) and text.strip() != "" and not text.startswith("-")
            and " " in text and not any(c in text for c in "\0\n\r"))


def resume_prompt(cls, override=None):
    reason = REASONS[cls]
    if override and prompt_override_ok(override):
        return override.replace("{reason}", reason)
    where = "" if cls in ("crash", "transient") else " on another account"
    return PROMPT_TEMPLATE.format(where=where, reason=reason)


def prompt_prefixes(override=None):
    out = [PROMPT_PREFIX]
    if override and prompt_override_ok(override):
        out.append(override.strip()[:40])
    return tuple(out)


def relaunch_argv(provider, argv, sid, cls, prompt=None):
    """The argv the relaunched shim runs, built from the ORIGINAL argv.

    claude: the kept options + ``--resume <sid> <prompt>`` — no --model added: Claude
    restores the session's own model on --resume ([1m] included), and an explicit
    --model would switch that restore off.
    codex: ``resume <kept options> <sid> <prompt>``.
    None when the argv is outside the allowlist or the id is not a session id."""
    if not uuid_ok(sid) or cls not in REASONS:
        return None
    text = resume_prompt(cls, prompt)
    if provider == "claude":
        kept = claude_argv_parse(argv)
        if kept is None:
            return None
        return kept + ["--resume", sid, text]
    if provider == "codex":
        parsed = codex_argv_parse(argv)
        if parsed is None:
            return None
        return ["resume"] + parsed[0] + [sid, text]
    return None


# ---- relaunch / state files ---------------------------------------------------------

def format_relaunch(fields):
    """Relaunch file text in RELAUNCH_KEYS order. The shim reads it line by line, so a
    value carrying a line break (or NUL) cannot be written faithfully: ValueError."""
    lines = []
    for key in RELAUNCH_KEYS:
        val = "" if fields.get(key) is None else str(fields[key])
        if any(c in val for c in "\0\n\r"):
            raise ValueError("unwritable value for " + key)
        lines.append("%s=%s" % (key, val))
    return "\n".join(lines) + "\n"


def encode_argv(argv):
    """NUL-separated, NUL-terminated — what `while IFS= read -r -d ''` reads back."""
    out = b""
    for arg in argv:
        raw = arg.encode("utf-8", "surrogateescape")
        if b"\0" in raw:
            raise ValueError("NUL in argv")
        out += raw + b"\0"
    return out


def decode_argv(data):
    parts = data.split(b"\0")
    if parts and parts[-1] == b"":
        parts.pop()
    return [p.decode("utf-8", "surrogateescape") for p in parts]


def _atomic_write(path, data):
    tmp = "%s.tmp.%d" % (path, os.getpid())
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
        os.replace(tmp, path)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def write_relaunch(dirpath, token, fields, argv):
    """Write ``r-<token>.argv`` then ``r-<token>.relaunch`` (the file the shim tests for
    first), each atomically. Returns both paths; raises OSError/ValueError."""
    if not TOKEN_RE.match(token or ""):
        raise ValueError("bad token")
    text = format_relaunch(fields).encode("utf-8", "surrogateescape")
    data = encode_argv(argv)
    os.makedirs(dirpath, mode=0o700, exist_ok=True)
    base = os.path.join(dirpath, "r-" + token)
    _atomic_write(base + ".argv", data)
    try:
        _atomic_write(base + ".relaunch", text)
    except BaseException:
        remove_quietly(base + ".argv")
        raise
    return base + ".relaunch", base + ".argv"


def remove_quietly(*paths):
    for path in paths:
        try:
            os.unlink(path)
        except OSError:
            pass


def relaunch_fields(provider, verdict, acct, sid, cwd, depth, avoid, hist, chain):
    """The relaunch file's fields for one verdict. reset/rtype/marked only mean something
    to the shim for a quota verdict (claude marks ``client:<rtype>`` until ``reset``;
    codex writes its fixed cooldown)."""
    cls = verdict["class"]
    quota = cls == "quota"
    return {
        "v": "1", "provider": provider, "class": cls, "acct": acct,
        "reset": str(int(verdict["reset"])) if quota and verdict.get("reset") else "",
        "rtype": verdict.get("rtype", "") if quota and provider == "claude" else "",
        "marked": verdict.get("marked") or "",
        "sid": sid, "cwd": cwd, "depth": str(int(depth) + 1), "avoid": avoid,
        "hist": hist, "chain": chain,
    }


def argv_path_for(state_path):
    return state_path[:-len(".state")] + ".argv" if state_path.endswith(".state") \
        else state_path + ".argv"


def load_state(path):
    """The shim's ``<pid>.state`` -> typed dict, or None on anything malformed (the
    watcher then exits silently: an unsupervised session is today's behaviour)."""
    try:
        with open(path, "rb") as fh:
            text = fh.read(65536).decode("utf-8", "surrogateescape")
    except OSError:
        return None
    kv = parse_kv(text, STATE_KEYS)
    try:
        if kv.get("v") != "1" or kv.get("provider") not in PROVIDERS:
            return None
        for key in ("pid", "ppid", "launched"):
            if not num_ok(kv.get(key, "")):
                return None
        pid, ppid = int(kv["pid"]), int(kv["ppid"])
        if pid <= 1 or ppid <= 1:
            return None
        # A relative pool root (bin/claude writes $ACC_ROOT as given) resolves against
        # the watcher's cwd, which is the shim's own $PWD at spawn time.
        acct = kv.get("acct", "")
        acct_dir = os.path.abspath(kv["acct_dir"]) if kv.get("acct_dir") else ""
        acc_root = os.path.abspath(kv["acc_root"]) if kv.get("acc_root") else ""
        if not ACCT_RE.match(acct) or not acct_dir or not acc_root \
                or os.path.basename(acct_dir) != acct:
            return None
        for key in ("cwd", "self"):
            if not os.path.isabs(kv.get(key, "")):
                return None
        if tmux_socket(kv.get("tmux", "")) is None or not PANE_RE.match(kv.get("pane", "")):
            return None
        depth = kv.get("depth", "") or "0"
        if not num_ok(depth):
            return None
        chain = kv.get("chain", "")
        if chain and not CHAIN_RE.match(chain):
            chain = ""
    except (TypeError, ValueError):
        return None
    return {
        "provider": kv["provider"], "pid": pid, "ppid": ppid, "acct": acct,
        "acct_dir": acct_dir, "cwd": kv["cwd"], "launched": int(kv["launched"]),
        "tmux": kv["tmux"], "pane": kv["pane"], "self": kv["self"],
        "acc_root": acc_root, "depth": int(depth),
        "avoid": kv.get("avoid", ""), "hist": kv.get("hist", ""), "chain": chain,
    }


def load_argv(path):
    try:
        with open(path, "rb") as fh:
            data = fh.read(1 << 20)
    except OSError:
        return None
    return decode_argv(data)


PID_FILE = re.compile(r"^(\d{1,9})\.(state|argv|rollout|log)$")


def _pid_alive(pid):
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except OSError:
        return True
    return True


def prune(dirpath, now, keep=(), age=2 * 86400):
    """Remove state/relaunch/log leftovers older than two days (crashed watchers, tokens
    nobody typed). A per-pid file whose client still runs is a session that simply lasts
    longer than that — its ``.rollout`` claim is what keeps a neighbour's watcher off its
    rollout — so it stays."""
    try:
        entries = list(os.scandir(dirpath))
    except OSError:
        return
    for ent in entries:
        try:
            if ent.path in keep or not ent.is_file(follow_symlinks=False):
                continue
            if now - ent.stat(follow_symlinks=False).st_mtime > age:
                m = PID_FILE.match(ent.name)
                if m and int(m.group(1)) > 1 and _pid_alive(int(m.group(1))):
                    continue
                os.unlink(ent.path)
        except OSError:
            continue


# ---- tmux ---------------------------------------------------------------------------

def tmux_socket(tmux_env):
    """$TMUX is ``<socket>,<server pid>,<session>``: the socket is everything before the
    first comma, and it is always an absolute path."""
    if not isinstance(tmux_env, str):
        return None
    sock = tmux_env.split(",", 1)[0]
    return sock if sock.startswith("/") else None


def tmux_cmd(sock, *args):
    return ["tmux", "-S", sock] + list(args)


# '|' separated, the command LAST (it is the only field that may hold anything): the
# pane's root pid, whether a tmux mode (copy/view/tree/clock...) owns its keys, whether
# synchronize-panes would copy typed keys into every pane of the window, and the
# foreground program. An option or format an old tmux lacks expands to "" — read as off.
PANE_FORMAT = "#{pane_pid}|#{pane_in_mode}|#{synchronize-panes}|#{pane_current_command}"


def pane_query_argv(sock, pane):
    return tmux_cmd(sock, "display-message", "-p", "-t", pane, PANE_FORMAT)


def parse_pane_reply(out):
    """PANE_FORMAT's reply -> (pid, command, in_mode, synchronized) or None."""
    line = (out or "").strip().split("\n", 1)[0].strip()
    parts = line.split("|", 3)
    if len(parts) != 4 or not parts[0].isdigit():
        return None
    mode, sync = parts[1].strip(), parts[2].strip()
    return (int(parts[0]), parts[3].strip(), mode not in ("", "0"),
            sync not in ("", "0", "off"))


def mode_cancel_argv(sock, pane):
    """Leaves copy/view mode by a mode COMMAND, never a keystroke: a key sent to a pane in
    a mode is eaten by the mode, and one sent after the mode ended would reach whatever
    runs there. A mode this does not end is read again and the watcher gives up."""
    return tmux_cmd(sock, "send-keys", "-X", "-t", pane, "cancel")


def is_shell(command):
    """A shell from SHELLS, as ps or tmux name it (``-zsh`` for a login shell, or a
    path)."""
    return os.path.basename((command or "").strip().lstrip("-")) in SHELLS


NO_TTY = ("", "?", "??", "-")


def ps_foreground(status):
    """From Runner.proc_status: is the process in its terminal's foreground process group
    (ps marks it '+')? None when that cannot be told — no ps answer, or no controlling
    terminal to be in front of."""
    if not status or status[1] in NO_TTY:
        return None
    return "+" in status[0]


def typeable(path):
    """An absolute path the relaunch line can carry as one bare word in every shell of
    SHELLS: nothing to quote, expand or glob."""
    return isinstance(path, str) and bool(TYPEABLE_RE.match(path))


def relaunch_command(provider, token, sid, target, root=None):
    """What is typed into the pane's shell:
    `` <P>_MULTIACC_AR=<token>:<sid> [<P>_ACCOUNTS_ROOT=<root>] <target>``.
    The leading space keeps it out of history where HIST_IGNORE_SPACE is set. ``target``
    is the shim that launched the session (its own path, so neither an alias nor the
    shell's PATH can send it elsewhere); ``root`` is the pool, named only when it is not
    the default (the pane's shell does not have a one-shot `CLAUDE_ACCOUNTS_ROOT=/x
    claude`'s variable). The session id rides along so a token the shim cannot honour
    still names what to resume."""
    if provider not in PROVIDERS or not TOKEN_RE.match(token or "") or not uuid_ok(sid) \
            or not typeable(target) or (root is not None and not typeable(root)):
        raise ValueError("bad relaunch command")
    up = provider.upper()
    line = " %s_MULTIACC_AR=%s:%s" % (up, token, sid)
    if root is not None:
        line += " %s_ACCOUNTS_ROOT=%s" % (up, root)
    return line + " " + target


def send_keys_argvs(sock, pane, provider, token, sid, target, root=None):
    """-R resets the pane's terminal modes first (codex dies on SIGTERM without restoring
    them), C-u clears anything typed at the prompt meanwhile, then the command, then
    Enter."""
    return [
        tmux_cmd(sock, "send-keys", "-R", "-t", pane),
        tmux_cmd(sock, "send-keys", "-t", pane, "C-u"),
        tmux_cmd(sock, "send-keys", "-t", pane, "-l",
                 relaunch_command(provider, token, sid, target, root)),
        tmux_cmd(sock, "send-keys", "-t", pane, "Enter"),
    ]


def resume_hint(provider, sid):
    return ("claude --resume %s" if provider == "claude" else "codex resume %s") % sid


def notice_argv(sock, pane, provider, sid, sticky=True):
    """A status-line message (display-message without -p): shown to whoever watches the
    pane, never keyed into it. For a session the watcher stopped but could not relaunch.
    ``sticky``: ``-d 0`` keeps it up until a key is pressed (tmux >= 3.2; an older tmux
    refuses the flag, and the caller retries without it)."""
    return tmux_cmd(sock, "display-message", *(("-d", "0") if sticky else ()), "-t", pane,
                    "claude-multiacc: auto-resume failed — run: " + resume_hint(provider, sid))


# ---- probe --------------------------------------------------------------------------

PROBE_DROP = {
    "claude": ("CLAUDE_CONFIG_DIR", "CLAUDE_CODE_OAUTH_TOKEN", "CLAUDE_SHIM_ACTIVE",
               "CLAUDE_ACCOUNT", "CLAUDE_MULTIACC_AR"),
    "codex": ("CODEX_HOME", "CODEX_SHIM_ACTIVE", "CODEX_ACCOUNT", "CODEX_MULTIACC_AR"),
}


def probe_env(environ, provider, avoid):
    """The watcher inherited the picked account's exports; the probe must select from
    scratch, so they go, and probe mode + the AVOID list come in."""
    env = {k: v for k, v in environ.items() if k not in PROBE_DROP[provider]}
    up = provider.upper()
    env[up + "_MULTIACC_AR_PROBE"] = "1"
    env[up + "_MULTIACC_AR_AVOID"] = avoid or ""
    return env


def parse_probe(out):
    """The probe's ``pick=<acct> tier=<eligible|soft|hard|none>`` -> (pick, tier)."""
    pick, tier = "", ""
    for line in (out or "").split("\n"):
        m = re.match(r"^pick=(\S*) tier=(\S+)\s*$", line.strip())
        if m:
            pick, tier = m.group(1), m.group(2)
    if pick and not ACCT_RE.match(pick):
        return "", ""
    if tier not in ("eligible", "soft", "hard", "none"):
        return "", ""
    return pick, tier


def probe_allows(cls, pick, tier, current):
    """Rotate classes need an ELIGIBLE account that is not the one being left (holding
    beats landing on a soft/hard fallback); transient/crash may retry anywhere usable."""
    if cls in ROTATE:
        return tier == "eligible" and bool(pick) and pick != current
    return tier in ("eligible", "soft") and bool(pick)


# ---- side effects -------------------------------------------------------------------

def descendants(table, pid):
    """Every descendant pid of ``pid`` in a ps table [(pid, ppid, lstart)]."""
    kids = {}
    for p, pp, _ in table:
        kids.setdefault(pp, []).append(p)
    out, todo, seen = [], [pid], {pid}
    while todo:
        for child in kids.get(todo.pop(), ()):
            if child not in seen:
                seen.add(child)
                out.append(child)
                todo.append(child)
    return out


class Runner:
    """Every process side effect the watcher has. Tests substitute a fake."""

    def now(self):
        return time.time()

    def sleep(self, seconds):
        time.sleep(max(0.0, seconds))

    def alive(self, pid):
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            return False
        except PermissionError:
            return True
        except OSError:
            return False
        return True

    def gone(self, pid):
        """Dead, or a zombie nobody has reaped yet: it still owns the pid, so kill -0
        succeeds, but it runs nothing. Only the stop/wait sequence asks this — the pane's
        shell reaps its foreground job at once, so the per-tick check stays fork-free."""
        if not self.alive(pid):
            return True
        try:
            with open("/proc/%d/stat" % pid) as fh:
                return fh.read().rsplit(")", 1)[1].split()[0] == "Z"
        except (OSError, IndexError):
            pass
        rc, out = self.run(["ps", "-o", "stat=", "-p", str(pid)], 5)
        if rc != 0:
            return not self.alive(pid)
        return out.strip().startswith("Z")

    def signal(self, pid, sig):
        # Never init, never a process group (pid <= 0), never this watcher itself.
        if not isinstance(pid, int) or pid <= 1 or pid == os.getpid():
            return False
        try:
            os.kill(pid, sig)
        except OSError:
            return False
        return True

    def pgid(self, pid):
        try:
            return os.getpgid(pid)
        except OSError:
            return None

    def run(self, argv, timeout, env=None, cwd=None):
        """(returncode or None, stdout). stdout goes to a temp FILE, not a pipe: a
        grandchild left in the background would hold a pipe open and turn a bounded call
        into an unbounded one. The child gets its own process group so a timeout takes
        everything it started with it."""
        try:
            with tempfile.TemporaryFile() as out:
                proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=out,
                                        stderr=subprocess.DEVNULL, env=env, cwd=cwd,
                                        start_new_session=True, close_fds=True)
                try:
                    rc = proc.wait(timeout=timeout)
                except subprocess.TimeoutExpired:
                    try:
                        os.killpg(proc.pid, signal.SIGKILL)
                    except OSError:
                        pass
                    proc.wait()
                    rc = None
                out.seek(0)
                data = out.read(1 << 20)
        except (OSError, ValueError, subprocess.SubprocessError):
            return None, ""
        return rc, data.decode("utf-8", "replace")

    def proc_status(self, pid):
        """(stat, tty, comm) as ps prints them — the command last, it may hold spaces —
        or None when ps has nothing for the pid."""
        rc, out = self.run(["ps", "-o", "stat=,tty=,comm=", "-p", str(pid)], 10)
        parts = out.strip().split("\n", 1)[0].split(None, 2) if rc == 0 else []
        return tuple(parts) if len(parts) == 3 else None

    def can_list_open_files(self):
        return os.path.isdir("/proc/self/fd") or bool(
            shutil.which("lsof") or os.path.exists("/usr/sbin/lsof"))

    def lstart(self, pid):
        """The process's start time as ps prints it: with the pid, an identity that pid
        reuse cannot fake."""
        rc, out = self.run(["ps", "-o", "lstart=", "-p", str(pid)], 10)
        text = " ".join(out.split())
        return text if rc == 0 and text else None

    def ps_table(self):
        rc, out = self.run(["ps", "-A", "-o", "pid=,ppid=,lstart="], 10)
        rows = []
        for line in out.split("\n"):
            parts = line.split(None, 2)
            if len(parts) >= 2 and parts[0].isdigit() and parts[1].isdigit():
                rows.append((int(parts[0]), int(parts[1]),
                             " ".join(parts[2].split()) if len(parts) > 2 else ""))
        return rows

    def open_files(self, pid):
        """The paths ``pid`` holds open; None when the lookup itself failed."""
        if os.path.isdir("/proc/self/fd"):
            fd_dir = "/proc/%d/fd" % pid
            out = []
            try:
                names = os.listdir(fd_dir)
            except OSError:
                return None
            for name in names:
                try:
                    out.append(os.readlink(os.path.join(fd_dir, name)))
                except OSError:
                    continue
            return out
        lsof = shutil.which("lsof") or ("/usr/sbin/lsof" if os.path.exists("/usr/sbin/lsof")
                                        else None)
        if not lsof:
            return None
        rc, out = self.run([lsof, "-n", "-P", "-p", str(pid), "-Fn"], 10)
        if rc != 0:
            return None
        return [line[1:] for line in out.split("\n") if line.startswith("n")]


# ---- transcript / rollout tailing ---------------------------------------------------

class JsonlTail:
    """Appended complete lines of a JSONL file, as bytes. A partial trailing line waits
    for its newline; a truncated or replaced file starts over; one absurdly long line is
    skipped rather than buffered forever."""

    MAX_READ = 32 << 20
    MAX_LINE = 64 << 20

    def __init__(self, path):
        self.path = path
        self.off = 0
        self.buf = b""
        self.skip = False
        self.ino = None

    def read_lines(self):
        try:
            fh = open(self.path, "rb")
        except OSError:
            return []
        with fh:
            try:
                st = os.fstat(fh.fileno())
            except OSError:
                return []
            if (self.ino is not None and st.st_ino != self.ino) or st.st_size < self.off:
                self.off, self.buf, self.skip = 0, b"", False
            self.ino = st.st_ino
            if st.st_size == self.off:
                return []
            fh.seek(self.off)
            data = fh.read(self.MAX_READ)
        self.off += len(data)
        if self.skip:
            cut = data.find(b"\n")
            if cut < 0:
                return []
            data, self.skip = data[cut + 1:], False
        parts = (self.buf + data).split(b"\n")
        self.buf = parts.pop()
        if len(self.buf) > self.MAX_LINE:
            self.buf, self.skip = b"", True
        return [p for p in parts if p.strip()]


def read_json_line(raw):
    try:
        rec = json.loads(raw.decode("utf-8", "replace"))
    except ValueError:
        return None
    return rec if isinstance(rec, dict) else None


def read_first_line(path, limit=4 << 20):
    """Line 1 of a rollout (its session_meta), bounded: base instructions ride in it."""
    try:
        with open(path, "rb") as fh:
            line = fh.readline(limit)
    except OSError:
        return None
    return read_json_line(line.rstrip(b"\n"))


def rollout_meta(path):
    """The session_meta payload of a MAIN rollout (id == session_id), else None —
    subagent and fork files carry their parent's session_id under their own id."""
    rec = read_first_line(path)
    if not rec or rec.get("type") != "session_meta" or not isinstance(rec.get("payload"),
                                                                      dict):
        return None
    meta = rec["payload"]
    sid = meta.get("id")
    if not uuid_ok(sid) or meta.get("session_id") != sid:
        return None
    return meta


ROLLOUT_NAME = re.compile(r"^rollout-.*\.jsonl$")


def codex_fallback_candidates(sessions_dir, launched, cwd, exclude=()):
    """Rollouts under the two newest day dirs that could only be this launch's: a
    codex-tui main session in its cwd, CREATED since the launch (session_meta's own
    timestamp — a file's mtime only says someone wrote to it, and a neighbour session
    writes all day). The caller needs EXACTLY one."""
    days = sorted(glob.glob(os.path.join(sessions_dir, "[0-9]*", "[0-9]*", "[0-9]*")))[-2:]
    want = os.path.realpath(cwd)
    out = []
    for day in days:
        for path in glob.glob(os.path.join(day, "rollout-*.jsonl")):
            try:
                if os.stat(path).st_mtime < launched:
                    continue
            except OSError:
                continue
            if os.path.realpath(path) in exclude:
                continue
            meta = rollout_meta(path)
            if not meta or meta.get("originator") != "codex-tui":
                continue
            created = iso_to_epoch(meta.get("timestamp"))
            if created is None or created < launched - 2:
                continue
            mcwd = meta.get("cwd")
            if not isinstance(mcwd, str) or os.path.realpath(mcwd) != want:
                continue
            out.append(path)
    return out


# ---- the watcher --------------------------------------------------------------------

class GiveUp(Exception):
    pass


class Quoted(str):
    """A log value written as ``key="value"`` with its spaces kept (the resume command)."""


class Watcher:
    """One supervised session: tail, classify, decide, stop, relaunch — or just exit."""

    PLACEHOLDER_SID = "00000000-0000-0000-0000-000000000000"

    def __init__(self, statefile, st, argv, runner=None, environ=None):
        self.statefile = os.path.abspath(statefile)
        self.st = st
        self.provider = st["provider"]
        self.env = dict(os.environ if environ is None else environ)
        self.cfg = Config(self.provider, self.env)
        self.r = runner or Runner()
        self.argv = list(argv)
        self.pid = st["pid"]
        self.acct = st["acct"]
        self.acct_dir = st["acct_dir"]
        self.launched = st["launched"]
        self.stdir = os.path.dirname(self.statefile)
        self.ar_dir = os.path.join(st["acc_root"], "tmp", "autoresume")
        self.chain = st["chain"] or make_token(8)
        self.tracker = make_tracker(self.provider, self.launched - 2,
                                    prompt_prefixes(self.cfg.prompt))
        self.sid = None
        self.cwd = None
        self.tail = None
        self.transcript = None
        self.lstart = None
        self.started = None
        self.next_find = 0.0
        self.next_lsof = 0.0
        self.next_fallback = 0.0
        self.never_seen = set()
        self.hold_logged = False
        self.last_probe = None
        self.claim = os.path.join(self.stdir, "%d.rollout" % self.pid)
        self.root = None       # the pool, when the relaunch has to name it
        self.stopped = False   # True once this watcher has signalled the client
        self.tree = []         # [(pid, lstart)] of the client's descendants at the stop
        self._open_files_ok = None
        self.lookup_failed_at = None

    # -- logging --

    def log(self, event, **fields):
        """selection.log grammar: ``<UTC ISO> autoresume <event> k=v ...``. Field 2 is
        the word autoresume, never an account id — lib/report.py and the *-accounts tools
        read an acct-NN there as a pick. No transcript text, tokens or secrets."""
        parts = ["autoresume", event, "provider=" + self.provider, "chain=" + self.chain]
        for key, val in fields.items():
            if isinstance(val, Quoted):
                text = "".join(c for c in val if c.isprintable() and c != '"')[:120]
                parts.append('%s="%s"' % (key, text))
                continue
            text = re.sub(r"\s+", "_", str(val if val is not None and val != "" else "-"))
            parts.append("%s=%s" % (key, "".join(c for c in text if c.isprintable())[:120]))
        line = "%s %s\n" % (utc_iso(self.r.now()), " ".join(parts))
        try:
            with open(os.path.join(self.st["acc_root"], "selection.log"), "a") as fh:
                fh.write(line)
        except OSError:
            pass
        self.dbg(line.strip())

    def dbg(self, msg):
        if not self.cfg.debug:
            return
        try:
            with open(os.path.join(self.stdir, "%d.log" % self.pid), "a") as fh:
                fh.write("%s %s\n" % (utc_iso(time.time()), msg))
        except OSError:
            pass

    # -- the loop --

    def switched_off(self):
        return os.path.exists(os.path.join(self.st["acc_root"], "autoresume.off"))

    def run(self):
        if not self.r.alive(self.pid):
            return 0
        self.lstart = self.r.lstart(self.pid)
        if not self.lstart:
            return 0  # no identity to re-check before a signal: stay passive
        self.started = self.r.now()
        if relaunch_argv(self.provider, self.argv, self.PLACEHOLDER_SID, "quota") is None:
            self.log("giveup", reason="argv", pid=self.pid)
            return 0
        if not typeable(self.st["self"]) or not typeable(self.st["acc_root"]):
            # The relaunch line carries both bare (the shims' gate refuses the same
            # paths): never stop a session whose way back cannot be typed.
            self.log("giveup", reason="path", pid=self.pid)
            return 0
        default = os.path.join(self.env.get("HOME") or "/nonexistent",
                               ".%s-accounts" % self.provider)
        if os.path.normpath(self.st["acc_root"]) != os.path.normpath(default):
            self.root = self.st["acc_root"]
        self.log("watch", pid=self.pid, acct=self.acct, depth=self.st["depth"])
        try:
            while True:
                if self.switched_off():
                    self.dbg("autoresume.off: exiting")
                    return 0
                if not self.r.alive(self.pid):
                    return self.on_exit()
                self.read()
                if self.tracker.pending is not None:
                    rc = self.decide()
                    if rc is not None:
                        return rc
                self.r.sleep(self.cfg.poll)
        except GiveUp:
            return 0

    def giveup(self, reason, **fields):
        if self.stopped:
            # This watcher stopped the session and cannot bring it back: say so where
            # the operator looks — the log line carries the command that resumes it,
            # and so does the pane's status line (display-message, never a keystroke).
            fields["stopped"] = 1
            if self.sid:
                fields["resume"] = Quoted(resume_hint(self.provider, self.sid))
            self.notify_stopped()
        self.log("giveup", reason=reason, **fields)
        raise GiveUp(reason)

    def notify_stopped(self):
        sock = tmux_socket(self.st["tmux"])
        if sock and self.sid:
            rc, _ = self.r.run(notice_argv(sock, self.st["pane"], self.provider, self.sid), 10)
            if rc != 0:  # tmux < 3.2 has no -d: the plain notice, gone after display-time
                self.r.run(notice_argv(sock, self.st["pane"], self.provider, self.sid,
                                       sticky=False), 10)

    # -- readers --

    def read(self):
        if self.provider == "claude":
            self.read_claude()
        else:
            self.read_codex()

    def registry(self):
        """``<acct>/sessions/<pid>.json`` -> (sessionId, cwd) of THIS run, or None. A file
        older than the launch belongs to a crashed run whose pid was reused."""
        path = os.path.join(self.acct_dir, "sessions", "%d.json" % self.pid)
        try:
            with open(path, "rb") as fh:
                doc = json.loads(fh.read(1 << 20).decode("utf-8", "replace"))
        except (OSError, ValueError):
            return None
        if not isinstance(doc, dict) or not uuid_ok(doc.get("sessionId")):
            return None
        if isinstance(doc.get("pid"), int) and doc["pid"] != self.pid:
            return None
        started = doc.get("startedAt")
        if isinstance(started, (int, float)) and not isinstance(started, bool) \
                and started / 1000.0 < self.launched - 30:
            return None
        cwd = doc.get("cwd")
        return doc["sessionId"], (cwd if isinstance(cwd, str) and os.path.isabs(cwd) else None)

    def read_claude(self):
        now = self.r.now()
        reg = self.registry()
        if reg:
            sid, cwd = reg
            if sid != self.sid:
                # /clear, /resume: a different conversation — whatever was pending
                # belonged to the old one.
                if self.sid is not None:
                    self.dbg("session changed %s -> %s" % (self.sid, sid))
                    # A conversation picked with /resume carries whatever happened to it
                    # elsewhere since this launch — another process's limit error
                    # included, which is not this client's and not this account's.
                    # Only what is written from now on is ours.
                    self.tracker.since = max(self.tracker.since, now - 2)
                self.tracker.clear()
                self.sid, self.tail, self.transcript = sid, None, None
            if cwd:
                self.cwd = cwd
        if self.sid and self.tail is None and now >= self.next_find:
            self.next_find = now + 2.0
            found = sorted(glob.glob(os.path.join(self.acct_dir, "projects", "*",
                                                  self.sid + ".jsonl")))
            if found:
                self.transcript = found[0]
                self.tail = JsonlTail(found[0])
                self.dbg("transcript %s" % found[0])
        if self.tail is not None:
            self.pump()

    def read_codex(self):
        if self.tail is None and not self.discover():
            return
        self.pump()

    def pump(self):
        now = self.r.now()
        for raw in self.tail.read_lines():
            if not self.tracker.wants(raw):
                continue
            rec = read_json_line(raw)
            if rec is None:
                continue
            before = self.tracker.pending
            ev = self.tracker.feed(rec, now)
            if ev is not None:
                self.on_event(ev, before)

    def on_event(self, ev, before):
        kind, val = ev
        if kind == "never":
            if val not in self.never_seen:
                self.never_seen.add(val)
                self.log("never", code=val, sid=self.sid, acct=self.acct)
            return
        # Any change of state re-arms the hold log and probes immediately.
        self.hold_logged = False
        self.last_probe = None
        if kind == "cancel":
            self.dbg("verdict cancelled by %s" % val)
            return
        if before is None or before["class"] != val["class"]:
            self.log("detect", **{"class": val["class"], "code": val["code"],
                                  "sid": self.sid, "acct": self.acct,
                                  "reset": val.get("reset") or "-"})

    # -- codex rollout discovery --

    def claimed_rollouts(self):
        """Rollouts another LIVE watcher already follows, so the cwd fallback cannot hand
        this session a neighbour's rollout that happens to be the only candidate."""
        out = set()
        for path in glob.glob(os.path.join(self.stdir, "*.rollout")):
            if path == self.claim:
                continue
            name = os.path.basename(path)[:-len(".rollout")]
            if not name.isdigit() or not self.r.alive(int(name)):
                continue
            try:
                with open(path) as fh:
                    out.add(os.path.realpath(fh.read(4096).strip()))
            except OSError:
                continue
        return out

    def attach(self, path, meta, how):
        self.tail = JsonlTail(path)
        self.transcript = path
        self.sid = meta["id"]
        mcwd = meta.get("cwd")
        self.cwd = mcwd if isinstance(mcwd, str) and os.path.isabs(mcwd) else None
        try:
            _atomic_write(self.claim, os.path.realpath(path).encode("utf-8", "surrogateescape"))
        except OSError:
            pass
        self.dbg("rollout %s (%s)" % (path, how))
        return True

    def open_files_work(self):
        if self._open_files_ok is None:
            self._open_files_ok = bool(self.r.can_list_open_files())
        return self._open_files_ok

    def discover(self):
        """Find this session's rollout. codex writes it only once the session has a first
        message, which can be long after the launch, so the search never simply ends
        while the open-file lookup works: after DISCOVER_TIMEOUT only that lookup runs,
        every DISCOVER_SLOW s, for as long as the client lives."""
        sessions = os.path.join(self.acct_dir, "sessions")
        now = self.r.now()
        early = now - self.started <= self.cfg.discover_timeout
        if not early:
            if not self.open_files_work():
                self.giveup("discover", pid=self.pid)
            if now < self.next_lsof:
                return False
        # (a) the id the launch resumes by name.
        rid = codex_resume_id(self.argv) if early else None
        if rid:
            for path in sorted(glob.glob(os.path.join(sessions, "*", "*", "*",
                                                      "rollout-*-%s.jsonl" % rid))):
                meta = rollout_meta(path)
                if meta and meta["id"] == rid:
                    return self.attach(path, meta, "resume-id")
        # (b) the file the native child actually holds open (lsof is slow: every 2 s,
        #     every DISCOVER_SLOW s once the launch is old).
        if now >= self.next_lsof:
            self.next_lsof = now + (2.0 if early else DISCOVER_SLOW)
            table = self.r.ps_table()
            for proc in [self.pid] + descendants(table, self.pid):
                files = self.r.open_files(proc)
                if proc == self.pid:
                    if files is not None:
                        self.lookup_failed_at = None
                    elif self.lookup_failed_at is None:
                        self.lookup_failed_at = now
                for path in files or ():
                    if ROLLOUT_NAME.match(os.path.basename(path)):
                        meta = rollout_meta(path)
                        if meta:
                            return self.attach(path, meta, "open-file")
        # (c) only where (b) cannot see open files (no /proc, no lsof, or a lookup that
        #     has failed for LOOKUP_FAILED s): exactly one plausible new rollout for this
        #     cwd. Where (b) works it is the whole answer — a guess would adopt a
        #     neighbour session's rollout while this one has none yet, and a limit there
        #     would stop THIS session and resume the other one here.
        failing = self.lookup_failed_at is not None \
            and now - self.lookup_failed_at >= LOOKUP_FAILED
        if early and now >= self.next_fallback and (not self.open_files_work() or failing):
            self.next_fallback = now + 2.0
            cands = codex_fallback_candidates(sessions, self.launched, self.st["cwd"],
                                              self.claimed_rollouts())
            if len(cands) == 1:
                meta = rollout_meta(cands[0])
                if meta:
                    return self.attach(cands[0], meta, "cwd-fallback")
        return False

    # -- deciding --

    def subagents_busy(self, now):
        """A model/transient error on the main thread while subagents are still writing
        is not a dead session yet — stopping it would kill their work."""
        if self.provider != "claude" or not self.sid:
            return False
        newest, budget = 0.0, 5000  # stat at most this many files per check
        for base in glob.glob(os.path.join(self.acct_dir, "projects", "*", self.sid,
                                           "subagents")):
            for root, _dirs, files in os.walk(base):
                for name in files[:budget]:
                    try:
                        newest = max(newest, os.stat(os.path.join(root, name)).st_mtime)
                    except OSError:
                        pass
                budget -= min(len(files), budget)
                if budget <= 0:
                    return newest > 0 and now - newest < self.cfg.subagent_quiet
        return newest > 0 and now - newest < self.cfg.subagent_quiet

    def session_cwd(self):
        """Where the probe and the relaunch run. claude: the registry's cwd first
        (``--resume`` finds the transcript by it). codex: the launch directory first, so
        the relaunch repeats the user's own invocation; the rollout's cwd after it."""
        order = (self.cwd, self.st["cwd"]) if self.provider == "claude" \
            else (self.st["cwd"], self.cwd)
        for cand in order:
            if cand and os.path.isdir(cand):
                return cand
        return None

    def build_relaunch_argv(self, cls):
        return relaunch_argv(self.provider, self.argv, self.sid, cls, self.cfg.prompt)

    def hold(self, verdict, **fields):
        if not self.hold_logged:
            self.hold_logged = True
            self.log("hold", **dict({"class": verdict["class"], "sid": self.sid,
                                     "acct": self.acct}, **fields))

    def decide(self):
        """Act on the pending verdict once it has stood long enough; None = keep going."""
        verdict = self.tracker.pending
        now = self.r.now()
        cls = verdict["class"]
        age = now - self.tracker.pending_since
        need = self.cfg.grace
        if cls == "transient":
            need = max(need, ladder_wait(self.st["hist"], now, self.cfg.transient_ladder))
        if age < need:
            return None
        ok, why = budget_check(cls, self.st["depth"], self.st["hist"], now, self.cfg)
        if not ok:
            if why == "depth":
                self.giveup("depth", depth=self.st["depth"], sid=self.sid)
            self.hold(verdict, reason=why)
            return None
        if cls in ("model", "transient") and age < self.cfg.subagent_cap \
                and self.subagents_busy(now):
            return None
        if self.last_probe is not None and now - self.last_probe < self.cfg.hold_reprobe:
            return None
        self.last_probe = now
        argv = self.build_relaunch_argv(cls)
        if argv is None:
            self.giveup("argv", sid=self.sid)
        if cls in ROTATE:
            avoid = avoid_merge(self.st["avoid"], now, self.acct, avoid_until(verdict, now))
        else:
            avoid = avoid_merge(self.st["avoid"], now)
        cwd = self.session_cwd()
        if cwd is None:
            self.giveup("cwd", sid=self.sid)
        rc, out = self.r.run([self.st["self"]] + argv, self.cfg.probe_timeout,
                             env=probe_env(self.env, self.provider, avoid), cwd=cwd)
        pick, tier = parse_probe(out)
        self.dbg("probe rc=%s pick=%s tier=%s" % (rc, pick, tier))
        if not probe_allows(cls, pick, tier, self.acct):
            # Claude keeps its own auto-continue meanwhile; a cancel ends the hold.
            self.hold(verdict, reason="no-room", pick=pick or "-", tier=tier or "-")
            return None
        # The probe can take seconds: whatever the session wrote meanwhile (the user
        # typing, auto-continue getting through) is read before anything is stopped.
        self.read()
        if self.tracker.pending is not verdict or not self.r.alive(self.pid):
            return None
        return self.stop_and_relaunch(verdict, argv, avoid, probe_pick=pick)

    # -- stopping and relaunching --

    def pane_info(self, sock):
        rc, out = self.r.run(pane_query_argv(sock, self.st["pane"]), 10)
        return parse_pane_reply(out) if rc == 0 else None

    def pane_is_ours(self, info):
        return info is not None and (self.cfg.anypane or info[0] == self.st["ppid"])

    def pane_problem(self, sock, info):
        """None when the pane can take the relaunch; else why not. It must still be the
        launching shell's pane; with synchronize-panes on, typed keys would reach every
        pane of the window — other TUIs included; and a tmux mode (a pane scrolled back
        in copy mode stays there after the operator switches away) would eat the keys,
        so it is cancelled by command and read again."""
        if not self.pane_is_ours(info):
            return "pane"
        if info[3]:
            return "sync"
        if info[2]:
            self.r.run(mode_cancel_argv(sock, self.st["pane"]), 10)
            info = self.pane_info(sock)
            if not self.pane_is_ours(info):
                return "pane"
            if info[3]:
                return "sync"
            if info[2]:
                return "mode"
        return None

    def other_holder(self):
        """Another live claude whose registry (any account of this pool) names this
        session: stopping this one would hand the conversation to two clients."""
        if self.provider != "claude" or not self.sid:
            return None
        for path in glob.glob(os.path.join(self.st["acc_root"], "acct-*", "sessions",
                                           "*.json")):
            name = os.path.basename(path)[:-len(".json")]
            if not name.isdigit() or int(name) in (self.pid, 0, 1):
                continue
            try:
                with open(path, "rb") as fh:
                    doc = json.loads(fh.read(1 << 20).decode("utf-8", "replace"))
            except (OSError, ValueError):
                continue
            if isinstance(doc, dict) and doc.get("sessionId") == self.sid \
                    and self.r.alive(int(name)):
                return int(name)
        return None

    def ours_alive(self, pid, started):
        """``pid`` still runs (a zombie does not) and is still the process recorded with
        start time ``started`` — a pid reused since names someone else's process."""
        return not self.r.gone(pid) and self.r.lstart(pid) == started

    def job_gone(self):
        """The stopped client and the descendants recorded right before the signal: none
        of them left. Only that tree — never the client's whole process group, which also
        holds what the shim started in the background before its exec (the detached
        `limits --quiet` refresh), reparented away and none of this session's business."""
        return not any(self.ours_alive(p, s) for p, s in [(self.pid, self.lstart)] + self.tree)

    def shell_ready(self, sock):
        """One reading: the client's job gone, and the pane back at the launching shell's
        prompt — its program in front, and the shell itself holding the terminal (a job
        it runs next would take it away)."""
        if not self.job_gone():
            return False
        info = self.pane_info(sock)
        if not self.pane_is_ours(info) or not is_shell(info[1]):
            return False
        shell = self.r.proc_status(self.st["ppid"])
        return shell is not None and ps_foreground(shell) is not False

    def stop_process(self):
        """SIGTERM to the client, then after TERM_GRACE SIGKILL for whatever is left of it
        and of the descendants recorded before the TERM (codex's node wrapper and native
        binary, claude's tool and MCP children) — each re-verified by its start time
        right before the signal. False when the process was already gone (or replaced)
        before the first one."""
        if self.r.lstart(self.pid) != self.lstart:
            return False
        table = self.r.ps_table()
        starts = {p: ls for p, _pp, ls in table}
        # Only ever the client's own job. A process-tree read that goes wrong must not be
        # able to turn a session stop into a machine-wide SIGKILL (a broken copy of this
        # code under test once did exactly that): a descendant is kept only while it
        # shares the client's process group, it is never pid 1, this watcher, its parent
        # or the launching shell, and a "tree" too large to be one TUI's is not trusted at
        # all. A descendant without a start time could never be re-verified: left alone.
        job = self.r.pgid(self.pid)
        never = {1, os.getpid(), os.getppid(), self.st.get("ppid")}
        tree = [(p, starts[p]) for p in descendants(table, self.pid)
                if starts.get(p) and p > 1 and p not in never
                and job is not None and self.r.pgid(p) == job]
        self.tree = tree if len(tree) <= MAX_TREE else []
        if not self.r.signal(self.pid, signal.SIGTERM):
            return False
        self.stopped = True
        deadline = self.r.now() + self.cfg.term_grace
        while not self.job_gone() and self.r.now() < deadline:
            self.r.sleep(min(0.2, self.cfg.poll))
        for pid, started in [(self.pid, self.lstart)] + self.tree:
            if pid != self.pid and self.r.pgid(pid) != job:
                continue
            if self.ours_alive(pid, started):
                self.r.signal(pid, signal.SIGKILL)
        return True

    def wait_shell(self, sock):
        """Until shell_ready holds on two readings SHELL_SETTLE apart (one reading can
        fall between two programs of a `while :; do claude; done`), or PANE_WAIT ends."""
        deadline = self.r.now() + self.cfg.pane_wait
        ready_since = None
        while True:
            now = self.r.now()
            if self.shell_ready(sock):
                if ready_since is None:
                    ready_since = now
                elif now - ready_since >= SHELL_SETTLE:
                    return True
            else:
                ready_since = None
            if now >= deadline:
                return False
            self.r.sleep(min(0.2, self.cfg.poll))

    def stop_and_relaunch(self, verdict, argv, avoid, probe_pick="", kill=True):
        """0 once the typed relaunch has consumed its token; GiveUp otherwise. Every check
        that can refuse runs BEFORE the kill: a stopped session must always have its
        relaunch."""
        cls = verdict["class"]
        sock = tmux_socket(self.st["tmux"])
        # 1. The pane must still be the shell that launched the shim — otherwise the
        #    typed command would land somewhere else (a subshell, another program) — and
        #    that shell must be one the typed syntax works in: stopping a session whose
        #    pane runs nu or tcsh would leave it stopped with nowhere to relaunch.
        info = self.pane_info(sock) if sock else None
        if not self.pane_is_ours(info):
            self.giveup("pane", sid=self.sid, **{"class": cls})
        shell = self.r.proc_status(self.st["ppid"])
        if shell is None or not is_shell(shell[2]):
            self.giveup("pane-shell", sid=self.sid, **{"class": cls})
        #    Only a client in the foreground of its terminal, and not stopped, is killed:
        #    one suspended with Ctrl-Z leaves another program in front, which the relaunch
        #    must not be typed into.
        if kill:
            client = self.r.proc_status(self.pid)
            flags = client[0] if client else ""
            if not client or "T" in flags or flags.startswith("t") \
                    or ("+" not in flags and not self.cfg.test_tty):
                self.giveup("background", sid=self.sid, **{"class": cls})
            #    ...and only one that leads its own process group: a job of a job-control
            #    shell. Anything else (a client started by a script, `sh -c`, a pipeline
            #    stage) is not the prompt's job, and no prompt comes back when it ends.
            #    The suites' no-terminal knob waives this too: their harness has no job
            #    control to give the client a group of its own.
            if not self.cfg.test_tty and self.r.pgid(self.pid) != self.pid:
                self.giveup("pgrp", sid=self.sid, **{"class": cls})
        #    One conversation, one client: another live process holding this session
        #    means stopping this one would not free it.
        holder = self.other_holder()
        if holder:
            self.giveup("holder", sid=self.sid, holder=holder, **{"class": cls})
        cwd = self.session_cwd()
        if cwd is None:
            self.giveup("cwd", sid=self.sid, **{"class": cls})
        #    Last, as it is the one check that touches the pane: synchronized panes, or a
        #    tmux mode that a cancel does not end, refuse.
        why = self.pane_problem(sock, info)
        if why:
            self.giveup(why, sid=self.sid, **{"class": cls})
        if self.switched_off():
            raise GiveUp("off")
        # 2. The relaunch file exists BEFORE the kill: a kill always has its relaunch.
        now = self.r.now()
        token = make_token()
        fields = relaunch_fields(self.provider, verdict, self.acct, self.sid, cwd,
                                 self.st["depth"], avoid,
                                 hist_append(self.st["hist"], cls, now), self.chain)
        try:
            keys = send_keys_argvs(sock, self.st["pane"], self.provider, token, self.sid,
                                   self.st["self"], self.root)
            paths = write_relaunch(self.ar_dir, token, fields, argv)
        except (OSError, ValueError):
            self.giveup("write", sid=self.sid, **{"class": cls})
        # 3. Stop the client (a crash needs no stopping).
        if kill and not self.stop_process():
            remove_quietly(*paths)
            self.giveup("gone", sid=self.sid, **{"class": cls})
        # 4. Only a shell prompt may receive the command, once nothing of the stopped job
        #    is left — and the pane is read once more right before typing.
        if not self.wait_shell(sock):
            remove_quietly(*paths)
            self.giveup("shell" if self.job_gone() else "tree", sid=self.sid,
                        **{"class": cls})
        why = self.pane_problem(sock, self.pane_info(sock))
        if why:
            remove_quietly(*paths)
            self.giveup(why, sid=self.sid, **{"class": cls})
        # 5. Type the relaunch into the shell: through the shim that launched the session
        #    (and the pool it used), not whatever the shell's PATH says today.
        for i, cmd in enumerate(keys):
            rc, _ = self.r.run(cmd, 10)
            if rc != 0:
                # Before the command text is in, the token is useless; after it, the
                # line sits at the prompt and Enter would still work — the file expires
                # on its own (600 s) and the watcher's prune removes it.
                if i < 2:
                    remove_quietly(*paths)
                self.giveup("send-keys", sid=self.sid, **{"class": cls})
        # 6. The relaunched shim consumes the token first thing: a relaunch file still
        #    there after RELAUNCH_WAIT means the line never ran (a shell that ignored it, a
        #    prompt that was not one). The token is withdrawn — a line run later gets the
        #    shim's resume hint — and the operator is told how to resume.
        deadline = self.r.now() + self.cfg.relaunch_wait
        while os.path.exists(paths[0]):
            if self.r.now() >= deadline:
                remove_quietly(*paths)
                self.giveup("relaunch", sid=self.sid, **{"class": cls})
            self.r.sleep(min(0.2, self.cfg.poll))
        # 7.
        self.log("switch", **{"from": self.acct, "class": cls, "sid": self.sid,
                              "depth": int(self.st["depth"]) + 1, "probe": probe_pick or "-"})
        return 0

    def on_exit(self):
        """The client is gone and not by our hand. A clean exit (/exit, SIGHUP) deletes
        claude's registry; a crash or SIGKILL leaves it behind — that is the only crash
        signal, and it is claude-only."""
        if self.provider != "claude":
            return 0
        try:
            self.read()
        except GiveUp:
            return 0
        if self.tracker.pending is not None:
            return 0
        reg = self.registry()
        if reg is None:
            return 0
        now = self.r.now()
        if now - self.launched < self.cfg.crash_min_runtime:
            self.dbg("exit after %ds: too early to call a crash" % (now - self.launched))
            return 0
        self.sid = reg[0]
        if reg[1]:
            self.cwd = reg[1]
        ok, why = budget_check("crash", self.st["depth"], self.st["hist"], now, self.cfg)
        if not ok:
            self.log("giveup", reason=why, sid=self.sid, **{"class": "crash"})
            return 0
        verdict = _verdict("crash", "crash", now)
        argv = self.build_relaunch_argv("crash")
        if argv is None:
            self.log("giveup", reason="argv", sid=self.sid, **{"class": "crash"})
            return 0
        self.log("crash", sid=self.sid, acct=self.acct, runtime=int(now - self.launched))
        try:
            return self.stop_and_relaunch(verdict, argv, avoid_merge(self.st["avoid"], now),
                                          kill=False)
        except GiveUp:
            return 0

    def cleanup(self):
        remove_quietly(self.statefile, argv_path_for(self.statefile), self.claim)


# ---- CLI ----------------------------------------------------------------------------

def daemonize():
    """Own session, deaf to the terminal's signals: closing the pane or Ctrl-C in it
    must not take the watcher along before it can decide anything."""
    try:
        os.setsid()
    except OSError:
        pass
    for sig in (signal.SIGINT, signal.SIGHUP, signal.SIGTSTP):
        try:
            signal.signal(sig, signal.SIG_IGN)
        except (OSError, ValueError):
            pass


def cmd_watch(state_path, runner=None, environ=None):
    st = load_state(state_path)
    if st is None:
        return 0
    argv = load_argv(argv_path_for(state_path))
    if argv is None:
        return 0
    stdir = os.path.dirname(os.path.abspath(state_path))
    prune(stdir, time.time(),
          keep=(os.path.abspath(state_path), os.path.abspath(argv_path_for(state_path))))
    watcher = Watcher(state_path, st, argv, runner, environ)
    try:
        return watcher.run()
    except Exception:  # noqa: BLE001 — fail open: the session simply goes unsupervised
        watcher.dbg("watcher error: " + traceback.format_exc().replace("\n", " | "))
        return 0
    finally:
        watcher.cleanup()


def cmd_classify(provider, path, since):
    tracker = make_tracker(provider, since)
    events = []
    now = time.time()
    with open(path, "rb") as fh:
        for raw in fh:
            rec = read_json_line(raw.rstrip(b"\n"))
            if rec is None:
                continue
            ev = tracker.feed(rec, now)
            if ev is None:
                continue
            kind, val = ev
            if kind == "error":
                events.append(dict(verdict_public(val), kind="verdict"))
            elif kind == "never":
                events.append({"kind": "never", "code": val})
            else:
                events.append({"kind": "cancel", "by": val})
    print(json.dumps({"events": events, "pending": verdict_public(tracker.pending)},
                     sort_keys=True))
    return 0


def cmd_relaunch_argv(provider, state_path, sid, cls):
    st = load_state(state_path)
    argv = load_argv(argv_path_for(state_path))
    if st is None or argv is None or st["provider"] != provider:
        print("null")
        return 1
    cfg = Config(provider)
    out = relaunch_argv(provider, argv, sid, cls, cfg.prompt)
    print(json.dumps(out))
    return 0 if out is not None else 1


def cmd_trust(acct_dir, cwd):
    """Mark ``cwd`` trusted in ``<acct_dir>/.claude.json`` (projects[cwd].
    hasTrustDialogAccepted), for a relaunch landing on an account that never opened that
    directory: Claude asks "Is this a project you trust?" there even under
    --dangerously-skip-permissions, and nobody is watching to answer. The shim passes the
    logical $PWD, and Claude may key ``projects`` by the physical path, so both are marked
    when they differ. Every other key is kept; the write is atomic and keeps the file's
    mode. A missing or unreadable file, anything unexpected, or a file some other process
    (a running Claude) rewrote meanwhile, changes nothing."""
    path = os.path.join(acct_dir, ".claude.json")
    if not os.path.isabs(cwd or ""):
        return
    try:
        with open(path, "rb") as fh:
            st = os.fstat(fh.fileno())
            doc = json.loads(fh.read().decode("utf-8"))
    except (OSError, ValueError):
        return
    if not isinstance(doc, dict):
        return
    projects = doc.setdefault("projects", {})
    if not isinstance(projects, dict):
        return
    keys = [cwd]
    real = os.path.realpath(cwd)
    if real != cwd:
        keys.append(real)
    entries = [projects.setdefault(key, {}) for key in keys]
    if not all(isinstance(entry, dict) for entry in entries):
        return
    todo = [entry for entry in entries if entry.get("hasTrustDialogAccepted") is not True]
    if not todo:
        return
    for entry in todo:
        entry["hasTrustDialogAccepted"] = True
    handle, scratch = tempfile.mkstemp(prefix=".claude.json.", dir=os.path.dirname(path))
    try:
        with os.fdopen(handle, "w", encoding="utf-8") as fh:
            os.fchmod(fh.fileno(), stat.S_IMODE(st.st_mode))
            fh.write(json.dumps(doc, indent=2) + "\n")
        # Claude rewrites this file whenever it likes: one written since it was read
        # would lose that write to a stale copy, so it is left as it is.
        now = os.stat(path)
        if (now.st_mtime_ns, now.st_size) != (st.st_mtime_ns, st.st_size):
            remove_quietly(scratch)
            return
        os.replace(scratch, path)
    except BaseException:
        remove_quietly(scratch)
        raise


def _opts(args, names):
    """Minimal ``--name value`` parser (argparse would print usage to a daemon's
    /dev/null and exit 2 on anything odd; this just returns None)."""
    out, rest, i = {}, [], 0
    while i < len(args):
        if args[i] in names:
            if i + 1 >= len(args):
                return None, None
            out[args[i][2:]] = args[i + 1]
            i += 2
        else:
            rest.append(args[i])
            i += 1
    return out, rest


def main(argv=None):
    args = list(sys.argv[1:] if argv is None else argv)
    if not args:
        sys.stderr.write(__doc__.split("CLI", 1)[1].split("Python 3.9", 1)[0])
        return 2
    cmd, args = args[0], args[1:]
    if cmd == "watch":
        opts, _ = _opts(args, ("--state",))
        if not opts or "state" not in opts:
            return 2
        daemonize()
        try:
            return cmd_watch(opts["state"])
        except Exception:  # noqa: BLE001 — a watcher never surfaces a traceback
            return 0
    if cmd == "classify":
        opts, rest = _opts(args, ("--provider", "--since"))
        if opts is None or opts.get("provider") not in PROVIDERS or len(rest) != 1:
            return 2
        try:
            since = float(opts["since"]) if opts.get("since") else 0.0
            return cmd_classify(opts["provider"], rest[0], since)
        except (OSError, ValueError) as exc:
            sys.stderr.write("autoresume classify: %s\n" % exc)
            return 2
    if cmd == "relaunch-argv":
        opts, _ = _opts(args, ("--provider", "--state", "--sid", "--class"))
        if opts is None or opts.get("provider") not in PROVIDERS \
                or not all(k in opts for k in ("state", "sid", "class")):
            return 2
        return cmd_relaunch_argv(opts["provider"], opts["state"], opts["sid"], opts["class"])
    if cmd == "trust":
        # Called by the shim right before its exec: always 0, never a word of output.
        try:
            opts, _ = _opts(args, ("--acct-dir", "--cwd"))
            if opts and opts.get("acct-dir") and opts.get("cwd"):
                cmd_trust(opts["acct-dir"], opts["cwd"])
        except Exception:  # noqa: BLE001 — fail open
            pass
        return 0
    return 2


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