fix(reset): SKILL.md version follows main's lockstep 4.13.0 after the rebase (#78)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.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

.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):
+            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]:

.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

.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

.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 "<empty>"),
+            })
+            continue
+
+        agent = agent_of_row(row)
+        role = str(row.get("role") or "").strip()
+        if not agent:
+            errors.append(
+                "running session %s has no recognisable agent type (agent=%r, pane.cmd=%r); "
+                "fix its registry row first"
+                % (name, row.get("agent"), (row.get("pane") or {}).get("cmd"))
+            )
+            continue
+        if not role:
+            errors.append(
+                "running session %s (%s) has an empty role; fix its registry row first"
+                % (name, agent)
+            )
+            continue
+
+        key = own_key(agent)
+        raw_id = row.get(key)
+        old_id = str(raw_id).strip() if isinstance(raw_id, str) and raw_id.strip() else None
+        targets.append({
+            "name": name,
+            "agent": agent,
+            "role": role,
+            "herdr_server": str(row_server(row)),
+            "herdr_workspace": str(row.get("herdr_workspace") or ""),
+            "old_id": old_id,
+            "old_created_at": str(row.get("herdr_session_created_at") or ""),
+        })
+
+    orchestrators = [str(u) for u in (state.get("orchestrator_uuids") or []) if u]
+    return {
+        "targets": targets,
+        "skipped": skipped,
+        "orchestrators": orchestrators,
+        "errors": errors,
+    }
+
+
+# --------------------------------------------------------------------------- D5: spec file
+
+def write_spec(workspace: str, agents: List[dict]) -> str:
+    """Atomically write .mam/reset-spec.json (mkstemp -> fsync -> replace -> dir fsync)."""
+    ws_real = _real(workspace)
+    mam_dir = os.path.join(ws_real, ".mam")
+    os.makedirs(mam_dir, exist_ok=True)
+    path = os.path.join(mam_dir, "reset-spec.json")
+    now_iso = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    doc = {
+        "version": SPEC_VERSION,
+        "workspace": ws_real,
+        "written_at": now_iso,
+        "agents": agents,
+    }
+    payload = (json.dumps(doc, indent=2) + "\n").encode("utf-8")
+    fd, tmp_path = tempfile.mkstemp(prefix=".reset-spec.", suffix=".tmp", dir=mam_dir)
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            fh.write(payload)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp_path, path)
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+    try:
+        dir_fd = os.open(mam_dir, os.O_RDONLY)
+        try:
+            os.fsync(dir_fd)
+        finally:
+            os.close(dir_fd)
+    except OSError:
+        pass
+    return path
+
+
+def read_spec(workspace: str) -> dict:
+    """Read and validate .mam/reset-spec.json; raise SpecError if invalid."""
+    ws_real = _real(workspace)
+    path = spec_path(ws_real)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            doc = json.load(fh)
+    except (OSError, ValueError) as exc:
+        raise SpecError("spec file %s cannot be read (%s); check it, then remove it by hand" % (path, exc))
+
+    if not isinstance(doc, dict) or doc.get("version") != SPEC_VERSION:
+        raise SpecError(
+            "spec file %s has unsupported version %r; check it, then remove it by hand"
+            % (path, doc.get("version") if isinstance(doc, dict) else None)
+        )
+    spec_ws = _real(str(doc.get("workspace") or ""))
+    if spec_ws != ws_real:
+        raise SpecError(
+            "spec file %s is for workspace %r, not %r; check it, then remove it by hand"
+            % (path, doc.get("workspace"), ws_real)
+        )
+    agents = doc.get("agents")
+    if not isinstance(agents, list):
+        raise SpecError("spec file %s has no agents list; check it, then remove it by hand" % path)
+    for item in agents:
+        if not isinstance(item, dict):
+            raise SpecError("spec file %s has a non-object agent entry; check it, then remove it by hand" % path)
+        name = item.get("name")
+        agent = item.get("agent")
+        role = item.get("role")
+        if not (isinstance(name, str) and name and isinstance(agent, str) and agent and isinstance(role, str) and role):
+            raise SpecError("spec file %s has an incomplete agent entry; check it, then remove it by hand" % path)
+        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())

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

.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

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/<session>/` 으로 옮겨 보관합니다. |
 | **`multi-agent-mux-orc-onboard`** | 오케스트레이터 자신의 세션 ID 를 등록해, 세션 탐색이 그것을 하위 에이전트로 착각하지 않게 합니다. |
 | **`multi-agent-mux-update`** | 설치된 작업 공간을 제자리에서 업데이트합니다. 실행 중인 세션·열린 잡·루프가 있으면 거부하고, 모니터를 멈췄다가 원래대로 되돌리며, 설치가 실패하면 되돌리고, 새 코드로 레지스트리를 변환합니다. |
+| **`multi-agent-mux-reset`** | 제어 루트의 실행 중인 상주 에이전트를 하나씩 정리(purge)하고 같은 이름·에이전트·역할·herdr 서버·작업 공간 라벨로 다시 만들며 온보딩합니다(`.mam/reset-spec.json` 으로 중단된 지점부터 재개). |
 
 ---
 

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/<session>/`. |
 | **`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). |
 
 ---
 

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

deploy/install.sh

@@ -668,5 +668,8 @@ cat <<EOF
 
   3. Update installed skills in the future:
      $ bash .agents/skills/multi-agent-mux-update/scripts/update.sh
+
+  4. Reset running control-root agents with fresh conversations:
+     $ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --dry-run
 --------------------------------------------------------------------------------
 EOF

deploy/remove.sh

@@ -93,6 +93,7 @@ else
     ".agents/skills/multi-agent-mux-status"
     ".agents/skills/multi-agent-mux-stop"
     ".agents/skills/multi-agent-mux-update"
+    ".agents/skills/multi-agent-mux-reset"
     ".agents/hooks"
     ".agents/hooks.json"
     ".agents/MULTI_AGENT_RULES.md"

tests/data/skill_code_blocks.json

@@ -477,6 +477,20 @@
    ],
    "mismatch": null,
    "note": "entry point call (Issue #72)"
+  },
+  {
+   "skill": "multi-agent-mux-reset",
+   "start": 17,
+   "end": 19,
+   "lang": "bash",
+   "sha": "3e9ea3c85bb0",
+   "class": "c",
+   "kind": "invocation",
+   "evidence": [
+    "multi-agent-mux-reset/scripts/reset.sh:6"
+   ],
+   "mismatch": null,
+   "note": "entry point call (Issue #78)"
   }
  ],
  "removed": [
@@ -635,6 +649,19 @@
    "passthrough": {
     "4": "multi-agent-mux-create/scripts/create_session.sh:554"
    }
+  },
+  "multi-agent-mux-reset/scripts/reset.sh": {
+   "skill": "multi-agent-mux-reset",
+   "exit_codes": [
+    "0",
+    "1",
+    "2",
+    "3",
+    "4",
+    "5",
+    "other"
+   ],
+   "passthrough": {}
   }
  },
  "prose_mismatches": [

tests/test_issue42_skill_blocks.py

@@ -116,7 +116,7 @@ def test_inventory_matches_skill_md(skill):
 
 
 def test_inventory_covers_nine_skills():
-    assert len(SKILL_NAMES) == 9  # Issue #72: + multi-agent-mux-update
+    assert len(SKILL_NAMES) == 10  # Issue #78: + multi-agent-mux-reset
     assert {b["skill"] for b in DATA["blocks"]} == set(SKILL_NAMES)
 
 

tests/test_issue78_hints_codex.py

@@ -0,0 +1,338 @@
+"""Issue #78: in_use_hints lists lsof holders, and codex purge removes
+tui-thread-reference-capabilities/<uuid>.
+"""
+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 <uuid> --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)