From ca2532307fff67d58cfec6d2984ff2f7a3240c78 Mon Sep 17 00:00:00 2001 From: Godopu Date: Thu, 29 Oct 2026 05:05:06 +0000 Subject: [PATCH] feat(reset): lib_py.reset - targets, spec file, progress, result table (#78) --- .agents/skills/lib_py/reset.py | 709 +++++++++++++++++++++++++++++++++ tests/test_issue78_reset.py | 378 ++++++++++++++++++ 2 files changed, 1087 insertions(+) create mode 100644 .agents/skills/lib_py/reset.py create mode 100644 tests/test_issue78_reset.py diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py new file mode 100644 index 0000000..a33fa0a --- /dev/null +++ b/.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 ""), + }) + 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()) diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py new file mode 100644 index 0000000..6b85351 --- /dev/null +++ b/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