"""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


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))

    # 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() + 15.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() + 15.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()
    )