feat(reset): multi-agent-mux-reset skill - reset.sh, SKILL.md, packaging (#78)

tests/test_issue78_reset.py

@@ -1,17 +1,21 @@
 """Issue #78: multi-agent-mux-reset — unit tests for lib_py.reset and
-end-to-end tests for reset.sh (populated across Commits 3 and 4).
+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"
@@ -27,6 +31,8 @@ UORC = "e7800005-0000-4000-8000-000000000005"
 pytestmark = pytest.mark.usefixtures("own_workspace_root")
 
 
+# --------------------------------------------------------------------------- Helpers
+
 def _row(
     name: str,
     agent: str,
@@ -36,8 +42,9 @@ def _row(
     created_at: str = "2026-10-01T00:00:00Z",
     server: str = "default",
     label: str = "ws",
-    own_id: str | None = None,
+    own_id: Optional[str] = None,
     pid: int = 4242,
+    epoch: int = 1700000000,
 ) -> dict:
     from lib_py.agents.registry import own_key
 
@@ -51,12 +58,20 @@ def _row(
         "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: dict | None = None, stdin_text: str | None = None, env_extra: dict | None = None):
+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"
@@ -74,11 +89,307 @@ def _run_reset_py(args: list[str], state: dict | None = None, stdin_text: str |
     )
 
 
+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_reset_py_python39_syntax():
-    src = (SKILLS / "lib_py" / "reset.py").read_text(encoding="utf-8")
-    ast.parse(src, filename="reset.py", feature_version=(3, 9))
+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):
@@ -186,12 +497,12 @@ def test_spec_file_write_read_and_validation(tmp_path: Path):
         read_spec(str(other_ws))
 
     # Wrong version
-    spath.write_text(json.dumps({"version": 99, "workspace": str(ws), "agents": []}))
+    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))
 
     # Corrupt JSON
-    spath.write_text("{corrupt")
+    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={})
@@ -262,7 +573,6 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
 
     state = {
         "herdr_sessions": [
-            # #2: normal stop wrote UA onto original row (same created_at, status stopped) -> still pending!
             _row(
                 "s2-same-inst-newly-recorded",
                 "agy",
@@ -271,7 +581,6 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
                 created_at="2026-10-01T00:00:00Z",
                 own_id=UA,
             ),
-            # #3 (9b): resumed into a new herdr session (different created_at), but same conversation ID UB -> pending!
             _row(
                 "s3-resumed-same-id",
                 "claude",
@@ -280,7 +589,6 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
                 created_at="2026-10-02T00:00:00Z",
                 own_id=UB,
             ),
-            # #4: newly created session, ID not discovered yet -> created
             _row(
                 "s4-created-no-id",
                 "hermes",
@@ -289,7 +597,6 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
                 created_at="2026-10-02T00:00:00Z",
                 own_id=None,
             ),
-            # #5: newly created session, new ID UD recorded (old_id was None) -> done
             _row(
                 "s5-done",
                 "codex",
@@ -315,7 +622,6 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
     assert parsed["s4-created-no-id"]["R_STATE"] == "created"
     assert parsed["s5-done"]["R_STATE"] == "done"
 
-    # Check D8 table formatting for these 5 rows (including 9b NEW!=OLD == "no" and 3-1 OLD_ID="-", NEW!=OLD="yes")
     res_tbl = _run_reset_py(
         [
             "table",
@@ -326,53 +632,519 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path):
         ],
         state=state,
     )
-    assert res_tbl.returncode == 4  # 4 takes precedence over 5
+    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
 
-    # s2: original row where step 1 captured UA, then step 3 refused purge -> NEW_ID is '-', RECORDED 'no', NEW!=OLD '-'
     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"
-
-    # s3 (9b): resumed row with same ID UB -> NEW_ID is UB, RECORDED 'yes', NEW!=OLD 'no'
     assert rows_by_name["s3-resumed-same-id"][1:6] == ["claude", UB, UB, "yes", "no"]
-
-    # s4: created, no ID yet -> NEW_ID '-', RECORDED 'no', NEW!=OLD '-', RESULT 'created; id not recorded yet'
     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"
-
-    # s5 (3-1): old_id was None, new_id is UD -> OLD_ID '-', NEW_ID UD, RECORDED 'yes', NEW!=OLD 'yes', RESULT 'done'
     assert rows_by_name["s5-done"][1:6] == ["codex", "-", UD, "yes", "yes"]
     assert rows_by_name["s5-done"][6] == "done"
 
 
-def test_precheck_open_jobs_and_caller_target(tmp_path: Path):
-    """D4-5: precheck reports open jobs with close command and refuses when caller is a target."""
-    ws = tmp_path / "ws"
-    jobs_dir = ws / ".mam" / "jobs"
-    jobs_dir.mkdir(parents=True)
-    (jobs_dir / "j123.json").write_text(
-        json.dumps(
-            {
-                "job_id": "j123",
-                "agent_session": "herdr:reviewer-claude-01",
-                "status": "running",
-            }
+# --------------------------------------------------------------------------- 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()
+
+
+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; 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()
+
+        # 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='-', 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", "-", 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()
     )
-    state = {
-        "herdr_sessions": [
-            _row("reviewer-claude-01", "claude", ws, own_id=UA, pid=os.getpid()),
-        ]
-    }
-    res = _run_reset_py(
-        ["precheck", "--workspace", str(ws)],
-        state=state,
-        env_extra={"CLAUDE_CODE_SESSION_ID": UA},
+    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.returncode == 1
-    assert "job j123" in res.stdout
-    assert "close --job j123 --request cancelled" in res.stdout
-    assert "you are reviewer-claude-01, one of the targets; run the reset from the orchestrator" in res.stdout
+    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)