#!/usr/bin/env python3
"""Unit tests for lib/autoresume.py — the auto-resume watcher (docs/AUTORESUME.md).

Sandboxed: no real claude/codex, no network, no quota, never the operator's pools. Every
pool is a temp dir; fixture records are inline, in the shapes the live clients write
(claude 2.1.280 transcripts, codex 0.156.0 rollouts, 2026-09-22); every process side
effect of the watcher goes through a fake Runner with a fake clock. One test drives the
real ``watch`` CLI end to end against a disposable ``sleep`` process, a fake ``tmux`` on
PATH and a fake probe script. Runs under python3.9+.
"""

from __future__ import annotations

import contextlib
import importlib.util
import json
import os
from pathlib import Path
import re
import shutil
import signal
import stat
import subprocess
import sys
import tempfile
import time
import unittest
from unittest import mock

REPO = Path(__file__).resolve().parents[1]
LIB = REPO / "lib" / "autoresume.py"

_spec = importlib.util.spec_from_file_location("autoresume", str(LIB))
ar = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(ar)

SID = "55f5e7a8-955c-4ab0-a533-08852c296213"
SID2 = "30316873-ce79-4132-aa64-30702aaee95b"
CX_SID = "01a0cb0f-a719-7461-bba6-a59bdf2412cc"
CX_SID2 = "01a0cb0e-685c-79c3-a64f-618c4ada3bca"
LSTART = "Tue Sep 22 21:52:12 2026"


def iso(epoch):
    """The transcripts' own timestamp shape: UTC, milliseconds, Z."""
    return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(epoch)) + ".009Z"


# ---- claude transcript records (claude 2.1.280 shapes) ------------------------------

def cc_base(ts, typ, **extra):
    rec = {"parentUuid": "06c84ec0-a9db-4387-b695-a6e0bdf91685", "isSidechain": False,
           "type": typ, "uuid": "3a416522-6c97-44cf-adfe-00b44d2dcad5", "timestamp": iso(ts),
           "userType": "external", "entrypoint": "cli", "cwd": "/w", "sessionId": SID,
           "version": "2.1.280", "gitBranch": "HEAD"}
    rec.update(extra)
    return rec


def cc_api_error(ts, code, text="API Error: 529 Overloaded.", **extra):
    msg = {"id": "71cbd922-b089-4a6a-a628-acb45a61dc63", "model": "<synthetic>",
           "role": "assistant", "stop_reason": "stop_sequence", "type": "message",
           "usage": {"input_tokens": 0, "output_tokens": 0},
           "content": [{"type": "text", "text": text}]}
    rec = cc_base(ts, "assistant", message=msg, isApiErrorMessage=True, apiErrorStatus=500,
                  perTurnEffort=None, session_id=SID)
    if code is not None:
        rec["error"] = code
    rec.update(extra)
    return rec


def cc_quota(ts, rtype="five_hour", resets=None, **extra):
    rec = cc_api_error(ts, "rate_limit",
                       "You've hit your limit · resets 3:36am (Europe/Istanbul)",
                       apiErrorStatus=429,
                       quotaLimits={"status": "rejected", "resetsAt": resets,
                                    "unifiedRateLimitFallbackAvailable": False,
                                    "rateLimitType": rtype, "overageStatus": "rejected",
                                    "isUsingOverage": False})
    rec.update(extra)
    return rec


def cc_system(ts):
    return cc_base(ts, "system", subtype="informational", level="notice", isMeta=False,
                   content="Usage limit reached · continuing automatically at 3:36am "
                           "· esc or type to cancel")


def cc_user(ts, content, **extra):
    rec = cc_base(ts, "user", promptId="07b69536-f70e-474e-b2bb-666b0f9bb539",
                  message={"role": "user", "content": content})
    rec.update(extra)
    return rec


def cc_reply(ts, text="Done.", **extra):
    rec = cc_base(ts, "assistant", message={"model": "claude-opus-5-5", "role": "assistant",
                                            "type": "message",
                                            "content": [{"type": "text", "text": text}]})
    rec.update(extra)
    return rec


# ---- codex rollout records (codex 0.156.0 shapes) ------------------------------------

def cx_meta(ts, sid=CX_SID, session_id=None, cwd="/w", originator="codex-tui"):
    return {"timestamp": iso(ts), "type": "session_meta",
            "payload": {"session_id": sid if session_id is None else session_id, "id": sid,
                        "timestamp": iso(ts), "cwd": cwd, "runtime_workspace_roots": [cwd],
                        "originator": originator, "cli_version": "0.156.0", "source": "cli",
                        "thread_source": "user", "model_provider": "openai",
                        "history_mode": "paginated"}}


def cx_event(ts, payload):
    return {"timestamp": iso(ts), "ordinal": 9, "type": "event_msg", "payload": payload}


def cx_error(ts, info, message="boom", completed_at=None):
    done = int(ts) if completed_at is None else completed_at
    return cx_event(ts, {"type": "task_complete", "turn_id": "01a0cb0f-a763-7310-92e6",
                         "last_agent_message": None,
                         "error": {"message": message, "codex_error_info": info},
                         "started_at": done, "completed_at": done, "duration_ms": 32})


LIMIT_MSG = ("You’ve hit your usage limit. Visit https://chatgpt.com/codex/settings/usage "
             "to purchase more credits or try again at Sep 26th, 2026 12:39 AM.")


def cx_tokens(ts, primary=None, secondary=None, limit_id="codex", info=None):
    return cx_event(ts, {"type": "token_count", "info": info,
                         "rate_limits": {"limit_id": limit_id, "limit_name": None,
                                         "primary": primary, "secondary": secondary,
                                         "credits": None, "individual_limit": None,
                                         "spend_control_reached": None, "plan_type": None,
                                         "rate_limit_reached_type": None}})


def jl(records):
    return "".join(json.dumps(r) + "\n" for r in records)


def append(path, records):
    with open(path, "a") as fh:
        fh.write(jl(records))


@contextlib.contextmanager
def local_tz(name):
    old = os.environ.get("TZ")
    os.environ["TZ"] = name
    time.tzset()
    try:
        yield
    finally:
        if old is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = old
        time.tzset()


def cfg(provider="claude", **env):
    return ar.Config(provider, {"%s_MULTIACC_AR_%s" % (provider.upper(), k): str(v)
                                for k, v in env.items()})


# =====================================================================================
# classifiers
# =====================================================================================

class ClaudeClassifierTests(unittest.TestCase):
    T = 1790113000

    def verdict(self, rec):
        ev = ar.classify_claude(rec)
        self.assertIsNotNone(ev)
        self.assertEqual(ev[0], "error", ev)
        return ev[1]

    def test_session_and_weekly_rejections_are_quota(self):
        for rtype in ("five_hour", "seven_day"):
            v = self.verdict(cc_quota(self.T, rtype, resets=self.T + 9000))
            self.assertEqual(v["class"], "quota")
            self.assertEqual(v["rtype"], rtype)
            self.assertEqual(v["reset"], self.T + 9000)
            self.assertEqual(v["marked"], ar.utc_iso(self.T))
            self.assertEqual(v["code"], "rate_limit")

    def test_other_rejected_bucket_is_model_scoped(self):
        for rtype in ("seven_day_opus", "seven_day_overage_included", None):
            self.assertEqual(self.verdict(cc_quota(self.T, rtype, resets=self.T + 60))["class"],
                             "model", rtype)

    def test_model_limit_without_quota_limits(self):
        for text in ("You've reached your Fable limit. Run /usage-credits to keep going.",
                     "You've reached your Fable limit. Switch to another model to continue."):
            self.assertEqual(self.verdict(cc_api_error(self.T, "rate_limit", text))["class"],
                             "model")
        rec = cc_api_error(self.T, "rate_limit", "nope", apiError="model_requires_usage_credits")
        self.assertEqual(self.verdict(rec)["class"], "model")

    def test_other_rate_limit_is_transient(self):
        rec = cc_api_error(self.T, "rate_limit", "We're experiencing a temporary capacity issue.")
        self.assertEqual(self.verdict(rec)["class"], "transient")
        allowed = cc_quota(self.T)
        allowed["quotaLimits"]["status"] = "allowed_warning"
        self.assertEqual(self.verdict(allowed)["class"], "transient")

    def test_server_side_errors_are_transient(self):
        for code in ("overloaded", "server_error", "unknown"):
            self.assertEqual(self.verdict(cc_api_error(self.T, code))["class"], "transient")

    def test_auth_and_blocked(self):
        self.assertEqual(self.verdict(cc_api_error(
            self.T, "authentication_failed", "Not logged in · Please run /login"))["class"],
            "auth")
        for code in ("oauth_org_not_allowed", "account_on_hold", "billing_error",
                     "verification_required"):
            self.assertEqual(self.verdict(cc_api_error(self.T, code))["class"], "blocked", code)

    def test_everything_else_is_never(self):
        for code in ("invalid_request", "max_output_tokens", "model_not_found",
                     "cloud_credential_error", "brand_new_error"):
            self.assertEqual(ar.classify_claude(cc_api_error(self.T, code)), ("never", code))
        self.assertEqual(ar.classify_claude(cc_api_error(self.T, None)), ("never", "?"))

    def test_system_records_never_count(self):
        self.assertIsNone(ar.classify_claude(cc_system(self.T)))

    def test_sidechain_records_are_ignored(self):
        self.assertIsNone(ar.classify_claude(cc_quota(self.T, isSidechain=True)))
        self.assertIsNone(ar.classify_claude(cc_user(self.T, "go on then", isSidechain=True)))
        self.assertIsNone(ar.classify_claude(cc_reply(self.T, isSidechain=True)))

    def test_what_counts_as_the_user_typing(self):
        cancel = ("cancel", "user")
        self.assertEqual(ar.classify_claude(cc_user(self.T, "please continue")), cancel)
        self.assertEqual(ar.classify_claude(cc_user(
            self.T, "<command-name>/model</command-name>\n<command-args></command-args>")),
            cancel)
        self.assertEqual(ar.classify_claude(cc_user(
            self.T, [{"type": "text", "text": "look at this"},
                     {"type": "image", "source": {"type": "base64", "data": "AA=="}}])), cancel)
        self.assertEqual(ar.classify_claude(cc_user(
            self.T, [{"type": "image", "source": {"type": "base64", "data": "AA=="}}])), cancel)
        not_human = [
            cc_user(self.T, "<local-command-caveat>Caveat</local-command-caveat>", isMeta=True),
            cc_user(self.T, [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "ok"}]),
            cc_user(self.T, "result", toolUseResult={"stdout": ""}),
            cc_user(self.T, ar.resume_prompt("quota")),
            cc_user(self.T, "<task-notification><task-id>b1</task-id></task-notification>"),
            cc_user(self.T, "<local-command-stdout></local-command-stdout>"),
            cc_user(self.T, "This session is being continued from a previous conversation",
                    isCompactSummary=True),
            cc_user(self.T, []),
        ]
        for rec in not_human:
            self.assertIsNone(ar.classify_claude(rec), rec["message"])

    def test_a_real_reply_cancels(self):
        self.assertEqual(ar.classify_claude(cc_reply(self.T)), ("cancel", "assistant"))


class ClaudeTrackerTests(unittest.TestCase):
    LAUNCHED = 1790113000

    def tracker(self):
        return ar.ClaudeTracker(self.LAUNCHED - 2)

    def test_old_rejection_in_a_resumed_transcript_never_fires(self):
        tr = self.tracker()
        self.assertIsNone(tr.feed(cc_quota(self.LAUNCHED - 3, resets=self.LAUNCHED + 99), 1))
        self.assertIsNone(tr.feed(cc_quota(self.LAUNCHED - 86400), 1))
        self.assertIsNone(tr.pending)
        # Two seconds of slack for the shim's clock read.
        ev = tr.feed(cc_quota(self.LAUNCHED - 1, resets=self.LAUNCHED + 99), 5)
        self.assertEqual(ev[0], "error")
        self.assertEqual(tr.pending["class"], "quota")
        self.assertEqual(tr.pending_since, 5)

    def test_error_without_a_timestamp_cannot_prove_it_is_new(self):
        rec = cc_quota(self.LAUNCHED + 5)
        del rec["timestamp"]
        tr = self.tracker()
        self.assertIsNone(tr.feed(rec, 1))
        self.assertIsNone(tr.pending)

    def test_cancel_and_rearm(self):
        tr = self.tracker()
        self.assertIsNone(tr.feed(cc_user(self.LAUNCHED + 1, "hi"), 1))  # nothing to cancel
        tr.feed(cc_quota(self.LAUNCHED + 2), 2)
        self.assertIsNone(tr.feed(cc_system(self.LAUNCHED + 2), 3))
        self.assertIsNotNone(tr.pending)
        self.assertIsNone(tr.feed(cc_user(self.LAUNCHED + 3, ar.resume_prompt("quota")), 3))
        self.assertIsNotNone(tr.pending)
        self.assertEqual(tr.feed(cc_user(self.LAUNCHED + 4, "wait, stop"), 4), ("cancel", "user"))
        self.assertIsNone(tr.pending)
        tr.feed(cc_api_error(self.LAUNCHED + 5, "overloaded"), 5)
        self.assertEqual(tr.pending["class"], "transient")
        self.assertEqual(tr.feed(cc_reply(self.LAUNCHED + 6), 6), ("cancel", "assistant"))
        self.assertIsNone(tr.pending)

    def test_a_cancel_older_than_the_launch_does_not_count(self):
        tr = self.tracker()
        tr.feed(cc_quota(self.LAUNCHED + 2), 2)
        self.assertIsNone(tr.feed(cc_user(self.LAUNCHED - 60, "old"), 3))
        self.assertIsNotNone(tr.pending)

    def test_pending_since_restarts_only_on_a_class_change(self):
        tr = self.tracker()
        tr.feed(cc_quota(self.LAUNCHED + 1), 10)
        tr.feed(cc_quota(self.LAUNCHED + 2), 20)
        self.assertEqual(tr.pending_since, 10)
        tr.feed(cc_api_error(self.LAUNCHED + 3, "authentication_failed"), 30)
        self.assertEqual((tr.pending["class"], tr.pending_since), ("auth", 30))

    def test_never_leaves_the_pending_verdict_alone(self):
        tr = self.tracker()
        tr.feed(cc_quota(self.LAUNCHED + 1), 10)
        self.assertEqual(tr.feed(cc_api_error(self.LAUNCHED + 2, "invalid_request"), 11),
                         ("never", "invalid_request"))
        self.assertEqual(tr.pending["class"], "quota")

    def test_prefilter(self):
        tr = self.tracker()
        err = json.dumps(cc_quota(self.LAUNCHED + 1), separators=(",", ":")).encode()
        user = json.dumps(cc_user(self.LAUNCHED + 1, "hi")).encode()
        self.assertTrue(tr.wants(err))
        self.assertFalse(tr.wants(user))
        tr.feed(json.loads(err), 1)
        self.assertTrue(tr.wants(user))


class CodexClassifierTests(unittest.TestCase):
    LAUNCHED = 1790113100

    def tracker(self):
        return ar.CodexTracker(self.LAUNCHED - 2)

    def test_error_info_string_or_one_key_object(self):
        cases = {
            "usage_limit_exceeded": "quota", "rate_limit_exceeded": "quota",
            "unauthorized": "auth", "server_overloaded": "transient",
            "internal_server_error": "transient", "http_connection_failed": "transient",
            "response_stream_connection_failed": "transient",
            "response_stream_disconnected": "transient",
            "response_too_many_failed_attempts": "transient",
        }
        for code, cls in cases.items():
            for info in (code, {code: {"http_status_code": None}}):
                ev = ar.classify_codex(cx_error(self.LAUNCHED, info))
                self.assertEqual((ev[0], ev[1]["class"], ev[1]["code"]), ("error", cls, code))
        for code in ("cyber_policy", "context_window_exceeded", "bad_request", "other"):
            self.assertEqual(ar.classify_codex(cx_error(self.LAUNCHED, code)), ("never", code))
            self.assertEqual(ar.classify_codex(cx_error(self.LAUNCHED, {code: None})),
                             ("never", code))
        self.assertEqual(ar.classify_codex(cx_error(self.LAUNCHED, None)), ("never", "?"))

    def test_a_clean_turn_is_nothing(self):
        ok = cx_event(self.LAUNCHED, {"type": "task_complete", "last_agent_message": "hi",
                                      "completed_at": self.LAUNCHED})
        self.assertIsNone(ar.classify_codex(ok))
        self.assertEqual(ar.classify_codex(cx_meta(self.LAUNCHED))[0], "meta")
        self.assertIsNone(ar.classify_codex({"type": "response_item", "payload": {}}))

    def test_cancel_events(self):
        tr = self.tracker()
        for ptype in ("user_message", "task_started", "turn_aborted"):
            tr.feed(cx_error(self.LAUNCHED + 1, "unauthorized"), 1)
            self.assertEqual(tr.feed(cx_event(self.LAUNCHED + 2, {"type": ptype}), 2),
                             ("cancel", ptype))
            self.assertIsNone(tr.pending)

    def test_time_filter_uses_completed_at(self):
        tr = self.tracker()
        # A resumed rollout: the record is re-read now, but the turn finished long ago.
        self.assertIsNone(tr.feed(cx_error(self.LAUNCHED + 50, "usage_limit_exceeded",
                                           completed_at=self.LAUNCHED - 3600), 1))
        self.assertIsNone(tr.pending)
        tr.feed(cx_error(self.LAUNCHED + 50, "usage_limit_exceeded"), 1)
        self.assertEqual(tr.pending["at"], self.LAUNCHED + 50)

    def test_premium_null_line_is_skipped_and_the_message_date_is_used(self):
        with local_tz("UTC"):
            tr = self.tracker()
            tr.feed(cx_tokens(self.LAUNCHED + 1, limit_id="premium"), 1)
            tr.feed(cx_tokens(self.LAUNCHED + 1), 1)  # codex, both windows null
            self.assertIsNone(tr.snapshot)
            tr.feed(cx_error(self.LAUNCHED + 1, "usage_limit_exceeded", LIMIT_MSG), 1)
            self.assertEqual(tr.pending["reset"], 1790383140)  # 2026-09-26 00:39 UTC
            self.assertNotIn("_message", ar.verdict_public(tr.pending))

    def test_codex_window_snapshot_wins(self):
        tr = self.tracker()
        now = self.LAUNCHED + 5
        tr.feed(cx_tokens(self.LAUNCHED + 1,
                          primary={"used_percent": 42.0, "window_minutes": 300,
                                   "resets_at": now + 600},
                          secondary={"used_percent": 100.0, "window_minutes": 10080,
                                     "resets_at": now + 86400}), now)
        tr.feed(cx_tokens(self.LAUNCHED + 2, limit_id="premium"), now)  # must not erase it
        tr.feed(cx_error(self.LAUNCHED + 3, "usage_limit_exceeded", LIMIT_MSG), now)
        self.assertEqual(tr.pending["reset"], now + 86400)
        self.assertEqual(tr.pending["rtype_window"], "7d")

    def test_stale_snapshot_falls_back(self):
        now = self.LAUNCHED + 5
        # older than the launch: ignored
        tr = self.tracker()
        tr.feed(cx_tokens(self.LAUNCHED - 600, primary={"used_percent": 99.0,
                                                        "window_minutes": 300,
                                                        "resets_at": now + 50}), now)
        self.assertIsNone(tr.snapshot)
        # a reset already behind us, and no date in the message: an hour
        tr.feed(cx_tokens(self.LAUNCHED + 1, primary={"used_percent": 99.0,
                                                      "window_minutes": 300,
                                                      "resets_at": now - 5}), now)
        tr.feed(cx_error(self.LAUNCHED + 2, "rate_limit_exceeded", "slow down"), now)
        self.assertEqual(tr.pending["reset"], now + 3600)

    def test_try_again_at_is_local_time(self):
        with local_tz("UTC"):
            self.assertEqual(ar.parse_try_again("try again at Sep 26th, 2026 11:22 AM."),
                             1790421720)
            self.assertEqual(ar.parse_try_again("try again at Sep 26th, 2026 12:39 AM."),
                             1790383140)
            self.assertEqual(ar.parse_try_again("try again at Oct 1st, 2026 12:05 PM"),
                             1790856300)
            self.assertEqual(ar.parse_try_again("try again at Nov 2nd, 2026 3:00 pm."),
                             1793631600)
            self.assertEqual(ar.parse_try_again("try again at December 3rd, 2026 9:15 AM"),
                             1796289300)
        if os.path.exists("/usr/share/zoneinfo/America/New_York"):
            with local_tz("America/New_York"):  # EDT, UTC-4
                self.assertEqual(ar.parse_try_again(LIMIT_MSG), 1790383140 + 4 * 3600)
        for bad in (None, "", "try again later", "try again at Foo 3rd, 2026 1:00 AM",
                    "try again at Sep 26th, 2026 13:00 PM", "try again at Sep 40th, 2026 1:00 AM"):
            self.assertIsNone(ar.parse_try_again(bad), bad)

    def test_prefilter(self):
        tr = self.tracker()
        self.assertTrue(tr.wants(json.dumps(cx_tokens(1)).encode()))
        self.assertTrue(tr.wants(json.dumps(cx_error(1, "unauthorized")).encode()))
        start = json.dumps(cx_event(1, {"type": "task_started"})).encode()
        self.assertFalse(tr.wants(start))
        self.assertFalse(tr.wants(b'{"type":"response_item","payload":{"type":"message"}}'))
        tr.feed(cx_error(self.LAUNCHED, "unauthorized"), 1)
        self.assertTrue(tr.wants(start))


# =====================================================================================
# argv
# =====================================================================================

class ArgvTests(unittest.TestCase):
    def rl(self, provider, argv, cls="quota", prompt=None, sid=None):
        sid = sid or (SID if provider == "claude" else CX_SID)
        return ar.relaunch_argv(provider, argv, sid, cls, prompt)

    def test_claude_relaunch_drops_resume_continue_and_the_prompt(self):
        p = ar.resume_prompt("quota")
        self.assertEqual(self.rl("claude", ["--dangerously-skip-permissions", "fix the tests"]),
                         ["--dangerously-skip-permissions", "--resume", SID, p])
        self.assertEqual(self.rl("claude", ["-c"]), ["--resume", SID, p])
        self.assertEqual(self.rl("claude", ["--continue", "--effort", "high"]),
                         ["--effort", "high", "--resume", SID, p])
        self.assertEqual(self.rl("claude", ["-r", SID2, "--permission-mode", "plan"]),
                         ["--permission-mode", "plan", "--resume", SID, p])
        self.assertEqual(self.rl("claude", ["--resume=" + SID2,
                                            "--allow-dangerously-skip-permissions"]),
                         ["--allow-dangerously-skip-permissions", "--resume", SID, p])
        self.assertEqual(self.rl("claude", []), ["--resume", SID, p])

    def test_claude_model_is_only_the_launchs_own(self):
        # Claude restores the session's model on --resume ([1m] included); an added
        # --model would switch that off. An explicit one from the launch is kept as is.
        p = ar.resume_prompt("model")
        self.assertEqual(self.rl("claude", ["hello there"], "model"), ["--resume", SID, p])
        self.assertEqual(self.rl("claude", ["--model", "sonnet"], "model"),
                         ["--model", "sonnet", "--resume", SID, p])
        self.assertEqual(self.rl("claude", ["--model=sonnet"], "model"),
                         ["--model=sonnet", "--resume", SID, p])

    def test_claude_allowlist_rejects(self):
        for argv in (["-p", "hi there"], ["--print"], ["doctor"], ["mcp", "list"],
                     ["one prompt", "two prompts"], ["-r"], ["-r", "not-hex-zz"],
                     ["-r", "abcdef"], ["--resume=-x"], ["--model"], ["--model", "-p"],
                     ["--model="], ["--verbose"], ["--", "hi there"], ["-x"]):
            self.assertIsNone(ar.claude_argv_parse(argv), argv)
            self.assertIsNone(self.rl("claude", argv), argv)

    def test_codex_relaunch(self):
        p = ar.resume_prompt("quota")
        self.assertEqual(self.rl("codex", ["--dangerously-bypass-approvals-and-sandbox"]),
                         ["resume", "--dangerously-bypass-approvals-and-sandbox", CX_SID, p])
        self.assertEqual(self.rl("codex", ["--yolo", "resume", CX_SID2, "keep going now"]),
                         ["resume", "--yolo", CX_SID, p])
        self.assertEqual(self.rl("codex", ["resume", "--last", "-m", "gpt-5.5"]),
                         ["resume", "-m", "gpt-5.5", CX_SID, p])
        self.assertEqual(self.rl("codex", ["--model=gpt-5", "fix the build"]),
                         ["resume", "--model=gpt-5", CX_SID, p])
        self.assertEqual(ar.codex_resume_id(["--yolo", "resume", CX_SID2]), CX_SID2)
        self.assertIsNone(ar.codex_resume_id(["resume", "--last"]))
        self.assertIsNone(ar.codex_resume_id(["do the thing"]))

    def test_codex_allowlist_rejects(self):
        for argv in (["exec", "hi there"], ["e"], ["-p", "work"], ["--profile", "x"],
                     ["resume"], ["--last"], ["resume", "--last", CX_SID],
                     ["resume", "notauuid"], ["hi there", "resume", CX_SID],
                     ["resume", CX_SID, "a b", "c d"], ["-c", "model=x"], ["-m"],
                     ["-m", "-x"], ["resume", "resume"], ["login"], ["app-server"]):
            self.assertIsNone(ar.codex_argv_parse(argv), argv)
            self.assertIsNone(self.rl("codex", argv), argv)

    def test_relaunch_refuses_a_bad_session_id_or_class(self):
        for sid in ("", "-rf", "abcdef", "../x", SID + "zz", "a-" * 40):
            self.assertIsNone(ar.relaunch_argv("claude", [], sid, "quota"), sid)
        self.assertIsNone(ar.relaunch_argv("claude", [], SID, "nope"))

    def test_prompt(self):
        q = ar.resume_prompt("quota")
        self.assertTrue(q.startswith(ar.PROMPT_PREFIX + " This session was restarted "
                                     "automatically on another account because the previous "
                                     "account hit its usage limit. Continue the task"))
        self.assertIn("the user has not sent a new message", q)
        self.assertIn("do not repeat work that is already done.", q)
        self.assertIn("this model's usage limit", ar.resume_prompt("model"))
        self.assertIn("login failed", ar.resume_prompt("auth"))
        self.assertIn("was refused", ar.resume_prompt("blocked"))
        for cls in ("crash", "transient"):
            text = ar.resume_prompt(cls)
            self.assertNotIn("on another account", text)
            self.assertIn("was restarted automatically because", text)
        self.assertIn("exited unexpectedly", ar.resume_prompt("crash"))
        self.assertIn("an API error ended the last turn", ar.resume_prompt("transient"))
        self.assertEqual(ar.resume_prompt("auth", "Resume: {reason} ok"),
                         "Resume: the previous account's login failed ok")
        for bad in ("-p boom", "oneword", "two\nlines", "   "):
            self.assertEqual(ar.resume_prompt("quota", bad), q, bad)
        self.assertEqual(ar.Config("claude", {"CLAUDE_MULTIACC_AUTORESUME_PROMPT": "go on now"})
                         .prompt, "go on now")
        self.assertIsNone(ar.Config("codex", {"CODEX_MULTIACC_AUTORESUME_PROMPT": "-x y"}).prompt)


# =====================================================================================
# relaunch file, state file, budgets, tmux, tokens
# =====================================================================================

class RelaunchFileTests(unittest.TestCase):
    def fields(self, **over):
        verdict = {"class": "quota", "reset": 1790123817, "rtype": "five_hour",
                   "marked": "2026-09-22T21:36:58Z"}
        verdict.update(over.pop("verdict", {}))
        args = dict(provider="claude", verdict=verdict, acct="acct-01", sid=SID, cwd="/w d",
                    depth=2, avoid="acct-01:1790123817", hist="quota:1790100000,quota:17901",
                    chain="Ab3dEf9h")
        args.update(over)
        return ar.relaunch_fields(**args)

    def test_format(self):
        text = ar.format_relaunch(self.fields())
        self.assertEqual(text.split("\n"), [
            "v=1", "provider=claude", "class=quota", "acct=acct-01", "reset=1790123817",
            "rtype=five_hour", "marked=2026-09-22T21:36:58Z", "sid=" + SID, "cwd=/w d",
            "depth=3", "avoid=acct-01:1790123817", "hist=quota:1790100000,quota:17901",
            "chain=Ab3dEf9h", ""])

    def test_reset_and_rtype_belong_to_a_claude_quota_only(self):
        kv = ar.parse_kv(ar.format_relaunch(self.fields(verdict={"class": "auth"})),
                         ar.RELAUNCH_KEYS)
        self.assertEqual((kv["class"], kv["reset"], kv["rtype"]), ("auth", "", ""))
        kv = ar.parse_kv(ar.format_relaunch(self.fields(provider="codex")), ar.RELAUNCH_KEYS)
        self.assertEqual((kv["provider"], kv["reset"], kv["rtype"]),
                         ("codex", "1790123817", ""))

    def test_write_is_atomic_private_and_bash_readable(self):
        argv = ["--dangerously-skip-permissions", "--resume", SID,
                "multi\nline prompt with 'quotes' and é"]
        with tempfile.TemporaryDirectory() as tmp:
            d = os.path.join(tmp, "tmp", "autoresume")
            rel, av = ar.write_relaunch(d, "Tok3nTok3nTok3n0", self.fields(), argv)
            self.assertEqual(rel, os.path.join(d, "r-Tok3nTok3nTok3n0.relaunch"))
            self.assertEqual(av, os.path.join(d, "r-Tok3nTok3nTok3n0.argv"))
            self.assertEqual(sorted(os.listdir(d)),
                             ["r-Tok3nTok3nTok3n0.argv", "r-Tok3nTok3nTok3n0.relaunch"])
            for path in (rel, av):
                self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
            self.assertEqual(ar.decode_argv(Path(av).read_bytes()), argv)
            # The shim's own reader: while IFS= read -r -d '' a
            out = subprocess.run(
                ["bash", "-c", "n=0; while IFS= read -r -d '' a; do n=$((n+1)); "
                               "printf '%s\\0' \"$a\"; done < \"$1\"", "_", av],
                stdout=subprocess.PIPE, check=True, timeout=10).stdout
            self.assertEqual(ar.decode_argv(out), argv)

    def test_unwritable_values_leave_nothing(self):
        with tempfile.TemporaryDirectory() as tmp:
            with self.assertRaises(ValueError):
                ar.write_relaunch(tmp, "Tok3nTok3nTok3n0", self.fields(cwd="/a\nb"), ["x"])
            with self.assertRaises(ValueError):
                ar.write_relaunch(tmp, "bad/token", self.fields(), ["x"])
            with self.assertRaises(ValueError):
                ar.write_relaunch(tmp, "Tok3nTok3nTok3n0", self.fields(), ["a\0b"])
            self.assertEqual(os.listdir(tmp), [])


class StateTests(unittest.TestCase):
    def good(self):
        return {"v": "1", "provider": "claude", "pid": "4242", "ppid": "777",
                "acct": "acct-01", "acct_dir": "/p/acct-01", "cwd": "/w", "launched": "1790113000",
                "tmux": "/private/tmp/tmux-501/default,12345,0", "pane": "%3",
                "self": "/p/bin/claude", "acc_root": "/p", "depth": "0", "avoid": "",
                "hist": "", "chain": ""}

    def load(self, kv, extra=""):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "4242.state")
            Path(path).write_text("".join("%s=%s\n" % i for i in kv.items()) + extra)
            return ar.load_state(path)

    def test_valid_state(self):
        st = self.load(self.good(), "future_key=whatever\ngarbage line\n")
        self.assertEqual((st["pid"], st["ppid"], st["depth"], st["launched"], st["pane"]),
                         (4242, 777, 0, 1790113000, "%3"))
        self.assertEqual(ar.tmux_socket(st["tmux"]), "/private/tmp/tmux-501/default")

    def test_malformed_state_is_refused(self):
        for key, val in (("v", "2"), ("provider", "gemini"), ("pid", "x"), ("pid", "1"),
                         ("ppid", ""), ("acct", "../etc"), ("acct_dir", "/p/acct-02"),
                         ("acct_dir", ""), ("cwd", "rel"), ("launched", "-1"),
                         ("tmux", "default,1,0"), ("pane", "3"), ("self", "claude"),
                         ("acc_root", ""), ("depth", "x")):
            kv = self.good()
            kv[key] = val
            self.assertIsNone(self.load(kv), (key, val))
        kv = self.good()
        del kv["pid"]
        self.assertIsNone(self.load(kv))
        self.assertIsNone(ar.load_state("/nonexistent/4242.state"))

    def test_relative_pool_paths_resolve_against_the_cwd(self):
        kv = self.good()
        kv.update(acct_dir="pool/acct-01", acc_root="pool")
        st = self.load(kv)
        self.assertEqual((st["acct_dir"], st["acc_root"]),
                         (os.path.join(os.getcwd(), "pool", "acct-01"),
                          os.path.join(os.getcwd(), "pool")))

    def test_bad_chain_is_dropped_not_fatal(self):
        kv = self.good()
        kv["chain"] = "a b"
        self.assertEqual(self.load(kv)["chain"], "")

    def test_probe_env_and_answer(self):
        env = {"PATH": "/bin", "CLAUDE_CONFIG_DIR": "/p/acct-01", "CLAUDE_CODE_OAUTH_TOKEN": "t",
               "CLAUDE_SHIM_ACTIVE": "1", "CLAUDE_ACCOUNT": "acct-01", "CLAUDE_MULTIACC_AR": "x",
               "CLAUDE_MULTIACC_AR_AVOID": "stale"}
        out = ar.probe_env(env, "claude", "acct-01:99")
        self.assertEqual(out, {"PATH": "/bin", "CLAUDE_MULTIACC_AR_PROBE": "1",
                               "CLAUDE_MULTIACC_AR_AVOID": "acct-01:99"})
        cx = ar.probe_env({"CODEX_HOME": "/h", "CODEX_SHIM_ACTIVE": "1", "CODEX_ACCOUNT": "a",
                           "CODEX_MULTIACC_AR": "t", "HOME": "/u"}, "codex", "")
        self.assertEqual(cx, {"HOME": "/u", "CODEX_MULTIACC_AR_PROBE": "1",
                              "CODEX_MULTIACC_AR_AVOID": ""})
        self.assertEqual(ar.parse_probe("noise\npick=acct-02 tier=eligible\n"),
                         ("acct-02", "eligible"))
        self.assertEqual(ar.parse_probe("pick= tier=none\n"), ("", "none"))
        for bad in ("", "pick=acct-02", "pick=../x tier=eligible", "pick=acct-02 tier=great"):
            self.assertEqual(ar.parse_probe(bad), ("", ""), bad)
        self.assertTrue(ar.probe_allows("quota", "acct-02", "eligible", "acct-01"))
        self.assertFalse(ar.probe_allows("quota", "acct-01", "eligible", "acct-01"))
        self.assertFalse(ar.probe_allows("auth", "acct-02", "soft", "acct-01"))
        self.assertTrue(ar.probe_allows("transient", "acct-01", "soft", "acct-01"))
        self.assertFalse(ar.probe_allows("transient", "acct-01", "hard", "acct-01"))
        self.assertFalse(ar.probe_allows("transient", "", "none", "acct-01"))


class BudgetTests(unittest.TestCase):
    NOW = 1790113000

    def test_depth_is_final(self):
        c = cfg()
        self.assertEqual(ar.budget_check("quota", 20, "", self.NOW, c), (False, "depth"))
        self.assertEqual(ar.budget_check("crash", 19, "", self.NOW, c), (True, ""))
        self.assertEqual(ar.budget_check("quota", 3, "", self.NOW, cfg(MAX_DEPTH=3)),
                         (False, "depth"))

    def test_hourly_windows(self):
        c = cfg()
        recent = ",".join("%s:%d" % (k, self.NOW - 60 * i) for i, k in
                          enumerate(["quota", "auth", "model", "blocked"] * 2))
        self.assertEqual(ar.budget_check("quota", 0, recent, self.NOW, c),
                         (False, "rotate-budget"))
        old = ",".join("quota:%d" % (self.NOW - 3600 - i) for i in range(8))
        self.assertEqual(ar.budget_check("auth", 0, old, self.NOW, c), (True, ""))
        # rotations do not spend the transient budget and vice versa
        self.assertEqual(ar.budget_check("transient", 0, recent, self.NOW, c), (True, ""))
        tr = "transient:%d,transient:%d,transient:%d" % (self.NOW - 10, self.NOW - 20,
                                                         self.NOW - 30)
        self.assertEqual(ar.budget_check("transient", 0, tr, self.NOW, c),
                         (False, "transient-budget"))
        self.assertEqual(ar.budget_check("quota", 0, tr, self.NOW, c), (True, ""))
        crash = "crash:%d,crash:%d" % (self.NOW - 100, self.NOW - 500)
        self.assertEqual(ar.budget_check("crash", 0, crash, self.NOW, c),
                         (False, "crash-budget"))
        self.assertEqual(ar.budget_check("crash", 0, "crash:%d,crash:%d" % (
            self.NOW - 100, self.NOW - 700), self.NOW, c), (True, ""))

    def test_transient_ladder(self):
        ladder = cfg().transient_ladder
        self.assertEqual(ladder, (30.0, 60.0, 120.0))
        self.assertEqual(ar.ladder_wait("", self.NOW, ladder), 30.0)
        self.assertEqual(ar.ladder_wait("transient:%d" % (self.NOW - 5), self.NOW, ladder), 60.0)
        many = ",".join("transient:%d" % (self.NOW - i) for i in range(5))
        self.assertEqual(ar.ladder_wait(many, self.NOW, ladder), 120.0)
        self.assertEqual(ar.ladder_wait("transient:%d" % (self.NOW - 4000), self.NOW, ladder),
                         30.0)
        self.assertEqual(cfg(TRANSIENT_LADDER="1,2").transient_ladder, (1.0, 2.0))
        self.assertEqual(cfg(TRANSIENT_LADDER="1,x").transient_ladder, (30.0, 60.0, 120.0))

    def test_knobs(self):
        c = cfg(POLL="0.5", GRACE="bad", MAX_DEPTH="7", TERM_GRACE="-3")
        self.assertEqual((c.poll, c.grace, c.max_depth, c.term_grace), (0.5, 5.0, 7, 10.0))
        self.assertEqual(ar.Config("claude", {}).grace, 5.0)
        self.assertEqual(cfg(POLL="0").poll, 0.02)
        d = ar.Config("codex", {})
        self.assertEqual((d.hold_reprobe, d.rotate_per_hour, d.transient_per_hour,
                          d.crash_per_10min, d.crash_min_runtime, d.pane_wait,
                          d.probe_timeout, d.discover_timeout),
                         (60.0, 8, 3, 2, 60.0, 15.0, 30.0, 300.0))
        self.assertFalse(d.anypane)
        self.assertTrue(ar.Config("codex", {"CODEX_MULTIACC_AR_TEST_TMUX_ANYPANE": "1"}).anypane)

    def test_avoid_and_hist(self):
        now = self.NOW
        raw = "acct-02:%d,acct-01:%d,acct-01:%d,bogus,acct-03:%d,../x:%d" % (
            now + 50, now + 10, now + 99, now - 1, now + 5)
        self.assertEqual(ar.avoid_merge(raw, now), "acct-01:%d,acct-02:%d" % (now + 99, now + 50))
        self.assertEqual(ar.avoid_merge(raw, now, "acct-01", now + 20),
                         "acct-01:%d,acct-02:%d" % (now + 99, now + 50))
        self.assertEqual(ar.avoid_merge("", now, "acct-04", now + 7), "acct-04:%d" % (now + 7))
        self.assertEqual(ar.avoid_until({"class": "quota", "reset": now + 500}, now), now + 500)
        self.assertEqual(ar.avoid_until({"class": "quota", "reset": now - 5}, now), now + 3600)
        self.assertEqual(ar.avoid_until({"class": "quota", "reset": None}, now), now + 3600)
        self.assertEqual(ar.avoid_until({"class": "auth"}, now), now + 3600)
        self.assertEqual(ar.avoid_until({"class": "blocked"}, now), now + 6 * 3600)
        self.assertEqual(ar.avoid_until({"class": "model"}, now), now + 5 * 3600)
        hist = ",".join("quota:%d" % i for i in range(40))
        out = ar.hist_append(hist + ",BAD:1,x", "auth", now)
        self.assertEqual(len(out.split(",")), 32)
        self.assertTrue(out.endswith(",auth:%d" % now))
        self.assertTrue(out.startswith("quota:9,"))


class TmuxAndTokenTests(unittest.TestCase):
    SOCK = "/private/tmp/tmux-501/default"

    def test_socket(self):
        self.assertEqual(ar.tmux_socket(self.SOCK + ",4211,0"), self.SOCK)
        self.assertEqual(ar.tmux_socket(self.SOCK), self.SOCK)
        for bad in ("", ",1,0", "default,1,0", None):
            self.assertIsNone(ar.tmux_socket(bad), bad)

    def test_pane_query(self):
        self.assertEqual(ar.pane_query_argv(self.SOCK, "%7"),
                         ["tmux", "-S", self.SOCK, "display-message", "-p", "-t", "%7",
                          "#{pane_pid}|#{pane_in_mode}|#{synchronize-panes}"
                          "|#{pane_current_command}"])
        # (pid, command, a tmux mode owns the keys, keys go to every pane)
        self.assertEqual(ar.parse_pane_reply("8123|0|0|zsh\n"), (8123, "zsh", False, False))
        self.assertEqual(ar.parse_pane_reply("8123|0|0|2.1.280\n"),
                         (8123, "2.1.280", False, False))
        self.assertEqual(ar.parse_pane_reply("8123|1|0|claude"), (8123, "claude", True, False))
        # modes stack: tree + clock + copy reads 3
        self.assertEqual(ar.parse_pane_reply("8123|3|1|zsh"), (8123, "zsh", True, True))
        # an option/format an old tmux lacks expands to nothing: off
        self.assertEqual(ar.parse_pane_reply("8123|||zsh"), (8123, "zsh", False, False))
        self.assertEqual(ar.parse_pane_reply("8123|0|0|a|b c"), (8123, "a|b c", False, False))
        for bad in ("", "no server running on /x\n", "zsh 8123", "8123 zsh", "x|0|0|zsh"):
            self.assertIsNone(ar.parse_pane_reply(bad), bad)
        for sh in ("zsh", "bash", "sh", "dash", "ksh", "-zsh", "/bin/bash", "-/bin/zsh"):
            self.assertTrue(ar.is_shell(sh), sh)
        # fish < 3.1 cannot run `VAR=x cmd`; the rest never can
        for other in ("claude", "node", "codex", "2.1.280", "", "-", "vim", "nu", "tcsh",
                      "-tcsh", "pwsh", "xonsh", "fish", "-fish", "elvish", "mksh"):
            self.assertFalse(ar.is_shell(other), other)
        # a mode is only ever left by a mode command, never a key
        self.assertEqual(ar.mode_cancel_argv(self.SOCK, "%7"),
                         ["tmux", "-S", self.SOCK, "send-keys", "-X", "-t", "%7", "cancel"])

    def test_foreground(self):
        self.assertTrue(ar.ps_foreground(("Ss+", "ttys003", "-zsh")))
        self.assertFalse(ar.ps_foreground(("Ss", "ttys003", "-zsh")))
        self.assertFalse(ar.ps_foreground(("T", "pts/3", "claude")))
        for no_answer in (None, ("S", "??", "bash"), ("S", "?", "bash"), ("S", "-", "sh")):
            self.assertIsNone(ar.ps_foreground(no_answer), no_answer)

    def test_typeable_paths(self):
        for good in ("/Users/gas/.claude-accounts", "/opt/x_y/v1.2+b-c/bin/claude", "/"):
            self.assertTrue(ar.typeable(good), good)
        for bad in ("", None, "rel/path", "~/.claude-accounts", "/Users/John Doe/pool",
                    "/a/$(rm -rf x)", "/a/it's", "/a\\b", "/a\nb", "/a*b", "/a;b", "/a=b",
                    "/a,b", "/a:b", "/a@b", "/a%b", "/é"):
            self.assertFalse(ar.typeable(bad), bad)

    def test_send_keys(self):
        tok = "AbCdEfGh12345678"
        shim = "/opt/cm/lib/node_modules/claude-multiacc/bin/claude"
        self.assertEqual(ar.send_keys_argvs(self.SOCK, "%7", "claude", tok, SID, shim), [
            ["tmux", "-S", self.SOCK, "send-keys", "-R", "-t", "%7"],
            ["tmux", "-S", self.SOCK, "send-keys", "-t", "%7", "C-u"],
            ["tmux", "-S", self.SOCK, "send-keys", "-t", "%7", "-l",
             " CLAUDE_MULTIACC_AR=AbCdEfGh12345678:%s %s" % (SID, shim)],
            ["tmux", "-S", self.SOCK, "send-keys", "-t", "%7", "Enter"]])
        # the pool is named only when the relaunch has to (codex reads CODEX_ACCOUNTS_ROOT)
        self.assertEqual(ar.relaunch_command("codex", tok, CX_SID, "/x/bin/codex", "/srv/p-1"),
                         " CODEX_MULTIACC_AR=%s:%s CODEX_ACCOUNTS_ROOT=/srv/p-1 /x/bin/codex"
                         % (tok, CX_SID))
        self.assertEqual(ar.relaunch_command("claude", tok, SID, "/x/claude", "/p"),
                         " CLAUDE_MULTIACC_AR=%s:%s CLAUDE_ACCOUNTS_ROOT=/p /x/claude"
                         % (tok, SID))
        for args in (("claude", "short", SID, shim), ("claude", "a;rm -rf ~ #xxxxx", SID, shim),
                     ("gemini", tok, SID, shim), ("claude", tok, "abcdef", shim),
                     ("claude", tok, SID, "claude"), ("claude", tok, SID, "/a b/claude"),
                     ("claude", tok, SID, shim, "/p q"), ("claude", tok, SID, shim, "")):
            with self.assertRaises(ValueError):
                ar.relaunch_command(*args)

    def test_notice(self):
        # -d 0: the message stays until a key is pressed (tmux >= 3.2)...
        self.assertEqual(ar.notice_argv(self.SOCK, "%7", "claude", SID), [
            "tmux", "-S", self.SOCK, "display-message", "-d", "0", "-t", "%7",
            "claude-multiacc: auto-resume failed — run: claude --resume " + SID])
        # ...and the plain form is the fallback for an older tmux
        self.assertEqual(ar.notice_argv(self.SOCK, "%7", "claude", SID, sticky=False), [
            "tmux", "-S", self.SOCK, "display-message", "-t", "%7",
            "claude-multiacc: auto-resume failed — run: claude --resume " + SID])
        cx = ar.notice_argv(self.SOCK, "%7", "codex", CX_SID)
        self.assertEqual(cx[-1], "claude-multiacc: auto-resume failed — run: codex resume "
                         + CX_SID)
        self.assertNotIn("-p", cx)  # the status line, never the pane's input

    def test_tokens(self):
        seen = set()
        for _ in range(500):
            tok = ar.make_token()
            self.assertRegex(tok, r"^[A-Za-z0-9]{16}$")
            seen.add(tok)
        self.assertEqual(len(seen), 500)
        self.assertRegex(ar.make_token(8), r"^[A-Za-z0-9]{8}$")

    def test_misc_parsers(self):
        self.assertEqual(ar.iso_to_epoch("2026-09-22T21:36:58.009Z"), 1790113018.009)
        self.assertEqual(ar.iso_to_epoch("2026-09-22T21:36:58+00:00"), 1790113018)
        self.assertEqual(ar.iso_to_epoch("2026-09-23T00:36:58+03:00"), 1790113018)
        for bad in (None, 5, "", "yesterday", "2026-13-40T99:00:00Z"):
            self.assertIsNone(ar.iso_to_epoch(bad), bad)
        self.assertEqual(ar.utc_iso(1790113018), "2026-09-22T21:36:58Z")
        self.assertEqual(ar.epoch_of(1790123817), 1790123817)
        self.assertEqual(ar.epoch_of("1790123817"), 1790123817)
        for bad in (True, None, -1, 0, "x", float("nan"), 10 ** 12):
            self.assertIsNone(ar.epoch_of(bad), bad)
        self.assertEqual(ar.descendants([(10, 1, ""), (11, 10, ""), (12, 11, ""), (13, 1, ""),
                                         (14, 10, "")], 10), [11, 14, 12])
        self.assertTrue(ar.uuid_ok(SID))
        self.assertFalse(ar.uuid_ok("abcdef"))


# =====================================================================================
# the watcher, with a fake runner
# =====================================================================================

class FakeRunner:
    """Processes, clock and tmux, simulated. ``hooks`` fire as the fake clock passes
    their time, so a scenario can make the client exit, the user type, etc."""

    def __init__(self, t0, pid, ppid, self_path):
        self.t = float(t0)
        self.pid = pid
        self.live = {pid}
        self.starts = {pid: LSTART}
        self.ppid = ppid
        self.pane_pid = ppid
        self.pane_cmd = "claude"
        self.pane_mode = 0         # #{pane_in_mode}: >0 while a tmux mode owns the keys
        self.mode_sticky = False   # True: `send-keys -X cancel` cannot end the mode
        self.pane_sync = 0         # #{synchronize-panes}
        self.shell_comm = "-zsh"   # what ps says the launching shell runs
        self.client_stat = "S+"    # the client's ps stat (T = stopped with Ctrl-Z)
        self.tty = "ttys003"
        self.open_files_ok = True  # /proc or lsof available
        self.lookup_fails = False  # the lookup is there but every call fails
        self.pgids = {}            # pid -> process group; default its own (a job leader)
        self.stuck = set()         # pids no signal ends (uninterruptible sleep)
        self.old_tmux = False      # True: display-message refuses -d (tmux < 3.2)
        self.ar_dir = None         # the pool's tmp/autoresume (PoolCase.watcher sets it)
        self.consume = True        # False: the typed line never runs (token left behind)
        self.consumed = {}         # token -> (relaunch text, argv bytes) the shim took
        self.line = ""             # the last -l text typed
        self.self_path = self_path
        self.probe_out = "pick=acct-02 tier=eligible\n"
        self.term_kills = True
        self.table = []
        self.files = {}
        self.calls = []
        self.signals = []
        self.hooks = []
        self.limit = t0 + 4000

    def at(self, when, fn):
        self.hooks.append((when, fn))

    def now(self):
        return self.t

    def sleep(self, seconds):
        self.t += max(seconds, 0.001)
        if self.t > self.limit:
            raise RuntimeError("fake clock ran away")
        for hook in sorted(self.hooks, key=lambda h: h[0]):
            if self.t >= hook[0]:
                self.hooks.remove(hook)
                hook[1]()

    def alive(self, pid):
        return pid in self.live

    def gone(self, pid):
        return pid not in self.live

    def exit(self, pid):
        self.live.discard(pid)
        if pid == self.pid:
            self.pane_cmd = "zsh"

    def signal(self, pid, sig):
        self.signals.append((pid, sig))
        self.calls.append({"argv": ["<signal>", pid, sig], "env": None, "cwd": None,
                           "t": self.t})
        if pid not in self.live:
            return False
        if pid not in self.stuck and (
                sig == signal.SIGKILL or (sig == signal.SIGTERM and self.term_kills)):
            self.exit(pid)
        return True

    def pgid(self, pid):
        return self.pgids.get(pid, pid) if pid in self.live else None

    def lstart(self, pid):
        return self.starts.get(pid) if pid in self.live else None

    def ps_table(self):
        return [row for row in self.table if row[0] in self.live]

    def open_files(self, pid):
        return None if self.lookup_fails else list(self.files.get(pid, ()))

    def can_list_open_files(self):
        return self.open_files_ok

    def proc_status(self, pid):
        if pid == self.ppid and self.shell_comm is not None:
            # the shell holds the terminal again once its foreground job is gone
            return ("Ss" if self.pid in self.live else "Ss+", self.tty, self.shell_comm)
        if pid == self.pid and pid in self.live:
            return (self.client_stat, self.tty, "claude")
        return None

    def consume_token(self):
        """Enter at the shell: the relaunched shim reads its relaunch files and removes
        them (single use), first thing. Kept here for the assertions."""
        if not self.consume or not self.ar_dir or "=" not in self.line:
            return
        tok = self.line.split("=", 1)[1].split()[0].split(":")[0]
        base = os.path.join(str(self.ar_dir), "r-" + tok)
        try:
            with open(base + ".relaunch") as fh:
                text = fh.read()
            with open(base + ".argv", "rb") as fh:
                data = fh.read()
        except OSError:
            return
        self.consumed[tok] = (text, data)
        ar.remove_quietly(base + ".relaunch", base + ".argv")

    def run(self, argv, timeout, env=None, cwd=None):
        self.calls.append({"argv": list(argv), "env": env, "cwd": cwd, "t": self.t})
        if argv[0] == "tmux":
            if "display-message" in argv and "-p" in argv:
                return 0, "%d|%d|%d|%s\n" % (self.pane_pid, self.pane_mode, self.pane_sync,
                                             self.pane_cmd)
            if "send-keys" in argv and "-X" in argv:
                if argv[-1] == "cancel" and not self.mode_sticky:
                    self.pane_mode = 0
                return 0, ""
            if "send-keys" in argv and "-l" in argv:
                self.line = argv[-1]
            elif "send-keys" in argv and argv[-1] == "Enter":
                self.consume_token()
            if "display-message" in argv and "-d" in argv and self.old_tmux:
                return 1, ""
            return 0, ""
        if argv[0] == self.self_path:
            return (3 if "tier=none" in self.probe_out else 0), self.probe_out
        return 1, ""

    @property
    def probes(self):
        return [c for c in self.calls if c["argv"][0] == self.self_path]

    @property
    def sends(self):
        """What was keyed into the pane (a -X mode command is not a key)."""
        return [c["argv"] for c in self.calls if c["argv"][0] == "tmux"
                and "send-keys" in c["argv"] and "-X" not in c["argv"]]

    @property
    def notices(self):
        return [c["argv"] for c in self.calls if c["argv"][0] == "tmux"
                and "display-message" in c["argv"] and "-p" not in c["argv"]]

    def index(self, pred):
        return next(i for i, c in enumerate(self.calls) if pred(c["argv"]))


class PoolCase(unittest.TestCase):
    PID = 4242
    PPID = 777

    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.tmp = Path(self._tmp.name).resolve()
        self.root = self.tmp / "pool"
        self.ar_dir = self.root / "tmp" / "autoresume"
        self.ar_dir.mkdir(parents=True)
        self.work = self.tmp / "work dir"
        self.work.mkdir()
        self.shared = self.tmp / "shared"
        for acct in ("acct-01", "acct-02"):
            d = self.root / acct
            (d / "sessions").mkdir(parents=True)
        self.proj = self.shared / "projects" / "-work-dir"
        self.proj.mkdir(parents=True)
        for acct in ("acct-01", "acct-02"):
            os.symlink(str(self.shared / "projects"), str(self.root / acct / "projects"))
        self.self_path = str(self.tmp / "bin" / "claude")
        self.t0 = float(int(time.time()))
        self.launched = int(self.t0) - 5

    def tearDown(self):
        self._tmp.cleanup()

    # -- fixtures --

    def write_state(self, argv, provider="claude", depth=0, avoid="", hist="", chain="",
                    launched=None, pid=None, acc_root=None):
        pid = pid or self.PID
        kv = {"v": "1", "provider": provider, "pid": pid, "ppid": self.PPID,
              "acct": "acct-01", "acct_dir": str(self.root / "acct-01"), "cwd": str(self.work),
              "launched": self.launched if launched is None else launched,
              "tmux": "/private/tmp/tmux-501/default,4211,0", "pane": "%7",
              "self": self.self_path, "acc_root": str(acc_root or self.root), "depth": depth,
              "avoid": avoid, "hist": hist, "chain": chain}
        path = self.ar_dir / ("%d.state" % pid)
        path.write_text("".join("%s=%s\n" % i for i in kv.items()))
        (self.ar_dir / ("%d.argv" % pid)).write_bytes(ar.encode_argv(argv))
        return str(path)

    def registry(self, sid=SID, pid=None):
        pid = pid or self.PID
        doc = {"pid": pid, "sessionId": sid, "cwd": str(self.work),
               "startedAt": int(self.launched * 1000 + 700), "version": "2.1.280",
               "kind": "interactive", "entrypoint": "cli", "status": "idle"}
        path = self.root / "acct-01" / "sessions" / ("%d.json" % pid)
        path.write_text(json.dumps(doc))
        return path

    def transcript(self, records, sid=SID):
        path = self.proj / (sid + ".jsonl")
        with open(path, "a") as fh:
            fh.write(jl(records))
        return path

    def watcher(self, state, env=None, runner=None):
        environ = {"PATH": "/usr/bin:/bin", "CLAUDE_CONFIG_DIR": str(self.root / "acct-01"),
                   "CLAUDE_SHIM_ACTIVE": "1", "CODEX_HOME": str(self.root / "acct-01")}
        environ.update(env or {})
        runner = runner or FakeRunner(self.t0, self.PID, self.PPID, self.self_path)
        runner.ar_dir = self.ar_dir
        w = ar.Watcher(state, ar.load_state(state), ar.load_argv(ar.argv_path_for(state)),
                       runner=runner, environ=environ)
        return w, runner

    def log_lines(self):
        path = self.root / "selection.log"
        return path.read_text().splitlines() if path.exists() else []

    def events(self):
        return [line.split()[2] for line in self.log_lines()]

    def relaunch_files(self):
        return sorted(p.name for p in self.ar_dir.glob("r-*"))

    def token_of(self, runner):
        typed = [a[-1] for a in runner.sends if "-l" in a]
        self.assertEqual(len(typed), 1)
        return typed[0].split("=", 1)[1].split()[0].split(":")[0]

    def relaunch_of(self, runner):
        """The relaunch files as the relaunched shim consumed them (the fake shell runs
        the typed line at Enter, and the token is single use)."""
        tok = self.token_of(runner)
        self.assertEqual(self.relaunch_files(), [])
        text, data = runner.consumed[tok]
        return ar.parse_kv(text, ar.RELAUNCH_KEYS), ar.decode_argv(data)

    def assert_sequence(self, runner, provider="claude", sid=None, root="pool"):
        """The four send-keys calls. The suites' pool is not $HOME/.<p>-accounts, so the
        typed line names it (root=None: a default pool, nothing named)."""
        tok = self.token_of(runner)
        sid = sid or (SID if provider == "claude" else CX_SID)
        self.assertEqual(runner.sends, ar.send_keys_argvs(
            "/private/tmp/tmux-501/default", "%7", provider, tok, sid, self.self_path,
            str(self.root) if root == "pool" else root))


class ClaudeWatcherTests(PoolCase):
    def test_quota_rotates_to_another_account(self):
        reset = int(self.t0) + 7200
        state = self.write_state(["--dangerously-skip-permissions", "fix the flaky test"])
        self.registry()
        self.transcript([cc_user(self.launched - 4000, "old prompt"),
                         cc_quota(self.launched - 3600, resets=self.launched - 100),
                         cc_quota(self.t0 - 1, resets=reset), cc_system(self.t0 - 1)])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        prompt = ar.resume_prompt("quota")
        # probe: selection from scratch, the old account avoided until its reset
        self.assertEqual(len(r.probes), 1)
        probe = r.probes[0]
        self.assertGreaterEqual(probe["t"] - self.t0, 5.0)  # GRACE
        self.assertEqual(probe["argv"], [self.self_path, "--dangerously-skip-permissions",
                                         "--resume", SID, prompt])
        self.assertEqual(probe["cwd"], str(self.work))
        self.assertEqual(probe["env"]["CLAUDE_MULTIACC_AR_PROBE"], "1")
        self.assertEqual(probe["env"]["CLAUDE_MULTIACC_AR_AVOID"], "acct-01:%d" % reset)
        self.assertNotIn("CLAUDE_CONFIG_DIR", probe["env"])
        self.assertNotIn("CLAUDE_SHIM_ACTIVE", probe["env"])
        # stop, then type into the shell: token + session id, the pool, the shim itself
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assert_sequence(r)
        self.assertEqual([a[-1] for a in r.sends if "-l" in a],
                         [" CLAUDE_MULTIACC_AR=%s:%s CLAUDE_ACCOUNTS_ROOT=%s %s"
                          % (self.token_of(r), SID, self.root, self.self_path)])
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["--dangerously-skip-permissions", "--resume", SID, prompt])
        self.assertEqual({k: kv[k] for k in ("v", "provider", "class", "acct", "reset", "rtype",
                                             "marked", "sid", "cwd", "depth", "avoid")},
                         {"v": "1", "provider": "claude", "class": "quota", "acct": "acct-01",
                          "reset": str(reset), "rtype": "five_hour",
                          "marked": ar.utc_iso(self.t0 - 1), "sid": SID, "cwd": str(self.work),
                          "depth": "1", "avoid": "acct-01:%d" % reset})
        self.assertRegex(kv["hist"], r"^quota:\d+$")
        self.assertRegex(kv["chain"], r"^[A-Za-z0-9]{8}$")
        # log grammar: field 2 is always the word autoresume
        self.assertEqual(self.events(), ["watch", "detect", "switch"])
        for line in self.log_lines():
            self.assertEqual(line.split()[1], "autoresume")
            self.assertNotIn("hit your limit", line)
        self.assertIn("from=acct-01 class=quota sid=%s depth=1 probe=acct-02" % SID,
                      self.log_lines()[-1])

    def test_chain_state_is_carried(self):
        far = int(self.t0) + 50000
        state = self.write_state(["-c"], depth=4, avoid="acct-05:%d,acct-06:1" % far,
                                 hist="quota:%d" % (self.t0 - 7200), chain="ChainId1")
        self.registry()
        self.transcript([cc_api_error(self.t0, "authentication_failed",
                                      "Not logged in · Please run /login")])
        w, r = self.watcher(state)
        w.run()
        until = int(r.probes[0]["t"]) + 3600
        self.assertEqual(r.probes[0]["env"]["CLAUDE_MULTIACC_AR_AVOID"],
                         "acct-01:%d,acct-05:%d" % (until, far))
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["--resume", SID, ar.resume_prompt("auth")])
        self.assertEqual((kv["class"], kv["reset"], kv["rtype"], kv["depth"], kv["chain"]),
                         ("auth", "", "", "5", "ChainId1"))
        self.assertTrue(kv["hist"].startswith("quota:%d,auth:" % (self.t0 - 7200)))
        self.assertIn("chain=ChainId1", self.log_lines()[0])

    def test_old_rejection_in_a_resumed_transcript_does_not_fire(self):
        state = self.write_state(["--resume", SID])
        self.registry()
        self.transcript([cc_quota(self.launched - 3, resets=int(self.t0) + 999),
                         cc_system(self.launched - 3)])
        w, r = self.watcher(state)
        reg = self.root / "acct-01" / "sessions" / ("%d.json" % self.PID)
        r.at(self.t0 + 20, lambda: (reg.unlink(), r.exit(self.PID)))
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals, r.sends), ([], [], []))
        self.assertEqual(self.events(), ["watch"])

    def test_no_alternative_holds_and_reprobes(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_HOLD_REPROBE": "5"})
        r.probe_out = "pick=acct-01 tier=soft\n"
        reg = self.root / "acct-01" / "sessions" / ("%d.json" % self.PID)
        r.at(self.t0 + 16.5, lambda: (reg.unlink(), r.exit(self.PID)))
        self.assertEqual(w.run(), 0)
        self.assertEqual(len(r.probes), 3)  # ~5 s, ~10 s, ~15 s
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertEqual(self.events(), ["watch", "detect", "hold"])
        self.assertIn("reason=no-room pick=acct-01 tier=soft", self.log_lines()[-1])

    def test_nobody_left_holds(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_api_error(self.t0, "account_on_hold")])
        w, r = self.watcher(state)
        r.probe_out = "pick= tier=none\n"
        r.at(self.t0 + 10, lambda: r.exit(self.PID))
        w.st["launched"] = w.launched = int(self.t0) - 5  # too young to be a crash
        self.assertEqual(w.run(), 0)
        self.assertEqual(r.signals, [])
        self.assertIn("hold", self.events())

    def test_a_cancel_ends_a_hold_and_nothing_is_stopped(self):
        state = self.write_state([])
        self.registry()
        tpath = self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        # the user types before GRACE runs out
        r.at(self.t0 + 1.5, lambda: append(tpath, [cc_user(self.t0 + 1, "never mind")]))
        r.at(self.t0 + 9, lambda: r.exit(self.PID))
        w.run()
        self.assertEqual((r.probes, r.signals), ([], []))

    def test_input_that_arrives_during_the_probe_wins(self):
        state = self.write_state([])
        self.registry()
        tpath = self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        real_run = r.run

        def run(argv, timeout, env=None, cwd=None):
            if argv[0] == self.self_path and not r.probes:
                append(tpath, [cc_user(self.t0 + 3, "actually, let me handle this")])
            return real_run(argv, timeout, env, cwd)
        r.run = run
        r.at(self.t0 + 20, lambda: r.exit(self.PID))
        w.run()
        self.assertEqual(len(r.probes), 1)
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))

    def test_kill_switch_file_reaches_a_running_watcher(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.at(self.t0 + 1, lambda: (self.root / "autoresume.off").write_text(""))
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals, r.sends), ([], [], []))

    def test_kill_switch_file_before_start(self):
        (self.root / "autoresume.off").write_text("")
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))

    def test_invalid_request_is_logged_once_and_left_alone(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_api_error(self.t0, "invalid_request", "Autocompact is thrashing"),
                         cc_api_error(self.t0, "invalid_request", "again")])
        w, r = self.watcher(state)
        r.at(self.t0 + 8, lambda: r.exit(self.PID))
        w.run()
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertEqual(self.events(), ["watch", "never"])
        self.assertIn("code=invalid_request", self.log_lines()[1])

    def test_crash_relaunches_through_normal_selection(self):
        state = self.write_state(["--dangerously-skip-permissions"],
                                 avoid="acct-05:%d" % (self.t0 + 900))
        self.registry()
        self.transcript([cc_reply(self.t0, "working on it")])
        w, r = self.watcher(state)
        r.at(self.t0 + 70, lambda: r.exit(self.PID))  # registry left behind: SIGKILL/crash
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["--dangerously-skip-permissions", "--resume", SID,
                                ar.resume_prompt("crash")])
        self.assertEqual((kv["class"], kv["acct"], kv["avoid"], kv["depth"]),
                         ("crash", "acct-01", "acct-05:%d" % (self.t0 + 900), "1"))
        self.assert_sequence(r)
        self.assertEqual(self.events(), ["watch", "crash", "switch"])

    def test_a_clean_exit_or_an_early_death_is_final(self):
        state = self.write_state([])
        reg = self.registry()
        w, r = self.watcher(state)
        r.at(self.t0 + 70, lambda: (reg.unlink(), r.exit(self.PID)))  # /exit: registry gone
        w.run()
        self.assertEqual((r.sends, self.relaunch_files()), ([], []))
        state = self.write_state([])
        self.registry()
        w, r = self.watcher(state)
        r.at(self.t0 + 30, lambda: r.exit(self.PID))  # died after 35 s: not a crash yet
        w.run()
        self.assertEqual(r.sends, [])

    def test_crash_budget(self):
        hist = "crash:%d,crash:%d" % (self.t0 - 60, self.t0 - 120)
        state = self.write_state([], hist=hist)
        self.registry()
        w, r = self.watcher(state)
        r.at(self.t0 + 70, lambda: r.exit(self.PID))
        w.run()
        self.assertEqual(r.sends, [])
        self.assertIn("reason=crash-budget", self.log_lines()[-1])

    def test_pane_must_be_the_launching_shell(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_pid = 31337
        self.assertEqual(w.run(), 0)
        self.assertEqual(len(r.probes), 1)
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertIn("reason=pane", self.log_lines()[-1])
        # the test knob skips only the pid equality
        state = self.write_state([])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TEST_TMUX_ANYPANE": "1"})
        r.pane_pid = 31337
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertEqual(len(r.sends), 4)

    def test_depth_cap_gives_up_without_probing(self):
        state = self.write_state([], depth=20)
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertIn("reason=depth", self.log_lines()[-1])

    def test_rotate_budget_holds_without_probing(self):
        hist = ",".join("quota:%d" % (self.t0 - 60 * i) for i in range(1, 9))
        state = self.write_state([], hist=hist)
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.at(self.t0 + 10, lambda: r.exit(self.PID))
        w.run()
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertIn("reason=rotate-budget", self.log_lines()[-1])

    def test_transient_waits_its_ladder_step_and_may_stay_put(self):
        far = int(self.t0) + 9000
        state = self.write_state(["--model", "opus"], avoid="acct-09:%d" % far,
                                 hist="transient:%d" % (self.t0 - 100))
        self.registry()
        self.transcript([cc_api_error(self.t0, "server_error",
                                      "API Error: Connection lost mid-response.")])
        w, r = self.watcher(state)
        r.probe_out = "pick=acct-01 tier=soft\n"
        w.run()
        self.assertGreaterEqual(r.probes[0]["t"] - self.t0, 60.0)  # second ladder step
        self.assertEqual(r.probes[0]["env"]["CLAUDE_MULTIACC_AR_AVOID"], "acct-09:%d" % far)
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["--model", "opus", "--resume", SID,
                                ar.resume_prompt("transient")])
        self.assertEqual((kv["class"], kv["avoid"]), ("transient", "acct-09:%d" % far))

    def test_model_limit_waits_for_busy_subagents_and_adds_no_model(self):
        # settings.json names a model, but --resume restores the session's own
        (self.root / "acct-01" / "settings.json").write_text(json.dumps({"model": "opus[1m]"}))
        sub = self.proj / SID / "subagents" / "workflows" / "w1"
        sub.mkdir(parents=True)
        busy = sub / "agent-1.jsonl"
        busy.write_text("{}\n")
        os.utime(str(busy), (self.t0 + 100, self.t0 + 100))  # still writing until ~t0+100
        state = self.write_state(["open the report please"])
        self.registry()
        self.transcript([cc_api_error(self.t0, "rate_limit",
                                      "You've reached your Fable limit. Switch to another model")])
        w, r = self.watcher(state)
        w.run()
        self.assertGreaterEqual(r.probes[0]["t"], self.t0 + 120)  # 20 s of subagent quiet
        until = int(r.probes[0]["t"]) + 5 * 3600
        self.assertEqual(r.probes[0]["env"]["CLAUDE_MULTIACC_AR_AVOID"], "acct-01:%d" % until)
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["--resume", SID, ar.resume_prompt("model")])
        self.assertEqual(kv["class"], "model")

    def test_session_switch_drops_the_pending_verdict(self):
        state = self.write_state([])
        reg = self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        # /clear before GRACE: a new conversation, nothing wrong with it
        r.at(self.t0 + 1.5, lambda: self.registry(sid=SID2))
        r.at(self.t0 + 10, lambda: (reg.unlink(), r.exit(self.PID)))
        w.run()
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertEqual(w.sid, SID2)

    def test_stale_registry_of_a_reused_pid_is_ignored(self):
        state = self.write_state([])
        path = self.registry()
        doc = json.loads(path.read_text())
        doc["startedAt"] = (self.launched - 86400) * 1000
        path.write_text(json.dumps(doc))
        w, _ = self.watcher(state)
        self.assertIsNone(w.registry())

    REFRESH = 5002  # the shim's detached `limits --quiet`: same group, reparented to init

    def job(self, r):
        """The client's job as an interactive shell leaves it: the client leads the group;
        5001 is its child (a tool or MCP process); REFRESH shares the group but is not the
        client's — the shim started it in the background before its exec."""
        r.table = [(self.PID, self.PPID, LSTART), (5001, self.PID, "d"), (self.REFRESH, 1, "x")]
        r.live |= {5001, self.REFRESH}
        r.starts.update({5001: "d", self.REFRESH: "x"})
        r.pgids.update({5001: self.PID, self.REFRESH: self.PID})

    def test_a_tree_read_naming_foreign_processes_never_signals_them(self):
        # 2026-09-23: a deliberately broken copy of descendants() under test named every
        # process on the machine as the client's child, and the stop SIGKILLed them. The
        # stop itself must refuse anything outside the client's own process group, and
        # never touch the launching shell, whatever the tree read says.
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        self.job(r)
        foreign = [6001, 6002, 6003]
        r.table += [(p, self.PID, "f%d" % p) for p in foreign] + [(self.PPID, self.PID, "sh")]
        r.live |= set(foreign) | {self.PPID}
        r.starts.update({p: "f%d" % p for p in foreign})
        r.starts[self.PPID] = "sh"
        r.pgids.update({p: p for p in foreign})
        r.pgids[self.PPID] = self.PID   # even sharing the group, the shell is never a target
        r.term_kills = False
        w.run()
        hit = {pid for pid, _sig in r.signals}
        self.assertEqual(hit, {self.PID, 5001})
        self.assertTrue(set(foreign) | {self.PPID} <= r.live)

    def test_an_implausibly_large_tree_is_not_trusted(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        self.job(r)
        many = list(range(7000, 7000 + 70))
        r.table += [(p, self.PID, "m") for p in many]
        r.live |= set(many)
        r.starts.update({p: "m" for p in many})
        r.pgids.update({p: self.PID for p in many})
        r.term_kills = False
        w.run()
        self.assertEqual([pid for pid, _sig in r.signals], [self.PID, self.PID])
        self.assertTrue(set(many) <= r.live)

    def test_term_refused_kills_the_client_and_its_descendants_never_its_group(self):
        # Review 2026-09-23: the group-wide wait and killpg took the shim's background
        # limits refresh along (leaving its lock behind), and the refresh made every
        # relaunch wait out the whole TERM_GRACE.
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        self.job(r)
        r.term_kills = False
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM), (self.PID, signal.SIGKILL),
                                     (5001, signal.SIGKILL)])
        self.assertEqual(r.live, {self.REFRESH})
        self.assertEqual(len(r.sends), 4)

    def test_a_process_of_the_group_that_is_not_the_clients_is_never_waited_for(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)  # TERM_GRACE 10 s
        self.job(r)
        real_exit = r.exit

        def exit_with_its_child(pid):
            real_exit(pid)
            if pid == self.PID:
                r.live.discard(5001)
        r.exit = exit_with_its_child
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertIn(self.REFRESH, r.live)
        term = r.index(lambda a: a[0] == "<signal>")
        typed = r.index(lambda a: a[0] == "tmux" and "-l" in a)
        self.assertLess(r.calls[typed]["t"] - r.calls[term]["t"], 2)
        self.assert_sequence(r)

    def test_a_descendant_outliving_the_client_is_killed_after_the_grace(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        self.job(r)  # 5001 ignores the client's exit
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM), (5001, signal.SIGKILL)])
        term = r.index(lambda a: a[0] == "<signal>")
        kill = r.index(lambda a: a[:2] == ["<signal>", 5001])
        self.assertGreaterEqual(r.calls[kill]["t"], r.calls[term]["t"] + 2)  # TERM_GRACE
        self.assertEqual(r.live, {self.REFRESH})
        self.assert_sequence(r)

    def test_a_recorded_descendant_whose_pid_was_reused_is_left_alone(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        self.job(r)
        real_signal = r.signal

        def signal_and_reuse(pid, sig):
            out = real_signal(pid, sig)
            if sig == signal.SIGTERM:
                r.starts[5001] = "someone else's"  # 5001 died; its pid names another
            return out
        r.signal = signal_and_reuse
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertIn(5001, r.live)
        term = r.index(lambda a: a[0] == "<signal>")
        typed = r.index(lambda a: a[0] == "tmux" and "-l" in a)
        self.assertLess(r.calls[typed]["t"] - r.calls[term]["t"], 2)

    def test_nothing_is_typed_while_the_stopped_tree_lives(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TERM_GRACE": "1",
                                    "CLAUDE_MULTIACC_AR_PANE_WAIT": "2"})
        self.job(r)
        r.stuck.add(5001)  # not even SIGKILL ends it (uninterruptible sleep)
        w.run()
        self.assertEqual([a for a in r.sends if "-l" in a], [])
        self.assertEqual(self.relaunch_files(), [])
        line = self.log_lines()[-1]
        self.assertIn("reason=tree", line)
        self.assertIn('stopped=1 resume="claude --resume %s"' % SID, line)
        # the notice stays up until a key is pressed
        self.assertEqual(r.notices, [ar.notice_argv("/private/tmp/tmux-501/default", "%7",
                                                    "claude", SID)])
        self.assertIn("-d", r.notices[0])

    def test_a_client_that_does_not_lead_its_group_is_never_stopped(self):
        # Review 2026-09-23: a client that is not a job of a job-control shell (started
        # by a script, `sh -c`, a pipeline) gets no prompt back when it ends
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_mode = 1  # not even the mode is cancelled: nothing is touched
        r.live.add(self.PPID)
        r.pgids[self.PID] = self.PPID
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.signals, r.sends, self.relaunch_files(), r.notices),
                         ([], [], [], []))
        self.assertEqual([c for c in r.calls if "-X" in c["argv"]], [])
        self.assertIn("reason=pgrp", self.log_lines()[-1])
        self.assertNotIn("stopped=", self.log_lines()[-1])
        # the suites' no-terminal knob (their harness has no job control) waives it; the
        # client is then stopped alone, never the group it shares with the shell
        state = self.write_state([])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TEST_TTY": "1",
                                    "CLAUDE_MULTIACC_AR_TERM_GRACE": "2"})
        r.term_kills = False
        r.live.add(self.PPID)
        r.pgids[self.PID] = self.PPID
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM), (self.PID, signal.SIGKILL)])
        self.assertIn(self.PPID, r.live)
        self.assertEqual(len(r.sends), 4)

    def test_a_relaunch_that_never_takes_is_announced(self):
        # Review 2026-09-23: Enter was typed, but the line never ran (the relaunch file
        # is still there): withdraw the token and say how to resume, never log a switch
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.consume = False
        self.assertEqual(w.run(), 0)
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertEqual(len(r.sends), 4)
        self.assertEqual(self.relaunch_files(), [])
        self.assertNotIn("switch", self.events())
        line = self.log_lines()[-1]
        self.assertIn("autoresume giveup", line)
        self.assertIn("reason=relaunch", line)
        self.assertIn('stopped=1 resume="claude --resume %s"' % SID, line)
        self.assertEqual(r.notices, [ar.notice_argv("/private/tmp/tmux-501/default", "%7",
                                                    "claude", SID)])
        enter = r.index(lambda a: a[0] == "tmux" and a[-1] == "Enter")
        notice = r.index(lambda a: a[0] == "tmux" and "-d" in a)
        self.assertGreaterEqual(r.calls[notice]["t"] - r.calls[enter]["t"], 10)  # ~10 s
        # a relaunch that takes within the wait is a switch
        state = self.write_state([])
        w, r = self.watcher(state)
        r.consume = False
        real_run = r.run

        def slow_shell(argv, timeout, env=None, cwd=None):
            out = real_run(argv, timeout, env, cwd)
            if argv[0] == "tmux" and argv[-1] == "Enter":
                r.consume = True
                r.at(r.t + 3, r.consume_token)
            return out
        r.run = slow_shell
        self.assertEqual(w.run(), 0)
        self.assertEqual(self.events()[-1], "switch")
        self.assertEqual(r.notices, [])

    def test_process_replaced_before_the_signal(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        real_run = r.run

        def run(argv, timeout, env=None, cwd=None):
            out = real_run(argv, timeout, env, cwd)
            if argv[0] == self.self_path:
                r.starts[self.PID] = "Wed Sep 23 09:00:00 2026"  # pid reused
            return out
        r.run = run
        w.run()
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertIn("reason=gone", self.log_lines()[-1])

    def test_pane_never_returns_to_a_shell(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_PANE_WAIT": "3"})
        r.exit = lambda pid: r.live.discard(pid)  # pane keeps showing another program
        r.old_tmux = True
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertEqual((r.sends, self.relaunch_files()), ([], []))
        # the session is stopped and stays stopped: the log line and the pane's status
        # line say so, with the command that brings it back (tmux < 3.2 refuses -d 0:
        # the plain notice follows)
        self.assertIn("reason=shell", self.log_lines()[-1])
        self.assertIn('stopped=1 resume="claude --resume %s"' % SID, self.log_lines()[-1])
        self.assertEqual(r.notices, [
            ar.notice_argv("/private/tmp/tmux-501/default", "%7", "claude", SID),
            ar.notice_argv("/private/tmp/tmux-501/default", "%7", "claude", SID,
                           sticky=False)])

    def test_nothing_is_announced_for_a_session_that_was_never_stopped(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_pid = 31337
        w.run()
        self.assertIn("reason=pane", self.log_lines()[-1])
        self.assertNotIn("stopped=", self.log_lines()[-1])
        self.assertNotIn("resume=", self.log_lines()[-1])
        self.assertEqual(r.notices, [])

    def test_a_shell_loop_restarting_the_client_is_never_typed_into(self):
        # `while :; do claude; done`: one reading may catch the shell between two runs;
        # it must hold for SHELL_SETTLE before anything is typed.
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_PANE_WAIT": "3"})
        real_exit = r.exit

        def exit_into_the_next_run(pid):
            real_exit(pid)
            if pid == self.PID:
                r.at(r.t + 0.3, lambda: setattr(r, "pane_cmd", "claude"))
        r.exit = exit_into_the_next_run
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertEqual([a for a in r.sends if "-l" in a], [])
        self.assertIn("reason=shell", self.log_lines()[-1])
        self.assertIn("stopped=1", self.log_lines()[-1])

    def test_typing_waits_for_the_shell_to_hold_the_terminal(self):
        # the pane names the shell, but a job it started still owns the terminal
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        real_status = r.proc_status
        ready_at = []

        def status(pid):
            out = real_status(pid)
            if pid == self.PPID and out and not ready_at:
                ready_at.append(r.t + 2)
            if pid == self.PPID and out and r.t < (ready_at or [0])[0] and self.PID not in r.live:
                return ("Ss", out[1], out[2])
            return out
        r.proc_status = status
        w.run()
        typed = r.index(lambda a: a[0] == "tmux" and "-l" in a)
        self.assertGreaterEqual(r.calls[typed]["t"], ready_at[0] + ar.SHELL_SETTLE)
        self.assert_sequence(r)

    def test_a_conversation_picked_with_resume_brings_no_old_errors(self):
        # Review 2026-09-23 (blocker): another session, on another account, hit ITS limit
        # after this launch; the operator /resumes that conversation here. Its rejection
        # is not this client's: nothing is stopped and this account is never parked.
        state = self.write_state([])
        self.registry()
        self.transcript([cc_reply(self.t0, "hello")])
        other = self.transcript([cc_user(self.t0 - 2, "go"),
                                 cc_quota(self.t0 + 1, resets=int(self.t0) + 999)], sid=SID2)
        w, r = self.watcher(state)
        r.at(self.t0 + 20, lambda: self.registry(sid=SID2))
        # ...while a limit THIS client meets in that conversation still counts
        r.at(self.t0 + 60, lambda: append(other, [cc_quota(self.t0 + 60,
                                                            resets=int(self.t0) + 999)]))
        self.assertEqual(w.run(), 0)
        self.assertEqual(len(r.probes), 1)
        self.assertGreaterEqual(r.probes[0]["t"], self.t0 + 60)
        kv, argv = self.relaunch_of(r)
        self.assertEqual((kv["sid"], argv[-2]), (SID2, SID2))
        self.assertEqual(self.events(), ["watch", "detect", "switch"])

    def test_copy_mode_is_left_by_command_before_the_kill(self):
        # a pane scrolled back (copy mode) swallows keys: the mode is cancelled with a
        # tmux command first — never a keystroke — and only then is anything stopped
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_mode = 1
        w.run()
        cancel = r.index(lambda a: "-X" in a)
        self.assertEqual(r.calls[cancel]["argv"],
                         ar.mode_cancel_argv("/private/tmp/tmux-501/default", "%7"))
        self.assertLess(cancel, r.index(lambda a: a[0] == "<signal>"))
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assert_sequence(r)

    def test_a_mode_that_will_not_end_gives_up_before_the_kill(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_mode, r.mode_sticky = 2, True
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertIn("reason=mode", self.log_lines()[-1])
        self.assertEqual(self.events()[-1], "giveup")
        self.assertEqual(r.notices, [])

    def test_synchronized_panes_give_up_without_touching_anything(self):
        # with synchronize-panes on, typed keys reach EVERY pane of the window
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        r.pane_sync, r.pane_mode = 1, 1
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertEqual([c for c in r.calls if "copy-mode" in c["argv"]
                          or "-X" in c["argv"]], [])
        self.assertIn("reason=sync", self.log_lines()[-1])

    def test_a_mode_entered_after_the_kill_is_cancelled_before_typing(self):
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        real_exit = r.exit

        def scroll_back_on_exit(pid):
            real_exit(pid)
            r.pane_mode = 1  # the operator scrolls the pane while the shell comes back
        r.exit = scroll_back_on_exit
        w.run()
        cancel = r.index(lambda a: "-X" in a)
        self.assertEqual(r.calls[cancel]["argv"],
                         ar.mode_cancel_argv("/private/tmp/tmux-501/default", "%7"))
        self.assertLess(r.index(lambda a: a[0] == "<signal>"), cancel)
        self.assertLess(cancel, r.index(lambda a: a[0] == "tmux" and "-l" in a))
        self.assert_sequence(r)

    def test_a_mode_or_sync_that_appears_after_the_kill_is_announced_not_typed_into(self):
        for attr, sticky, reason in (("pane_mode", True, "mode"), ("pane_sync", False, "sync")):
            (self.root / "selection.log").unlink() if (self.root / "selection.log").exists() \
                else None
            state = self.write_state([])
            self.registry()
            self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
            w, r = self.watcher(state)
            r.mode_sticky = sticky
            real_exit = r.exit

            def on_exit(pid, r=r, real_exit=real_exit, attr=attr):
                real_exit(pid)
                setattr(r, attr, 1)
            r.exit = on_exit
            w.run()
            self.assertEqual([a for a in r.sends if "-l" in a], [], reason)
            self.assertEqual(self.relaunch_files(), [], reason)
            self.assertIn("reason=%s" % reason, self.log_lines()[-1])
            self.assertIn('resume="claude --resume %s"' % SID, self.log_lines()[-1])
            self.assertEqual(len(r.notices), 1, reason)

    def test_a_pane_shell_the_relaunch_cannot_be_typed_into_is_never_stopped(self):
        for comm in ("nu", "/usr/local/bin/tcsh", "-tcsh", "pwsh", "xonsh", "fish", "-fish"):
            (self.root / "selection.log").unlink() if (self.root / "selection.log").exists() \
                else None
            state = self.write_state([])
            self.registry()
            self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
            w, r = self.watcher(state)
            r.shell_comm = comm
            self.assertEqual(w.run(), 0)
            self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []), comm)
            self.assertIn("reason=pane-shell", self.log_lines()[-1], comm)

    def test_a_suspended_or_background_client_is_never_stopped(self):
        # Ctrl-Z'd; not in its terminal's foreground; no terminal at all
        for flags, tty in (("T", "ttys003"), ("T+", "ttys003"), ("S", "ttys003"),
                           ("S", "??")):
            (self.root / "selection.log").unlink() if (self.root / "selection.log").exists() \
                else None
            state = self.write_state([])
            self.registry()
            self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
            w, r = self.watcher(state)
            r.client_stat, r.tty = flags, tty
            self.assertEqual(w.run(), 0)
            self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []), flags)
            self.assertIn("reason=background", self.log_lines()[-1], flags)
        # the suites' no-terminal knob waives the foreground (never the stop) check
        for flags, stopped in (("S", False), ("T", True)):
            state = self.write_state([])
            self.registry()
            w, r = self.watcher(state, {"CLAUDE_MULTIACC_AR_TEST_TTY": "1"})
            r.client_stat, r.tty = flags, "??"
            w.run()
            self.assertEqual(r.signals, [] if stopped else [(self.PID, signal.SIGTERM)], flags)

    def test_another_live_holder_of_the_session_is_never_raced(self):
        # the same conversation open in a second client (any account of the pool)
        state = self.write_state([])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        other = self.root / "acct-02" / "sessions" / "6060.json"
        other.write_text(json.dumps({"pid": 6060, "sessionId": SID, "cwd": str(self.work)}))
        w, r = self.watcher(state)
        r.live.add(6060)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.signals, r.sends, self.relaunch_files()), ([], [], []))
        self.assertIn("reason=holder", self.log_lines()[-1])
        self.assertIn("holder=6060", self.log_lines()[-1])
        # a registry left behind by a dead client is no holder
        state = self.write_state([])
        w, r = self.watcher(state)
        w.run()
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assertEqual(len(r.sends), 4)

    def test_the_default_pool_is_not_named(self):
        # $HOME/.claude-accounts is what the relaunched shim finds on its own; a pool
        # anywhere else (the suites' pool, `CLAUDE_ACCOUNTS_ROOT=/other claude`) is named
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        home = self.tmp / "home"
        home.mkdir()
        os.symlink(str(self.root), str(home / ".claude-accounts"))
        state = self.write_state([], acc_root=str(home / ".claude-accounts"))
        w, r = self.watcher(state, {"HOME": str(home)})
        w.run()
        self.assertEqual([a[-1] for a in r.sends if "-l" in a],
                         [" CLAUDE_MULTIACC_AR=%s:%s %s" % (self.token_of(r), SID,
                                                             self.self_path)])
        self.assert_sequence(r, root=None)

    def test_a_path_the_relaunch_cannot_type_is_never_supervised(self):
        odd = self.tmp / "my pool"
        odd.mkdir()
        state = self.write_state([], acc_root=odd)
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))
        log = (odd / "selection.log").read_text()
        self.assertIn("autoresume giveup", log)
        self.assertIn("reason=path", log)
        # the shim's own path, likewise
        self.self_path = str(self.tmp / "b in" / "claude")
        state = self.write_state([])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertIn("reason=path", self.log_lines()[-1])

    def test_unsupported_argv_never_watches(self):
        state = self.write_state(["-p", "summarize this"])
        self.registry()
        self.transcript([cc_quota(self.t0, resets=int(self.t0) + 999)])
        w, r = self.watcher(state)
        self.assertEqual(w.run(), 0)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertEqual(self.events(), ["giveup"])

    def test_a_dead_pid_or_unknown_identity_exits_at_once(self):
        state = self.write_state([])
        w, r = self.watcher(state)
        r.live.clear()
        self.assertEqual(w.run(), 0)
        w, r = self.watcher(state)
        r.starts.clear()
        self.assertEqual(w.run(), 0)
        self.assertEqual(self.log_lines(), [])


class CodexWatcherTests(PoolCase):
    CHILD = 4300
    GRAND = 4301

    def rollout(self, records, sid=CX_SID, day="2026/09/23", name=None):
        d = self.shared / "sessions" / day
        d.mkdir(parents=True, exist_ok=True)
        path = d / (name or "rollout-2026-09-23T00-39-55-%s.jsonl" % sid)
        with open(path, "a") as fh:
            fh.write(jl(records))
        return path

    def setUp(self):
        super().setUp()
        self.self_path = str(self.tmp / "bin" / "codex")
        for acct in ("acct-01", "acct-02"):
            os.rmdir(str(self.root / acct / "sessions"))
            os.symlink(str(self.shared / "sessions"), str(self.root / acct / "sessions"))

    def run_codex(self, argv, records, env=None, setup=None):
        state = self.write_state(argv, provider="codex")
        path = self.rollout(records)
        w, r = self.watcher(state, env)
        r.pane_cmd = "node"
        if setup:
            setup(w, r, path)
        rc = w.run()
        return rc, w, r, path

    def test_quota_rotates_with_resume_and_the_message_date(self):
        with local_tz("UTC"):
            rc, w, r, _ = self.run_codex(
                ["--yolo", "resume", CX_SID], [
                    cx_meta(self.launched - 7200, cwd=str(self.work)),
                    cx_error(self.launched - 7000, "usage_limit_exceeded", LIMIT_MSG),
                    cx_event(self.t0 - 2, {"type": "task_started"}),
                    cx_tokens(self.t0 - 1, limit_id="premium"),
                    cx_error(self.t0 - 1, "usage_limit_exceeded", LIMIT_MSG)])
        self.assertEqual(rc, 0)
        reset = 1790383140  # Sep 26th, 2026 12:39 AM, read in the local (UTC) zone
        prompt = ar.resume_prompt("quota")
        probe = r.probes[0]
        self.assertEqual(probe["argv"], [self.self_path, "resume", "--yolo", CX_SID, prompt])
        self.assertEqual(probe["env"]["CODEX_MULTIACC_AR_PROBE"], "1")
        self.assertEqual(probe["env"]["CODEX_MULTIACC_AR_AVOID"], "acct-01:%d" % reset)
        self.assertNotIn("CODEX_HOME", probe["env"])
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM)])
        self.assert_sequence(r, "codex")
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["resume", "--yolo", CX_SID, prompt])
        self.assertEqual((kv["provider"], kv["class"], kv["reset"], kv["rtype"], kv["sid"],
                          kv["cwd"]),
                         ("codex", "quota", str(reset), "", CX_SID, str(self.work)))
        self.assertEqual(self.events(), ["watch", "detect", "switch"])
        self.assertEqual(r.sends[2][-1], " CODEX_MULTIACC_AR=%s:%s CODEX_ACCOUNTS_ROOT=%s %s"
                         % (self.token_of(r), CX_SID, self.root, self.self_path))

    def tree(self, r, pgid):
        r.term_kills = False
        r.table = [(self.PID, self.PPID, LSTART), (self.CHILD, self.PID, "c"),
                   (self.GRAND, self.CHILD, "g"), (9999, 1, "x")]
        r.live |= {self.CHILD, self.GRAND, 9999}
        r.starts.update({self.CHILD: "c", self.GRAND: "g", 9999: "x"})
        r.pgids.update({self.PID: pgid, self.CHILD: pgid, self.GRAND: pgid, 9999: 9999})
        if pgid != self.PID:
            r.live.add(pgid)

    def test_term_refused_kills_the_recorded_tree(self):
        # node wrapper + native binary: the client and the tree it spawned go, each
        # re-verified by start time; a process of the same group that is not the
        # client's (9999: the shim's detached limits refresh) stays
        def setup(w, r, p):
            self.tree(r, self.PID)
            r.pgids[9999] = self.PID
        rc, w, r, _ = self.run_codex(
            ["resume", CX_SID], [cx_meta(self.launched, cwd=str(self.work)),
                                 cx_error(self.t0, "unauthorized", "401")],
            {"CODEX_MULTIACC_AR_TERM_GRACE": "2"}, setup)
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM), (self.PID, signal.SIGKILL),
                                     (self.CHILD, signal.SIGKILL), (self.GRAND, signal.SIGKILL)])
        self.assertEqual(r.live, {9999})
        kv, _ = self.relaunch_of(r)
        self.assertEqual(kv["class"], "auth")

    def test_a_client_outside_its_own_group_is_never_stopped(self):
        rc, w, r, _ = self.run_codex(
            ["resume", CX_SID], [cx_meta(self.launched, cwd=str(self.work)),
                                 cx_error(self.t0, "unauthorized", "401")],
            {"CODEX_MULTIACC_AR_TERM_GRACE": "2"}, lambda w, r, p: self.tree(r, 7000))
        self.assertEqual((rc, r.signals, r.sends, self.relaunch_files()), (0, [], [], []))
        self.assertIn("reason=pgrp", self.log_lines()[-1])
        # under the suites' knob the recorded tree is stopped, never the group
        rc, w, r, _ = self.run_codex(
            ["resume", CX_SID], [], {"CODEX_MULTIACC_AR_TERM_GRACE": "2",
                                     "CODEX_MULTIACC_AR_TEST_TTY": "1"},
            lambda w, r, p: self.tree(r, 7000))
        self.assertEqual(r.signals, [(self.PID, signal.SIGTERM), (self.PID, signal.SIGKILL),
                                     (self.CHILD, signal.SIGKILL), (self.GRAND, signal.SIGKILL)])
        self.assertEqual(r.live, {9999, 7000})
        self.assertEqual(len(r.sends), 4)

    def test_discovery_by_the_native_childs_open_file(self):
        def setup(w, r, path):
            r.table = [(self.PID, self.PPID, LSTART), (self.CHILD, self.PID, "c")]
            r.live.add(self.CHILD)
            r.files[self.CHILD] = ["/dev/ttys003", str(path)]
            # a decoy: another new rollout in the same cwd — (c) alone would refuse
            self.rollout([cx_meta(self.t0, sid=CX_SID2, cwd=str(self.work))], sid=CX_SID2)
        rc, w, r, path = self.run_codex(
            ["--dangerously-bypass-approvals-and-sandbox"],
            [cx_meta(self.t0 - 1, cwd=str(self.work)),
             cx_error(self.t0, "server_overloaded")],
            {"CODEX_MULTIACC_AR_TRANSIENT_LADDER": "0"}, setup)
        self.assertEqual(w.sid, CX_SID)
        kv, argv = self.relaunch_of(r)
        self.assertEqual(argv, ["resume", "--dangerously-bypass-approvals-and-sandbox", CX_SID,
                                ar.resume_prompt("transient")])
        self.assertEqual(kv["class"], "transient")

    def no_open_files(self, w, r, path):
        r.open_files_ok = False  # no /proc, no lsof: the cwd fallback is all there is

    def test_discovery_fallback_needs_exactly_one_candidate(self):
        # one new rollout in this cwd: taken
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.t0 - 1, cwd=str(self.work)),
                                      cx_error(self.t0, "rate_limit_exceeded", "slow")],
                                     None, self.no_open_files)
        self.assertEqual(w.sid, CX_SID)
        self.assertEqual(len(r.sends), 4)

    def test_fallback_never_guesses_where_open_files_can_be_read(self):
        # Review 2026-09-23: a fresh, idle codex has no rollout yet (codex writes it at the
        # first message), and a NEIGHBOUR session in the same cwd — unsupervised, or
        # supervised past the 2-day prune — wrote one since this launch. The fallback
        # adopted it; a limit there stopped THIS session and typed the neighbour's
        # `codex resume` into its pane. Where the open-file lookup works it is the answer.
        def setup(w, r, path):
            r.at(self.t0 + 400, lambda: r.exit(self.PID))
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.t0 - 1, cwd=str(self.work)),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     {"CODEX_MULTIACC_AR_DISCOVER_TIMEOUT": "20"}, setup)
        self.assertEqual(rc, 0)
        self.assertIsNone(w.tail)
        self.assertEqual((r.probes, r.signals, r.sends), ([], [], []))
        # ...and it never stopped looking: past DISCOVER_TIMEOUT only more slowly
        self.assertNotIn("giveup", self.events())

    def test_session_first_used_after_the_discovery_window_is_still_found(self):
        # codex creates the rollout at the first message: a TUI opened now and used
        # ten minutes later is found by the (slowed) open-file lookup then.
        seen = []

        def setup(w, r, path):
            r.table = [(self.PID, self.PPID, LSTART), (self.CHILD, self.PID, "c")]
            r.live.add(self.CHILD)
            real = r.ps_table
            r.ps_table = lambda: (seen.append(r.t), real())[1]
            r.at(self.t0 + 600, lambda: r.files.__setitem__(self.CHILD, [str(path)]))
        rc, w, r, path = self.run_codex(
            ["--yolo"], [cx_meta(self.t0 + 599, cwd=str(self.work)),
                         cx_error(self.t0 + 599, "usage_limit_exceeded", LIMIT_MSG)],
            {"CODEX_MULTIACC_AR_DISCOVER_TIMEOUT": "20"}, setup)
        self.assertEqual(w.sid, CX_SID)
        self.assertEqual(len(r.sends), 4)
        late = [t for t in seen if self.t0 + 30 < t <= self.t0 + 601]  # discovery only
        self.assertTrue(late)
        gaps = [b - a for a, b in zip(late, late[1:])]
        self.assertTrue(all(g >= ar.DISCOVER_SLOW - 0.5 for g in gaps), gaps)

    def test_fallback_needs_a_session_created_since_the_launch(self):
        # a rollout WRITTEN since the launch but created before it is someone else's
        def setup(w, r, path):
            r.open_files_ok = False
            r.at(self.t0 + 30, lambda: r.exit(self.PID))
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.launched - 600, cwd=str(self.work)),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     {"CODEX_MULTIACC_AR_DISCOVER_TIMEOUT": "20"}, setup)
        self.assertIsNone(w.tail)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertIn("reason=discover", self.log_lines()[-1])

    def test_fallback_once_the_open_file_lookup_keeps_failing(self):
        # lsof is there but every call fails: after LOOKUP_FAILED s it counts as absent
        seen = []

        def setup(w, r, path):
            r.lookup_fails = True
            real = r.run
            r.run = lambda argv, *a, **k: (seen.append(r.t) if argv[0] == self.self_path
                                           else None, real(argv, *a, **k))[1]
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.t0 - 1, cwd=str(self.work)),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     None, setup)
        self.assertEqual(w.sid, CX_SID)
        self.assertGreaterEqual(seen[0], self.t0 + ar.LOOKUP_FAILED)
        self.assertEqual(len(r.sends), 4)

    def test_discovery_gives_up_when_ambiguous(self):
        def setup(w, r, path):
            r.open_files_ok = False
            self.rollout([cx_meta(self.t0, sid=CX_SID2, cwd=str(self.work))], sid=CX_SID2)
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.t0 - 1, cwd=str(self.work)),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     {"CODEX_MULTIACC_AR_DISCOVER_TIMEOUT": "20"}, setup)
        self.assertEqual(rc, 0)
        self.assertIsNone(w.tail)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertIn("reason=discover", self.log_lines()[-1])

    def test_fallback_skips_rollouts_other_watchers_follow_and_foreign_ones(self):
        def setup(w, r, path):
            r.open_files_ok = False
            other = self.rollout([cx_meta(self.t0, sid=CX_SID2, cwd=str(self.work))],
                                 sid=CX_SID2)
            (self.ar_dir / "5555.rollout").write_text(str(other))
            r.live.add(5555)
            # neither a subagent file, another cwd, nor an exec session is a candidate
            self.rollout([cx_meta(self.t0, sid="01a0cb0f-0000-7000-8000-000000000001",
                                  session_id=CX_SID, cwd=str(self.work))],
                         name="rollout-sub.jsonl")
            self.rollout([cx_meta(self.t0, sid="01a0cb0f-0000-7000-8000-000000000002",
                                  cwd="/elsewhere")], name="rollout-cwd.jsonl")
            self.rollout([cx_meta(self.t0, sid="01a0cb0f-0000-7000-8000-000000000003",
                                  cwd=str(self.work), originator="codex_exec")],
                         name="rollout-exec.jsonl")
        rc, w, r, _ = self.run_codex(["fix the build now"],
                                     [cx_meta(self.t0 - 1, cwd=str(self.work)),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     None, setup)
        self.assertEqual(w.sid, CX_SID)
        self.assertEqual(len(r.sends), 4)

    def test_codex_never_classes_and_cancel(self):
        def setup(w, r, path):
            r.at(self.t0 + 1.5, lambda: append(path, [
                cx_event(self.t0 + 1, {"type": "user_message", "message": "wait"})]))
            r.at(self.t0 + 9, lambda: r.exit(self.PID))
        rc, w, r, _ = self.run_codex(["resume", CX_SID],
                                     [cx_meta(self.launched, cwd=str(self.work)),
                                      cx_error(self.t0 - 2, "cyber_policy"),
                                      cx_error(self.t0, "usage_limit_exceeded", LIMIT_MSG)],
                                     None, setup)
        self.assertEqual((r.probes, r.signals), ([], []))
        self.assertEqual(self.events(), ["watch", "never", "detect"])

    def test_codex_exit_is_never_a_crash(self):
        def setup(w, r, path):
            r.at(self.t0 + 90, lambda: r.exit(self.PID))
        rc, w, r, _ = self.run_codex(["resume", CX_SID],
                                     [cx_meta(self.launched, cwd=str(self.work))], None, setup)
        self.assertEqual((rc, r.sends, self.relaunch_files()), (0, [], []))


# =====================================================================================
# the CLI, and the real watch process end to end
# =====================================================================================

class CliTests(PoolCase):
    def cli(self, *args, env=None, timeout=30):
        full = dict(os.environ)
        full.update(env or {})
        for key in [k for k, v in full.items() if v is None]:
            del full[key]  # None: this variable must not reach the child at all
        return subprocess.run([sys.executable, "-I", str(LIB)] + list(args),
                              stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=full,
                              timeout=timeout, universal_newlines=True)

    def test_classify(self):
        path = self.transcript([cc_quota(self.t0 - 10000, resets=1790123817),
                                cc_api_error(self.t0, "invalid_request"),
                                cc_quota(self.t0, "seven_day", resets=1790372787),
                                cc_user(self.t0 + 1, "go"),
                                cc_api_error(self.t0 + 2, "authentication_failed")])
        out = self.cli("classify", "--provider", "claude", str(path),
                       "--since", str(int(self.t0) - 5))
        self.assertEqual(out.returncode, 0, out.stderr)
        doc = json.loads(out.stdout)
        self.assertEqual([e["kind"] for e in doc["events"]],
                         ["never", "verdict", "cancel", "verdict"])
        self.assertEqual((doc["events"][1]["class"], doc["events"][1]["rtype"],
                          doc["events"][1]["reset"]), ("quota", "seven_day", 1790372787))
        self.assertEqual(doc["pending"]["class"], "auth")
        doc = json.loads(self.cli("classify", "--provider", "claude", str(path)).stdout)
        self.assertEqual(doc["events"][0]["kind"], "verdict")  # no --since: everything counts

    def test_classify_codex(self):
        path = self.tmp / "r.jsonl"
        path.write_text(jl([cx_meta(self.t0), cx_tokens(self.t0, limit_id="premium"),
                            cx_error(self.t0, {"usage_limit_exceeded": None}, LIMIT_MSG)]))
        doc = json.loads(self.cli("classify", "--provider", "codex", str(path),
                                  env={"TZ": "UTC"}).stdout)
        self.assertEqual(doc["pending"]["class"], "quota")
        self.assertEqual(doc["pending"]["reset"], 1790383140)
        self.assertNotIn("_message", doc["pending"])
        self.assertNotIn("usage limit", json.dumps(doc))

    def test_relaunch_argv(self):
        (self.root / "acct-01" / "settings.json").write_text(json.dumps({"model": "opus"}))
        state = self.write_state(["--dangerously-skip-permissions", "-c"])
        out = self.cli("relaunch-argv", "--provider", "claude", "--state", state, "--sid", SID,
                       "--class", "quota")
        self.assertEqual(out.returncode, 0, out.stderr)
        self.assertEqual(json.loads(out.stdout), ["--dangerously-skip-permissions", "--resume",
                                                  SID, ar.resume_prompt("quota")])
        out = self.cli("relaunch-argv", "--provider", "claude", "--state", state, "--sid", SID,
                       "--class", "crash",
                       env={"CLAUDE_MULTIACC_AUTORESUME_PROMPT": "carry on: {reason}"})
        self.assertEqual(json.loads(out.stdout)[-1],
                         "carry on: the previous process exited unexpectedly")
        bad = self.write_state(["mcp"])
        out = self.cli("relaunch-argv", "--provider", "claude", "--state", bad, "--sid", SID,
                       "--class", "quota")
        self.assertEqual((out.returncode, out.stdout.strip()), (1, "null"))
        cx = self.write_state(["resume", "--last"], provider="codex")
        out = self.cli("relaunch-argv", "--provider", "codex", "--state", cx, "--sid", CX_SID,
                       "--class", "auth")
        self.assertEqual(json.loads(out.stdout),
                         ["resume", CX_SID, ar.resume_prompt("auth")])

    def test_usage_errors(self):
        self.assertEqual(self.cli().returncode, 2)
        self.assertEqual(self.cli("bogus").returncode, 2)
        self.assertEqual(self.cli("watch").returncode, 2)
        self.assertEqual(self.cli("classify", "--provider", "gemini", "x").returncode, 2)
        missing = self.cli("classify", "--provider", "claude", str(self.tmp / "none.jsonl"))
        self.assertEqual(missing.returncode, 2)
        self.assertNotIn("Traceback", missing.stderr)
        path = self.transcript([])
        self.assertEqual(self.cli("classify", "--provider", "claude", str(path),
                                  "--since", "soon").returncode, 2)

    def trust(self, *args):
        out = self.cli("trust", *args)
        self.assertEqual((out.returncode, out.stdout, out.stderr), (0, "", ""), args)

    def test_trust_marks_the_cwd_and_keeps_everything_else(self):
        acct = self.root / "acct-02"
        conf = acct / ".claude.json"
        doc = {"numStartups": 3, "mcpServers": {"x": {"command": "y"}},
               "projects": {"/elsewhere": {"hasTrustDialogAccepted": True},
                            str(self.work): {"allowedTools": ["Bash"],
                                             "hasTrustDialogAccepted": False}},
               "oauthAccount": {"emailAddress": "a@b.c"}}
        conf.write_text(json.dumps(doc))
        conf.chmod(0o640)
        self.trust("--acct-dir", str(acct), "--cwd", str(self.work))
        got = json.loads(conf.read_text())
        doc["projects"][str(self.work)]["hasTrustDialogAccepted"] = True
        self.assertEqual(got, doc)
        self.assertEqual(list(got), list(doc))  # key order kept
        self.assertEqual(stat.S_IMODE(conf.stat().st_mode), 0o640)
        # a directory with no entry gets one holding just that key
        self.trust("--acct-dir", str(acct), "--cwd", "/new/dir")
        got = json.loads(conf.read_text())
        self.assertEqual(got["projects"]["/new/dir"], {"hasTrustDialogAccepted": True})
        self.assertEqual(got["projects"][str(self.work)]["allowedTools"], ["Bash"])
        # already trusted: the file is not rewritten
        before = conf.stat()
        os.utime(str(conf), (before.st_atime, before.st_mtime - 100))
        mtime = conf.stat().st_mtime
        self.trust("--acct-dir", str(acct), "--cwd", "/new/dir")
        self.assertEqual(conf.stat().st_mtime, mtime)
        self.assertEqual(sorted(p.name for p in acct.iterdir()),
                         [".claude.json", "projects", "sessions"])  # no temp file left

    def test_trust_without_projects_and_fail_open_cases(self):
        acct = self.root / "acct-02"
        conf = acct / ".claude.json"
        conf.write_text(json.dumps({"numStartups": 1}))
        self.trust("--acct-dir", str(acct), "--cwd", str(self.work))
        self.assertEqual(json.loads(conf.read_text()),
                         {"numStartups": 1,
                          "projects": {str(self.work): {"hasTrustDialogAccepted": True}}})
        # nothing to change, nothing written, always 0 and silent
        self.trust("--acct-dir", str(self.root / "acct-01"), "--cwd", str(self.work))
        self.assertFalse((self.root / "acct-01" / ".claude.json").exists())
        for text in ("{not json", "[1, 2]", json.dumps({"projects": []}),
                     json.dumps({"projects": {str(self.work): "odd"}})):
            conf.write_text(text)
            self.trust("--acct-dir", str(acct), "--cwd", str(self.work))
            self.assertEqual(conf.read_text(), text)
        conf.write_text("{}")
        self.trust("--acct-dir", str(acct), "--cwd", "relative/dir")
        self.trust("--acct-dir", str(acct))
        self.trust("--cwd", str(self.work))
        self.trust("--acct-dir")
        self.trust()
        self.assertEqual(conf.read_text(), "{}")

    def test_trust_marks_the_logical_and_the_physical_path(self):
        # the shim passes the logical $PWD; Claude may key projects by the physical one
        acct = self.root / "acct-02"
        conf = acct / ".claude.json"
        conf.write_text(json.dumps({"projects": {}}))
        link = self.tmp / "link to work"
        os.symlink(str(self.work), str(link))
        self.trust("--acct-dir", str(acct), "--cwd", str(link))
        projects = json.loads(conf.read_text())["projects"]
        self.assertEqual(projects, {str(link): {"hasTrustDialogAccepted": True},
                                    str(self.work): {"hasTrustDialogAccepted": True}})
        # one of the two already trusted: the other is added, the first left as it was
        conf.write_text(json.dumps({"projects": {str(self.work): {
            "hasTrustDialogAccepted": True, "allowedTools": []}}}))
        self.trust("--acct-dir", str(acct), "--cwd", str(link))
        projects = json.loads(conf.read_text())["projects"]
        self.assertEqual(projects, {str(self.work): {"hasTrustDialogAccepted": True,
                                                     "allowedTools": []},
                                    str(link): {"hasTrustDialogAccepted": True}})
        # an entry of an unexpected shape under either name: nothing is written
        text = json.dumps({"projects": {str(self.work): "odd"}})
        conf.write_text(text)
        self.trust("--acct-dir", str(acct), "--cwd", str(link))
        self.assertEqual(conf.read_text(), text)

    def test_trust_never_overwrites_a_file_rewritten_meanwhile(self):
        # Review 2026-09-23: a running Claude rewrites .claude.json whenever it likes; a
        # write between our read and our replace must survive, not be lost to a stale copy
        acct = self.root / "acct-02"
        conf = acct / ".claude.json"
        real_mkstemp = tempfile.mkstemp
        for concurrent in ('{"numStartups": 99, "projects": {}}',  # size differs
                           '{"numStartups": 2}'):                   # same size, newer mtime
            conf.write_text('{"numStartups": 1}')
            os.utime(str(conf), (self.t0 - 100, self.t0 - 100))

            def racing_mkstemp(*args, concurrent=concurrent, **kwargs):
                conf.write_text(concurrent)  # Claude writes right after our read
                return real_mkstemp(*args, **kwargs)
            with mock.patch.object(ar.tempfile, "mkstemp", racing_mkstemp):
                self.assertIsNone(ar.cmd_trust(str(acct), str(self.work)))
            self.assertEqual(conf.read_text(), concurrent)
            self.assertEqual(sorted(p.name for p in acct.iterdir()),
                             [".claude.json", "projects", "sessions"])  # no temp file left
        # untouched meanwhile: written
        conf.write_text('{"numStartups": 1}')
        ar.cmd_trust(str(acct), str(self.work))
        self.assertEqual(json.loads(conf.read_text())["projects"],
                         {str(self.work): {"hasTrustDialogAccepted": True}})

    def test_watch_on_a_malformed_state_exits_quietly(self):
        bad = self.ar_dir / "1234.state"
        bad.write_text("v=1\nprovider=claude\n")
        out = self.cli("watch", "--state", str(bad), timeout=15)
        self.assertEqual((out.returncode, out.stdout, out.stderr), (0, "", ""))

    def test_watch_prunes_leftovers(self):
        old = self.ar_dir / "r-OldTokenOldToken.relaunch"
        old.write_text("v=1\n")
        os.utime(str(old), (time.time() - 3 * 86400,) * 2)
        fresh = self.ar_dir / "r-NewTokenNewToken.relaunch"
        fresh.write_text("v=1\n")
        # an old claim of a session that still runs stays (a neighbour's watcher reads
        # it); a dead one's goes
        live = self.ar_dir / ("%d.rollout" % os.getpid())
        dead = self.ar_dir / ("%d.rollout" % (2 ** 22 + 12346))
        for path in (live, dead):
            path.write_text("/x/rollout.jsonl")
            os.utime(str(path), (time.time() - 3 * 86400,) * 2)
        state = self.write_state([], pid=2 ** 22 + 12345)  # no such process
        self.assertEqual(self.cli("watch", "--state", state, timeout=15).returncode, 0)
        self.assertFalse(old.exists())
        self.assertTrue(fresh.exists())
        self.assertTrue(live.exists())
        self.assertFalse(dead.exists())

    @unittest.skipUnless(shutil.which("ps") and shutil.which("sleep"), "needs ps and sleep")
    def test_watch_end_to_end_with_real_processes(self):
        """The real Runner: ps/lstart, kill, a probe subprocess and tmux on PATH."""
        bindir = self.tmp / "fakebin"
        bindir.mkdir()
        tmux_log = self.tmp / "tmux.log"
        probe_log = self.tmp / "probe.log"
        # Enter stands in for the relaunched shim, which consumes its token first thing
        (bindir / "tmux").write_text(
            "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$FAKE_TMUX_LOG\"\n"
            "case \"$*\" in *display-message\\ -p*) echo \"$FAKE_PANE_PID|0|0|zsh\" ;;\n"
            "  *send-keys*Enter) for f in \"$FAKE_AR_DIR\"/r-*.relaunch; do\n"
            "    [ -f \"$f\" ] && mv \"$f\" \"$f.used\"; done ;; esac\n"
            "exit 0\n")
        os.makedirs(os.path.dirname(self.self_path), exist_ok=True)
        Path(self.self_path).write_text(
            "#!/bin/sh\nprintf 'avoid=%s probe=%s cfg=%s args=%s\\n' "
            "\"$CLAUDE_MULTIACC_AR_AVOID\" \"$CLAUDE_MULTIACC_AR_PROBE\" "
            "\"${CLAUDE_CONFIG_DIR:-unset}\" \"$*\" >> \"$FAKE_PROBE_LOG\"\n"
            "echo 'pick=acct-02 tier=eligible'\n")
        for path in (bindir / "tmux", Path(self.self_path)):
            path.chmod(0o755)
        # A disposable stand-in for the TUI, reparented to init so its exit is reaped, and
        # one for the pane's shell (ps must name a shell there). Both in sessions of their
        # own: no controlling terminal, whatever terminal (or none) runs this suite.
        out = subprocess.run(["sh", "-c", "sleep 120 </dev/null >/dev/null 2>&1 & echo $!"],
                             stdout=subprocess.PIPE, universal_newlines=True, timeout=10,
                             start_new_session=True)
        pid = int(out.stdout.strip())
        self.addCleanup(lambda: _kill_quietly(pid))
        shell = subprocess.Popen(["sh", "-c", "sleep 120; :"], stdin=subprocess.DEVNULL,
                                 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                 start_new_session=True)
        self.addCleanup(lambda: (_kill_group_quietly(shell.pid), shell.wait()))
        self.PPID = shell.pid
        self.launched = int(time.time()) - 1
        state = self.write_state(["--dangerously-skip-permissions", "do the thing"], pid=pid)
        self.registry(pid=pid)
        self.transcript([cc_quota(time.time(), resets=int(time.time()) + 3600)])
        env = {"PATH": "%s:%s" % (bindir, os.environ.get("PATH", "/usr/bin:/bin")),
               "FAKE_TMUX_LOG": str(tmux_log), "FAKE_PROBE_LOG": str(probe_log),
               "FAKE_PANE_PID": str(self.PPID), "FAKE_AR_DIR": str(self.ar_dir),
               "CLAUDE_CONFIG_DIR": str(self.root / "acct-01"),
               "CLAUDE_ACCOUNTS_ROOT": str(self.root), "CLAUDE_MULTIACC_AR_TEST_TTY": "1",
               "CLAUDE_MULTIACC_AR_POLL": "0.1", "CLAUDE_MULTIACC_AR_GRACE": "0.3",
               "CLAUDE_MULTIACC_AR_TERM_GRACE": "3", "CLAUDE_MULTIACC_AR_PANE_WAIT": "5",
               "CLAUDE_MULTIACC_AUTORESUME_DEBUG": "1"}
        started = time.time()
        res = self.cli("watch", "--state", state, env=env, timeout=60)
        self.assertEqual(res.returncode, 0, res.stderr)
        self.assertLess(time.time() - started, 30)
        # The watcher returns once the pid is gone OR a zombie; reaping the orphan is
        # init's (or a CI subreaper's) job, and it may take a moment.
        deadline = time.time() + 10
        while time.time() < deadline:
            try:
                os.kill(pid, 0)
            except ProcessLookupError:
                break
            time.sleep(0.05)
        with self.assertRaises(ProcessLookupError):
            os.kill(pid, 0)
        probe = probe_log.read_text()
        self.assertIn("probe=1 cfg=unset args=--dangerously-skip-permissions --resume %s "
                      "(claude-multiacc auto-resume)" % SID, probe)
        self.assertIn("avoid=acct-01:", probe)
        lines = [l for l in tmux_log.read_text().splitlines() if "send-keys" in l]
        sock = "-S /private/tmp/tmux-501/default"
        self.assertEqual(len(lines), 4, lines)
        self.assertEqual(lines[0], "%s send-keys -R -t %%7" % sock)
        self.assertEqual(lines[1], "%s send-keys -t %%7 C-u" % sock)
        # typed through the shim that launched the session, naming the pool it used
        self.assertRegex(lines[2], r"^%s send-keys -t %%7 -l  CLAUDE_MULTIACC_AR="
                                   r"[A-Za-z0-9]{16}:%s CLAUDE_ACCOUNTS_ROOT=%s %s$"
                         % (sock, SID, re.escape(str(self.root)), re.escape(self.self_path)))
        self.assertEqual(lines[3], "%s send-keys -t %%7 Enter" % sock)
        tok = lines[2].split("CLAUDE_MULTIACC_AR=", 1)[1].split(":")[0]
        self.assertTrue((self.ar_dir / ("r-%s.relaunch.used" % tok)).exists())
        # the watcher's own state/argv files are gone; its debug log stays
        self.assertFalse(Path(state).exists())
        self.assertFalse(Path(ar.argv_path_for(state)).exists())
        self.assertTrue((self.ar_dir / ("%d.log" % pid)).exists())
        self.assertEqual(self.events(), ["watch", "detect", "switch"])


class RunnerTests(unittest.TestCase):
    """The real side-effect layer, on this machine's own ps/lsof or /proc."""

    def test_signal_refuses_init_groups_and_itself(self):
        # Signal 0 only: a regression here must never be able to hurt this machine.
        r = ar.Runner()
        for pid in (1, 0, -1, os.getpid(), "123", None):
            self.assertFalse(r.signal(pid, 0), pid)

    def test_run_is_bounded(self):
        r = ar.Runner()
        started = time.time()
        # a grandchild left in the background must not hold the call open
        rc, out = r.run(["sh", "-c", "sleep 3 >/dev/null 2>&1 & echo hi"], 10)
        self.assertEqual((rc, out), (0, "hi\n"))
        self.assertLess(time.time() - started, 2.5)
        started = time.time()
        rc, _ = r.run(["sh", "-c", "sleep 30"], 0.5)
        self.assertIsNone(rc)
        self.assertLess(time.time() - started, 5)
        self.assertEqual(r.run(["/nonexistent/binary"], 5), (None, ""))

    @unittest.skipUnless(shutil.which("ps"), "needs ps")
    def test_identity_and_zombies(self):
        r = ar.Runner()
        self.assertTrue(r.lstart(os.getpid()))
        self.assertIn((os.getpid(), os.getppid()),
                      [(p, pp) for p, pp, _ in r.ps_table()])
        child = subprocess.Popen(["sh", "-c", "exit 0"])
        try:
            deadline = time.time() + 10
            while not r.gone(child.pid) and time.time() < deadline:
                time.sleep(0.05)
            # not reaped yet: kill -0 still succeeds, but it is gone
            self.assertTrue(r.alive(child.pid))
            self.assertTrue(r.gone(child.pid))
        finally:
            child.wait()
        self.assertFalse(r.alive(child.pid))
        self.assertIsNone(r.lstart(child.pid))

    @unittest.skipUnless(os.path.isdir("/proc/self/fd") or shutil.which("lsof")
                         or os.path.exists("/usr/sbin/lsof"), "needs /proc or lsof")
    def test_open_files(self):
        with tempfile.NamedTemporaryFile(suffix=".jsonl") as fh:
            names = {os.path.realpath(p) for p in ar.Runner().open_files(os.getpid())}
            self.assertIn(os.path.realpath(fh.name), names)


def _kill_quietly(pid):
    try:
        os.kill(pid, signal.SIGKILL)
    except OSError:
        pass


def _kill_group_quietly(pgid):
    try:
        os.killpg(pgid, signal.SIGKILL)
    except OSError:
        pass


if __name__ == "__main__":
    unittest.main()