diff --git a/.agents/skills/lib_py/agents/adapters/agy.py b/.agents/skills/lib_py/agents/adapters/agy.py index febdef9..8ea8aa5 100644 --- a/.agents/skills/lib_py/agents/adapters/agy.py +++ b/.agents/skills/lib_py/agents/adapters/agy.py @@ -197,11 +197,14 @@ class AgyAgentAdapter(BaseAgentAdapter): # the DB already gone, and those orphans must still be reachable. return any(os.path.exists(p) for p in self._artifact_paths(uuid, ctx)) + def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list: + return self._artifact_paths(uuid, ctx) + def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool: # The holder may keep the presence lock (or a sidecar) open rather # than the DB, and may name the uuid nowhere; the lsof leg of the # check must see every path of the conversation. - return conversation_in_use(uuid, self._artifact_paths(uuid, ctx)) + return conversation_in_use(uuid, self.in_use_paths(uuid, ctx)) def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list: # Issue #10: a purge must leave nothing resumable behind. Besides the diff --git a/.agents/skills/lib_py/agents/adapters/codex.py b/.agents/skills/lib_py/agents/adapters/codex.py index 4b4fa53..d330550 100644 --- a/.agents/skills/lib_py/agents/adapters/codex.py +++ b/.agents/skills/lib_py/agents/adapters/codex.py @@ -1,6 +1,6 @@ -import os, json, glob, sqlite3, subprocess +import os, json, glob, shutil, sqlite3, subprocess from typing import Optional, Any, List -from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _ro_connect, scan_names, db_rows_naming +from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, _plausible_id, _ro_connect, scan_names, db_rows_naming from lib_py.verify_session import workspace_key class CodexAgentAdapter(BaseAgentAdapter): @@ -131,6 +131,22 @@ class CodexAgentAdapter(BaseAgentAdapter): except Exception: return False + def _purge_tui_thread_capability(self, uuid: str, ctx: DiscoveryContext, purged: List[str]) -> None: + if not _plausible_id(uuid): + return + cap = os.path.join(self._codex_home(ctx), "tui-thread-reference-capabilities", uuid) + if not os.path.lexists(cap): + return + try: + if os.path.isdir(cap) and not os.path.islink(cap): + shutil.rmtree(cap) + else: + os.unlink(cap) + if not os.path.lexists(cap): + purged.append(cap) + except Exception: + pass + def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]: purged = [] path = self._find_rollout(uuid, ctx) @@ -140,12 +156,16 @@ class CodexAgentAdapter(BaseAgentAdapter): if res.returncode == 0: if path: purged.append(path) + self._purge_tui_thread_capability(uuid, ctx, purged) return purged except Exception: pass if path and os.path.exists(path): - os.remove(path) - purged.append(path) + try: + os.remove(path) + purged.append(path) + except Exception: + pass db = self._state_db(ctx) if db and os.path.exists(db): try: @@ -156,6 +176,7 @@ class CodexAgentAdapter(BaseAgentAdapter): purged.append(f"state db row for thread: {uuid}") except Exception: pass + self._purge_tui_thread_capability(uuid, ctx, purged) return purged def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]: diff --git a/.agents/skills/lib_py/agents/adapters/opencode.py b/.agents/skills/lib_py/agents/adapters/opencode.py index 3ea89fb..be3f98f 100644 --- a/.agents/skills/lib_py/agents/adapters/opencode.py +++ b/.agents/skills/lib_py/agents/adapters/opencode.py @@ -13,6 +13,11 @@ _LEGACY_TABLES = (("message", "session_id"), ("messages", "session_id"), ("part" # (binary, HOME, XDG_DATA_HOME) -> the db `debug paths` printed for them. _DB_PATHS = {} + +def _is_shared_service(cmd: str) -> bool: + """The per-user opencode background service (`serve --service`).""" + return "serve --service" in (cmd or "") + class OpenCodeAgentAdapter(BaseAgentAdapter): @property def name(self) -> str: @@ -250,6 +255,13 @@ class OpenCodeAgentAdapter(BaseAgentAdapter): found += rows if i == uuid else [f"{r} [child session {i}]" for r in rows] return found + def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list: + db = self._resolve_db(ctx) + return [db] if db else [] + + def _holder_counts(self, cmd: str) -> bool: + return not _is_shared_service(cmd) + def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool: """A process naming the session, or an opencode process holding the shared database. The per-user background service (`serve --service`) @@ -270,13 +282,14 @@ class OpenCodeAgentAdapter(BaseAgentAdapter): timeout=10).stdout except Exception: return True - if cmd.strip() and "serve --service" not in cmd: + if cmd.strip() and not _is_shared_service(cmd): return True return False def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list: + db = self._resolve_db(ctx) return super().in_use_hints(uuid, ctx) + [ - f"another opencode session holding {self._resolve_db(ctx)}: lsof {self._resolve_db(ctx)}"] + f"another opencode session holding {db}: lsof {db}"] # opencode 2.x: the default agent is build (`--agent` is gone from the # TUI), and --standalone keeps the server a child of the pane's TUI diff --git a/.agents/skills/lib_py/agents/base.py b/.agents/skills/lib_py/agents/base.py index 35423e6..7438095 100644 --- a/.agents/skills/lib_py/agents/base.py +++ b/.agents/skills/lib_py/agents/base.py @@ -1,7 +1,7 @@ # base.py — BaseAgentAdapter abstract base class, SpawnSpec, and DiscoveryContext import os, json, sqlite3, shutil, subprocess, tempfile -from typing import Optional, Dict, Any, List +from typing import Optional, Dict, Any, List, Tuple from urllib.parse import quote from lib_py.paths import resolve_home from lib_py.verify_session import verify_session_uuid, workspace_key @@ -86,6 +86,34 @@ def conversation_in_use(uuid: str, paths=()) -> bool: return False +def file_holders(paths=()) -> List[Tuple[int, str]]: + """(pid, command_line) for each process holding one of `paths` open via lsof (#78). + Informational only: returns [] when lsof is absent or fails.""" + existing = [p for p in paths if p and os.path.exists(p)] + if not existing or not shutil.which("lsof"): + return [] + try: + r = subprocess.run(["lsof", "-t"] + existing, capture_output=True, text=True, timeout=10) + except Exception: + return [] + pids: List[int] = [] + for tok in (r.stdout or "").split(): + if tok.isdigit(): + pid = int(tok) + if pid not in pids: + pids.append(pid) + out: List[Tuple[int, str]] = [] + for pid in pids: + try: + cmd = subprocess.run(["ps", "-o", "command=", "-p", str(pid)], + capture_output=True, text=True, timeout=10).stdout.strip() + except Exception: + continue + if cmd: + out.append((pid, cmd)) + return out + + def _plausible_id(uuid: str) -> bool: """An id safe to build a path from and to search for: not empty, not made of dots alone ("." names the parent dir itself), no "/", ".." or "*".""" @@ -354,14 +382,25 @@ class BaseAgentAdapter: path = self.artifact_path(uuid, ctx) return uuid in path and os.path.exists(path) + def in_use_paths(self, uuid: str, ctx: DiscoveryContext) -> list: + """On-disk paths checked with lsof for conversation_in_use and in_use_hints (#78).""" + return [self.artifact_path(uuid, ctx)] + + def _holder_counts(self, cmd: str) -> bool: + return True + def conversation_in_use(self, uuid: str, ctx: DiscoveryContext) -> bool: """True while a live process still references the conversation.""" - return conversation_in_use(uuid, [self.artifact_path(uuid, ctx)]) + return conversation_in_use(uuid, self.in_use_paths(uuid, ctx)) def in_use_hints(self, uuid: str, ctx: DiscoveryContext) -> list: """Where to look for what holds the conversation, shown when a purge is refused because it is in use.""" - return [f"any process with it on its command line (an editor, a grep, ...): ps -axww | grep {uuid}"] + hints = [f"any process with it on its command line (an editor, a grep, ...): ps -axww | grep {uuid}"] + for pid, cmd in file_holders(self.in_use_paths(uuid, ctx)): + if self._holder_counts(cmd): + hints.append(f"held open by pid {pid}: {cmd} (not stopped by MAM; end it yourself if it is safe)") + return hints def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str: raise NotImplementedError diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py new file mode 100644 index 0000000..6aaf6e9 --- /dev/null +++ b/.agents/skills/lib_py/reset.py @@ -0,0 +1,812 @@ +"""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 shutil +import sqlite3 +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, get_adapter, 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(workspace: str = "") -> dict: + raw = os.environ.get("MAM_STATE_JSON") + if raw is not None: + return json.loads(raw or "{}") or {} + yaml_path = os.environ.get("AGENT_SESSIONS_YAML") or ( + os.path.join(_real(workspace), ".mam", "agent-sessions.yaml") if workspace else "" + ) + if not yaml_path: + return {} + db_path = os.path.splitext(yaml_path)[0] + ".db" + if os.path.exists(db_path): + tmp_dir = tempfile.mkdtemp(prefix="mam-reset-ro.") + try: + tmp_db = os.path.join(tmp_dir, "agent-sessions.db") + shutil.copyfile(db_path, tmp_db) + for suf in ("-wal", "-shm"): + if os.path.exists(db_path + suf): + shutil.copyfile(db_path + suf, tmp_db + suf) + conn = sqlite3.connect(tmp_db, timeout=60.0) + try: + d: dict = {} + row = conn.execute("SELECT data FROM state WHERE id=1").fetchone() + if row: + d = json.loads(row[0]) + cursor = conn.execute("SELECT data FROM sessions") + d["herdr_sessions"] = [json.loads(r[0]) for r in cursor.fetchall()] + return d + finally: + conn.close() + except Exception as exc: + raise ValueError("the registry %s cannot be read (%s); fix it first" % (yaml_path, exc)) + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + if os.path.exists(yaml_path): + try: + import yaml + with open(yaml_path, encoding="utf-8") as fh: + data = yaml.safe_load(fh) + if data is not None and not isinstance(data, dict): + raise ValueError("top-level YAML is not a mapping") + return data or {} + except Exception as exc: + raise ValueError("the registry %s cannot be read (%s); fix it first" % (yaml_path, exc)) + return {} + + +# --------------------------------------------------------------------------- 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) + adapter = get_adapter(agent) + if adapter is None or adapter.name != agent or not own_key(agent): + raise SpecError( + "spec file %s has unknown agent %r; check it, then remove it by hand" % (path, agent) + ) + return doc + + +def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[str]: + """Persist a newly captured conversation id onto an entry with old_id=None. + + Called after step 1 (normal stop) captures a POST_DISCOVERED session's id + onto the original instance's registry row, so a subsequent resume + rerun + cannot mistake the original conversation for a new one (review F1). + Reads the id from `state` (never from command-line args). + """ + ws_real = _real(workspace) + spec = read_spec(ws_real) + rows_by_name = { + str(s.get("name")): s + for s in (state.get("herdr_sessions") or []) + if isinstance(s, dict) and s.get("name") + } + row = rows_by_name.get(session_name) + if row is None: + return None + changed = False + recorded: Optional[str] = None + for entry in spec["agents"]: + if str(entry.get("name") or "") != session_name: + continue + raw_old = entry.get("old_id") + old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None + if old_id: + return old_id + cur_created_at = str(row.get("herdr_session_created_at") or "") + old_created_at = str(entry.get("old_created_at") or "") + if cur_created_at != old_created_at: + return None + key = own_key(entry["agent"]) + raw_cur = row.get(key) + cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None + if cur_id: + entry["old_id"] = cur_id + recorded = cur_id + changed = True + if changed: + write_spec(ws_real, spec["agents"]) + return recorded + + +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("record-id") + s.add_argument("--workspace", required=True) + s.add_argument("--session", required=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(ws) + except ValueError as exc: + if a.cmd == "dry-run": + refs = list(a.refusal or []) + [str(exc)] + out, rc = format_dry_run(ws, {}, os.environ, extra_refusals=refs) + print(out) + return rc + 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 == "record-id": + try: + record_old_id(ws, state, a.session) + except (SpecError, OSError) as exc: + print("ERROR: %s" % exc, file=sys.stderr) + return 1 + 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/.agents/skills/multi-agent-mux-reset/SKILL.md b/.agents/skills/multi-agent-mux-reset/SKILL.md new file mode 100644 index 0000000..7b74635 --- /dev/null +++ b/.agents/skills/multi-agent-mux-reset/SKILL.md @@ -0,0 +1,48 @@ +--- +name: multi-agent-mux-reset +description: "Purge the control root's running resident agents and recreate each with the same name, agent, role, herdr server and workspace label, one at a time, with onboarding. Persists .mam/reset-spec.json before the first purge so an interrupted or skipped run resumes where it left off." +version: 4.13.0 +author: godopu +license: MIT +platforms: [linux, macos] +environments: [terminal, herdr] +metadata: + hermes: + tags: [reset, purge, recreate, lifecycle, onboarding] + related_skills: [multi-agent-mux-stop, multi-agent-mux-create, multi-agent-mux-status] +--- + +# Multi-Agent Reset + +```bash +bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] +``` + +| Item | Behaviour | +|---|---| +| Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Worktree rows, `stopped`/`terminated` rows, and `orchestrator_uuids` are skipped and listed with their reason. | +| Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required (exits `3` otherwise). `--dry-run` inspects without `--yes` and changes nothing. | +| Prechecks | Refuses before any change when an open job exists (prints the `registry.py ... close --request cancelled` command), when `.mam/loop-guard-active` or `.mam/update-in-progress` is held, when a target row lacks a valid agent or role, or when the caller itself is one of the targets. | +| Lock & concurrency | Takes update admission (`mam_update_admit`), acquires `.mam/loop-guard-active` (`mam_acquire_loop_lock`), then releases update admission so `run_loop.sh`, `update.sh`, and concurrent resets refuse while the reset runs. | +| Spec file & rerun | Writes `.mam/reset-spec.json` atomically before the first stop/purge and removes it only when all targets reach `done`. Rerunning resumes from the spec file without re-purging already recreated agents (`created` or `done`). | +| Sequence per target | Sequential (never parallel): ① normal `stop_session.sh --reason reset` (captures verified ID onto the row while herdr is live and sets `status: stopped`), ② checks that the row has a recorded conversation ID, ③ `stop_session.sh --reason reset --purge-conversation --yes` (using the row's recorded ID, never a CLI UUID), then `create_session.sh` with the original name, agent, role, `--herdr-session`, and `--herdr-workspace` (onboarding enabled by default). | +| When skipped (exit 4) | If step ② finds no recorded ID or step ③ exits `4` (in-use gate), the target is skipped and the reset continues to the next agent: the skipped agent is stopped but resumable (its row keeps its id) and its conversation files are untouched; rerun once its holders end, or resume it. | + +| Exit | Meaning | +|---|---| +| 0 | All targeted agents reached `done` and `.mam/reset-spec.json` was removed; or nothing to reset, cancelled at prompt, or `--dry-run` with no blockers | +| 1 | Refused by precheck (nothing changed), corrupt/mismatched `.mam/reset-spec.json`, or a `stop`/`create` step failed (`.mam/reset-spec.json` is kept so rerunning resumes) | +| 2 | Invalid arguments | +| 3 | Not a terminal and `-y`/`--yes` was not passed; nothing was changed | +| 4 | Completed the pass, but one or more agents were skipped (purge refused with stop exit 4, or no conversation ID recorded); `.mam/reset-spec.json` is kept for rerun | +| 5 | No agent was skipped or failed, but one or more newly created post-discovered agents do not have their new conversation ID recorded yet; `.mam/reset-spec.json` is kept so rerunning re-checks without re-purging | +| other | Under `set -e`, an unexpected sub-step failure passes its nonzero exit code through | + +## Safety Notes + +- **Irreversible**: purged conversations cannot be recovered. Confirm with `--dry-run` first. +- **No external process termination**: processes holding conversation files open (reported via `lsof` in `in_use_hints`) are never signalled by MAM; end them yourself when safe, then rerun. +- **Shared stores preserved**: Grok's shared `memory-v2.db` is left intact during purge; only session rows and per-session files are removed. +- **Orchestrator hook during reset**: while `.mam/loop-guard-active` is held by the reset, the AGY orchestrator's direct file-edit hook (`loop_delegation_guard.sh`) blocks direct edits until the reset finishes. +- **Post-precheck job submissions**: `multi-agent-mux-delegate-job submit` does not check `.mam/loop-guard-active`; do not submit new jobs to resident agents while a reset is running. +- **Same-type post-discovered ordering**: when resetting multiple post-discovered agents of the same type, the second `create_session.sh` waits up to `MAM_ID_ORDER_WAIT_S` for the first agent's ID to be recorded; if it times out, rerun the reset once the ID is recorded. diff --git a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh new file mode 100755 index 0000000..5963263 --- /dev/null +++ b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh @@ -0,0 +1,286 @@ +#!/usr/bin/env bash +# reset.sh — multi-agent-mux-reset (Issue #78): purge the control root's +# running resident agents and recreate each with the same name, agent, role, +# herdr server and workspace label, one at a time, with onboarding. +# Usage: +# bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh [-y|--yes] [--dry-run] +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +SKILL_DIR="$(cd "$SCRIPT_DIR/../.." && pwd)" +WS="$(cd -P "$SCRIPT_DIR/../../../.." && pwd -P)" +export WORKSPACE_ROOT="$WS" +export AGENT_SESSIONS_YAML="${AGENT_SESSIONS_YAML:-$WS/.mam/agent-sessions.yaml}" +export PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" + +usage() { + cat <<'USAGE' +Usage: bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh [-y|--yes] [--dry-run] + +Options: + -y, --yes Confirm purging and recreating the control root's running agents + --dry-run Inspect targets, skipped rows, spec progress and blockers without changing anything + -h, --help Show this help +USAGE +} + +YES=0 +DRY_RUN=0 +while [ $# -gt 0 ]; do + case "$1" in + -y|--yes) YES=1; shift ;; + --dry-run) DRY_RUN=1; shift ;; + -h|--help) usage; exit 0 ;; + *) echo "ERROR: unknown arg: $1" >&2; usage >&2; exit 2 ;; + esac +done + +if [ "$DRY_RUN" != "1" ] && [ "$YES" != "1" ] && [ ! -t 0 ]; then + echo "ERROR: not a terminal; run with -y/--yes (or --dry-run to inspect)" >&2 + exit 3 +fi + +MAM="$WS/.mam" +MARKER="${MAM_LOOP_MARKER:-$MAM/loop-guard-active}" +SPEC_FILE="$MAM/reset-spec.json" + +# shellcheck disable=SC1091 +source "$SKILL_DIR/multi-agent-mux-loop/scripts/loop_lock.sh" + +if [ "$DRY_RUN" = "1" ]; then + if [ -x "$WS/.venv/bin/python" ]; then + PY="$WS/.venv/bin/python" + elif command -v python3 >/dev/null 2>&1; then + PY="python3" + else + PY="python" + fi + DRY_ARGS=() + if [ -f "$MAM/update-in-progress" ]; then + upd_st="$(_mam_lock_holder_state "$MAM/update-in-progress")" + case "$upd_st" in + stale) echo "NOTE: stale update marker $MAM/update-in-progress (its update is gone); not a reason to refuse" ;; + *) DRY_ARGS+=(--refusal "an update of $WS is in progress (pid $(_mam_lock_field "$MAM/update-in-progress" pid), holder $upd_st); wait for it to finish") ;; + esac + fi + if [ -f "$MARKER" ]; then + loop_st="$(_mam_lock_holder_state "$MARKER")" + case "$loop_st" in + stale) echo "NOTE: stale loop marker $MARKER (its loop is gone); not a reason to refuse" ;; + *) DRY_ARGS+=(--refusal "a review loop or another reset is running (pid $(_mam_lock_field "$MARKER" pid), holder $loop_st); wait for it to end") ;; + esac + fi + dry_rc=0 + "$PY" -m lib_py.reset dry-run --workspace "$WS" ${DRY_ARGS[@]+"${DRY_ARGS[@]}"} || dry_rc=$? + if [ "$dry_rc" -eq 0 ]; then + exit 0 + fi + exit 1 +fi + +# shellcheck disable=SC1091 +source "$WS/.agents/skills/lib.sh" +PY="$(_delegate_py_bin)" + +mam_update_admit "$WS" || exit 1 +_acquire_rc=0 +mam_acquire_loop_lock "$MARKER" || _acquire_rc=$? +case "$_acquire_rc" in + 0) ;; + 3) + mam_update_admit_release + _holder_pid="$(_mam_lock_field "$MARKER" "pid" 2>/dev/null || echo "unknown")" + echo "ERROR: reset refused; nothing was changed:" >&2 + echo " - a review loop or another reset is running (pid $_holder_pid); wait for it to end" >&2 + exit 1 + ;; + *) + mam_update_admit_release + echo "ERROR: reset refused; nothing was changed:" >&2 + echo " - cannot determine whether a review loop or another reset is running (ps unavailable?)" >&2 + exit 1 + ;; +esac +_release_reset_lock() { + mam_release_loop_lock "$MARKER" || true +} +_on_reset_signal() { + local sig_rc="$1" + _release_reset_lock + exit "$sig_rc" +} +trap _release_reset_lock EXIT +trap '_on_reset_signal 130' INT +trap '_on_reset_signal 143' TERM +trap '_on_reset_signal 129' HUP +mam_update_admit_release + +STATE_JSON="" +if ! STATE_JSON="$(load_state_json strict)"; then + echo "ERROR: reset refused; nothing was changed:" >&2 + echo " - the registry $AGENT_SESSIONS_YAML cannot be read; fix it first" >&2 + exit 1 +fi + +pre_out="" +pre_rc=0 +pre_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset precheck --workspace "$WS")" || pre_rc=$? +if [ "$pre_rc" -ne 0 ]; then + echo "ERROR: reset refused; nothing was changed:" >&2 + while IFS= read -r line; do + [ -n "$line" ] && echo " - $line" >&2 + done <<< "${pre_out:-lib_py.reset precheck failed (rc $pre_rc)}" + exit 1 +fi + +if [ -f "$SPEC_FILE" ]; then + MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset dry-run --workspace "$WS" || exit 1 + if [ "$YES" != "1" ]; then + ans="" + read -r -p "Resume reset from $SPEC_FILE? [y/N]: " ans || ans="" + case "$ans" in + y|Y|yes|Yes|YES) ;; + *) echo "Reset cancelled; nothing was changed"; exit 0 ;; + esac + fi +else + PLAN_JSON="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset plan --workspace "$WS")" || exit 1 + TARGET_COUNT="$(printf '%s' "$PLAN_JSON" | "$PY" -c 'import json,sys; print(len(json.load(sys.stdin).get("targets", [])))')" + MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset dry-run --workspace "$WS" || exit 1 + if [ "$TARGET_COUNT" = "0" ]; then + echo "nothing to reset" + exit 0 + fi + if [ "$YES" != "1" ]; then + ans="" + read -r -p "Purge and recreate $TARGET_COUNT agents? [y/N]: " ans || ans="" + case "$ans" in + y|Y|yes|Yes|YES) ;; + *) echo "Reset cancelled; nothing was changed"; exit 0 ;; + esac + fi + printf '%s' "$PLAN_JSON" | "$PY" -m lib_py.reset write-spec --workspace "$WS" || exit 1 +fi + +ENTRIES_OUT="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --warn-new)" || exit 1 +ENTRY_NAMES=() +while IFS= read -r line; do + [ -n "$line" ] || continue + eval "$line" + ENTRY_NAMES+=("$R_NAME") +done <<< "$ENTRIES_OUT" + +SKIPPED_ARGS=() +FAILED_SPEC="" + +for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do + if ! STATE_JSON="$(load_state_json strict)"; then + FAILED_SPEC="$target_name:stop:1" + break + fi + entry_line="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --session "$target_name")" || { + FAILED_SPEC="$target_name:stop:1" + break + } + eval "$entry_line" + + case "$R_STATE" in + done|created) + continue + ;; + pending) + STOP_CMD=(bash "$SKILL_DIR/multi-agent-mux-stop/scripts/stop_session.sh" --session "$R_NAME" --agent "$R_AGENT") + if [ -n "$R_SERVER" ]; then + STOP_CMD+=(--herdr-session "$R_SERVER") + fi + STOP_CMD+=(--reason reset) + + echo "reset [$R_NAME]: ${STOP_CMD[*]}" + stop_rc=0 + "${STOP_CMD[@]}" || stop_rc=$? + if [ "$stop_rc" -ne 0 ]; then + FAILED_SPEC="$R_NAME:stop:$stop_rc" + break + fi + + if ! STATE_JSON="$(load_state_json strict)"; then + FAILED_SPEC="$R_NAME:stop:1" + break + fi + if [ -z "$R_OLD_ID" ]; then + MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset record-id --workspace "$WS" --session "$R_NAME" || { + FAILED_SPEC="$R_NAME:stop:1" + break + } + fi + entry_line="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --session "$R_NAME")" || { + FAILED_SPEC="$R_NAME:stop:1" + break + } + eval "$entry_line" + if [ -z "$R_ROW_ID" ]; then + echo "skipped $R_NAME: no conversation id is recorded, so a purge would drop the row and leave the conversation on disk; resume it, or purge it by hand (stop_session.sh --session $R_NAME --purge-conversation --yes) and rerun the reset to recreate it" + SKIPPED_ARGS+=(--skipped "$R_NAME:no_id") + continue + fi + + echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes" + purge_rc=0 + "${STOP_CMD[@]}" --purge-conversation --yes || purge_rc=$? + if [ "$purge_rc" -eq 4 ]; then + echo "skipped $R_NAME: its purge was refused (stop exit 4); the row (now stopped, id recorded) and the conversation are kept; the processes above hold it - they are not killed; rerun the reset once they end" + SKIPPED_ARGS+=(--skipped "$R_NAME:purge_refused") + continue + fi + if [ "$purge_rc" -ne 0 ]; then + FAILED_SPEC="$R_NAME:stop:$purge_rc" + break + fi + R_STATE="purged" + ;; + esac + + if [ "$R_STATE" = "purged" ]; then + CREATE_CMD=(bash "$SKILL_DIR/multi-agent-mux-create/scripts/create_session.sh" --workspace "$WS" --agent "$R_AGENT" --role "$R_ROLE" --session "$R_NAME") + if [ -n "$R_SERVER" ]; then + CREATE_CMD+=(--herdr-session "$R_SERVER") + fi + if [ -n "$R_LABEL" ]; then + CREATE_CMD+=(--herdr-workspace "$R_LABEL") + fi + + echo "reset [$R_NAME]: ${CREATE_CMD[*]}" + create_rc=0 + "${CREATE_CMD[@]}" || create_rc=$? + if [ "$create_rc" -ne 0 ]; then + FAILED_SPEC="$R_NAME:create:$create_rc" + break + fi + fi +done + +STATE_JSON="$(load_state_json strict)" || exit 1 +TABLE_ARGS=(--workspace "$WS") +if [ ${#SKIPPED_ARGS[@]} -gt 0 ]; then + TABLE_ARGS+=("${SKIPPED_ARGS[@]}") +fi +if [ -n "$FAILED_SPEC" ]; then + TABLE_ARGS+=(--failed "$FAILED_SPEC") +fi +table_rc=0 +MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset table "${TABLE_ARGS[@]}" || table_rc=$? +case "$table_rc" in + 0) + rm -f "$SPEC_FILE" + exit 0 + ;; + 4) + exit 4 + ;; + 5) + exit 5 + ;; + *) + exit 1 + ;; +esac diff --git a/README.ko.md b/README.ko.md index 2e6832d..8eb829e 100644 --- a/README.ko.md +++ b/README.ko.md @@ -57,6 +57,7 @@ bash deploy/install.sh | **`multi-agent-mux-loop`** | 계획·구현·리뷰: 선택적인 planner 와 creator, 그다음 선택한 리뷰어(`--reviewer A,B`, 또는 `--all-reviewer` 로 creator 를 뺀 실행 중인 리뷰어 전원)가 모두 `PASS` 를 낼 때까지(또는 `--max-loop` 까지). 둘 다 없으면 creator 가 스스로 리뷰합니다. 루프 밖에서 끝난 작업은 `--review-only` 로 리뷰합니다. 통과한 리포트는 `.agents/reports//` 으로 옮겨 보관합니다. | | **`multi-agent-mux-orc-onboard`** | 오케스트레이터 자신의 세션 ID 를 등록해, 세션 탐색이 그것을 하위 에이전트로 착각하지 않게 합니다. | | **`multi-agent-mux-update`** | 설치된 작업 공간을 제자리에서 업데이트합니다. 실행 중인 세션·열린 잡·루프가 있으면 거부하고, 모니터를 멈췄다가 원래대로 되돌리며, 설치가 실패하면 되돌리고, 새 코드로 레지스트리를 변환합니다. | +| **`multi-agent-mux-reset`** | 제어 루트의 실행 중인 상주 에이전트를 하나씩 정리(purge)하고 같은 이름·에이전트·역할·herdr 서버·작업 공간 라벨로 다시 만들며 온보딩합니다(`.mam/reset-spec.json` 으로 중단된 지점부터 재개). | --- diff --git a/README.md b/README.md index 9d596f1..c2f974c 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,7 @@ Every skill lives under `.agents/skills/` with a `SKILL.md` that documents its o | **`multi-agent-mux-loop`** | Plan, implement, review: an optional planner and a creator, then the selected reviewers (`--reviewer A,B`, or `--all-reviewer` for every running reviewer except the creator) until all of them `PASS` or `--max-loop` is reached. With neither option the creator reviews its own work. `--review-only` reviews work finished outside the loop. Passing reports are promoted to `.agents/reports//`. | | **`multi-agent-mux-orc-onboard`** | Registers the orchestrator's own session id, so session discovery never mistakes it for a sub-agent. | | **`multi-agent-mux-update`** | Updates an installed workspace in place: refuses while sessions, open jobs or a loop are live; stops and restores the monitor; rolls back on a failed install; converts the registry with the new code. | +| **`multi-agent-mux-reset`** | Purges the control root's running resident agents and recreates each with the same name, agent, role, herdr server and workspace label, one at a time, with onboarding (`.mam/reset-spec.json` for resumable reruns). | --- diff --git a/deploy/gitea-ci.yml b/deploy/gitea-ci.yml index 1e33c70..178f999 100644 --- a/deploy/gitea-ci.yml +++ b/deploy/gitea-ci.yml @@ -42,6 +42,7 @@ jobs: shellcheck deploy/install_mam.sh shellcheck deploy/generate-env.sh shellcheck .agents/skills/multi-agent-mux-update/scripts/*.sh + shellcheck .agents/skills/multi-agent-mux-reset/scripts/*.sh shellcheck deploy/remove.sh echo "✅ ShellCheck completed successfully." diff --git a/deploy/install.sh b/deploy/install.sh index 4f7ace2..a047099 100644 --- a/deploy/install.sh +++ b/deploy/install.sh @@ -668,5 +668,8 @@ cat <. +""" +from __future__ import annotations + +import json +import os +import sqlite3 +import subprocess +import sys +import uuid as uuidlib +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +SKILLS = REPO_ROOT / ".agents" / "skills" +if str(SKILLS) not in sys.path: + sys.path.insert(0, str(SKILLS)) + +from lib_py.agents.base import DiscoveryContext # noqa: E402 +from lib_py.agents.registry import get_adapter # noqa: E402 + +pytestmark = pytest.mark.usefixtures("own_workspace_root") + + +def _clean_env(extra: dict | None = None) -> dict: + env = { + k: v + for k, v in os.environ.items() + if not k.startswith("XDG_") + and not k.startswith("HERDR_") + and k + not in ( + "CLAUDE_CODE_SESSION_ID", + "ANTIGRAVITY_CONVERSATION_ID", + "HERMES_SESSION_ID", + "GROK_SESSION_ID", + "OPENCODE_SESSION_ID", + "CLAUDE_CONFIG_DIR", + "CLAUDE_PROJECT_DIR", + "CODEX_HOME", + ) + } + env["PYTHONPATH"] = str(SKILLS) + if extra: + env.update(extra) + return env + + +def _write_fake_lsof(bindir: Path, mapping: dict[str, list[int]]) -> Path: + """Fake lsof that prints mapped PIDs when any argument matches a key substring.""" + bindir.mkdir(parents=True, exist_ok=True) + cases = [] + for needle, pids in mapping.items(): + echoes = "\n".join(f" echo {p}" for p in pids) + cases.append(f" *{needle}*)\n{echoes}\n found=1 ;;") + body = "\n".join(cases) + exe = bindir / "lsof" + exe.write_text( + "#!/bin/sh\n" + "found=0\n" + 'for a in "$@"; do\n' + ' case "$a" in\n' + f"{body}\n" + " esac\n" + "done\n" + '[ "$found" = "1" ] && exit 0\n' + "exit 1\n", + encoding="utf-8", + ) + exe.chmod(0o755) + return exe + + +def _materialize_artifact(agent: str, conv: str, ctx: DiscoveryContext) -> str: + adapter = get_adapter(agent) + if agent == "agy": + # Hold the presence lock so agy's in_use_paths (_artifact_paths) is exercised too. + paths = adapter._artifact_paths(conv, ctx) + target = paths[-1] + else: + target = adapter.artifact_path(conv, ctx) + Path(target).parent.mkdir(parents=True, exist_ok=True) + Path(target).write_text("{}\n", encoding="utf-8") + return target + + +@pytest.mark.parametrize("agent", ["claude", "agy", "codex", "grok", "hermes"]) +def test_in_use_hints_list_lsof_holders(tmp_path, monkeypatch, agent): + home = tmp_path / "home" + ws = tmp_path / "ws" + bindir = tmp_path / "bin" + no_lsof_bin = tmp_path / "no-lsof-bin" + for d in (home, ws, bindir, no_lsof_bin): + d.mkdir(parents=True, exist_ok=True) + + # Populate no_lsof_bin with symlinks to basic tools (ps) but not lsof. + for tool in ("ps", "sh", "bash"): + for sysdir in ("/bin", "/usr/bin"): + cand = Path(sysdir) / tool + if cand.exists() and not (no_lsof_bin / tool).exists(): + (no_lsof_bin / tool).symlink_to(cand) + + conv = str(uuidlib.uuid4()) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("HOME_DIR", str(home)) + monkeypatch.setenv("CODEX_HOME", str(home / ".codex")) + ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home)) + target = _materialize_artifact(agent, conv, ctx) + adapter = get_adapter(agent) + + holder = subprocess.Popen(["sleep", "300"]) + try: + _write_fake_lsof(bindir, {conv: [holder.pid]}) + env = _clean_env({ + "HOME": str(home), + "HOME_DIR": str(home), + "CODEX_HOME": str(home / ".codex"), + "PATH": f"{bindir}:{os.environ.get('PATH', '')}", + }) + res = subprocess.run( + [sys.executable, "-m", "lib_py.agents", "in-use", agent, conv, str(ws)], + capture_output=True, + text=True, + env=env, + timeout=30, + ) + assert res.returncode == 4, res.stdout + res.stderr + assert json.loads(res.stdout) == {"agent": agent, "in_use": True} + assert f"ps -axww | grep {conv}" in res.stderr + assert f"held open by pid {holder.pid}:" in res.stderr + assert "sleep 300" in res.stderr + assert "not stopped by MAM" in res.stderr + + monkeypatch.setenv("PATH", env["PATH"]) + hints = adapter.in_use_hints(conv, ctx) + if agent == "claude": + assert hints[0].startswith("a claude process whose session file under ") + assert f"ps -axww | grep {conv}" in hints[1] + else: + assert f"ps -axww | grep {conv}" in hints[0] + assert any(f"held open by pid {holder.pid}:" in h and "sleep 300" in h for h in hints) + assert target in adapter.in_use_paths(conv, ctx) + finally: + holder.kill() + holder.wait() + + +def test_in_use_hints_no_lsof_unchanged(tmp_path, monkeypatch): + """§5.1 regression control: without lsof on PATH, hints match pre-#78 lines.""" + home = tmp_path / "home" + ws = tmp_path / "ws" + no_lsof_bin = tmp_path / "no-lsof-bin" + for d in (home, ws, no_lsof_bin): + d.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("HOME_DIR", str(home)) + monkeypatch.setenv("CODEX_HOME", str(home / ".codex")) + monkeypatch.setenv("PATH", str(no_lsof_bin)) + + for agent in ("claude", "agy", "codex", "grok", "hermes"): + conv = str(uuidlib.uuid4()) + ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home)) + _materialize_artifact(agent, conv, ctx) + adapter = get_adapter(agent) + hints = adapter.in_use_hints(conv, ctx) + assert not any("held open by pid" in h for h in hints) + assert any(f"ps -axww | grep {conv}" in h for h in hints) + + +def test_in_use_hints_opencode_skip_service(tmp_path, monkeypatch): + home = tmp_path / "home" + ws = tmp_path / "ws" + bindir = tmp_path / "bin" + for d in (home, ws, bindir): + d.mkdir(parents=True, exist_ok=True) + + db = home / ".local" / "share" / "opencode" / "opencode.db" + db.parent.mkdir(parents=True, exist_ok=True) + db.write_text("", encoding="utf-8") + + adapter = get_adapter("opencode") + monkeypatch.setattr(adapter, "_bin", lambda ctx: None) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("HOME_DIR", str(home)) + ctx = DiscoveryContext(workspace=str(ws), agent_name="opencode", home_dir=str(home)) + + service_proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(300)", "serve", "--service"] + ) + holder_proc = subprocess.Popen(["sleep", "300"]) + try: + _write_fake_lsof(bindir, {"opencode.db": [service_proc.pid, holder_proc.pid]}) + monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}") + + hints = adapter.in_use_hints("ses_test78", ctx) + assert hints[0] == "any process with it on its command line (an editor, a grep, ...): ps -axww | grep ses_test78" + assert any(f"held open by pid {holder_proc.pid}:" in h and "sleep 300" in h for h in hints) + assert not any(f"pid {service_proc.pid}" in h or "serve --service" in h for h in hints) + assert hints[-1] == f"another opencode session holding {db}: lsof {db}" + finally: + for p in (service_proc, holder_proc): + p.kill() + p.wait() + + +def test_stop_purge_refusal_prints_holders(mam_sandbox, mock_herdr, monkeypatch): + conv = str(uuidlib.uuid4()) + name = "issue78-reviewer-claude" + ws = mam_sandbox + ctx = DiscoveryContext(workspace=str(ws), agent_name="claude", home_dir=str(mam_sandbox)) + transcript = Path(get_adapter("claude").artifact_path(conv, ctx)) + transcript.parent.mkdir(parents=True, exist_ok=True) + transcript.write_text(json.dumps({"sessionId": conv}) + "\n", encoding="utf-8") + + lib = mam_sandbox / ".agents" / "skills" / "lib.sh" + yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" + mutation = ( + f"d['herdr_sessions'] = [{{'name': {name!r}, 'status': 'running', 'agent': 'claude', " + f"'role': 'reviewer', 'claude_session_id_own': {conv!r}, " + f"'pane': {{'cwd': {os.path.realpath(str(ws))!r}}}}}]\n" + ) + res_init = subprocess.run( + ["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'], + input=mutation, + capture_output=True, + text=True, + env=_clean_env({ + "HOME": str(mam_sandbox), + "HOME_DIR": str(mam_sandbox), + "WORKSPACE_ROOT": str(mam_sandbox), + "AGENT_SESSIONS_YAML": str(yaml_path), + }), + ) + assert res_init.returncode == 0, res_init.stdout + res_init.stderr + + holder = subprocess.Popen(["sleep", "300"]) + try: + bindir = mam_sandbox / ".local" / "bin" + _write_fake_lsof(bindir, {conv: [holder.pid]}) + stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" + env = _clean_env({ + "HOME": str(mam_sandbox), + "HOME_DIR": str(mam_sandbox), + "CLAUDE_PROJECT_DIR": str(mam_sandbox / ".claude" / "projects"), + "WORKSPACE_ROOT": str(mam_sandbox), + "AGENT_SESSIONS_YAML": str(yaml_path), + "PATH": f"{bindir}:{os.environ.get('PATH', '')}", + }) + res = subprocess.run( + ["bash", str(stop_sh), "--session", name, "--agent", "claude", "--purge-conversation", "--yes"], + capture_output=True, + text=True, + cwd=str(mam_sandbox), + env=env, + timeout=60, + ) + assert res.returncode == 4, res.stdout + res.stderr + assert f"held open by pid {holder.pid}:" in res.stderr + assert "sleep 300" in res.stderr + assert transcript.exists() + assert name in yaml_path.read_text(encoding="utf-8") + finally: + holder.kill() + holder.wait() + + +@pytest.mark.parametrize("delete_ok", [True, False], ids=["delete_ok", "delete_fails"]) +def test_codex_purge_removes_tui_thread_capability(tmp_path, monkeypatch, delete_ok): + home = tmp_path / "home" + codex_home = tmp_path / "codex-home" + ws = tmp_path / "ws" + bindir = tmp_path / "bin" + for d in (home, codex_home, ws, bindir): + d.mkdir(parents=True, exist_ok=True) + + conv = str(uuidlib.uuid4()) + other = str(uuidlib.uuid4()) + + rollout = codex_home / "sessions" / "2026" / "10" / "28" / f"rollout-2026-10-28T00-00-00-{conv}.jsonl" + rollout.parent.mkdir(parents=True, exist_ok=True) + rollout.write_text(json.dumps({"payload": {"id": conv, "cwd": str(ws)}}) + "\n", encoding="utf-8") + + cap_dir = codex_home / "tui-thread-reference-capabilities" + cap_dir.mkdir(parents=True, exist_ok=True) + cap_mine = cap_dir / conv + cap_other = cap_dir / other + cap_mine.write_text("cap-mine", encoding="utf-8") + cap_other.write_text("cap-other", encoding="utf-8") + + state_db = codex_home / "state_5.sqlite" + conn = sqlite3.connect(str(state_db)) + conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT, cwd TEXT)") + conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (conv, str(rollout), str(ws))) + conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (other, "", str(ws))) + conn.commit() + conn.close() + + fake_codex = bindir / "codex" + if delete_ok: + # Simulate `codex delete --force` removing rollout + DB row, leaving tui-thread-reference-capabilities. + fake_codex.write_text( + f"#!{sys.executable}\n" + "import os, sqlite3, sys\n" + "if len(sys.argv) >= 3 and sys.argv[1] == 'delete':\n" + f" p = {str(rollout)!r}\n" + " if os.path.exists(p):\n" + " os.remove(p)\n" + f" conn = sqlite3.connect({str(state_db)!r})\n" + " conn.execute('DELETE FROM threads WHERE id=?', (sys.argv[2],))\n" + " conn.commit()\n" + " conn.close()\n" + " sys.exit(0)\n" + "sys.exit(1)\n", + encoding="utf-8", + ) + else: + fake_codex.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8") + fake_codex.chmod(0o755) + + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("HOME_DIR", str(home)) + monkeypatch.setenv("CODEX_HOME", str(codex_home)) + monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}") + + adapter = get_adapter("codex") + ctx = DiscoveryContext(workspace=str(ws), agent_name="codex", home_dir=str(home)) + assert str(cap_mine) in adapter.residual_artifacts(conv, ctx) + + purged = adapter.purge_artifacts(conv, ctx) + + assert str(cap_mine) in purged + assert not cap_mine.exists() + assert cap_other.exists() + assert not rollout.exists() + assert adapter.residual_artifacts(conv, ctx) == [] + assert str(cap_other) in adapter.residual_artifacts(other, ctx) diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py new file mode 100644 index 0000000..8befe59 --- /dev/null +++ b/tests/test_issue78_reset.py @@ -0,0 +1,1416 @@ +"""Issue #78: multi-agent-mux-reset — unit tests for lib_py.reset and +end-to-end tests for reset.sh. +""" +from __future__ import annotations + +import ast +import json +import os +import re +import shlex +import subprocess +import sys +import time +from pathlib import Path +from typing import Dict, List, Optional + +import pytest +import yaml + +REPO = Path(__file__).resolve().parents[1] +SKILLS = REPO / ".agents" / "skills" +if str(SKILLS) not in sys.path: + sys.path.insert(0, str(SKILLS)) + +UA = "a7800001-0000-4000-8000-000000000001" +UB = "b7800002-0000-4000-8000-000000000002" +UC = "c7800003-0000-4000-8000-000000000003" +UD = "d7800004-0000-4000-8000-000000000004" +UORC = "e7800005-0000-4000-8000-000000000005" + +pytestmark = pytest.mark.usefixtures("own_workspace_root") + + +# --------------------------------------------------------------------------- Helpers + +def _row( + name: str, + agent: str, + cwd: Path, + role: str = "reviewer", + status: str = "running", + created_at: str = "2026-10-01T00:00:00Z", + server: str = "default", + label: str = "ws", + own_id: Optional[str] = None, + pid: int = 4242, + epoch: int = 1700000000, +) -> dict: + from lib_py.agents.registry import own_key + + key = own_key(agent) if agent in ("claude", "agy", "codex", "grok", "hermes", "opencode") else "unknown_id" + return { + "name": name, + "status": status, + "role": role, + "agent": agent, + "herdr_session": server, + "herdr_server": server, + "herdr_workspace": label, + "herdr_session_created_at": created_at, + "herdr_session_epoch": epoch, + key: own_id, + "session_id_source": "discovered" if own_id else "pending-discovery", + "session_id_verified": bool(own_id), + "pane": {"cwd": str(cwd), "pid": pid, "cmd": agent}, + } + + +def _run_reset_py( + args: List[str], + state: Optional[dict] = None, + stdin_text: Optional[str] = None, + env_extra: Optional[dict] = None, +): + env = dict(os.environ) + env["PYTHONPATH"] = str(SKILLS) + env["PYTHONDONTWRITEBYTECODE"] = "1" + if state is not None: + env["MAM_STATE_JSON"] = json.dumps(state) + if env_extra: + env.update(env_extra) + return subprocess.run( + [sys.executable, "-m", "lib_py.reset", *args], + env=env, + input=stdin_text, + capture_output=True, + text=True, + timeout=15, + ) + + +def _write_rows(sandbox: Path, rows: list, orchestrator_uuids: Optional[List[str]] = None) -> Path: + lib = sandbox / ".agents" / "skills" / "lib.sh" + yaml_path = sandbox / ".mam" / "agent-sessions.yaml" + mutation = f"d['herdr_sessions'] = {repr(rows)}\n" + if orchestrator_uuids is not None: + mutation += f"d['orchestrator_uuids'] = {repr(orchestrator_uuids)}\n" + res = subprocess.run( + ["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'], + input=mutation, + capture_output=True, + text=True, + cwd=str(sandbox), + timeout=60, + ) + assert res.returncode == 0, res.stdout + res.stderr + return yaml_path + + +def _read_rows(sandbox: Path) -> Dict[str, dict]: + yaml_path = sandbox / ".mam" / "agent-sessions.yaml" + data = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) or {} + return {r["name"]: r for r in data.get("herdr_sessions") or []} + + +def _db_session_names(sandbox: Path) -> Dict[str, str]: + import sqlite3 + + db_path = sandbox / ".mam" / "agent-sessions.db" + if not db_path.exists(): + return {} + conn = sqlite3.connect(str(db_path)) + try: + return dict(conn.execute("SELECT name, status FROM sessions").fetchall()) + finally: + conn.close() + + +def _claude_transcript(sandbox: Path, ws: Path, uuid: str, mtime: Optional[float] = None) -> Path: + from lib_py.verify_session import workspace_key + + pdir = sandbox / ".claude" / "projects" / workspace_key(str(ws)) + pdir.mkdir(parents=True, exist_ok=True) + f = pdir / f"{uuid}.jsonl" + f.write_text(json.dumps({"sessionId": uuid, "cwd": str(ws), "type": "user"}) + "\n", encoding="utf-8") + if mtime is not None: + os.utime(f, (mtime, mtime)) + return f + + +def _agy_full_artifacts(home: Path, uuid: str, mtime: Optional[float] = None) -> List[Path]: + import sqlite3 + + base = Path(home) / ".gemini" / "antigravity-cli" + db = base / "conversations" / f"{uuid}.db" + db.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db)) + conn.execute("CREATE TABLE steps (id INTEGER)") + conn.execute("INSERT INTO steps VALUES (1)") + conn.commit() + conn.close() + if mtime is not None: + os.utime(db, (mtime, mtime)) + wal = Path(str(db) + "-wal") + shm = Path(str(db) + "-shm") + wal.write_text("wal", encoding="utf-8") + shm.write_text("shm", encoding="utf-8") + brain = base / "brain" / uuid + brain.mkdir(parents=True, exist_ok=True) + (brain / "note.md").write_text("note", encoding="utf-8") + ann = base / "annotations" / f"{uuid}.pbtxt" + ann.parent.mkdir(parents=True, exist_ok=True) + ann.write_text("pbtxt", encoding="utf-8") + lock = base / "presence" / f"{uuid}.lock" + lock.parent.mkdir(parents=True, exist_ok=True) + lock.write_text("", encoding="utf-8") + return [db, wal, shm, brain, ann, lock] + + +def _agy_cache(home: Path, mapping: dict) -> None: + lc = Path(home) / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json" + lc.parent.mkdir(parents=True, exist_ok=True) + lc.write_text( + json.dumps({os.path.realpath(str(k)): v for k, v in mapping.items()}), + encoding="utf-8", + ) + + +def _fake_lsof_multi(bindir: Path, mapping: dict) -> Path: + cases = "\n".join(f" *{k}*) echo {v}; found=1 ;;" for k, v in mapping.items()) + exe = Path(bindir) / "lsof" + exe.write_text( + "#!/bin/sh\n" + "found=0\n" + 'for a in "$@"; do\n' + ' case "$a" in\n' + f"{cases}\n" + " esac\n" + "done\n" + '[ "$found" = "1" ] && exit 0\n' + "exit 1\n", + encoding="utf-8", + ) + exe.chmod(0o755) + return exe + + +def _mock_herdr_agent( + state_file: Path, + name: str, + cwd: Path, + pid: int = 4242, + epoch: int = 1700000000, + agent: str = "claude", + pane_id: str = "", +) -> None: + state = json.loads(Path(state_file).read_text(encoding="utf-8")) + pid_str = pane_id or f"w1:p_{name}" + state.setdefault("agents", {})[name] = { + "pid": pid, + "cwd": str(cwd), + "command": agent, + "agent": agent, + "pane_id": pid_str, + "workspace_id": "w1", + "created": epoch, + "buffer": f"Ready in {cwd}\n>", + } + Path(state_file).write_text(json.dumps(state), encoding="utf-8") + + +def _wait_reconcile_idle(sandbox: Path, timeout: float = 30.0) -> None: + """Wait until any background `reconcile.sh` spawned under `sandbox` finishes. + Uses `ps -eo pid=,args=` (never `pgrep -f`) so it never matches itself. + """ + sb = os.path.realpath(str(sandbox)) + deadline = time.time() + timeout + while time.time() < deadline: + res = subprocess.run( + ["ps", "-eo", "pid=,args="], + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=5, + ) + active = [] + for line in (res.stdout or "").splitlines(): + line = line.strip() + if not line: + continue + parts = line.split(None, 1) + if len(parts) < 2 or not parts[0].isdigit(): + continue + pid = int(parts[0]) + if pid == os.getpid(): + continue + args = parts[1] + if "reconcile.sh" in args and (sb in args or str(sandbox) in args): + active.append((pid, args)) + if not active: + return + time.sleep(0.2) + pytest.fail(f"reconcile.sh still running under {sandbox} after {timeout}s: {active}") + + +def _install_herdr_spy(sandbox: Path) -> Path: + """Wrap sandbox/bin/herdr to log verb sequence, spec-at-first-kill, and optional + run_loop.sh probe at first agent-start. + """ + bin_dir = sandbox / "bin" + herdr_bin = bin_dir / "herdr" + herdr_real = bin_dir / "herdr.real" + if not herdr_real.exists(): + herdr_bin.rename(herdr_real) + spy_log = sandbox / "herdr_spy.jsonl" + probe_file = sandbox / "herdr_spy_loop_probe.json" + + wrapper = f"""#!{sys.executable} +import json, os, subprocess, sys + +real = {repr(str(herdr_real))} +spy_log = {repr(str(spy_log))} +probe_file = {repr(str(probe_file))} +sandbox = {repr(str(sandbox))} + +raw_args = sys.argv[1:] +args = list(raw_args) +while args and args[0] in ("-L", "--server", "-s", "--session"): + args = args[2:] if len(args) > 1 else args[1:] + +if args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls"): + verb = args[0] +else: + verb = " ".join(args[:2]) if len(args) >= 2 else (args[0] if args else "") +rec = {{"verb": verb, "args": args}} + +if verb in ("kill-session", "pane close"): + spec_p = os.path.join(sandbox, ".mam", "reset-spec.json") + rec["spec_exists"] = os.path.isfile(spec_p) + if rec["spec_exists"]: + try: + with open(spec_p, encoding="utf-8") as fh: + rec["spec"] = json.load(fh) + except Exception as exc: + rec["spec_error"] = str(exc) + +if verb == "agent start" and os.environ.get("SPY_PROBE_RUN_LOOP") == "1" and not os.path.exists(probe_file): + env = dict(os.environ, MAM_LOOP_NO_FREEZE="1") + loop_sh = os.path.join(sandbox, ".agents", "skills", "multi-agent-mux-loop", "scripts", "run_loop.sh") + res = subprocess.run( + ["bash", loop_sh, "--creator", "x", "--task", "t"], + cwd=sandbox, + env=env, + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=30, + ) + with open(probe_file, "w", encoding="utf-8") as pf: + json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr}}, pf) + +with open(spy_log, "a", encoding="utf-8") as lf: + lf.write(json.dumps(rec) + "\\n") + +os.execv(real, [real, *raw_args]) +""" + herdr_bin.write_text(wrapper, encoding="utf-8") + herdr_bin.chmod(0o755) + return spy_log + + +def _read_spy_log(spy_log: Path) -> List[dict]: + if not spy_log.exists(): + return [] + return [json.loads(line) for line in spy_log.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def _run_reset( + sandbox: Path, + *args: str, + stdin_text: Optional[str] = None, + env_extra: Optional[dict] = None, +) -> subprocess.CompletedProcess: + script = sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh" + env = dict( + os.environ, + SKS_QUIESCENT_TRIES="0", + SKS_DIALOG_TIMEOUT="0", + SKS_BUSY_TIMEOUT="0", + MAM_ONBOARD_ACK_S="3", + MAM_ONBOARD_WAIT_S="3", + MAM_NO_SUBSCRIBER_DAEMON="1", + PYTHONDONTWRITEBYTECODE="1", + ) + if env_extra: + env.update(env_extra) + return subprocess.run( + ["bash", str(script), *args], + cwd=str(sandbox), + env=env, + input=stdin_text, + stdin=subprocess.DEVNULL if stdin_text is None else None, + capture_output=True, + text=True, + timeout=180, + ) + + +# --------------------------------------------------------------------------- Unit tests: lib_py.reset + +def test_static_checks_and_skill_version(): + """§5.2-12: + - bash -n reset.sh + - ast.parse(feature_version=(3, 9)) on lib_py/reset.py + - no `kill` command in reset.sh + - multi-agent-mux-reset/SKILL.md frontmatter version == lib.sh MAM_VERSION + """ + reset_sh = SKILLS / "multi-agent-mux-reset" / "scripts" / "reset.sh" + res = subprocess.run(["bash", "-n", str(reset_sh)], capture_output=True, text=True, timeout=10) + assert res.returncode == 0, res.stderr + + src_py = (SKILLS / "lib_py" / "reset.py").read_text(encoding="utf-8") + ast.parse(src_py, filename="reset.py", feature_version=(3, 9)) + + src_sh = reset_sh.read_text(encoding="utf-8") + for lineno, line in enumerate(src_sh.splitlines(), 1): + stripped = line.strip() + if stripped.startswith("#"): + continue + assert not re.search(r"\bkill\b", stripped), f"reset.sh:{lineno} contains forbidden `kill`: {line}" + + # Self-contained version check (no import of test_version_consistency) + lib_text = (SKILLS / "lib.sh").read_text(encoding="utf-8") + m_lib = re.search(r'^MAM_VERSION="([^"]+)"', lib_text, re.M) + assert m_lib, "MAM_VERSION not found in lib.sh" + mam_version = m_lib.group(1) + + skill_text = (SKILLS / "multi-agent-mux-reset" / "SKILL.md").read_text(encoding="utf-8") + frontmatter = skill_text.split("---", 2)[1] + m_skill = re.search(r"^version:\s*(\S+)\s*$", frontmatter, re.M) + assert m_skill, "version not found in multi-agent-mux-reset/SKILL.md frontmatter" + assert m_skill.group(1) == mam_version + + +def test_select_targets_and_plan_cli(tmp_path: Path): + """§5.2-1: select_targets chooses only status==running rows at realpath(WS), + resolving symlinks, listing skipped rows with reasons and orchestrator_uuids, + and refusing running control-root rows with missing agent or empty role. + """ + ws = tmp_path / "ws" + ws.mkdir() + wt = ws / "wt" / "x" + wt.mkdir(parents=True) + sym_ws = tmp_path / "sym-ws" + sym_ws.symlink_to(ws) + + state = { + "orchestrator_uuids": [UORC], + "herdr_sessions": [ + _row("reviewer-claude-01", "claude", ws, role="planner,reviewer", own_id=UA, server="srv1", label="lbl1"), + _row("reviewer-grok-01", "grok", sym_ws, role="reviewer", own_id=UB, server="default", label="lbl2"), + _row("creator-claude-wt", "claude", wt, role="creator", own_id=UC), + _row("stopped-codex-01", "codex", ws, status="stopped", own_id=UD), + _row("dead-hermes-01", "hermes", ws, status="terminated", own_id=None), + ], + } + + res = _run_reset_py(["plan", "--workspace", str(ws)], state=state) + assert res.returncode == 0, res.stderr + plan = json.loads(res.stdout) + assert [t["name"] for t in plan["targets"]] == ["reviewer-claude-01", "reviewer-grok-01"] + assert plan["targets"][0] == { + "name": "reviewer-claude-01", + "agent": "claude", + "role": "planner,reviewer", + "herdr_server": "srv1", + "herdr_workspace": "lbl1", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + } + skipped = {s["name"]: s["reason"] for s in plan["skipped"]} + assert "not the control root" in skipped["creator-claude-wt"] + assert skipped["stopped-codex-01"] == "status stopped" + assert skipped["dead-hermes-01"] == "status terminated" + assert plan["orchestrators"] == [UORC] + assert plan["errors"] == [] + + # Empty role or unknown agent on a running control-root row causes plan & precheck to return 1 + bad_state = { + "herdr_sessions": [ + _row("bad-role-claude", "claude", ws, role="", own_id=UA), + { + "name": "mystery-row", + "status": "running", + "role": "reviewer", + "agent": "not-an-agent", + "pane": {"cwd": str(ws), "pid": 123, "cmd": "bash"}, + }, + ] + } + res_bad = _run_reset_py(["plan", "--workspace", str(ws)], state=bad_state) + assert res_bad.returncode == 1 + bad_plan = json.loads(res_bad.stdout) + assert len(bad_plan["errors"]) == 2 + + res_pre = _run_reset_py(["precheck", "--workspace", str(ws)], state=bad_state) + assert res_pre.returncode == 1 + assert "bad-role-claude" in res_pre.stdout + assert "mystery-row" in res_pre.stdout + + +def test_spec_file_write_read_and_validation(tmp_path: Path): + """D5: write-spec atomically writes .mam/reset-spec.json; corrupt, wrong-version, + or wrong-workspace spec files are rejected with 'check it, then remove it by hand'. + """ + from lib_py.reset import SpecError, read_spec + + ws = tmp_path / "ws" + ws.mkdir() + targets = [ + { + "name": "reviewer-claude-01", + "agent": "claude", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + } + ] + res = _run_reset_py( + ["write-spec", "--workspace", str(ws)], + stdin_text=json.dumps({"targets": targets}), + ) + assert res.returncode == 0, res.stderr + spath = ws / ".mam" / "reset-spec.json" + assert spath.is_file() + doc = read_spec(str(ws)) + assert doc["version"] == 1 + assert doc["workspace"] == os.path.realpath(str(ws)) + assert doc["agents"] == targets + + # Wrong workspace + other_ws = tmp_path / "other" + other_ws.mkdir() + with pytest.raises(SpecError, match="check it, then remove it by hand"): + read_spec(str(other_ws)) + + # Wrong version + spath.write_text(json.dumps({"version": 99, "workspace": str(ws), "agents": []}), encoding="utf-8") + with pytest.raises(SpecError, match="check it, then remove it by hand"): + read_spec(str(ws)) + + # Unknown or non-canonical agent in saved spec (review P2) + for bad_agent in ("not-an-agent", "CLAUDE"): + spath.write_text( + json.dumps( + { + "version": 1, + "workspace": str(ws), + "agents": [ + { + "name": "reviewer-bad-01", + "agent": bad_agent, + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + } + ], + } + ), + encoding="utf-8", + ) + with pytest.raises(SpecError, match="unknown agent"): + read_spec(str(ws)) + + # Corrupt JSON + spath.write_text("{corrupt", encoding="utf-8") + with pytest.raises(SpecError, match="check it, then remove it by hand"): + read_spec(str(ws)) + res_entries = _run_reset_py(["entries", "--workspace", str(ws)], state={}) + assert res_entries.returncode == 1 + assert "check it, then remove it by hand" in res_entries.stderr + + +def test_entries_5_states_and_resumed_row_9b(tmp_path: Path): + """D6 1-5 & §5.2-9b: + #1 row absent -> purged + #2 same created_at (even if normal stop recorded a new id onto old_id=None) -> pending + #3 new created_at and id == old_id -> pending (resumed original conversation, 9b) + #4 new created_at and id empty -> created + #5 new created_at and id != old_id -> done + """ + from lib_py.reset import write_spec + + ws = tmp_path / "ws" + ws.mkdir() + spec_agents = [ + { + "name": "s1-purged", + "agent": "claude", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "s2-same-inst-newly-recorded", + "agent": "agy", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": None, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "s3-resumed-same-id", + "agent": "claude", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UB, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "s4-created-no-id", + "agent": "hermes", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UC, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "s5-done", + "agent": "codex", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": None, + "old_created_at": "2026-10-01T00:00:00Z", + }, + ] + write_spec(str(ws), spec_agents) + + state = { + "herdr_sessions": [ + _row( + "s2-same-inst-newly-recorded", + "agy", + ws, + status="stopped", + created_at="2026-10-01T00:00:00Z", + own_id=UA, + ), + _row( + "s3-resumed-same-id", + "claude", + ws, + status="running", + created_at="2026-10-02T00:00:00Z", + own_id=UB, + ), + _row( + "s4-created-no-id", + "hermes", + ws, + status="running", + created_at="2026-10-02T00:00:00Z", + own_id=None, + ), + _row( + "s5-done", + "codex", + ws, + status="running", + created_at="2026-10-02T00:00:00Z", + own_id=UD, + ), + ] + } + + res = _run_reset_py(["entries", "--workspace", str(ws)], state=state) + assert res.returncode == 0, res.stderr + parsed = {} + for line in res.stdout.strip().splitlines(): + kv = dict(tok.split("=", 1) for tok in shlex.split(line)) + parsed[kv["R_NAME"]] = kv + + assert parsed["s1-purged"]["R_STATE"] == "purged" + assert parsed["s2-same-inst-newly-recorded"]["R_STATE"] == "pending" + assert parsed["s2-same-inst-newly-recorded"]["R_ROW_ID"] == UA + assert parsed["s3-resumed-same-id"]["R_STATE"] == "pending" + assert parsed["s4-created-no-id"]["R_STATE"] == "created" + assert parsed["s5-done"]["R_STATE"] == "done" + + res_tbl = _run_reset_py( + [ + "table", + "--workspace", + str(ws), + "--skipped", + "s2-same-inst-newly-recorded:purge_refused", + ], + state=state, + ) + assert res_tbl.returncode == 4 + rows_by_name = {} + for line in res_tbl.stdout.strip().splitlines()[1:]: + cols = line.split(None, 6) + rows_by_name[cols[0]] = cols + + assert rows_by_name["s2-same-inst-newly-recorded"][1:6] == ["agy", "-", "-", "no", "-"] + assert rows_by_name["s2-same-inst-newly-recorded"][6] == "skipped: purge refused" + assert rows_by_name["s3-resumed-same-id"][1:6] == ["claude", UB, UB, "yes", "no"] + assert rows_by_name["s4-created-no-id"][1:6] == ["hermes", UC, "-", "no", "-"] + assert rows_by_name["s4-created-no-id"][6] == "created; id not recorded yet" + assert rows_by_name["s5-done"][1:6] == ["codex", "-", UD, "yes", "yes"] + assert rows_by_name["s5-done"][6] == "done" + + # Review F1: (a) old_id=None + same instance + id recorded on row -> record-id persists old_id=UA + from lib_py.reset import read_spec + + res_rec = _run_reset_py( + ["record-id", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"], + state=state, + ) + assert res_rec.returncode == 0, res_rec.stderr + spec_after = read_spec(str(ws)) + s2_entry = next(e for e in spec_after["agents"] if e["name"] == "s2-same-inst-newly-recorded") + assert s2_entry["old_id"] == UA + + # Review F1: (b) after resuming that skipped row (new created_at, id still UA), it is classified + # as 'pending' (NOT falsely 'done') and table exits non-zero (4) with NEW!=OLD == 'no' + state["herdr_sessions"][0] = _row( + "s2-same-inst-newly-recorded", + "agy", + ws, + status="running", + created_at="2026-10-02T00:00:00Z", + own_id=UA, + ) + res_f1_entries = _run_reset_py( + ["entries", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"], + state=state, + ) + assert res_f1_entries.returncode == 0 + kv_f1 = dict(tok.split("=", 1) for tok in shlex.split(res_f1_entries.stdout.strip())) + assert kv_f1["R_STATE"] == "pending" + assert kv_f1["R_OLD_ID"] == UA + + res_f1_tbl = _run_reset_py(["table", "--workspace", str(ws)], state=state) + assert res_f1_tbl.returncode == 4 + rows_f1 = { + line.split(None, 6)[0]: line.split(None, 6) + for line in res_f1_tbl.stdout.strip().splitlines()[1:] + } + assert rows_f1["s2-same-inst-newly-recorded"][1:7] == ["agy", UA, UA, "yes", "no", "pending"] + + +# --------------------------------------------------------------------------- Integration tests: reset.sh + +def test_no_yes_non_tty_exits_3_without_touching_locks(mam_sandbox: Path, mock_herdr: Path, mock_agents): + """§5.2-3: Without --yes and with non-TTY stdin, reset.sh exits 3 before touching + any lock (.mam/loop-guard-active or .mam/update-admission.lock) or registry. + """ + _write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)]) + yaml_before = (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() + + res = _run_reset(mam_sandbox) + assert res.returncode == 3 + assert "not a terminal" in res.stderr + assert (mam_sandbox / ".mam" / "agent-sessions.yaml").read_bytes() == yaml_before + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + assert not (mam_sandbox / ".mam" / "update-admission.lock").exists() + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + + +def test_dry_run_changes_nothing_in_mam_tree(mam_sandbox: Path, mock_herdr: Path, mock_agents): + """§5.2-1 & §5.2-4: --dry-run prints targets (with all spec fields), skipped rows, + and orchestrator_uuids, exits 0, and leaves the entire .mam tree (names, sizes, + mtimes) and mock_herdr state untouched. + """ + wt = mam_sandbox / "wt" / "issue-1" + wt.mkdir(parents=True) + _write_rows( + mam_sandbox, + [ + _row("reviewer-claude-01", "claude", mam_sandbox, role="planner,reviewer", server="srv-main", label="lbl-main", own_id=UA), + _row("reviewer-grok-01", "grok", mam_sandbox, role="reviewer", server="default", label="lbl-main", own_id=UB), + _row("creator-claude-wt", "claude", wt, role="creator", own_id=UC), + _row("stopped-codex-01", "codex", mam_sandbox, status="stopped", own_id=UD), + ], + orchestrator_uuids=[UORC], + ) + mam_dir = mam_sandbox / ".mam" + + def _snapshot_mam(): + return { + str(p.relative_to(mam_dir)): (p.stat().st_size, p.stat().st_mtime_ns) + for p in sorted(mam_dir.rglob("*")) + } + + snap_before = _snapshot_mam() + yaml_before = (mam_dir / "agent-sessions.yaml").read_bytes() + + res = _run_reset(mam_sandbox, "--dry-run") + assert res.returncode == 0, res.stdout + res.stderr + assert _snapshot_mam() == snap_before + assert (mam_dir / "agent-sessions.yaml").read_bytes() == yaml_before + + out = res.stdout + assert "target: reviewer-claude-01" in out + assert "agent=claude" in out + assert "role=planner,reviewer" in out + assert "herdr_server=srv-main" in out + assert "herdr_workspace=lbl-main" in out + assert f"old_id={UA}" in out + assert "old_created_at=2026-10-01T00:00:00Z" in out + assert "target: reviewer-grok-01" in out + assert "skip: creator-claude-wt" in out and "not the control root" in out + assert "skip: stopped-codex-01 (status stopped)" in out + assert f"orchestrator {UORC} (not a row; never reset)" in out + + herdr_state = json.loads(mock_herdr.read_text(encoding="utf-8")) + assert herdr_state.get("calls") == [] + + +def test_precheck_refusals_change_nothing(mam_sandbox: Path, mock_herdr: Path, mock_agents): + """§5.2-2: Precheck refusals (open job, live loop marker, live update marker, + unrecreatable target, caller self-target) exit 1 and change nothing; stale + markers do not refuse. + """ + tf = _claude_transcript(mam_sandbox, mam_sandbox, UA) + _mock_herdr_agent(mock_herdr, "reviewer-claude-01", mam_sandbox, agent="claude") + _write_rows(mam_sandbox, [_row("reviewer-claude-01", "claude", mam_sandbox, own_id=UA)]) + spy_log = _install_herdr_spy(mam_sandbox) + + yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml" + yaml_before = yaml_path.read_bytes() + db_before = _db_session_names(mam_sandbox) + tf_before = tf.read_bytes() + + def _assert_nothing_changed(expect_loop_marker: bool = False): + assert yaml_path.read_bytes() == yaml_before + assert _db_session_names(mam_sandbox) == db_before + assert tf.read_bytes() == tf_before + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + calls = _read_spy_log(spy_log) + assert not any(c["verb"] in ("kill-session", "pane close", "agent start") for c in calls) + if not expect_loop_marker: + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + + # (a) Open job + jobs_dir = mam_sandbox / ".mam" / "jobs" + jobs_dir.mkdir(parents=True, exist_ok=True) + job_file = jobs_dir / "job78a.json" + job_file.write_text( + json.dumps({"job_id": "job78a", "agent_session": "herdr:reviewer-claude-01", "status": "running"}), + encoding="utf-8", + ) + res_job = _run_reset(mam_sandbox, "--yes") + assert res_job.returncode == 1 + assert "job78a" in res_job.stderr + assert "close --job job78a --request cancelled" in res_job.stderr + _assert_nothing_changed() + job_file.unlink() + + # Helper to get lstart for a live process + holder = subprocess.Popen(["sleep", "300"]) + try: + lstart = subprocess.run( + ["ps", "-p", str(holder.pid), "-o", "lstart="], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + lstart = " ".join(lstart.split()) + + # (b) Live loop marker + loop_marker = mam_sandbox / ".mam" / "loop-guard-active" + loop_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8") + res_loop = _run_reset(mam_sandbox, "--yes") + assert res_loop.returncode == 1 + assert "loop" in res_loop.stderr + assert str(holder.pid) in res_loop.stderr + _assert_nothing_changed(expect_loop_marker=True) + loop_marker.unlink() + + # (c) Live update marker + upd_marker = mam_sandbox / ".mam" / "update-in-progress" + upd_marker.write_text(f"pid={holder.pid}\nlstart={lstart}\nstarted=2026-10-29T00:00:00Z\n", encoding="utf-8") + res_upd = _run_reset(mam_sandbox, "--yes") + assert res_upd.returncode == 1 + assert "update" in res_upd.stderr + _assert_nothing_changed() + upd_marker.unlink() + finally: + holder.kill() + holder.wait() + + # Stale control: dead PID in loop and update markers does not refuse --dry-run + (mam_sandbox / ".mam" / "loop-guard-active").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8") + (mam_sandbox / ".mam" / "update-in-progress").write_text("pid=999999\nlstart=Thu Jan 1 00:00:00 1970\n", encoding="utf-8") + res_stale = _run_reset(mam_sandbox, "--dry-run") + assert res_stale.returncode == 0, res_stale.stdout + res_stale.stderr + (mam_sandbox / ".mam" / "loop-guard-active").unlink() + (mam_sandbox / ".mam" / "update-in-progress").unlink() + + # (d) Caller self-target (D4-4) + res_self = _run_reset(mam_sandbox, "--yes", env_extra={"CLAUDE_CODE_SESSION_ID": UA}) + assert res_self.returncode == 1 + assert "you are reviewer-claude-01, one of the targets" in res_self.stderr + _assert_nothing_changed() + + # (e) Saved spec with valid pending target first and unknown agent second (review P2) + spec_file = mam_sandbox / ".mam" / "reset-spec.json" + bad_spec_text = json.dumps( + { + "version": 1, + "workspace": str(mam_sandbox), + "written_at": "2026-10-29T00:00:00Z", + "agents": [ + { + "name": "reviewer-claude-01", + "agent": "claude", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "reviewer-bad-01", + "agent": "not-an-agent", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UB, + "old_created_at": "2026-10-01T00:00:00Z", + }, + ], + } + ) + spec_file.write_text(bad_spec_text, encoding="utf-8") + res_bad_spec = _run_reset(mam_sandbox, "--yes") + assert res_bad_spec.returncode == 1 + assert "unknown agent" in res_bad_spec.stderr + assert "check it, then remove it by hand" in res_bad_spec.stderr + assert spec_file.read_text(encoding="utf-8") == bad_spec_text + spec_file.unlink() + _assert_nothing_changed() + + +def test_full_reset_success_spec_ordering_table_and_loop_block( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """§5.2-5, §5.2-6, §5.2-10: + - .mam/reset-spec.json is written atomically with both targets before the first kill call + - Both claude sessions are stopped (normal then purge, no UUID on CLI) and recreated sequentially + - Old transcripts are removed, new IDs recorded != old IDs, result table shows RECORDED=yes, NEW!=OLD=yes, done + - Spec file and loop marker are cleaned up on exit 0 + - During the reset, run_loop.sh is blocked by .mam/loop-guard-active ("Another multi-agent-mux-loop is already running") + """ + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA) + tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [ + _row(name_a, "claude", mam_sandbox, role="planner,reviewer", server="default", label="lbl-a", own_id=UA), + _row(name_b, "claude", mam_sandbox, role="reviewer", server="default", label="lbl-b", own_id=UB), + ], + ) + spy_log = _install_herdr_spy(mam_sandbox) + + res = _run_reset(mam_sandbox, "--yes", env_extra={"SPY_PROBE_RUN_LOOP": "1"}) + _wait_reconcile_idle(mam_sandbox) + assert res.returncode == 0, res.stdout + res.stderr + + # Spec file & loop marker removed after completion + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + + # Old transcripts purged + assert not tf_a.exists() + assert not tf_b.exists() + + # Recreated rows have same name, role, herdr_server, herdr_workspace, and new IDs != old IDs + rows = _read_rows(mam_sandbox) + assert set(rows.keys()) == {name_a, name_b} + assert rows[name_a]["role"] == "planner,reviewer" + assert rows[name_a]["herdr_server"] == "default" + assert rows[name_a]["herdr_workspace"] == "lbl-a" + new_a = rows[name_a]["claude_session_id_own"] + assert new_a and new_a != UA + + assert rows[name_b]["role"] == "reviewer" + assert rows[name_b]["herdr_server"] == "default" + assert rows[name_b]["herdr_workspace"] == "lbl-b" + new_b = rows[name_b]["claude_session_id_own"] + assert new_b and new_b != UB + + # Result table in stdout + out_lines = [l for l in res.stdout.splitlines() if l.startswith(name_a) or l.startswith(name_b)] + assert len(out_lines) == 2 + assert out_lines[0].split() == [name_a, "claude", UA, new_a, "yes", "yes", "done"] + assert out_lines[1].split() == [name_b, "claude", UB, new_b, "yes", "yes", "done"] + + # Stop invoked twice per agent (normal, then --purge-conversation --yes), no UUID on CLI + stop_traces = [l for l in res.stdout.splitlines() if l.startswith("reset [") and "stop_session.sh" in l] + assert len(stop_traces) == 4 + for st in stop_traces: + assert UA not in st and UB not in st + + # §5.2-5 & §5.2-6: Spy log verifies spec file existed at first kill and sequential order + calls = _read_spy_log(spy_log) + first_kill = next(c for c in calls if c["verb"] in ("pane close", "kill-session")) + assert first_kill["spec_exists"] is True + assert first_kill["spec"]["version"] == 1 + assert [a["name"] for a in first_kill["spec"]["agents"]] == [name_a, name_b] + + # Sequential order: kill A -> start A -> kill B -> start B + events = [] + for c in calls: + if c["verb"] == "kill-session": + tgt = c["args"][2] if len(c["args"]) > 2 and c["args"][1] == "-t" else c["args"][1] + if not events or events[-1] != ("kill", tgt): + events.append(("kill", tgt)) + elif c["verb"] == "agent start": + events.append(("start", c["args"][2])) + assert events == [ + ("kill", name_a), + ("start", name_a), + ("kill", name_b), + ("start", name_b), + ] + + # §5.2-10: During the reset, run_loop.sh was refused by loop-guard-active + probe_data = json.loads((mam_sandbox / "herdr_spy_loop_probe.json").read_text(encoding="utf-8")) + assert probe_data["rc"] == 1 + combined_probe = probe_data["stdout"] + probe_data["stderr"] + assert "Another multi-agent-mux-loop is already running" in combined_probe + + # Control after reset finishes: run_loop.sh passes the loop lock and fails at "is not registered" + loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh" + post_loop = subprocess.run( + ["bash", str(loop_sh), "--creator", "x", "--task", "t"], + cwd=str(mam_sandbox), + env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=30, + ) + assert "is not registered" in (post_loop.stdout + post_loop.stderr) + + +def test_mid_run_failure_and_rerun_resumes_from_spec( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """§5.2-7: A (claude) succeeds, B (grok) fails at create (grok preflight fails). + Run 1 exits 1 with 'failed at create (rc 1)' and leaves .mam/reset-spec.json. + Restoring grok binary and rerunning skips A (0 stop calls for A) and only creates B. + """ + name_a = "reviewer-claude-01" + name_b = "reviewer-grok-01" + _claude_transcript(mam_sandbox, mam_sandbox, UA) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="grok", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [ + _row(name_a, "claude", mam_sandbox, own_id=UA), + _row(name_b, "grok", mam_sandbox, own_id=UB), + ], + ) + spy_log = _install_herdr_spy(mam_sandbox) + + # Make bin/grok exit 1 (and XAI_API_KEY="") so B's create_session.sh preflight fails with exit 1 + grok_bin = mam_sandbox / "bin" / "grok" + grok_bin.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8") + grok_bin.chmod(0o755) + + res1 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""}) + _wait_reconcile_idle(mam_sandbox) + assert res1.returncode == 1 + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + assert "failed at create (rc 1)" in res1.stdout + + rows1 = _read_rows(mam_sandbox) + assert name_a in rows1 and rows1[name_a]["claude_session_id_own"] != UA + assert name_b not in rows1 # B was purged before its create failed + + # Restore grok binary, clear spy log, and rerun + grok_bin.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + grok_bin.chmod(0o755) + spy_log.unlink() + + res2 = _run_reset(mam_sandbox, "--yes", env_extra={"XAI_API_KEY": ""}) + _wait_reconcile_idle(mam_sandbox) + assert res2.returncode == 0, res2.stdout + res2.stderr + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + + # Spy confirms A had 0 stop/kill/start calls on rerun, and B was only created + calls2 = _read_spy_log(spy_log) + assert not any(c["verb"] in ("pane close", "kill-session") for c in calls2) + starts2 = [c["args"][2] for c in calls2 if c["verb"] == "agent start"] + assert starts2 == [name_b] + assert f"reset [{name_a}]:" not in res2.stdout + + +def test_purge_refused_exit_4_preserves_stopped_row_and_rerun_completes( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """§5.2-8: A's transcript is held open by a real sleep 300 process (via fake lsof). + Run 1: A is stopped in step 1, refused at purge in step 3 (exit 4 -> skipped: purge refused), + prints holder PID and 'sleep 300', keeps A's row (status=stopped, own_id=UA) and transcript, + leaves sleep alive, completes B, and exits 4 with spec file intact. + Run 2 (after holder ends): A is purged and recreated, B is untouched, exits 0. + """ + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA) + tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [ + _row(name_a, "claude", mam_sandbox, own_id=UA), + _row(name_b, "claude", mam_sandbox, own_id=UB), + ], + ) + spy_log = _install_herdr_spy(mam_sandbox) + + holder = subprocess.Popen(["sleep", "300"]) + try: + lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid}) + res1 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res1.returncode == 4, res1.stdout + res1.stderr + combined1 = res1.stdout + res1.stderr + assert f"pid {holder.pid}" in combined1 + assert "sleep 300" in combined1 + assert "skipped: purge refused" in res1.stdout + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + + # Holder process is still alive (reset never kills external processes) + assert holder.poll() is None + + # A's row is stopped with UA preserved and transcript intact; B is done + rows1 = _read_rows(mam_sandbox) + assert rows1[name_a]["status"] == "stopped" + assert rows1[name_a]["claude_session_id_own"] == UA + assert tf_a.is_file() + assert not tf_b.exists() + assert rows1[name_b]["claude_session_id_own"] != UB + + # Release holder and rerun + lsof_exe.unlink() + finally: + holder.kill() + holder.wait() + + spy_log.unlink() + res2 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res2.returncode == 0, res2.stdout + res2.stderr + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + assert not tf_a.exists() + + rows2 = _read_rows(mam_sandbox) + assert rows2[name_a]["claude_session_id_own"] != UA + assert f"reset [{name_b}]:" not in res2.stdout + + +def test_post_discovered_without_recorded_id_8b_and_exit_5( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """§5.2-8b (i) & (ii) and §5.2-9: + - 8b(i): agy row has no recorded own_id while herdr is live and cache points to UA. + Run 1 with UA held by fake lsof: step 1 captures UA onto the row (status=stopped), + step 3 refuses purge (exit 4). + Run 2 after releasing lsof: step 3 purges UA (all UA artifacts deleted!) and recreates + the agy session. Because mock agy does not record a new ID, exits 5 (created). + Run 3 (while still without ID): 0 stop calls (does not re-purge the new session!), exits 5. + Run 4 (after recording new ID UB): exits 0, removes spec file, table shows OLD_ID='-' and NEW!=OLD='yes'. + - 8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips with + 'skipped: no conversation id recorded' (exit 4), keeping the stopped row without calling purge. + """ + name_a = "reviewer-agy-01" + now = time.time() + epoch = int(now) - 100 + a_files = _agy_full_artifacts(mam_sandbox, UA, mtime=now) + _agy_cache(mam_sandbox, {mam_sandbox: UA}) + + holder = subprocess.Popen(["sleep", "300"]) + try: + lsof_exe = _fake_lsof_multi(mam_sandbox / "bin", {UA: holder.pid}) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch, agent="agy", pane_id="w1:p_a") + _write_rows( + mam_sandbox, + [_row(name_a, "agy", mam_sandbox, own_id=None, pid=holder.pid, epoch=epoch)], + ) + spy_log = _install_herdr_spy(mam_sandbox) + + # Run 1: step 1 captures UA onto the row and record-id persists UA into reset-spec.json; + # step 3 refuses purge because lsof holds UA + res1 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res1.returncode == 4, res1.stdout + res1.stderr + rows1 = _read_rows(mam_sandbox) + assert rows1[name_a]["status"] == "stopped" + assert rows1[name_a]["agy_conversation_id_own"] == UA + assert a_files[0].exists() + spec1 = json.loads((mam_sandbox / ".mam" / "reset-spec.json").read_text(encoding="utf-8")) + assert spec1["agents"][0]["old_id"] == UA + + # Simulate resuming the skipped session (new created_at, own_id still UA) before Run 2 (review F1): + # because record-id persisted old_id=UA, Run 2 still treats it as 'pending' and purges UA + rows1[name_a]["status"] = "running" + rows1[name_a]["herdr_session_created_at"] = "2026-10-02T00:00:00Z" + _write_rows(mam_sandbox, [rows1[name_a]]) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch + 10, agent="agy", pane_id="w1:p_a") + + # Release lsof and Run 2: step 3 purges UA (deletes all UA artifacts!) and creates new agy + lsof_exe.unlink() + finally: + holder.kill() + holder.wait() + + res2 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res2.returncode == 5, res2.stdout + res2.stderr + assert "created; id not recorded yet" in res2.stdout + for p in a_files: + assert not p.exists(), f"UA artifact should have been purged on rerun: {p}" + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + + # Run 3 (§5.2-9): rerunning while in `created` state performs 0 stop/create calls and exits 5 + spy_log.unlink() + res3 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res3.returncode == 5 + assert _read_spy_log(spy_log) == [] + assert f"reset [{name_a}]:" not in res3.stdout + + # Record new ID UB onto the recreated row and Run 4 -> exits 0, OLD_ID=UA, NEW!=OLD='yes' + rows3 = _read_rows(mam_sandbox) + rows3[name_a]["agy_conversation_id_own"] = UB + _write_rows(mam_sandbox, [rows3[name_a]]) + + res4 = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res4.returncode == 0, res4.stdout + res4.stderr + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + tbl_line = [l for l in res4.stdout.splitlines() if l.startswith(name_a)][0] + assert tbl_line.split() == [name_a, "agy", UA, UB, "yes", "yes", "done"] + + +def test_post_discovered_uncapturable_id_8b_ii( + mam_sandbox: Path, mock_herdr: Path, mock_agents +): + """§5.2-8b(ii): agy row has no recorded own_id and no conversation on disk -> step 2 skips + with 'skipped: no conversation id recorded' (exit 4), keeping the stopped row without + calling purge; manual purge followed by rerun recreates via D6 #1 (purged). + """ + name_b = "reviewer-agy-02" + epoch = int(time.time()) - 100 + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, epoch=epoch, agent="agy", pane_id="w1:p_b") + _write_rows(mam_sandbox, [_row(name_b, "agy", mam_sandbox, own_id=None, epoch=epoch)]) + spy_log = _install_herdr_spy(mam_sandbox) + + res_ii = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res_ii.returncode == 4, res_ii.stdout + res_ii.stderr + assert "skipped: no conversation id recorded" in res_ii.stdout + assert not any( + l.startswith("reset [") and "--purge-conversation" in l for l in res_ii.stdout.splitlines() + ) + rows_ii = _read_rows(mam_sandbox) + assert name_b in rows_ii and rows_ii[name_b]["status"] == "stopped" + + # Manual purge drops the row; rerunning reset then sees D6 #1 (purged) and runs create only + stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh" + res_manual = subprocess.run( + ["bash", str(stop_sh), "--session", name_b, "--agent", "agy", "--purge-conversation", "--yes"], + cwd=str(mam_sandbox), + env=dict(os.environ, SKS_QUIESCENT_TRIES="0", SKS_DIALOG_TIMEOUT="0", SKS_BUSY_TIMEOUT="0"), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=60, + ) + assert res_manual.returncode == 0, res_manual.stdout + res_manual.stderr + assert name_b not in _read_rows(mam_sandbox) + + spy_log.unlink() + res_ii_rerun = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res_ii_rerun.returncode == 5, res_ii_rerun.stdout + res_ii_rerun.stderr + calls_ii = _read_spy_log(spy_log) + assert not any(c["verb"] in ("pane close", "kill-session") for c in calls_ii) + assert [c["args"][2] for c in calls_ii if c["verb"] == "agent start"] == [name_b] + assert name_b in _read_rows(mam_sandbox) + + # Recording new ID UB when old_id remained None produces OLD_ID='-' and NEW!=OLD='yes' + rows_ii_after = _read_rows(mam_sandbox) + rows_ii_after[name_b]["agy_conversation_id_own"] = UB + _write_rows(mam_sandbox, [rows_ii_after[name_b]]) + res_ii_done = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res_ii_done.returncode == 0, res_ii_done.stdout + res_ii_done.stderr + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + tbl_ii = [l for l in res_ii_done.stdout.splitlines() if l.startswith(name_b)][0] + assert tbl_ii.split() == [name_b, "agy", "-", UB, "yes", "yes", "done"] + + +@pytest.mark.parametrize( + "sig_name,expected_rc", + [("SIGTERM", 143), ("SIGINT", 130), ("SIGHUP", 129)], +) +def test_signal_interrupts_reset_preserves_spec_and_holds_lock_until_child_exits( + mam_sandbox: Path, mock_herdr: Path, mock_agents, sig_name: str, expected_rc: int +): + """Review P1 / N1: + When SIGTERM, SIGINT, or SIGHUP arrives while an in-flight stop/create child is running: + - reset.sh keeps holding .mam/loop-guard-active (so a competing run_loop.sh is still refused) + until the in-flight child finishes; + - once the child finishes, reset.sh immediately exits with the signal's nonzero code (143/130/129), + releasing .mam/loop-guard-active on exit, preserving .mam/reset-spec.json, and NOT continuing + to purge/create the current or subsequent targets. + """ + import signal + + sig = getattr(signal, sig_name) + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA) + tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [ + _row(name_a, "claude", mam_sandbox, own_id=UA), + _row(name_b, "claude", mam_sandbox, own_id=UB), + ], + ) + + # Wrap bin/herdr so the first kill-session blocks until release_gate is created + bin_dir = mam_sandbox / "bin" + herdr_bin = bin_dir / "herdr" + herdr_real = bin_dir / "herdr.real" + if not herdr_real.exists(): + herdr_bin.rename(herdr_real) + entered_gate = mam_sandbox / "child_entered" + release_gate = mam_sandbox / "child_release" + spy_log = mam_sandbox / "herdr_spy.jsonl" + + herdr_bin.write_text( + f"""#!{sys.executable} +import json, os, sys, time + +real = {repr(str(herdr_real))} +entered = {repr(str(entered_gate))} +release = {repr(str(release_gate))} +spy_log = {repr(str(spy_log))} + +raw_args = sys.argv[1:] +args = list(raw_args) +while args and args[0] in ("-L", "--server", "-s", "--session"): + args = args[2:] if len(args) > 1 else args[1:] +verb = args[0] if (args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls")) else (" ".join(args[:2]) if len(args) >= 2 else "") +with open(spy_log, "a", encoding="utf-8") as lf: + lf.write(json.dumps({{"verb": verb, "args": args}}) + "\\n") + +if verb == "kill-session" and not os.path.exists(entered): + with open(entered, "w", encoding="utf-8") as fh: + fh.write("1") + deadline = time.time() + 15.0 + while not os.path.exists(release) and time.time() < deadline: + time.sleep(0.05) + +os.execv(real, [real, *raw_args]) +""", + encoding="utf-8", + ) + herdr_bin.chmod(0o755) + + script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh" + env = dict( + os.environ, + SKS_QUIESCENT_TRIES="0", + SKS_DIALOG_TIMEOUT="0", + SKS_BUSY_TIMEOUT="0", + MAM_ONBOARD_ACK_S="3", + MAM_ONBOARD_WAIT_S="3", + MAM_NO_SUBSCRIBER_DAEMON="1", + PYTHONDONTWRITEBYTECODE="1", + ) + proc = subprocess.Popen( + ["bash", str(script), "--yes"], + cwd=str(mam_sandbox), + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.time() + 15.0 + while not entered_gate.exists() and time.time() < deadline: + assert proc.poll() is None, "reset.sh exited before reaching kill-session" + time.sleep(0.05) + assert entered_gate.exists(), "timed out waiting for kill-session child" + + # Signal the reset parent while the foreground child is still blocked + proc.send_signal(sig) + time.sleep(0.2) + + # While the child is still running, reset.sh has not exited and still holds the loop lock + assert proc.poll() is None + assert (mam_sandbox / ".mam" / "loop-guard-active").is_file() + if sig_name == "SIGTERM": + loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh" + competing = subprocess.run( + ["bash", str(loop_sh), "--creator", "x", "--task", "t"], + cwd=str(mam_sandbox), + env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=20, + ) + assert competing.returncode == 1 + assert "Another multi-agent-mux-loop is already running" in (competing.stdout + competing.stderr) + + # Release the child and wait for reset.sh to terminate + release_gate.write_text("1", encoding="utf-8") + stdout, stderr = proc.communicate(timeout=30) + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + _wait_reconcile_idle(mam_sandbox) + + assert proc.returncode == expected_rc, f"rc={proc.returncode}, stdout={stdout}, stderr={stderr}" + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + # Neither transcript was purged, and no create or second-target operation ran + assert tf_a.is_file() + assert tf_b.is_file() + calls = _read_spy_log(spy_log) + assert not any(c["verb"] == "agent start" for c in calls) + assert f"reset [{name_b}]:" not in stdout + assert not any( + l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines() + )