"""Issue #78: multi-agent-mux-reset — unit tests for lib_py.reset and
end-to-end tests for reset.sh.
"""
from __future__ import annotations

import ast
import json
import os
import re
import shlex
import subprocess
import sys
import time
from pathlib import Path
from typing import Dict, List, Optional

import pytest
import yaml

REPO = Path(__file__).resolve().parents[1]
SKILLS = REPO / ".agents" / "skills"
if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

UA = "a7800001-0000-4000-8000-000000000001"
UB = "b7800002-0000-4000-8000-000000000002"
UC = "c7800003-0000-4000-8000-000000000003"
UD = "d7800004-0000-4000-8000-000000000004"
UORC = "e7800005-0000-4000-8000-000000000005"

pytestmark = pytest.mark.usefixtures("own_workspace_root")


# --------------------------------------------------------------------------- Helpers

def _row(
    name: str,
    agent: str,
    cwd: Path,
    role: str = "reviewer",
    status: str = "running",
    created_at: str = "2026-10-01T00:00:00Z",
    server: str = "default",
    label: str = "ws",
    own_id: Optional[str] = None,
    pid: int = 4242,
    epoch: int = 1700000000,
) -> dict:
    from lib_py.agents.registry import own_key

    key = own_key(agent) if agent in ("claude", "agy", "codex", "grok", "hermes", "opencode") else "unknown_id"
    return {
        "name": name,
        "status": status,
        "role": role,
        "agent": agent,
        "herdr_session": server,
        "herdr_server": server,
        "herdr_workspace": label,
        "herdr_session_created_at": created_at,
        "herdr_session_epoch": epoch,
        key: own_id,
        "session_id_source": "discovered" if own_id else "pending-discovery",
        "session_id_verified": bool(own_id),
        "pane": {"cwd": str(cwd), "pid": pid, "cmd": agent},
    }


def _run_reset_py(
    args: List[str],
    state: Optional[dict] = None,
    stdin_text: Optional[str] = None,
    env_extra: Optional[dict] = None,
):
    env = dict(os.environ)
    env["PYTHONPATH"] = str(SKILLS)
    env["PYTHONDONTWRITEBYTECODE"] = "1"
    if state is not None:
        env["MAM_STATE_JSON"] = json.dumps(state)
    if env_extra:
        env.update(env_extra)
    return subprocess.run(
        [sys.executable, "-m", "lib_py.reset", *args],
        env=env,
        input=stdin_text,
        capture_output=True,
        text=True,
        timeout=15,
    )


def _write_rows(sandbox: Path, rows: list, orchestrator_uuids: Optional[List[str]] = None) -> Path:
    lib = sandbox / ".agents" / "skills" / "lib.sh"
    yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
    mutation = f"d['herdr_sessions'] = {repr(rows)}\n"
    if orchestrator_uuids is not None:
        mutation += f"d['orchestrator_uuids'] = {repr(orchestrator_uuids)}\n"
    res = subprocess.run(
        ["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'],
        input=mutation,
        capture_output=True,
        text=True,
        cwd=str(sandbox),
        timeout=60,
    )
    assert res.returncode == 0, res.stdout + res.stderr
    return yaml_path


def _read_rows(sandbox: Path) -> Dict[str, dict]:
    yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
    data = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) or {}
    return {r["name"]: r for r in data.get("herdr_sessions") or []}


def _db_session_names(sandbox: Path) -> Dict[str, str]:
    import sqlite3

    db_path = sandbox / ".mam" / "agent-sessions.db"
    if not db_path.exists():
        return {}
    conn = sqlite3.connect(str(db_path))
    try:
        return dict(conn.execute("SELECT name, status FROM sessions").fetchall())
    finally:
        conn.close()


def _claude_transcript(sandbox: Path, ws: Path, uuid: str, mtime: Optional[float] = None) -> Path:
    from lib_py.verify_session import workspace_key

    pdir = sandbox / ".claude" / "projects" / workspace_key(str(ws))
    pdir.mkdir(parents=True, exist_ok=True)
    f = pdir / f"{uuid}.jsonl"
    f.write_text(json.dumps({"sessionId": uuid, "cwd": str(ws), "type": "user"}) + "\n", encoding="utf-8")
    if mtime is not None:
        os.utime(f, (mtime, mtime))
    return f


def _agy_full_artifacts(home: Path, uuid: str, mtime: Optional[float] = None) -> List[Path]:
    import sqlite3

    base = Path(home) / ".gemini" / "antigravity-cli"
    db = base / "conversations" / f"{uuid}.db"
    db.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(db))
    conn.execute("CREATE TABLE steps (id INTEGER)")
    conn.execute("INSERT INTO steps VALUES (1)")
    conn.commit()
    conn.close()
    if mtime is not None:
        os.utime(db, (mtime, mtime))
    wal = Path(str(db) + "-wal")
    shm = Path(str(db) + "-shm")
    wal.write_text("wal", encoding="utf-8")
    shm.write_text("shm", encoding="utf-8")
    brain = base / "brain" / uuid
    brain.mkdir(parents=True, exist_ok=True)
    (brain / "note.md").write_text("note", encoding="utf-8")
    ann = base / "annotations" / f"{uuid}.pbtxt"
    ann.parent.mkdir(parents=True, exist_ok=True)
    ann.write_text("pbtxt", encoding="utf-8")
    lock = base / "presence" / f"{uuid}.lock"
    lock.parent.mkdir(parents=True, exist_ok=True)
    lock.write_text("", encoding="utf-8")
    return [db, wal, shm, brain, ann, lock]


def _agy_cache(home: Path, mapping: dict) -> None:
    lc = Path(home) / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
    lc.parent.mkdir(parents=True, exist_ok=True)
    lc.write_text(
        json.dumps({os.path.realpath(str(k)): v for k, v in mapping.items()}),
        encoding="utf-8",
    )


def _fake_lsof_multi(bindir: Path, mapping: dict) -> Path:
    cases = "\n".join(f"    *{k}*) echo {v}; found=1 ;;" for k, v in mapping.items())
    exe = Path(bindir) / "lsof"
    exe.write_text(
        "#!/bin/sh\n"
        "found=0\n"
        'for a in "$@"; do\n'
        '  case "$a" in\n'
        f"{cases}\n"
        "  esac\n"
        "done\n"
        '[ "$found" = "1" ] && exit 0\n'
        "exit 1\n",
        encoding="utf-8",
    )
    exe.chmod(0o755)
    return exe


def _mock_herdr_agent(
    state_file: Path,
    name: str,
    cwd: Path,
    pid: int = 4242,
    epoch: int = 1700000000,
    agent: str = "claude",
    pane_id: str = "",
) -> None:
    state = json.loads(Path(state_file).read_text(encoding="utf-8"))
    pid_str = pane_id or f"w1:p_{name}"
    state.setdefault("agents", {})[name] = {
        "pid": pid,
        "cwd": str(cwd),
        "command": agent,
        "agent": agent,
        "pane_id": pid_str,
        "workspace_id": "w1",
        "created": epoch,
        "buffer": f"Ready in {cwd}\n>",
    }
    Path(state_file).write_text(json.dumps(state), encoding="utf-8")


def _wait_reconcile_idle(sandbox: Path, timeout: float = 30.0) -> None:
    """Wait until any background `reconcile.sh` spawned under `sandbox` finishes.
    Uses `ps -eo pid=,args=` (never `pgrep -f`) so it never matches itself.
    """
    sb = os.path.realpath(str(sandbox))
    deadline = time.time() + timeout
    while time.time() < deadline:
        res = subprocess.run(
            ["ps", "-eo", "pid=,args="],
            stdin=subprocess.DEVNULL,
            capture_output=True,
            text=True,
            timeout=5,
        )
        active = []
        for line in (res.stdout or "").splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split(None, 1)
            if len(parts) < 2 or not parts[0].isdigit():
                continue
            pid = int(parts[0])
            if pid == os.getpid():
                continue
            args = parts[1]
            if "reconcile.sh" in args and (sb in args or str(sandbox) in args):
                active.append((pid, args))
        if not active:
            return
        time.sleep(0.2)
    pytest.fail(f"reconcile.sh still running under {sandbox} after {timeout}s: {active}")


def _install_herdr_spy(sandbox: Path) -> Path:
    """Wrap sandbox/bin/herdr to log verb sequence, spec-at-first-kill, and optional
    run_loop.sh probe at first agent-start.
    """
    bin_dir = sandbox / "bin"
    herdr_bin = bin_dir / "herdr"
    herdr_real = bin_dir / "herdr.real"
    if not herdr_real.exists():
        herdr_bin.rename(herdr_real)
    spy_log = sandbox / "herdr_spy.jsonl"
    probe_file = sandbox / "herdr_spy_loop_probe.json"

    wrapper = f"""#!{sys.executable}
import json, os, subprocess, sys

real = {repr(str(herdr_real))}
spy_log = {repr(str(spy_log))}
probe_file = {repr(str(probe_file))}
sandbox = {repr(str(sandbox))}

raw_args = sys.argv[1:]
args = list(raw_args)
while args and args[0] in ("-L", "--server", "-s", "--session"):
    args = args[2:] if len(args) > 1 else args[1:]

if args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls"):
    verb = args[0]
else:
    verb = " ".join(args[:2]) if len(args) >= 2 else (args[0] if args else "")
rec = {{"verb": verb, "args": args}}

if verb in ("kill-session", "pane close"):
    spec_p = os.path.join(sandbox, ".mam", "reset-spec.json")
    rec["spec_exists"] = os.path.isfile(spec_p)
    if rec["spec_exists"]:
        try:
            with open(spec_p, encoding="utf-8") as fh:
                rec["spec"] = json.load(fh)
        except Exception as exc:
            rec["spec_error"] = str(exc)

if verb == "agent start" and os.environ.get("SPY_PROBE_RUN_LOOP") == "1" and not os.path.exists(probe_file):
    env = dict(os.environ, MAM_LOOP_NO_FREEZE="1")
    loop_sh = os.path.join(sandbox, ".agents", "skills", "multi-agent-mux-loop", "scripts", "run_loop.sh")
    res = subprocess.run(
        ["bash", loop_sh, "--creator", "x", "--task", "t"],
        cwd=sandbox,
        env=env,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=30,
    )
    with open(probe_file, "w", encoding="utf-8") as pf:
        json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr}}, pf)

with open(spy_log, "a", encoding="utf-8") as lf:
    lf.write(json.dumps(rec) + "\\n")

os.execv(real, [real, *raw_args])
"""
    herdr_bin.write_text(wrapper, encoding="utf-8")
    herdr_bin.chmod(0o755)
    return spy_log


def _read_spy_log(spy_log: Path) -> List[dict]:
    if not spy_log.exists():
        return []
    return [json.loads(line) for line in spy_log.read_text(encoding="utf-8").splitlines() if line.strip()]


def _run_reset(
    sandbox: Path,
    *args: str,
    stdin_text: Optional[str] = None,
    env_extra: Optional[dict] = None,
) -> subprocess.CompletedProcess:
    script = sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
    env = dict(
        os.environ,
        SKS_QUIESCENT_TRIES="0",
        SKS_DIALOG_TIMEOUT="0",
        SKS_BUSY_TIMEOUT="0",
        MAM_ONBOARD_ACK_S="3",
        MAM_ONBOARD_WAIT_S="3",
        MAM_NO_SUBSCRIBER_DAEMON="1",
        PYTHONDONTWRITEBYTECODE="1",
    )
    if env_extra:
        env.update(env_extra)
    return subprocess.run(
        ["bash", str(script), *args],
        cwd=str(sandbox),
        env=env,
        input=stdin_text,
        stdin=subprocess.DEVNULL if stdin_text is None else None,
        capture_output=True,
        text=True,
        timeout=180,
    )


# --------------------------------------------------------------------------- Unit tests: lib_py.reset

def test_static_checks_and_skill_version():
    """§5.2-12:
    - bash -n reset.sh
    - ast.parse(feature_version=(3, 9)) on lib_py/reset.py
    - no `kill` command in reset.sh
    - multi-agent-mux-reset/SKILL.md frontmatter version == lib.sh MAM_VERSION
    """
    reset_sh = SKILLS / "multi-agent-mux-reset" / "scripts" / "reset.sh"
    res = subprocess.run(["bash", "-n", str(reset_sh)], capture_output=True, text=True, timeout=10)
    assert res.returncode == 0, res.stderr

    src_py = (SKILLS / "lib_py" / "reset.py").read_text(encoding="utf-8")
    ast.parse(src_py, filename="reset.py", feature_version=(3, 9))

    src_sh = reset_sh.read_text(encoding="utf-8")
    for lineno, line in enumerate(src_sh.splitlines(), 1):
        stripped = line.strip()
        if stripped.startswith("#"):
            continue
        assert not re.search(r"\bkill\b", stripped), f"reset.sh:{lineno} contains forbidden `kill`: {line}"

    # Self-contained version check (no import of test_version_consistency)
    lib_text = (SKILLS / "lib.sh").read_text(encoding="utf-8")
    m_lib = re.search(r'^MAM_VERSION="([^"]+)"', lib_text, re.M)
    assert m_lib, "MAM_VERSION not found in lib.sh"
    mam_version = m_lib.group(1)

    skill_text = (SKILLS / "multi-agent-mux-reset" / "SKILL.md").read_text(encoding="utf-8")
    frontmatter = skill_text.split("---", 2)[1]
    m_skill = re.search(r"^version:\s*(\S+)\s*$", frontmatter, re.M)
    assert m_skill, "version not found in multi-agent-mux-reset/SKILL.md frontmatter"
    assert m_skill.group(1) == mam_version


def test_select_targets_and_plan_cli(tmp_path: Path):
    """§5.2-1: select_targets chooses only status==running rows at realpath(WS),
    resolving symlinks, listing skipped rows with reasons and orchestrator_uuids,
    and refusing running control-root rows with missing agent or empty role.
    """
    ws = tmp_path / "ws"
    ws.mkdir()
    wt = ws / "wt" / "x"
    wt.mkdir(parents=True)
    sym_ws = tmp_path / "sym-ws"
    sym_ws.symlink_to(ws)

    state = {
        "orchestrator_uuids": [UORC],
        "herdr_sessions": [
            _row("reviewer-claude-01", "claude", ws, role="planner,reviewer", own_id=UA, server="srv1", label="lbl1"),
            _row("reviewer-grok-01", "grok", sym_ws, role="reviewer", own_id=UB, server="default", label="lbl2"),
            _row("creator-claude-wt", "claude", wt, role="creator", own_id=UC),
            _row("stopped-codex-01", "codex", ws, status="stopped", own_id=UD),
            _row("dead-hermes-01", "hermes", ws, status="terminated", own_id=None),
        ],
    }

    res = _run_reset_py(["plan", "--workspace", str(ws)], state=state)
    assert res.returncode == 0, res.stderr
    plan = json.loads(res.stdout)
    assert [t["name"] for t in plan["targets"]] == ["reviewer-claude-01", "reviewer-grok-01"]
    assert plan["targets"][0] == {
        "name": "reviewer-claude-01",
        "agent": "claude",
        "role": "planner,reviewer",
        "herdr_server": "srv1",
        "herdr_workspace": "lbl1",
        "old_id": UA,
        "old_created_at": "2026-10-01T00:00:00Z",
    }
    skipped = {s["name"]: s["reason"] for s in plan["skipped"]}
    assert "not the control root" in skipped["creator-claude-wt"]
    assert skipped["stopped-codex-01"] == "status stopped"
    assert skipped["dead-hermes-01"] == "status terminated"
    assert plan["orchestrators"] == [UORC]
    assert plan["errors"] == []

    # Empty role or unknown agent on a running control-root row causes plan & precheck to return 1
    bad_state = {
        "herdr_sessions": [
            _row("bad-role-claude", "claude", ws, role="", own_id=UA),
            {
                "name": "mystery-row",
                "status": "running",
                "role": "reviewer",
                "agent": "not-an-agent",
                "pane": {"cwd": str(ws), "pid": 123, "cmd": "bash"},
            },
        ]
    }
    res_bad = _run_reset_py(["plan", "--workspace", str(ws)], state=bad_state)
    assert res_bad.returncode == 1
    bad_plan = json.loads(res_bad.stdout)
    assert len(bad_plan["errors"]) == 2

    res_pre = _run_reset_py(["precheck", "--workspace", str(ws)], state=bad_state)
    assert res_pre.returncode == 1
    assert "bad-role-claude" in res_pre.stdout
    assert "mystery-row" in res_pre.stdout

    # Review B2: claude/grok get their id at spawn, so a running row without one
    # is refused (stop would fill it with a folder-wide guess); a post-discovered
    # agent without one stays a target (its id is captured by its own pane).
    no_id_state = {
        "herdr_sessions": [
            _row("reviewer-claude-noid", "claude", ws, own_id=None),
            _row("reviewer-grok-noid", "grok", ws, own_id=None),
            _row("reviewer-agy-noid", "agy", ws, own_id=None),
        ]
    }
    res_noid = _run_reset_py(["plan", "--workspace", str(ws)], state=no_id_state)
    assert res_noid.returncode == 1
    noid_plan = json.loads(res_noid.stdout)
    assert [t["name"] for t in noid_plan["targets"]] == ["reviewer-agy-noid"]
    assert len(noid_plan["errors"]) == 2
    for name, agent in (("reviewer-claude-noid", "claude"), ("reviewer-grok-noid", "grok")):
        assert any(
            "running session %s (%s) has no recorded conversation id" % (name, agent) in e
            for e in noid_plan["errors"]
        )


def test_spec_file_write_read_and_validation(tmp_path: Path):
    """D5: write-spec atomically writes .mam/reset-spec.json; corrupt, wrong-version,
    or wrong-workspace spec files are rejected with 'check it, then remove it by hand'.
    """
    from lib_py.reset import SpecError, read_spec

    ws = tmp_path / "ws"
    ws.mkdir()
    targets = [
        {
            "name": "reviewer-claude-01",
            "agent": "claude",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": UA,
            "old_created_at": "2026-10-01T00:00:00Z",
        }
    ]
    res = _run_reset_py(
        ["write-spec", "--workspace", str(ws)],
        stdin_text=json.dumps({"targets": targets}),
    )
    assert res.returncode == 0, res.stderr
    spath = ws / ".mam" / "reset-spec.json"
    assert spath.is_file()
    doc = read_spec(str(ws))
    assert doc["version"] == 1
    assert doc["workspace"] == os.path.realpath(str(ws))
    assert doc["agents"] == targets

    # Wrong workspace
    other_ws = tmp_path / "other"
    other_ws.mkdir()
    with pytest.raises(SpecError, match="check it, then remove it by hand"):
        read_spec(str(other_ws))

    # Wrong version
    spath.write_text(json.dumps({"version": 99, "workspace": str(ws), "agents": []}), encoding="utf-8")
    with pytest.raises(SpecError, match="check it, then remove it by hand"):
        read_spec(str(ws))

    # Unknown or non-canonical agent in saved spec (review P2)
    for bad_agent in ("not-an-agent", "CLAUDE"):
        spath.write_text(
            json.dumps(
                {
                    "version": 1,
                    "workspace": str(ws),
                    "agents": [
                        {
                            "name": "reviewer-bad-01",
                            "agent": bad_agent,
                            "role": "reviewer",
                            "herdr_server": "default",
                            "herdr_workspace": "ws",
                            "old_id": UA,
                            "old_created_at": "2026-10-01T00:00:00Z",
                        }
                    ],
                }
            ),
            encoding="utf-8",
        )
        with pytest.raises(SpecError, match="unknown agent"):
            read_spec(str(ws))

    # Review B2 (codex): every stored field is validated; a lost herdr_server is
    # never replaced by "default", and a name listed twice is refused
    good = dict(targets[0])
    for mutate in (
        lambda e: e.pop("herdr_server"),
        lambda e: e.update(herdr_server=""),
        lambda e: e.update(herdr_server=None),
        lambda e: e.pop("herdr_workspace"),
        lambda e: e.update(herdr_workspace=None),
        lambda e: e.pop("old_id"),
        lambda e: e.update(old_id=""),
        lambda e: e.update(old_id=7),
        lambda e: e.pop("old_created_at"),
        lambda e: e.update(old_created_at=None),
    ):
        bad = dict(good)
        mutate(bad)
        spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [bad]}), encoding="utf-8")
        with pytest.raises(SpecError, match="incomplete agent entry"):
            read_spec(str(ws))
    spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [good, dict(good)]}), encoding="utf-8")
    with pytest.raises(SpecError, match="lists reviewer-claude-01 twice"):
        read_spec(str(ws))
    # Control: legal empty values (label, created_at) and an explicit null old_id pass
    ok = dict(good, herdr_workspace="", old_created_at="", old_id=None)
    spath.write_text(json.dumps({"version": 1, "workspace": str(ws), "agents": [ok]}), encoding="utf-8")
    assert read_spec(str(ws))["agents"] == [ok]

    # Corrupt JSON
    spath.write_text("{corrupt", encoding="utf-8")
    with pytest.raises(SpecError, match="check it, then remove it by hand"):
        read_spec(str(ws))
    res_entries = _run_reset_py(["entries", "--workspace", str(ws)], state={})
    assert res_entries.returncode == 1
    assert "check it, then remove it by hand" in res_entries.stderr


def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
    """D6 1-5 & §5.2-9b:
    #1 row absent -> purged
    #2 same created_at (even if normal stop recorded a new id onto old_id=None) -> pending
    #3 new created_at and id == old_id -> pending (resumed original conversation, 9b)
    #4 new created_at and id empty -> created
    #5 new created_at and id != old_id -> done
    """
    from lib_py.reset import write_spec

    ws = tmp_path / "ws"
    ws.mkdir()
    spec_agents = [
        {
            "name": "s1-purged",
            "agent": "claude",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": UA,
            "old_created_at": "2026-10-01T00:00:00Z",
        },
        {
            "name": "s2-same-inst-newly-recorded",
            "agent": "agy",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": None,
            "old_created_at": "2026-10-01T00:00:00Z",
        },
        {
            "name": "s3-resumed-same-id",
            "agent": "claude",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": UB,
            "old_created_at": "2026-10-01T00:00:00Z",
        },
        {
            "name": "s4-created-no-id",
            "agent": "hermes",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": UC,
            "old_created_at": "2026-10-01T00:00:00Z",
        },
        {
            "name": "s5-done",
            "agent": "codex",
            "role": "reviewer",
            "herdr_server": "default",
            "herdr_workspace": "ws",
            "old_id": None,
            "old_created_at": "2026-10-01T00:00:00Z",
        },
    ]
    write_spec(str(ws), spec_agents)

    state = {
        "herdr_sessions": [
            _row(
                "s2-same-inst-newly-recorded",
                "agy",
                ws,
                status="stopped",
                created_at="2026-10-01T00:00:00Z",
                own_id=UA,
            ),
            _row(
                "s3-resumed-same-id",
                "claude",
                ws,
                status="running",
                created_at="2026-10-02T00:00:00Z",
                own_id=UB,
            ),
            _row(
                "s4-created-no-id",
                "hermes",
                ws,
                status="running",
                created_at="2026-10-02T00:00:00Z",
                own_id=None,
            ),
            _row(
                "s5-done",
                "codex",
                ws,
                status="running",
                created_at="2026-10-02T00:00:00Z",
                own_id=UD,
            ),
        ]
    }

    res = _run_reset_py(["entries", "--workspace", str(ws)], state=state)
    assert res.returncode == 0, res.stderr
    parsed = {}
    for line in res.stdout.strip().splitlines():
        kv = dict(tok.split("=", 1) for tok in shlex.split(line))
        parsed[kv["R_NAME"]] = kv

    assert parsed["s1-purged"]["R_STATE"] == "purged"
    assert parsed["s2-same-inst-newly-recorded"]["R_STATE"] == "pending"
    assert parsed["s2-same-inst-newly-recorded"]["R_ROW_ID"] == UA
    assert parsed["s3-resumed-same-id"]["R_STATE"] == "pending"
    assert parsed["s4-created-no-id"]["R_STATE"] == "created"
    assert parsed["s5-done"]["R_STATE"] == "done"

    res_tbl = _run_reset_py(
        [
            "table",
            "--workspace",
            str(ws),
            "--skipped",
            "s2-same-inst-newly-recorded:purge_refused",
        ],
        state=state,
    )
    assert res_tbl.returncode == 4
    rows_by_name = {}
    for line in res_tbl.stdout.strip().splitlines()[1:]:
        cols = line.split(None, 6)
        rows_by_name[cols[0]] = cols

    assert rows_by_name["s2-same-inst-newly-recorded"][1:6] == ["agy", "-", "-", "no", "-"]
    assert rows_by_name["s2-same-inst-newly-recorded"][6] == "skipped: purge refused"
    assert rows_by_name["s3-resumed-same-id"][1:6] == ["claude", UB, UB, "yes", "no"]
    assert rows_by_name["s4-created-no-id"][1:6] == ["hermes", UC, "-", "no", "-"]
    assert rows_by_name["s4-created-no-id"][6] == "created; id not recorded yet"
    assert rows_by_name["s5-done"][1:6] == ["codex", "-", UD, "yes", "yes"]
    assert rows_by_name["s5-done"][6] == "done"

    # Review F1: (a) old_id=None + same instance + id recorded on row -> record-id persists old_id=UA
    from lib_py.reset import read_spec

    res_rec = _run_reset_py(
        ["record-id", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"],
        state=state,
    )
    assert res_rec.returncode == 0, res_rec.stderr
    spec_after = read_spec(str(ws))
    s2_entry = next(e for e in spec_after["agents"] if e["name"] == "s2-same-inst-newly-recorded")
    assert s2_entry["old_id"] == UA

    # Review F1: (b) after resuming that skipped row (new created_at, id still UA), it is classified
    # as 'pending' (NOT falsely 'done') and table exits non-zero (4) with NEW!=OLD == 'no'
    state["herdr_sessions"][0] = _row(
        "s2-same-inst-newly-recorded",
        "agy",
        ws,
        status="running",
        created_at="2026-10-02T00:00:00Z",
        own_id=UA,
    )
    res_f1_entries = _run_reset_py(
        ["entries", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"],
        state=state,
    )
    assert res_f1_entries.returncode == 0
    kv_f1 = dict(tok.split("=", 1) for tok in shlex.split(res_f1_entries.stdout.strip()))
    assert kv_f1["R_STATE"] == "pending"
    assert kv_f1["R_OLD_ID"] == UA

    res_f1_tbl = _run_reset_py(["table", "--workspace", str(ws)], state=state)
    assert res_f1_tbl.returncode == 4
    rows_f1 = {
        line.split(None, 6)[0]: line.split(None, 6)
        for line in res_f1_tbl.stdout.strip().splitlines()[1:]
    }
    assert rows_f1["s2-same-inst-newly-recorded"][1:7] == ["agy", UA, UA, "yes", "no", "pending"]


# --------------------------------------------------------------------------- Integration tests: reset.sh

def test_no_yes_non_tty_exits_3_without_touching_locks(mam_sandbox: Path, mock_herdr: Path, mock_agents):
    """§5.2-3: Without --yes and with non-TTY stdin, reset.sh exits 3 before touching
    any lock (.mam/loop-guard-active or .mam/update-admission.lock) or registry.
    """
    _write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)])
    yaml_before = (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes()

    res = _run_reset(mam_sandbox)
    assert res.returncode == 3
    assert "not a terminal" in res.stderr
    assert (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() == yaml_before
    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
    assert not (mam_sandbox / ".mam" / "update-admission.lock").exists()
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()


def test_dry_run_changes_nothing_in_mam_tree(mam_sandbox: Path, mock_herdr: Path, mock_agents):
    """§5.2-1 & §5.2-4: --dry-run prints targets (with all spec fields), skipped rows,
    and orchestrator_uuids, exits 0, and leaves the entire .mam tree (names, sizes,
    mtimes) and mock_herdr state untouched.
    """
    wt = mam_sandbox / "wt" / "issue-1"
    wt.mkdir(parents=True)
    _write_rows(
        mam_sandbox,
        [
            _row("reviewer-claude-01", "claude", mam_sandbox, role="planner,reviewer", server="srv-main", label="lbl-main", own_id=UA),
            _row("reviewer-grok-01", "grok", mam_sandbox, role="reviewer", server="default", label="lbl-main", own_id=UB),
            _row("creator-claude-wt", "claude", wt, role="creator", own_id=UC),
            _row("stopped-codex-01", "codex", mam_sandbox, status="stopped", own_id=UD),
        ],
        orchestrator_uuids=[UORC],
    )
    mam_dir = mam_sandbox / ".mam"

    def _snapshot_mam():
        return {
            str(p.relative_to(mam_dir)): (p.stat().st_size, p.stat().st_mtime_ns)
            for p in sorted(mam_dir.rglob("*"))
        }

    snap_before = _snapshot_mam()
    yaml_before = (mam_dir / "agent-sessions.yaml").read_bytes()

    res = _run_reset(mam_sandbox, "--dry-run")
    assert res.returncode == 0, res.stdout + res.stderr
    assert _snapshot_mam() == snap_before
    assert (mam_dir / "agent-sessions.yaml").read_bytes() == yaml_before

    out = res.stdout
    assert "target: reviewer-claude-01" in out
    assert "agent=claude" in out
    assert "role=planner,reviewer" in out
    assert "herdr_server=srv-main" in out
    assert "herdr_workspace=lbl-main" in out
    assert f"old_id={UA}" in out
    assert "old_created_at=2026-10-01T00:00:00Z" in out
    assert "target: reviewer-grok-01" in out
    assert "skip: creator-claude-wt" in out and "not the control root" in out
    assert "skip: stopped-codex-01 (status stopped)" in out
    assert f"orchestrator {UORC} (not a row; never reset)" in out

    herdr_state = json.loads(mock_herdr.read_text(encoding="utf-8"))
    assert herdr_state.get("calls") == []


def test_precheck_refusals_change_nothing(mam_sandbox: Path, mock_herdr: Path, mock_agents):
    """§5.2-2: Precheck refusals (open job, live loop marker, live update marker,
    unrecreatable target, caller self-target) exit 1 and change nothing; stale
    markers do not refuse.
    """
    tf = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    _mock_herdr_agent(mock_herdr, "reviewer-claude-01", mam_sandbox, agent="claude")
    _write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)])
    spy_log = _install_herdr_spy(mam_sandbox)

    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    yaml_before = yaml_path.read_bytes()
    db_before = _db_session_names(mam_sandbox)
    tf_before = tf.read_bytes()

    def _assert_nothing_changed(expect_loop_marker: bool = False):
        assert yaml_path.read_bytes() == yaml_before
        assert _db_session_names(mam_sandbox) == db_before
        assert tf.read_bytes() == tf_before
        assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
        calls = _read_spy_log(spy_log)
        assert not any(c["verb"] in ("kill-session", "pane close", "agent start") for c in calls)
        if not expect_loop_marker:
            assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()

    # (a) Open job
    jobs_dir = mam_sandbox / ".mam" / "jobs"
    jobs_dir.mkdir(parents=True, exist_ok=True)
    job_file = jobs_dir / "job78a.json"
    job_file.write_text(
        json.dumps({"job_id": "job78a", "agent_session": "herdr:reviewer-claude-01", "status": "running"}),
        encoding="utf-8",
    )
    res_job = _run_reset(mam_sandbox, "--yes")
    assert res_job.returncode == 1
    assert "job78a" in res_job.stderr
    assert "close --job job78a --request cancelled" in res_job.stderr
    _assert_nothing_changed()
    job_file.unlink()

    # Helper to get lstart for a live process
    holder = subprocess.Popen(["sleep", "300"])
    try:
        lstart = subprocess.run(
            ["ps", "-p", str(holder.pid), "-o", "lstart="],
            capture_output=True,
            text=True,
            check=True,
        ).stdout.strip()
        lstart = " ".join(lstart.split())

        # (b) Live loop marker
        loop_marker = mam_sandbox / ".mam" / "loop-guard-active"
        loop_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8")
        res_loop = _run_reset(mam_sandbox, "--yes")
        assert res_loop.returncode == 1
        assert "loop" in res_loop.stderr
        assert str(holder.pid) in res_loop.stderr
        _assert_nothing_changed(expect_loop_marker=True)
        loop_marker.unlink()

        # (c) Live update marker
        upd_marker = mam_sandbox / ".mam" / "update-in-progress"
        upd_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8")
        res_upd = _run_reset(mam_sandbox, "--yes")
        assert res_upd.returncode == 1
        assert "update" in res_upd.stderr
        _assert_nothing_changed()
        upd_marker.unlink()
    finally:
        holder.kill()
        holder.wait()

    # Stale control: dead PID in loop and update markers does not refuse --dry-run
    (mam_sandbox / ".mam" / "loop-guard-active").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8")
    (mam_sandbox / ".mam" / "update-in-progress").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8")
    res_stale = _run_reset(mam_sandbox, "--dry-run")
    assert res_stale.returncode == 0, res_stale.stdout + res_stale.stderr
    (mam_sandbox / ".mam" / "loop-guard-active").unlink()
    (mam_sandbox / ".mam" / "update-in-progress").unlink()

    # (d) Caller self-target (D4-4)
    res_self = _run_reset(mam_sandbox, "--yes", env_extra={"CLAUDE_CODE_SESSION_ID": UA})
    assert res_self.returncode == 1
    assert "you are reviewer-claude-01, one of the targets" in res_self.stderr
    _assert_nothing_changed()

    # (e) Saved spec with valid pending target first and unknown agent second (review P2)
    spec_file = mam_sandbox / ".mam" / "reset-spec.json"
    bad_spec_text = json.dumps(
        {
            "version": 1,
            "workspace": str(mam_sandbox),
            "written_at": "2026-10-29T00:00:00Z",
            "agents": [
                {
                    "name": "reviewer-claude-01",
                    "agent": "claude",
                    "role": "reviewer",
                    "herdr_server": "default",
                    "herdr_workspace": "ws",
                    "old_id": UA,
                    "old_created_at": "2026-10-01T00:00:00Z",
                },
                {
                    "name": "reviewer-bad-01",
                    "agent": "not-an-agent",
                    "role": "reviewer",
                    "herdr_server": "default",
                    "herdr_workspace": "ws",
                    "old_id": UB,
                    "old_created_at": "2026-10-01T00:00:00Z",
                },
            ],
        }
    )
    spec_file.write_text(bad_spec_text, encoding="utf-8")
    res_bad_spec = _run_reset(mam_sandbox, "--yes")
    assert res_bad_spec.returncode == 1
    assert "unknown agent" in res_bad_spec.stderr
    assert "check it, then remove it by hand" in res_bad_spec.stderr
    assert spec_file.read_text(encoding="utf-8") == bad_spec_text
    spec_file.unlink()
    _assert_nothing_changed()


def test_full_reset_success_spec_ordering_table_and_loop_block(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """§5.2-5, §5.2-6, §5.2-10:
    - .mam/reset-spec.json is written atomically with both targets before the first kill call
    - Both claude sessions are stopped (normal then purge, no UUID on CLI) and recreated sequentially
    - Old transcripts are removed, new IDs recorded != old IDs, result table shows RECORDED=yes, NEW!=OLD=yes, done
    - Spec file and loop marker are cleaned up on exit 0
    - During the reset, run_loop.sh is blocked by .mam/loop-guard-active ("Another multi-agent-mux-loop is already running")
    """
    name_a = "reviewer-claude-01"
    name_b = "reviewer-claude-02"
    tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [
            _row(name_a, "claude", mam_sandbox, role="planner,reviewer", server="default", label="lbl-a", own_id=UA),
            _row(name_b, "claude", mam_sandbox, role="reviewer", server="default", label="lbl-b", own_id=UB),
        ],
    )
    spy_log = _install_herdr_spy(mam_sandbox)

    res = _run_reset(mam_sandbox, "--yes", env_extra={"SPY_PROBE_RUN_LOOP": "1"})
    _wait_reconcile_idle(mam_sandbox)
    assert res.returncode == 0, res.stdout + res.stderr

    # Spec file & loop marker removed after completion
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()

    # Old transcripts purged
    assert not tf_a.exists()
    assert not tf_b.exists()

    # Recreated rows have same name, role, herdr_server, herdr_workspace, and new IDs != old IDs
    rows = _read_rows(mam_sandbox)
    assert set(rows.keys()) == {name_a, name_b}
    assert rows[name_a]["role"] == "planner,reviewer"
    assert rows[name_a]["herdr_server"] == "default"
    assert rows[name_a]["herdr_workspace"] == "lbl-a"
    new_a = rows[name_a]["claude_session_id_own"]
    assert new_a and new_a != UA

    assert rows[name_b]["role"] == "reviewer"
    assert rows[name_b]["herdr_server"] == "default"
    assert rows[name_b]["herdr_workspace"] == "lbl-b"
    new_b = rows[name_b]["claude_session_id_own"]
    assert new_b and new_b != UB

    # Result table in stdout
    out_lines = [l for l in res.stdout.splitlines() if l.startswith(name_a) or l.startswith(name_b)]
    assert len(out_lines) == 2
    assert out_lines[0].split() == [name_a, "claude", UA, new_a, "yes", "yes", "done"]
    assert out_lines[1].split() == [name_b, "claude", UB, new_b, "yes", "yes", "done"]

    # Stop invoked twice per agent (normal, then --purge-conversation --yes), no UUID on CLI
    stop_traces = [l for l in res.stdout.splitlines() if l.startswith("reset [") and "stop_session.sh" in l]
    assert len(stop_traces) == 4
    for st in stop_traces:
        assert UA not in st and UB not in st

    # §5.2-5 & §5.2-6: Spy log verifies spec file existed at first kill and sequential order
    calls = _read_spy_log(spy_log)
    first_kill = next(c for c in calls if c["verb"] in ("pane close", "kill-session"))
    assert first_kill["spec_exists"] is True
    assert first_kill["spec"]["version"] == 1
    assert [a["name"] for a in first_kill["spec"]["agents"]] == [name_a, name_b]

    # Sequential order: kill A -> start A -> kill B -> start B
    events = []
    for c in calls:
        if c["verb"] == "kill-session":
            tgt = c["args"][2] if len(c["args"]) > 2 and c["args"][1] == "-t" else c["args"][1]
            if not events or events[-1] != ("kill", tgt):
                events.append(("kill", tgt))
        elif c["verb"] == "agent start":
            events.append(("start", c["args"][2]))
    assert events == [
        ("kill", name_a),
        ("start", name_a),
        ("kill", name_b),
        ("start", name_b),
    ]

    # §5.2-10: During the reset, run_loop.sh was refused by loop-guard-active
    probe_data = json.loads((mam_sandbox / "herdr_spy_loop_probe.json").read_text(encoding="utf-8"))
    assert probe_data["rc"] == 1
    combined_probe = probe_data["stdout"] + probe_data["stderr"]
    assert "Another multi-agent-mux-loop is already running" in combined_probe

    # Control after reset finishes: run_loop.sh passes the loop lock and fails at "is not registered"
    loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
    post_loop = subprocess.run(
        ["bash", str(loop_sh), "--creator", "x", "--task", "t"],
        cwd=str(mam_sandbox),
        env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"),
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=30,
    )
    assert "is not registered" in (post_loop.stdout + post_loop.stderr)


def test_mid_run_failure_and_rerun_resumes_from_spec(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """§5.2-7: A (claude) succeeds, B (grok) fails at create (grok preflight fails).
    Run 1 exits 1 with 'failed at create (rc 1)' and leaves .mam/reset-spec.json.
    Restoring grok binary and rerunning skips A (0 stop calls for A) and only creates B.
    """
    name_a = "reviewer-claude-01"
    name_b = "reviewer-grok-01"
    _claude_transcript(mam_sandbox, mam_sandbox, UA)
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="grok", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [
            _row(name_a, "claude", mam_sandbox, own_id=UA),
            _row(name_b, "grok", mam_sandbox, own_id=UB),
        ],
    )
    spy_log = _install_herdr_spy(mam_sandbox)

    # Make bin/grok exit 1 (and XAI_API_KEY="") so B's create_session.sh preflight fails with exit 1
    grok_bin = mam_sandbox / "bin" / "grok"
    grok_bin.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
    grok_bin.chmod(0o755)

    res1 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""})
    _wait_reconcile_idle(mam_sandbox)
    assert res1.returncode == 1
    assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
    assert "failed at create (rc 1)" in res1.stdout

    rows1 = _read_rows(mam_sandbox)
    assert name_a in rows1 and rows1[name_a]["claude_session_id_own"] != UA
    assert name_b not in rows1  # B was purged before its create failed

    # Restore grok binary, clear spy log, and rerun
    grok_bin.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
    grok_bin.chmod(0o755)
    spy_log.unlink()

    res2 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""})
    _wait_reconcile_idle(mam_sandbox)
    assert res2.returncode == 0, res2.stdout + res2.stderr
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()

    # Spy confirms A had 0 stop/kill/start calls on rerun, and B was only created
    calls2 = _read_spy_log(spy_log)
    assert not any(c["verb"] in ("pane close", "kill-session") for c in calls2)
    starts2 = [c["args"][2] for c in calls2 if c["verb"] == "agent start"]
    assert starts2 == [name_b]
    assert f"reset [{name_a}]:" not in res2.stdout


def test_purge_refused_exit_4_preserves_stopped_row_and_rerun_completes(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """§5.2-8: A's transcript is held open by a real sleep 300 process (via fake lsof).
    Run 1: A is stopped in step 1, refused at purge in step 3 (exit 4 -> skipped: purge refused),
    prints holder PID and 'sleep 300', keeps A's row (status=stopped, own_id=UA) and transcript,
    leaves sleep alive, completes B, and exits 4 with spec file intact.
    Run 2 (after holder ends): A is purged and recreated, B is untouched, exits 0.
    """
    name_a = "reviewer-claude-01"
    name_b = "reviewer-claude-02"
    tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [
            _row(name_a, "claude", mam_sandbox, own_id=UA),
            _row(name_b, "claude", mam_sandbox, own_id=UB),
        ],
    )
    spy_log = _install_herdr_spy(mam_sandbox)

    holder = subprocess.Popen(["sleep", "300"])
    try:
        lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid})
        res1 = _run_reset(mam_sandbox, "--yes")
        _wait_reconcile_idle(mam_sandbox)
        assert res1.returncode == 4, res1.stdout + res1.stderr
        combined1 = res1.stdout + res1.stderr
        assert f"pid {holder.pid}" in combined1
        assert "sleep 300" in combined1
        assert "skipped: purge refused" in res1.stdout
        assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()

        # Holder process is still alive (reset never kills external processes)
        assert holder.poll() is None

        # A's row is stopped with UA preserved and transcript intact; B is done
        rows1 = _read_rows(mam_sandbox)
        assert rows1[name_a]["status"] == "stopped"
        assert rows1[name_a]["claude_session_id_own"] == UA
        assert tf_a.is_file()
        assert not tf_b.exists()
        assert rows1[name_b]["claude_session_id_own"] != UB

        # Release holder and rerun
        lsof_exe.unlink()
    finally:
        holder.kill()
        holder.wait()

    spy_log.unlink()
    res2 = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res2.returncode == 0, res2.stdout + res2.stderr
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
    assert not tf_a.exists()

    rows2 = _read_rows(mam_sandbox)
    assert rows2[name_a]["claude_session_id_own"] != UA
    assert f"reset [{name_b}]:" not in res2.stdout


def test_post_discovered_without_recorded_id_8b_and_exit_5(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """§5.2-8b (i) & (ii) and §5.2-9:
    - 8b(i): agy row has no recorded own_id while herdr is live and cache points to UA.
      Run 1 with UA held by fake lsof: step 1 captures UA onto the row (status=stopped),
      step 3 refuses purge (exit 4).
      Run 2 after releasing lsof: step 3 purges UA (all UA artifacts deleted!) and recreates
      the agy session. Because mock agy does not record a new ID, exits 5 (created).
      Run 3 (while still without ID): 0 stop calls (does not re-purge the new session!), exits 5.
      Run 4 (after recording new ID UB): exits 0, removes spec file, table shows OLD_ID='-' and NEW!=OLD='yes'.
    - 8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips with
      'skipped: no conversation id recorded' (exit 4), keeping the stopped row without calling purge.
    """
    name_a = "reviewer-agy-01"
    now = time.time()
    epoch = int(now) - 100
    a_files = _agy_full_artifacts(mam_sandbox, UA, mtime=now)
    _agy_cache(mam_sandbox, {mam_sandbox: UA})

    holder = subprocess.Popen(["sleep", "300"])
    try:
        lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid})
        _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch, agent="agy", pane_id="w1:p_a")
        _write_rows(
            mam_sandbox,
            [_row(name_a, "agy", mam_sandbox, own_id=None, pid=holder.pid, epoch=epoch)],
        )
        spy_log = _install_herdr_spy(mam_sandbox)

        # Run 1: step 1 captures UA onto the row and record-id persists UA into reset-spec.json;
        # step 3 refuses purge because lsof holds UA
        res1 = _run_reset(mam_sandbox, "--yes")
        _wait_reconcile_idle(mam_sandbox)
        assert res1.returncode == 4, res1.stdout + res1.stderr
        rows1 = _read_rows(mam_sandbox)
        assert rows1[name_a]["status"] == "stopped"
        assert rows1[name_a]["agy_conversation_id_own"] == UA
        assert a_files[0].exists()
        spec1 = json.loads((mam_sandbox / ".mam" / "reset-spec.json").read_text(encoding="utf-8"))
        assert spec1["agents"][0]["old_id"] == UA

        # Simulate resuming the skipped session (new created_at, own_id still UA) before Run 2 (review F1):
        # because record-id persisted old_id=UA, Run 2 still treats it as 'pending' and purges UA
        rows1[name_a]["status"] = "running"
        rows1[name_a]["herdr_session_created_at"] = "2026-10-02T00:00:00Z"
        _write_rows(mam_sandbox, [rows1[name_a]])
        _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch + 10, agent="agy", pane_id="w1:p_a")

        # Release lsof and Run 2: step 3 purges UA (deletes all UA artifacts!) and creates new agy
        lsof_exe.unlink()
    finally:
        holder.kill()
        holder.wait()

    res2 = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res2.returncode == 5, res2.stdout + res2.stderr
    assert "created; id not recorded yet" in res2.stdout
    for p in a_files:
        assert not p.exists(), f"UA artifact should have been purged on rerun: {p}"
    assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()

    # Run 3 (§5.2-9): rerunning while in `created` state performs 0 stop/create calls and exits 5
    spy_log.unlink()
    res3 = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res3.returncode == 5
    assert _read_spy_log(spy_log) == []
    assert f"reset [{name_a}]:" not in res3.stdout

    # Record new ID UB onto the recreated row and Run 4 -> exits 0, OLD_ID=UA, NEW!=OLD='yes'
    rows3 = _read_rows(mam_sandbox)
    rows3[name_a]["agy_conversation_id_own"] = UB
    _write_rows(mam_sandbox, [rows3[name_a]])

    res4 = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res4.returncode == 0, res4.stdout + res4.stderr
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
    tbl_line = [l for l in res4.stdout.splitlines() if l.startswith(name_a)][0]
    assert tbl_line.split() == [name_a, "agy", UA, UB, "yes", "yes", "done"]


def test_post_discovered_uncapturable_id_8b_ii(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """§5.2-8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips
    with 'skipped: no conversation id recorded' (exit 4), keeping the stopped row without
    calling purge; manual purge followed by rerun recreates via D6 #1 (purged).
    """
    name_b = "reviewer-agy-02"
    epoch = int(time.time()) - 100
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, epoch=epoch, agent="agy", pane_id="w1:p_b")
    _write_rows(mam_sandbox, [_row(name_b, "agy", mam_sandbox, own_id=None, epoch=epoch)])
    spy_log = _install_herdr_spy(mam_sandbox)

    res_ii = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res_ii.returncode == 4, res_ii.stdout + res_ii.stderr
    assert "skipped: no conversation id recorded" in res_ii.stdout
    assert not any(
        l.startswith("reset [") and "--purge-conversation" in l for l in res_ii.stdout.splitlines()
    )
    rows_ii = _read_rows(mam_sandbox)
    assert name_b in rows_ii and rows_ii[name_b]["status"] == "stopped"

    # Manual purge drops the row; rerunning reset then sees D6 #1 (purged) and runs create only
    stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
    res_manual = subprocess.run(
        ["bash", str(stop_sh), "--session", name_b, "--agent", "agy", "--purge-conversation", "--yes"],
        cwd=str(mam_sandbox),
        env=dict(os.environ, SKS_QUIESCENT_TRIES="0", SKS_DIALOG_TIMEOUT="0", SKS_BUSY_TIMEOUT="0"),
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=60,
    )
    assert res_manual.returncode == 0, res_manual.stdout + res_manual.stderr
    assert name_b not in _read_rows(mam_sandbox)

    spy_log.unlink()
    res_ii_rerun = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res_ii_rerun.returncode == 5, res_ii_rerun.stdout + res_ii_rerun.stderr
    calls_ii = _read_spy_log(spy_log)
    assert not any(c["verb"] in ("pane close", "kill-session") for c in calls_ii)
    assert [c["args"][2] for c in calls_ii if c["verb"] == "agent start"] == [name_b]
    assert name_b in _read_rows(mam_sandbox)

    # Recording new ID UB when old_id remained None produces OLD_ID='-' and NEW!=OLD='yes'
    rows_ii_after = _read_rows(mam_sandbox)
    rows_ii_after[name_b]["agy_conversation_id_own"] = UB
    _write_rows(mam_sandbox, [rows_ii_after[name_b]])
    res_ii_done = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res_ii_done.returncode == 0, res_ii_done.stdout + res_ii_done.stderr
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
    tbl_ii = [l for l in res_ii_done.stdout.splitlines() if l.startswith(name_b)][0]
    assert tbl_ii.split() == [name_b, "agy", "-", UB, "yes", "yes", "done"]


@pytest.mark.parametrize(
    "sig_name,expected_rc",
    [("SIGTERM", 143), ("SIGINT", 130), ("SIGHUP", 129)],
)
def test_signal_interrupts_reset_preserves_spec_and_holds_lock_until_child_exits(
    mam_sandbox: Path, mock_herdr: Path, mock_agents, sig_name: str, expected_rc: int
):
    """Review P1 / N1:
    When SIGTERM, SIGINT, or SIGHUP arrives while an in-flight stop/create child is running:
    - reset.sh keeps holding .mam/loop-guard-active (so a competing run_loop.sh is still refused)
      until the in-flight child finishes;
    - once the child finishes, reset.sh immediately exits with the signal's nonzero code (143/130/129),
      releasing .mam/loop-guard-active on exit, preserving .mam/reset-spec.json, and NOT continuing
      to purge/create the current or subsequent targets.
    """
    import signal

    sig = getattr(signal, sig_name)
    name_a = "reviewer-claude-01"
    name_b = "reviewer-claude-02"
    tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [
            _row(name_a, "claude", mam_sandbox, own_id=UA),
            _row(name_b, "claude", mam_sandbox, own_id=UB),
        ],
    )

    # Wrap bin/herdr so the first kill-session blocks until release_gate is created
    bin_dir = mam_sandbox / "bin"
    herdr_bin = bin_dir / "herdr"
    herdr_real = bin_dir / "herdr.real"
    if not herdr_real.exists():
        herdr_bin.rename(herdr_real)
    entered_gate = mam_sandbox / "child_entered"
    release_gate = mam_sandbox / "child_release"
    spy_log = mam_sandbox / "herdr_spy.jsonl"

    herdr_bin.write_text(
        f"""#!{sys.executable}
import json, os, sys, time

real = {repr(str(herdr_real))}
entered = {repr(str(entered_gate))}
release = {repr(str(release_gate))}
spy_log = {repr(str(spy_log))}

raw_args = sys.argv[1:]
args = list(raw_args)
while args and args[0] in ("-L", "--server", "-s", "--session"):
    args = args[2:] if len(args) > 1 else args[1:]
verb = args[0] if (args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls")) else (" ".join(args[:2]) if len(args) >= 2 else "")
with open(spy_log, "a", encoding="utf-8") as lf:
    lf.write(json.dumps({{"verb": verb, "args": args}}) + "\\n")

if verb == "kill-session" and not os.path.exists(entered):
    with open(entered, "w", encoding="utf-8") as fh:
        fh.write("1")
    deadline = time.time() + 60.0
    while not os.path.exists(release) and time.time() < deadline:
        time.sleep(0.05)

os.execv(real, [real, *raw_args])
""",
        encoding="utf-8",
    )
    herdr_bin.chmod(0o755)

    script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
    env = dict(
        os.environ,
        SKS_QUIESCENT_TRIES="0",
        SKS_DIALOG_TIMEOUT="0",
        SKS_BUSY_TIMEOUT="0",
        MAM_ONBOARD_ACK_S="3",
        MAM_ONBOARD_WAIT_S="3",
        MAM_NO_SUBSCRIBER_DAEMON="1",
        PYTHONDONTWRITEBYTECODE="1",
    )
    proc = subprocess.Popen(
        ["bash", str(script), "--yes"],
        cwd=str(mam_sandbox),
        env=env,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    try:
        deadline = time.time() + 60.0
        while not entered_gate.exists() and time.time() < deadline:
            assert proc.poll() is None, "reset.sh exited before reaching kill-session"
            time.sleep(0.05)
        assert entered_gate.exists(), "timed out waiting for kill-session child"

        # Signal the reset parent while the foreground child is still blocked
        proc.send_signal(sig)
        time.sleep(0.2)

        # While the child is still running, reset.sh has not exited and still holds the loop lock
        assert proc.poll() is None
        assert (mam_sandbox / ".mam" / "loop-guard-active").is_file()
        if sig_name == "SIGTERM":
            loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh"
            competing = subprocess.run(
                ["bash", str(loop_sh), "--creator", "x", "--task", "t"],
                cwd=str(mam_sandbox),
                env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"),
                stdin=subprocess.DEVNULL,
                capture_output=True,
                text=True,
                timeout=20,
            )
            assert competing.returncode == 1
            assert "Another multi-agent-mux-loop is already running" in (competing.stdout + competing.stderr)

        # Release the child and wait for reset.sh to terminate
        release_gate.write_text("1", encoding="utf-8")
        stdout, stderr = proc.communicate(timeout=30)
    finally:
        if proc.poll() is None:
            proc.kill()
            proc.wait()
        _wait_reconcile_idle(mam_sandbox)

    assert proc.returncode == expected_rc, f"rc={proc.returncode}, stdout={stdout}, stderr={stderr}"
    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
    assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
    # Neither transcript was purged, and no create or second-target operation ran
    assert tf_a.is_file()
    assert tf_b.is_file()
    calls = _read_spy_log(spy_log)
    assert not any(c["verb"] == "agent start" for c in calls)
    assert f"reset [{name_b}]:" not in stdout
    assert not any(
        l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines()
    )


# --------------------------------------------------------------------------- Review round 1 (PR #82)

def _write_spec_file(sandbox: Path, agents: List[dict]) -> Path:
    spec = sandbox / ".mam" / "reset-spec.json"
    spec.write_text(
        json.dumps({"version": 1, "workspace": os.path.realpath(str(sandbox)), "written_at": "x", "agents": agents}),
        encoding="utf-8",
    )
    return spec


def _spec_entry(name: str, agent: str, old_id: Optional[str], server: str = "default") -> dict:
    return {
        "name": name,
        "agent": agent,
        "role": "reviewer",
        "herdr_server": server,
        "herdr_workspace": "ws",
        "old_id": old_id,
        "old_created_at": "2026-10-01T00:00:00Z",
    }


def _result_of(stdout: str, name: str) -> str:
    line = [l for l in stdout.splitlines() if l.startswith(name + " ")][-1]
    return line.split(None, 6)[6]


def _live_marker(path: Path, holder: subprocess.Popen) -> None:
    ls = " ".join(
        subprocess.run(["ps", "-p", str(holder.pid), "-o", "lstart="], capture_output=True, text=True).stdout.split()
    )
    path.write_text("pid=%d\nlstart=%s\nstarted=x\n" % (holder.pid, ls), encoding="utf-8")


def test_record_id_never_records_a_pre_assigned_guess(tmp_path: Path):
    """Review B2 (claude): a claude/grok entry with old_id=None never takes the id
    a normal stop guessed onto its row; a post-discovered one still does."""
    from lib_py.reset import read_spec, write_spec

    ws = tmp_path / "ws"
    ws.mkdir()
    write_spec(str(ws), [_spec_entry("c-noid", "claude", None), _spec_entry("a-noid", "agy", None)])
    state = {
        "herdr_sessions": [
            _row("c-noid", "claude", ws, status="stopped", own_id=UA),
            _row("a-noid", "agy", ws, status="stopped", own_id=UB),
        ]
    }
    for name in ("c-noid", "a-noid"):
        res = _run_reset_py(["record-id", "--workspace", str(ws), "--session", name], state=state)
        assert res.returncode == 0, res.stderr
    old = {e["name"]: e["old_id"] for e in read_spec(str(ws))["agents"]}
    assert old == {"c-noid": None, "a-noid": UB}


def test_claude_without_recorded_id_is_refused_and_unregistered_conversation_kept(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """Review B2 (claude) (c): a running claude row with no id and an unregistered
    claude conversation in the same folder: the reset refuses before any change,
    so the conversation that is not provably this pane's is never purged."""
    name = "reviewer-claude-01"
    tf = _claude_transcript(mam_sandbox, mam_sandbox, UC)
    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
    _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=None)])
    spy_log = _install_herdr_spy(mam_sandbox)
    yaml_before = (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes()

    res = _run_reset(mam_sandbox, "--yes")
    assert res.returncode == 1, res.stdout + res.stderr
    assert "running session %s (claude) has no recorded conversation id" % name in res.stderr
    assert "nothing was changed" in res.stderr
    assert tf.is_file()
    assert (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() == yaml_before
    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
    assert not any(c["verb"] in ("kill-session", "pane close", "agent start") for c in _read_spy_log(spy_log))

    res_dry = _run_reset(mam_sandbox, "--dry-run")
    assert res_dry.returncode == 1
    assert "has no recorded conversation id" in res_dry.stdout


def test_rerun_skips_a_target_whose_row_id_is_not_the_recorded_one(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """Review B2 (claude) (b): after the normal stop the row's id must equal the
    spec's old_id. A: the row's id changed after the spec was written (UC, spec
    UA). B: an older spec holds a claude entry with old_id null, and the stop
    guesses an unregistered conversation onto its row. Neither is purged."""
    name_a = "reviewer-claude-01"
    name_b = "reviewer-claude-02"
    files = [_claude_transcript(mam_sandbox, mam_sandbox, u) for u in (UA, UC, UD)]
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [_row(name_a, "claude", mam_sandbox, own_id=UC), _row(name_b, "claude", mam_sandbox, own_id=None)],
    )
    _write_spec_file(mam_sandbox, [_spec_entry(name_a, "claude", UA), _spec_entry(name_b, "claude", None)])

    res = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res.returncode == 4, res.stdout + res.stderr
    assert _result_of(res.stdout, name_a) == "skipped: id not verified"
    assert _result_of(res.stdout, name_b) == "skipped: id not verified"
    assert not any(l.startswith("reset [") and "--purge-conversation" in l for l in res.stdout.splitlines())
    for f in files:
        assert f.is_file(), f
    rows = _read_rows(mam_sandbox)
    assert rows[name_a]["claude_session_id_own"] == UC
    assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()


def test_spec_server_is_validated_and_a_changed_row_is_not_stopped(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """Review B2 (codex): a spec entry that lost herdr_server is refused before any
    stop (never `--herdr-session default`); a valid spec whose server no longer
    matches the row skips that target untouched."""
    name = "reviewer-claude-01"
    tf = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
    _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
    spy_log = _install_herdr_spy(mam_sandbox)

    entry = _spec_entry(name, "claude", UA)
    entry.pop("herdr_server")
    spec = _write_spec_file(mam_sandbox, [entry])
    spec_text = spec.read_text(encoding="utf-8")
    res = _run_reset(mam_sandbox, "--yes")
    assert res.returncode == 1, res.stdout + res.stderr
    assert "incomplete agent entry" in res.stderr
    assert "--herdr-session" not in res.stdout
    assert spec.read_text(encoding="utf-8") == spec_text
    assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log))

    _write_spec_file(mam_sandbox, [_spec_entry(name, "claude", UA, server="isolated-server")])
    res_changed = _run_reset(mam_sandbox, "--yes")
    assert res_changed.returncode == 4, res_changed.stdout + res_changed.stderr
    assert _result_of(res_changed.stdout, name) == "skipped: row changed"
    assert "reset [%s]:" % name not in res_changed.stdout
    assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log))
    assert _read_rows(mam_sandbox)[name]["status"] == "running"
    assert tf.is_file()


def test_job_after_precheck_skips_its_target_and_submit_is_refused_during_reset(
    mam_sandbox: Path, mock_herdr: Path, mock_agents
):
    """Review B1 (codex), "reset first": while the reset runs (A's stop), a
    delegate submit to B is refused by .mam/reset-in-progress and registers
    nothing; a record for B written by other means is seen by B's re-check, so B
    is skipped with its job left open and its conversation kept."""
    name_a = "reviewer-claude-01"
    name_b = "reviewer-claude-02"
    tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB)
    _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a")
    _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b")
    _write_rows(
        mam_sandbox,
        [_row(name_a, "claude", mam_sandbox, own_id=UA), _row(name_b, "claude", mam_sandbox, own_id=UB)],
    )

    bin_dir = mam_sandbox / "bin"
    herdr_bin = bin_dir / "herdr"
    herdr_real = bin_dir / "herdr.real"
    herdr_bin.rename(herdr_real)
    probe = mam_sandbox / "submit_probe.json"
    jobs_dir = mam_sandbox / ".mam" / "jobs"
    job_file = jobs_dir / "job78late.json"
    submit = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
    herdr_bin.write_text(
        f"""#!{sys.executable}
import json, os, subprocess, sys

real = {repr(str(herdr_real))}
probe = {repr(str(probe))}
args = list(sys.argv[1:])
while args and args[0] in ("-L", "--server", "-s", "--session"):
    args = args[2:]
if args[:2] == ["pane", "close"] or args[:1] == ["kill-session"]:
    if not os.path.exists(probe):
        marker = {repr(str(mam_sandbox / ".mam" / "reset-in-progress"))}
        res = subprocess.run(
            ["bash", {repr(str(submit))}, "submit", "--agent", "claude-code",
             "--agent-session", "herdr:{name_b}", "--prompt", "hi"],
            cwd={repr(str(mam_sandbox))}, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=120,
        )
        jobs = sorted(n for n in os.listdir({repr(str(jobs_dir))}) if n.endswith(".json")) if os.path.isdir({repr(str(jobs_dir))}) else []
        with open(probe, "w", encoding="utf-8") as fh:
            json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr,
                        "marker": os.path.isfile(marker), "jobs": jobs}}, fh)
        os.makedirs({repr(str(jobs_dir))}, exist_ok=True)
        with open({repr(str(job_file))}, "w", encoding="utf-8") as fh:
            json.dump({{"job_id": "job78late", "agent_session": "herdr:{name_b}", "status": "pending"}}, fh)
os.execv(real, [real, *sys.argv[1:]])
""",
        encoding="utf-8",
    )
    herdr_bin.chmod(0o755)

    res = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    assert res.returncode == 4, res.stdout + res.stderr

    p = json.loads(probe.read_text(encoding="utf-8"))
    assert p["marker"] is True
    assert p["rc"] != 0, p
    assert "a reset of %s is in progress" % mam_sandbox in p["stderr"], p
    assert "registered job" not in p["stdout"]
    assert p["jobs"] == []

    assert _result_of(res.stdout, name_a) == "done"
    assert _result_of(res.stdout, name_b) == "skipped: job open"
    assert "job78late" in res.stdout
    assert "reset [%s]:" % name_b not in res.stdout
    assert not tf_a.exists()
    assert tf_b.is_file()
    assert json.loads(job_file.read_text(encoding="utf-8"))["status"] == "pending"
    rows = _read_rows(mam_sandbox)
    assert rows[name_b]["status"] == "running" and rows[name_b]["claude_session_id_own"] == UB
    assert not (mam_sandbox / ".mam" / "reset-in-progress").exists()
    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()


def _descendant_args(root_pid: int) -> List[str]:
    res = subprocess.run(["ps", "-eo", "pid=,ppid=,args="], capture_output=True, text=True, timeout=5)
    kids: Dict[int, List[int]] = {}
    args: Dict[int, str] = {}
    for line in res.stdout.splitlines():
        parts = line.split(None, 2)
        if len(parts) == 3 and parts[0].isdigit() and parts[1].isdigit():
            kids.setdefault(int(parts[1]), []).append(int(parts[0]))
            args[int(parts[0])] = parts[2]
    out, todo = [], [root_pid]
    while todo:
        for c in kids.get(todo.pop(), []):
            out.append(args.get(c, ""))
            todo.append(c)
    return out


def test_reset_waits_for_an_admitted_submit_and_sees_its_job(mam_sandbox: Path, mock_herdr: Path, mock_agents):
    """Review B1 (codex), "submit first": a submit that holds the admission lock
    shared until its record is written makes the reset wait before it publishes
    its markers; the precheck then sees the job and refuses, nothing changed."""
    import fcntl

    name = "reviewer-claude-01"
    tf = _claude_transcript(mam_sandbox, mam_sandbox, UA)
    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
    _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
    spy_log = _install_herdr_spy(mam_sandbox)

    lock = open(mam_sandbox / ".mam" / "update-admission.lock", "a")
    fcntl.flock(lock, fcntl.LOCK_SH)
    script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
    proc = subprocess.Popen(
        ["bash", str(script), "--yes"],
        cwd=str(mam_sandbox),
        env=dict(os.environ, MAM_UPDATE_ADMISSION_WAIT_S="60", PYTHONDONTWRITEBYTECODE="1"),
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    exclusive = "lib_py.pane_lock acquire --fd 7 --wait-env MAM_UPDATE_ADMISSION_WAIT_S"
    try:
        deadline = time.time() + 60
        while time.time() < deadline:
            assert proc.poll() is None, proc.communicate()
            if any(a.endswith(exclusive) for a in _descendant_args(proc.pid)):
                break
            time.sleep(0.1)
        else:
            pytest.fail("reset.sh never waited for the admission lock")
        # Still blocked a second later: a shared take would have returned at once
        time.sleep(1.0)
        assert any(a.endswith(exclusive) for a in _descendant_args(proc.pid))
        assert not (mam_sandbox / ".mam" / "reset-in-progress").exists()
        assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
        jobs_dir = mam_sandbox / ".mam" / "jobs"
        jobs_dir.mkdir(parents=True, exist_ok=True)
        (jobs_dir / "job78first.json").write_text(
            json.dumps({"job_id": "job78first", "agent_session": "herdr:%s" % name, "status": "pending"}),
            encoding="utf-8",
        )
    finally:
        fcntl.flock(lock, fcntl.LOCK_UN)
        lock.close()
    stdout, stderr = proc.communicate(timeout=120)
    assert proc.returncode == 1, stdout + stderr
    assert "job78first" in stderr and "nothing was changed" in stderr
    assert tf.is_file()
    assert not any(c["verb"] in ("kill-session", "pane close") for c in _read_spy_log(spy_log))
    assert not (mam_sandbox / ".mam" / "reset-in-progress").exists()
    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()


def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path):
    lib = mam_sandbox / ".agents" / "skills" / "lib.sh"
    marker = mam_sandbox / ".mam" / "reset-in-progress"

    def admit():
        return subprocess.run(
            ["bash", "-c", 'source "$1" && mam_reset_admit "$2"', "_", str(lib), str(mam_sandbox)],
            cwd=str(mam_sandbox), capture_output=True, text=True, timeout=60,
        )

    assert admit().returncode == 0
    holder = subprocess.Popen(["sleep", "120"])
    try:
        _live_marker(marker, holder)
        res = admit()
        assert res.returncode == 1
        assert "a reset of %s is in progress" % mam_sandbox in res.stderr
    finally:
        holder.kill()
        holder.wait()
    res_stale = admit()
    assert res_stale.returncode == 0, res_stale.stderr


@pytest.mark.parametrize("registry", ["canonical", "alias", "custom"])
def test_registry_cli_is_refused_after_the_target_recheck(
    mam_sandbox: Path, mock_herdr: Path, mock_agents, registry: str
):
    """Review rounds 2-3 R1 (codex): the documented registry CLI, run by a caller
    that is not a child of the reset, after the target's open-job re-check and
    before its stop, is refused and registers nothing - whether it names the
    registry by its canonical path, by a symlink alias, or the registry is the
    MAM_JOBS_DIR the reset runs with; the reset's own create still registers
    its onboarding job."""
    import threading

    name = "reviewer-claude-01"
    _claude_transcript(mam_sandbox, mam_sandbox, UA)
    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
    _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
    jobs_dir = mam_sandbox / ".mam" / "jobs"
    jobs_dir.mkdir(parents=True, exist_ok=True)
    env_extra = None
    cli_dir = jobs_dir
    if registry == "alias":
        cli_dir = mam_sandbox / "job-alias"
        cli_dir.symlink_to(jobs_dir)
    elif registry == "custom":
        jobs_dir = cli_dir = mam_sandbox / "elsewhere" / "jobs"
        env_extra = {"MAM_JOBS_DIR": str(jobs_dir)}
    stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
    original = stop.with_name("stop-original.sh")
    stop.rename(original)
    entered, release = mam_sandbox / "stop-entered", mam_sandbox / "stop-release"
    gate = mam_sandbox / "stop-gate.py"
    gate.write_text(
        f"""import os, pathlib, sys, time
entered = pathlib.Path({str(entered)!r})
release = pathlib.Path({str(release)!r})
if not entered.exists():
    entered.touch()
    end = time.monotonic() + 60
    while not release.exists() and time.monotonic() < end:
        time.sleep(0.02)
os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
""",
        encoding="utf-8",
    )
    stop.write_text("#!/bin/bash\nexec %s %s \"$@\"\n" % (shlex.quote(sys.executable), shlex.quote(str(gate))))
    stop.chmod(0o755)
    record: dict = {}

    def writer():
        try:
            end = time.monotonic() + 60
            while not entered.exists() and time.monotonic() < end:
                time.sleep(0.02)
            record["marker"] = (mam_sandbox / ".mam" / "reset-in-progress").is_file()
            before = sorted(p.name for p in jobs_dir.glob("*.json"))
            cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
            res = subprocess.run(
                [sys.executable, str(cli), "--registry-dir", str(cli_dir), "register",
                 "--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name],
                capture_output=True, text=True, timeout=30,
            )
            record.update(rc=res.returncode, stdout=res.stdout, stderr=res.stderr, before=before,
                          after=sorted(p.name for p in jobs_dir.glob("*.json")))
        finally:
            release.touch()

    worker = threading.Thread(target=writer)
    worker.start()
    try:
        res = _run_reset(mam_sandbox, "--yes", env_extra=env_extra)
    finally:
        worker.join(timeout=90)
        _wait_reconcile_idle(mam_sandbox)
    assert not worker.is_alive()
    assert record["marker"] is True, record
    assert record["rc"] == 1, record
    assert "a reset of %s is in progress" % mam_sandbox in record["stderr"], record
    assert record["stdout"] == "" and record["after"] == record["before"], record
    assert res.returncode == 0, res.stdout + res.stderr
    assert _result_of(res.stdout, name) == "done"
    onboarding = [json.loads(p.read_text(encoding="utf-8")) for p in jobs_dir.glob("*.json")]
    assert [j["agent_session"] for j in onboarding] == ["herdr:" + name], onboarding
    assert not (jobs_dir / ".reset-in-progress").exists()


def _registry_marker(jobs_dir: Path, holder: subprocess.Popen, ws: Path, sessions: Optional[str] = None) -> None:
    _live_marker(jobs_dir / ".reset-in-progress", holder)
    with open(jobs_dir / ".reset-in-progress", "a", encoding="utf-8") as fh:
        fh.write("workspace=%s\n" % ws)
        if sessions is not None:
            fh.write("sessions=%s\n" % sessions)


def test_register_job_refuses_only_a_live_reset_not_its_own(mam_sandbox: Path, monkeypatch):
    from lib_py.delegation import registry

    jobs_dir = mam_sandbox / ".mam" / "jobs"
    other = mam_sandbox / "other-control-plane" / "jobs"
    jobs_dir.mkdir(parents=True, exist_ok=True)
    other.mkdir(parents=True)
    monkeypatch.delenv(registry.RESET_OWNER_ENV, raising=False)
    holder = subprocess.Popen(["sleep", "120"])
    try:
        _registry_marker(jobs_dir, holder, mam_sandbox)
        _registry_marker(other, holder, mam_sandbox, sessions="herdr:a,herdr:b")
        for d in (jobs_dir, other):
            with pytest.raises(registry.ResetInProgressError, match="a reset of %s is in progress" % mam_sandbox):
                registry.register_job(prompt="p", agent_session="herdr:b", registry_dir=str(d))
        # Another control plane's registry refuses only the reset's sessions
        assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(other))
        assert not list(jobs_dir.glob("*.json"))
        monkeypatch.setenv(registry.RESET_OWNER_ENV, str(holder.pid))
        assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(jobs_dir))
        monkeypatch.delenv(registry.RESET_OWNER_ENV)
    finally:
        holder.kill()
        holder.wait()
    assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(jobs_dir))
    assert len(list(jobs_dir.glob("*.json"))) == 2


def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path: Path):
    """R1 lock order and scope: a registration that read no marker holds the
    registry lock until its record is written; fence writes the marker under
    that lock, so it waits, and the precheck after it sees the job. Fenced are
    .mam/jobs, MAM_JOBS_DIR (one marker for an alias of the same directory) and
    a row's other control-plane registry (for this registry's sessions only)."""
    import fcntl

    ws = tmp_path / "ws"
    jobs_dir = ws / ".mam" / "jobs"
    jobs_dir.mkdir(parents=True)
    (ws / "alias").symlink_to(jobs_dir)
    other = tmp_path / "cp" / "jobs"
    other.mkdir(parents=True)
    row = _row("reviewer-claude-01", "claude", ws, own_id=UA)
    row["control_plane"] = {"mam_jobs_dir": str(other)}
    state = {"herdr_sessions": [row]}
    env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON=json.dumps(state),
               MAM_JOBS_DIR=str(ws / "alias"))
    holder = subprocess.Popen(["sleep", "120"])
    try:
        _live_marker(ws / ".mam" / "reset-in-progress", holder)
        lock = open(jobs_dir / ".lock", "a")
        fcntl.flock(lock, fcntl.LOCK_EX)
        try:
            proc = subprocess.Popen(
                [sys.executable, "-m", "lib_py.reset", "fence", "--workspace", str(ws)],
                env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
            )
            time.sleep(1.5)
            assert proc.poll() is None, proc.communicate()
            assert not (jobs_dir / ".reset-in-progress").exists()
            (jobs_dir / "job78locked.json").write_text(
                json.dumps({"job_id": "job78locked", "agent_session": "herdr:reviewer-claude-01", "status": "pending"}),
                encoding="utf-8",
            )
        finally:
            fcntl.flock(lock, fcntl.LOCK_UN)
            lock.close()
        out, err = proc.communicate(timeout=30)
        assert proc.returncode == 0, out + err
        assert out.split() == [os.path.realpath(jobs_dir), os.path.realpath(other)]
        own = (jobs_dir / ".reset-in-progress").read_text(encoding="utf-8")
        assert "pid=%d\n" % holder.pid in own and "workspace=%s\n" % os.path.realpath(ws) in own
        assert "sessions=" not in own
        assert "sessions=herdr:reviewer-claude-01\n" in (other / ".reset-in-progress").read_text(encoding="utf-8")
        pre = subprocess.run([sys.executable, "-m", "lib_py.reset", "precheck", "--workspace", str(ws)],
                             env=env, capture_output=True, text=True, timeout=30)
        assert pre.returncode == 1 and "job job78locked" in pre.stdout, pre.stdout + pre.stderr
    finally:
        holder.kill()
        holder.wait()


@pytest.mark.parametrize("artifact_cwd", ["own", "foreign"])
def test_post_discovered_id_is_purged_only_when_it_verifies_for_its_folder(
    mam_sandbox: Path, mock_herdr: Path, mock_agents, monkeypatch, artifact_cwd: str
):
    """Review round 2 R2 (codex): a recorded codex id whose rollout names another
    folder is not purged (stop's own unverified-own purge is for a hand-run
    cleanup); the same rollout naming the row's folder is."""
    from lib_py.agents.base import DiscoveryContext
    from lib_py.agents.registry import get_adapter

    foreign = mam_sandbox / "another-workspace"
    foreign.mkdir()
    cwd = mam_sandbox if artifact_cwd == "own" else foreign
    home = mam_sandbox / ".codex"
    monkeypatch.setenv("CODEX_HOME", str(home))
    rollout = home / "sessions/2026/10/30" / ("rollout-2026-10-30T00-00-00-%s.jsonl" % UB)
    rollout.parent.mkdir(parents=True)
    rollout.write_text(json.dumps({"type": "session_meta", "payload": {"id": UB, "cwd": str(cwd)}}) + "\n")
    name = "reviewer-codex-01"
    row = _row(name, "codex", mam_sandbox, own_id=UB)
    row["session_id_verified"] = False
    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="codex")
    _write_rows(mam_sandbox, [row])
    # The CLI delete fails, so the adapter's own fallback removes the rollout
    codex = mam_sandbox / "bin" / "codex"
    codex.write_text('#!/bin/bash\nif [ "$1" = delete ]; then exit 1; fi\nexit 0\n')
    codex.chmod(0o755)
    ctx = DiscoveryContext(workspace=str(mam_sandbox), agent_name="codex", home_dir=str(mam_sandbox),
                           row=row, mode="revalidate")
    assert get_adapter("codex").verify_artifact(UB, ctx) is (artifact_cwd == "own")

    res = _run_reset(mam_sandbox, "--yes")
    _wait_reconcile_idle(mam_sandbox)
    if artifact_cwd == "own":
        assert res.returncode == 0, res.stdout + res.stderr
        assert _result_of(res.stdout, name) == "done"
        assert not rollout.exists()
    else:
        assert res.returncode == 4, res.stdout + res.stderr
        assert _result_of(res.stdout, name) == "skipped: id not verified"
        assert "--purge-conversation --yes" not in res.stdout
        assert rollout.is_file()
        kept = _read_rows(mam_sandbox)[name]
        assert kept["status"] == "stopped" and kept["codex_session_id_own"] == UB
        assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()