.agents/skills/lib_py/reset.py

@@ -0,0 +1,709 @@
+"""Decisions behind the multi-agent-mux-reset skill (Issue #78).
+
+reset.sh keeps the shell sequence (arg parsing, terminal check, update
+admission, loop lock, sequential stop/create invocations, exit codes) and
+delegates target selection, prechecks, atomic spec-file persistence, 5-state
+progress classification, and result-table formatting to this module.
+Standard library only, Python 3.9 compatible.
+"""
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+import shlex
+import subprocess
+import sys
+import tempfile
+from typing import Dict, List, Mapping, Optional, Tuple
+
+from lib_py.agents.base import AGENT_SESSION_ENV
+from lib_py.agents.registry import agent_of_row, own_key
+from lib_py.update import open_jobs, row_server
+
+SPEC_VERSION = 1
+SPEC_REL = os.path.join(".mam", "reset-spec.json")
+
+
+class SpecError(Exception):
+    pass
+
+
+def _real(p: str) -> str:
+    return os.path.realpath(p) if p else ""
+
+
+def spec_path(workspace: str) -> str:
+    return os.path.join(_real(workspace), SPEC_REL)
+
+
+def _state() -> dict:
+    return json.loads(os.environ.get("MAM_STATE_JSON") or "{}") or {}
+
+
+# --------------------------------------------------------------------------- D2: target selection
+
+def select_targets(workspace: str, state: dict) -> dict:
+    """Select running control-root sessions and classify non-targets.
+
+    Returns {"targets": [...], "skipped": [{"name", "reason"}],
+             "orchestrators": [...], "errors": [...]}.
+    """
+    ws_real = _real(workspace)
+    targets: List[dict] = []
+    skipped: List[dict] = []
+    errors: List[str] = []
+
+    for row in state.get("herdr_sessions") or []:
+        if not isinstance(row, dict):
+            continue
+        name = str(row.get("name") or "")
+        status = str(row.get("status") or "")
+        if status != "running":
+            skipped.append({"name": name, "reason": "status %s" % (status or "unknown")})
+            continue
+
+        cwd = str((row.get("pane") or {}).get("cwd") or "")
+        cwd_real = _real(cwd) if cwd else ""
+        if cwd_real != ws_real:
+            skipped.append({
+                "name": name,
+                "reason": "pane.cwd %s is not the control root (a worktree creator?)" % (cwd or "<empty>"),
+            })
+            continue
+
+        agent = agent_of_row(row)
+        role = str(row.get("role") or "").strip()
+        if not agent:
+            errors.append(
+                "running session %s has no recognisable agent type (agent=%r, pane.cmd=%r); "
+                "fix its registry row first"
+                % (name, row.get("agent"), (row.get("pane") or {}).get("cmd"))
+            )
+            continue
+        if not role:
+            errors.append(
+                "running session %s (%s) has an empty role; fix its registry row first"
+                % (name, agent)
+            )
+            continue
+
+        key = own_key(agent)
+        raw_id = row.get(key)
+        old_id = str(raw_id).strip() if isinstance(raw_id, str) and raw_id.strip() else None
+        targets.append({
+            "name": name,
+            "agent": agent,
+            "role": role,
+            "herdr_server": str(row_server(row)),
+            "herdr_workspace": str(row.get("herdr_workspace") or ""),
+            "old_id": old_id,
+            "old_created_at": str(row.get("herdr_session_created_at") or ""),
+        })
+
+    orchestrators = [str(u) for u in (state.get("orchestrator_uuids") or []) if u]
+    return {
+        "targets": targets,
+        "skipped": skipped,
+        "orchestrators": orchestrators,
+        "errors": errors,
+    }
+
+
+# --------------------------------------------------------------------------- D5: spec file
+
+def write_spec(workspace: str, agents: List[dict]) -> str:
+    """Atomically write .mam/reset-spec.json (mkstemp -> fsync -> replace -> dir fsync)."""
+    ws_real = _real(workspace)
+    mam_dir = os.path.join(ws_real, ".mam")
+    os.makedirs(mam_dir, exist_ok=True)
+    path = os.path.join(mam_dir, "reset-spec.json")
+    now_iso = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    doc = {
+        "version": SPEC_VERSION,
+        "workspace": ws_real,
+        "written_at": now_iso,
+        "agents": agents,
+    }
+    payload = (json.dumps(doc, indent=2) + "\n").encode("utf-8")
+    fd, tmp_path = tempfile.mkstemp(prefix=".reset-spec.", suffix=".tmp", dir=mam_dir)
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            fh.write(payload)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp_path, path)
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+    try:
+        dir_fd = os.open(mam_dir, os.O_RDONLY)
+        try:
+            os.fsync(dir_fd)
+        finally:
+            os.close(dir_fd)
+    except OSError:
+        pass
+    return path
+
+
+def read_spec(workspace: str) -> dict:
+    """Read and validate .mam/reset-spec.json; raise SpecError if invalid."""
+    ws_real = _real(workspace)
+    path = spec_path(ws_real)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            doc = json.load(fh)
+    except (OSError, ValueError) as exc:
+        raise SpecError("spec file %s cannot be read (%s); check it, then remove it by hand" % (path, exc))
+
+    if not isinstance(doc, dict) or doc.get("version") != SPEC_VERSION:
+        raise SpecError(
+            "spec file %s has unsupported version %r; check it, then remove it by hand"
+            % (path, doc.get("version") if isinstance(doc, dict) else None)
+        )
+    spec_ws = _real(str(doc.get("workspace") or ""))
+    if spec_ws != ws_real:
+        raise SpecError(
+            "spec file %s is for workspace %r, not %r; check it, then remove it by hand"
+            % (path, doc.get("workspace"), ws_real)
+        )
+    agents = doc.get("agents")
+    if not isinstance(agents, list):
+        raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path)
+    for item in agents:
+        if not isinstance(item, dict):
+            raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path)
+        name = item.get("name")
+        agent = item.get("agent")
+        role = item.get("role")
+        if not (isinstance(name, str) and name and isinstance(agent, str) and agent and isinstance(role, str) and role):
+            raise SpecError("spec file %s has an incomplete agent entry; check it, then remove it by hand" % path)
+        try:
+            own_key(agent)
+        except KeyError:
+            raise SpecError(
+                "spec file %s has unknown agent %r; check it, then remove it by hand" % (path, agent)
+            )
+    return doc
+
+
+def new_running_rows(workspace: str, state: dict, spec_agents: List[dict]) -> List[str]:
+    """Names of running control-root rows not present in spec_agents."""
+    known = {str(a.get("name") or "") for a in spec_agents}
+    sel = select_targets(workspace, state)
+    return [t["name"] for t in sel["targets"] if t["name"] not in known]
+
+
+# --------------------------------------------------------------------------- D6: progress classification
+
+def classify_entry(entry: dict, row: Optional[dict]) -> Tuple[str, Optional[str], bool]:
+    """Classify a spec entry against its current registry row (matched by name only).
+
+    Returns (r_state, new_id, same_instance):
+      1. row is None -> ("purged", None, False)
+      2. r.herdr_session_created_at == old_created_at -> ("pending", None, True)
+      3. not same_instance and r[id] == old_id (non-empty) -> ("pending", cur_id, False)
+      4. not same_instance and r[id] is empty -> ("created", None, False)
+      5. not same_instance and r[id] is non-empty and != old_id -> ("done", cur_id, False)
+    """
+    if row is None:
+        return "purged", None, False
+
+    key = own_key(entry["agent"])
+    cur_created_at = str(row.get("herdr_session_created_at") or "")
+    old_created_at = str(entry.get("old_created_at") or "")
+    raw_cur = row.get(key)
+    cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None
+    raw_old = entry.get("old_id")
+    old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
+
+    if cur_created_at == old_created_at:
+        return "pending", None, True
+    if cur_id and old_id and cur_id == old_id:
+        return "pending", cur_id, False
+    if not cur_id:
+        return "created", None, False
+    return "done", cur_id, False
+
+
+# --------------------------------------------------------------------------- D4: precheck
+
+def _ancestor_pids(start_pid: Optional[int] = None) -> set:
+    pid = os.getpid() if start_pid is None else start_pid
+    try:
+        res = subprocess.run(
+            ["ps", "-eo", "pid=,ppid="],
+            stdin=subprocess.DEVNULL,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return set()
+    if res.returncode != 0:
+        return set()
+    ppid_of: Dict[int, int] = {}
+    for line in (res.stdout or "").splitlines():
+        parts = line.split()
+        if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
+            ppid_of[int(parts[0])] = int(parts[1])
+    ancestors = set()
+    cur = pid
+    while cur and cur > 1 and cur not in ancestors:
+        ancestors.add(cur)
+        cur = ppid_of.get(cur, 0)
+    return ancestors
+
+
+def _caller_blockers(
+    targets: List[dict],
+    rows_by_name: Mapping[str, dict],
+    env: Mapping[str, str],
+    ancestors: Optional[set] = None,
+) -> List[str]:
+    env_ids = {
+        str(env.get(k) or "").strip()
+        for k in AGENT_SESSION_ENV
+        if str(env.get(k) or "").strip()
+    }
+    anc = _ancestor_pids() if ancestors is None else ancestors
+    out: List[str] = []
+    for t in targets:
+        name = t["name"]
+        row = rows_by_name.get(name)
+        r_state, _, _ = classify_entry(t, row)
+        if r_state != "pending":
+            continue
+        old_id = t.get("old_id")
+        key = own_key(t["agent"])
+        raw_cur = (row or {}).get(key)
+        cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None
+
+        matched = False
+        if env_ids and ((old_id and old_id in env_ids) or (cur_id and cur_id in env_ids)):
+            matched = True
+        elif row is not None and anc:
+            pane_pid = (row.get("pane") or {}).get("pid")
+            child_pid = row.get("child_pid")
+            for p in (pane_pid, child_pid):
+                if isinstance(p, int) and p > 1 and p in anc:
+                    matched = True
+                    break
+        if matched:
+            out.append("you are %s, one of the targets; run the reset from the orchestrator" % name)
+    return out
+
+
+def precheck(
+    workspace: str,
+    state: dict,
+    env: Mapping[str, str],
+    ancestors: Optional[set] = None,
+) -> List[str]:
+    """Return refusal lines for open jobs, unreadable spec, un-recreatable targets, or caller==target."""
+    ws_real = _real(workspace)
+    blockers: List[str] = []
+
+    spath = spec_path(ws_real)
+    targets: List[dict] = []
+    if os.path.lexists(spath):
+        try:
+            spec = read_spec(ws_real)
+            targets = spec["agents"]
+        except SpecError as exc:
+            blockers.append(str(exc))
+    else:
+        sel = select_targets(ws_real, state)
+        blockers.extend(sel["errors"])
+        targets = sel["targets"]
+
+    for j in open_jobs(ws_real, state, env):
+        if j.get("dir"):
+            blockers.append(
+                "jobs directory %s cannot be listed; fix its permissions, then try again"
+                % j["path"]
+            )
+        elif j.get("unreadable"):
+            blockers.append(
+                "job record %s cannot be read; check it, then move or delete it by hand"
+                % j["path"]
+            )
+        else:
+            blockers.append(
+                "job %s (%s, %s) is open; wait for it to end or close it: python "
+                ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
+                " --registry-dir %s close --job %s --request cancelled"
+                % (
+                    j["job_id"],
+                    j["agent_session"] or "-",
+                    j["status"],
+                    j["registry_dir"],
+                    j["job_id"],
+                )
+            )
+
+    rows_by_name = {
+        str(s.get("name")): s
+        for s in (state.get("herdr_sessions") or [])
+        if isinstance(s, dict) and s.get("name")
+    }
+    blockers.extend(_caller_blockers(targets, rows_by_name, env, ancestors=ancestors))
+    return blockers
+
+
+# --------------------------------------------------------------------------- D1: entries shell lines
+
+def format_entry_line(entry: dict, row: Optional[dict]) -> str:
+    r_state, _, _ = classify_entry(entry, row)
+    key = own_key(entry["agent"])
+    raw_cur = (row or {}).get(key)
+    row_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else ""
+    pairs = [
+        ("R_STATE", r_state),
+        ("R_NAME", str(entry.get("name") or "")),
+        ("R_AGENT", str(entry.get("agent") or "")),
+        ("R_ROLE", str(entry.get("role") or "")),
+        ("R_SERVER", str(entry.get("herdr_server") or "default")),
+        ("R_LABEL", str(entry.get("herdr_workspace") or "")),
+        ("R_OLD_ID", str(entry.get("old_id") or "")),
+        ("R_ROW_ID", row_id),
+    ]
+    return " ".join("%s=%s" % (k, shlex.quote(v)) for k, v in pairs)
+
+
+# --------------------------------------------------------------------------- D8: result table
+
+def _parse_skipped(items: Optional[List[str]]) -> Dict[str, str]:
+    out: Dict[str, str] = {}
+    for raw in items or []:
+        for token in raw.split(","):
+            token = token.strip()
+            if not token:
+                continue
+            if ":" in token:
+                name, kind = token.split(":", 1)
+                out[name] = kind
+            else:
+                out[token] = "purge_refused"
+    return out
+
+
+def _parse_failed(raw: Optional[str]) -> Optional[Tuple[str, str, str]]:
+    if not raw:
+        return None
+    parts = raw.split(":", 2)
+    if len(parts) == 3:
+        return parts[0], parts[1], parts[2]
+    if len(parts) == 2:
+        return parts[0], parts[1], "1"
+    return parts[0], "stop", "1"
+
+
+def build_table(
+    agents: List[dict],
+    state: dict,
+    skipped_items: Optional[List[str]] = None,
+    failed_spec: Optional[str] = None,
+) -> Tuple[str, int]:
+    """Format the D8 result table and compute the D9 exit code (0, 1, 4, or 5)."""
+    rows_by_name = {
+        str(s.get("name")): s
+        for s in (state.get("herdr_sessions") or [])
+        if isinstance(s, dict) and s.get("name")
+    }
+    skipped_map = _parse_skipped(skipped_items)
+    failed = _parse_failed(failed_spec)
+    failed_name = failed[0] if failed else None
+    after_failed = False
+
+    headers = ("NAME", "AGENT", "OLD_ID", "NEW_ID", "RECORDED", "NEW≠OLD", "RESULT")
+    table_rows: List[Tuple[str, str, str, str, str, str, str]] = []
+
+    has_skip_or_incomplete = bool(skipped_map)
+    has_created = False
+
+    for entry in agents:
+        name = str(entry.get("name") or "")
+        agent = str(entry.get("agent") or "")
+        raw_old = entry.get("old_id")
+        old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None
+        old_col = old_id if old_id else "-"
+
+        row = rows_by_name.get(name)
+        r_state, new_id, _ = classify_entry(entry, row)
+        new_col = new_id if new_id else "-"
+        recorded_col = "yes" if new_id else "no"
+        if not new_id:
+            diff_col = "-"
+        elif not old_id or new_id != old_id:
+            diff_col = "yes"
+        else:
+            diff_col = "no"
+
+        if failed and name == failed_name:
+            result_col = "failed at %s (rc %s)" % (failed[1], failed[2])
+            after_failed = True
+            has_skip_or_incomplete = True
+        elif after_failed and r_state in ("pending", "purged"):
+            result_col = "not reached"
+            has_skip_or_incomplete = True
+        elif name in skipped_map:
+            kind = skipped_map[name]
+            if kind == "no_id":
+                result_col = "skipped: no conversation id recorded"
+            else:
+                result_col = "skipped: purge refused"
+            has_skip_or_incomplete = True
+        elif r_state == "done":
+            result_col = "done"
+        elif r_state == "created":
+            result_col = "created; id not recorded yet"
+            has_created = True
+        elif r_state == "purged":
+            result_col = "purged"
+            has_skip_or_incomplete = True
+        else:
+            result_col = "pending"
+            has_skip_or_incomplete = True
+
+        table_rows.append((name, agent, old_col, new_col, recorded_col, diff_col, result_col))
+
+    widths = [len(h) for h in headers]
+    for r in table_rows:
+        for i in range(len(headers) - 1):
+            if len(r[i]) > widths[i]:
+                widths[i] = len(r[i])
+
+    def _fmt(cols: Tuple[str, ...]) -> str:
+        parts = [cols[i].ljust(widths[i]) for i in range(len(cols) - 1)]
+        parts.append(cols[-1])
+        return "  ".join(parts)
+
+    lines = [_fmt(headers)]
+    for r in table_rows:
+        lines.append(_fmt(r))
+
+    if failed:
+        rc = 1
+    elif has_skip_or_incomplete:
+        rc = 4
+    elif has_created:
+        rc = 5
+    else:
+        rc = 0
+    return "\n".join(lines), rc
+
+
+# --------------------------------------------------------------------------- D10: dry-run
+
+def format_dry_run(
+    workspace: str,
+    state: dict,
+    env: Mapping[str, str],
+    extra_refusals: Optional[List[str]] = None,
+) -> Tuple[str, int]:
+    ws_real = _real(workspace)
+    lines: List[str] = []
+    refusals: List[str] = list(extra_refusals or [])
+    refusals.extend(precheck(ws_real, state, env))
+
+    rows_by_name = {
+        str(s.get("name")): s
+        for s in (state.get("herdr_sessions") or [])
+        if isinstance(s, dict) and s.get("name")
+    }
+    sel = select_targets(ws_real, state)
+    spath = spec_path(ws_real)
+    action_map = {
+        "pending": "stop + purge + create",
+        "purged": "create",
+        "created": "none (waiting for id)",
+        "done": "none (done)",
+    }
+
+    if os.path.lexists(spath):
+        try:
+            spec = read_spec(ws_real)
+        except SpecError:
+            spec = None
+        if spec is not None:
+            lines.append("spec file: %s" % spath)
+            for e in spec["agents"]:
+                r_state, _, _ = classify_entry(e, rows_by_name.get(e["name"]))
+                lines.append(
+                    "target: %s  agent=%s  role=%s  herdr_server=%s  herdr_workspace=%s  "
+                    "old_id=%s  old_created_at=%s  state=%s  action=%s"
+                    % (
+                        e["name"],
+                        e["agent"],
+                        e["role"],
+                        e.get("herdr_server") or "default",
+                        e.get("herdr_workspace") or "",
+                        e.get("old_id") or "-",
+                        e.get("old_created_at") or "-",
+                        r_state,
+                        action_map.get(r_state, r_state),
+                    )
+                )
+            extra_running = new_running_rows(ws_real, state, spec["agents"])
+            if extra_running:
+                lines.append(
+                    "NOTE: running control-root session(s) not in reset-spec.json "
+                    "(not targeted by this rerun): %s" % ", ".join(extra_running)
+                )
+    else:
+        if sel["targets"]:
+            for t in sel["targets"]:
+                lines.append(
+                    "target: %s  agent=%s  role=%s  herdr_server=%s  herdr_workspace=%s  "
+                    "old_id=%s  old_created_at=%s"
+                    % (
+                        t["name"],
+                        t["agent"],
+                        t["role"],
+                        t["herdr_server"],
+                        t["herdr_workspace"],
+                        t["old_id"] or "-",
+                        t["old_created_at"] or "-",
+                    )
+                )
+        else:
+            lines.append("targets: (none)")
+
+    for sk in sel["skipped"]:
+        lines.append("skip: %s (%s)" % (sk["name"], sk["reason"]))
+    for u in sel["orchestrators"]:
+        lines.append("orchestrator %s (not a row; never reset)" % u)
+
+    if refusals:
+        lines.append("would refuse:")
+        for r in refusals:
+            lines.append("  - %s" % r)
+        return "\n".join(lines), 1
+    return "\n".join(lines), 0
+
+
+# --------------------------------------------------------------------------- CLI
+
+def main(argv: Optional[List[str]] = None) -> int:
+    p = argparse.ArgumentParser(prog="lib_py.reset")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    s = sub.add_parser("plan")
+    s.add_argument("--workspace", required=True)
+
+    s = sub.add_parser("precheck")
+    s.add_argument("--workspace", required=True)
+
+    s = sub.add_parser("write-spec")
+    s.add_argument("--workspace", required=True)
+
+    s = sub.add_parser("entries")
+    s.add_argument("--workspace", required=True)
+    s.add_argument("--session", default="")
+    s.add_argument("--warn-new", action="store_true")
+
+    s = sub.add_parser("table")
+    s.add_argument("--workspace", required=True)
+    s.add_argument("--skipped", action="append", default=[])
+    s.add_argument("--failed", default="")
+
+    s = sub.add_parser("dry-run")
+    s.add_argument("--workspace", required=True)
+    s.add_argument("--refusal", action="append", default=[])
+
+    a = p.parse_args(argv)
+    ws = _real(a.workspace)
+
+    if a.cmd == "write-spec":
+        try:
+            raw = json.load(sys.stdin)
+        except ValueError as exc:
+            print("ERROR: invalid targets JSON on stdin (%s)" % exc, file=sys.stderr)
+            return 1
+        agents = raw.get("targets") if isinstance(raw, dict) else raw
+        if not isinstance(agents, list):
+            print("ERROR: targets JSON must be a list", file=sys.stderr)
+            return 1
+        try:
+            write_spec(ws, agents)
+        except OSError as exc:
+            print("ERROR: cannot write %s (%s)" % (spec_path(ws), exc), file=sys.stderr)
+            return 1
+        return 0
+
+    try:
+        state = _state()
+    except ValueError as exc:
+        if a.cmd == "precheck":
+            print("registry: cannot parse MAM_STATE_JSON (%s)" % exc)
+            return 1
+        print("ERROR: cannot parse MAM_STATE_JSON (%s)" % exc, file=sys.stderr)
+        return 1
+
+    if a.cmd == "plan":
+        sel = select_targets(ws, state)
+        print(json.dumps(sel))
+        return 1 if sel["errors"] else 0
+
+    if a.cmd == "precheck":
+        lines = precheck(ws, state, os.environ)
+        for line in lines:
+            print(line)
+        return 1 if lines else 0
+
+    if a.cmd == "entries":
+        try:
+            spec = read_spec(ws)
+        except SpecError as exc:
+            print("ERROR: %s" % exc, file=sys.stderr)
+            return 1
+        if a.warn_new:
+            extra = new_running_rows(ws, state, spec["agents"])
+            if extra:
+                print(
+                    "NOTE: running control-root session(s) not in %s (not targeted by this rerun): %s"
+                    % (spec_path(ws), ", ".join(extra)),
+                    file=sys.stderr,
+                )
+        rows_by_name = {
+            str(s.get("name")): s
+            for s in (state.get("herdr_sessions") or [])
+            if isinstance(s, dict) and s.get("name")
+        }
+        for e in spec["agents"]:
+            if a.session and e.get("name") != a.session:
+                continue
+            print(format_entry_line(e, rows_by_name.get(e["name"])))
+        return 0
+
+    if a.cmd == "table":
+        try:
+            spec = read_spec(ws)
+        except SpecError as exc:
+            print("ERROR: %s" % exc, file=sys.stderr)
+            return 1
+        out, rc = build_table(
+            spec["agents"],
+            state,
+            skipped_items=a.skipped,
+            failed_spec=a.failed or None,
+        )
+        print(out)
+        return rc
+
+    if a.cmd == "dry-run":
+        out, rc = format_dry_run(ws, state, os.environ, extra_refusals=a.refusal)
+        print(out)
+        return rc
+
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_issue78_reset.py

@@ -0,0 +1,378 @@
+"""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).
+"""
+from __future__ import annotations
+
+import ast
+import json
+import os
+import shlex
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+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")
+
+
+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: str | None = None,
+    pid: int = 4242,
+) -> 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,
+        key: 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):
+    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,
+    )
+
+
+# --------------------------------------------------------------------------- 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_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": []}))
+    with pytest.raises(SpecError, match="check it, then remove it by hand"):
+        read_spec(str(ws))
+
+    # Corrupt JSON
+    spath.write_text("{corrupt")
+    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": [
+            # #2: normal stop wrote UA onto original row (same created_at, status stopped) -> still pending!
+            _row(
+                "s2-same-inst-newly-recorded",
+                "agy",
+                ws,
+                status="stopped",
+                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",
+                ws,
+                status="running",
+                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",
+                ws,
+                status="running",
+                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",
+                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"
+
+    # 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",
+            "--workspace",
+            str(ws),
+            "--skipped",
+            "s2-same-inst-newly-recorded:purge_refused",
+        ],
+        state=state,
+    )
+    assert res_tbl.returncode == 4  # 4 takes precedence over 5
+    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",
+            }
+        )
+    )
+    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},
+    )
+    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