test(issue31): stop purges a session kept under CLAUDE_CONFIG_DIR

End to end through stop_session.sh with CLAUDE_CONFIG_DIR alone and with a
matching CLAUDE_PROJECT_DIR; plus the DiscoveryContext default. The
config-only stop case and the default check fail on 872d5f1.

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

.agents/skills/lib.sh

@@ -74,7 +74,8 @@ _MAM_DIALOG_TOKENS="${_MAM_MODAL_TOKENS}|${_MAM_HINT_TOKENS}"
 
 # Workspace-relative defaults with environment overrides (Phase Z)
 HOME_DIR="${HOME_DIR:-$HOME}"
-CLAUDE_PROJECT_DIR="${CLAUDE_PROJECT_DIR:-$HOME/.claude/projects}"
+# claude keeps projects/ under its config root, which CLAUDE_CONFIG_DIR moves (Issue #31).
+CLAUDE_PROJECT_DIR="${CLAUDE_PROJECT_DIR:-${CLAUDE_CONFIG_DIR:-$HOME/.claude}/projects}"
 if [ -z "${LOCAL_BIN:-}" ]; then
   if [ -x "$HOME_DIR/bin/herdr" ]; then
     LOCAL_BIN="$HOME_DIR/bin"

.agents/skills/lib_py/agents/__main__.py

@@ -1,6 +1,6 @@
 # __main__.py — CLI bridge for lib_py.agents facts and resolution (N4)
 
-import os, sys, json, shlex
+import os, sys, json, shlex, time
 from lib_py.agents.registry import get_adapter, own_key, agent_of_row
 
 # `purge auto` probes adapters in this order; opencode last because its
@@ -151,6 +151,38 @@ def main():
         print(json.dumps({"agent": name, "purged": purged, "residual": residual, "in_use": False}))
         sys.exit(0)
 
+    elif cmd == 'in-use':
+        # in-use <agent> <uuid> [workspace] [--wait <sec>]  (Issue #31)
+        # The purge CLI's own in-use verdict, for stop_session.sh to ask
+        # before it purges. Re-checks every 0.25 s for up to --wait seconds.
+        # Prints {"agent", "in_use"}. Exit 0 free; 4 still in use; 1 bad args.
+        args = sys.argv[2:]
+        wait = 0.0
+        if '--wait' in args:
+            i = args.index('--wait')
+            try:
+                wait = float(args[i + 1])
+            except (IndexError, ValueError):
+                print("ERROR: --wait requires seconds", file=sys.stderr)
+                sys.exit(1)
+            del args[i:i + 2]
+        agent_name = args[0] if args else ''
+        uuid = args[1] if len(args) > 1 else ''
+        workspace = args[2] if len(args) > 2 and args[2] else os.getcwd()
+        adapter = get_adapter(agent_name)
+        if not adapter or not uuid:
+            print("ERROR: in-use requires <agent> <uuid>", file=sys.stderr)
+            sys.exit(1)
+        from lib_py.agents.base import DiscoveryContext
+        ctx = DiscoveryContext(workspace, agent_name)
+        deadline = time.monotonic() + wait
+        in_use = adapter.conversation_in_use(uuid, ctx)
+        while in_use and time.monotonic() < deadline:
+            time.sleep(0.25)
+            in_use = adapter.conversation_in_use(uuid, ctx)
+        print(json.dumps({"agent": agent_name, "in_use": in_use}))
+        sys.exit(4 if in_use else 0)
+
     else:
         print(f"ERROR: Unknown CLI command {cmd!r}", file=sys.stderr)
         sys.exit(1)

.agents/skills/lib_py/agents/adapters/claude.py

@@ -5,7 +5,21 @@ from lib_py.verify_session import workspace_key
 
 
 def _config_dir(ctx: DiscoveryContext) -> str:
-    return os.path.join(ctx.home_dir, ".claude")
+    """claude's config root: CLAUDE_CONFIG_DIR moves all of it (sessions/,
+    file-history/, session-env/, jobs/, projects/)."""
+    return os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(ctx.home_dir, ".claude")
+
+
+def _projects_dirs(ctx: DiscoveryContext) -> list:
+    """ctx.claude_dir (CLAUDE_PROJECT_DIR, else the config root's projects/)
+    and the config root's own projects/, each once."""
+    dirs, seen = [], set()
+    for d in (ctx.claude_dir, os.path.join(_config_dir(ctx), "projects")):
+        real = os.path.realpath(d)
+        if real not in seen:
+            seen.add(real)
+            dirs.append(d)
+    return dirs
 
 
 def _tmp_roots() -> list:
@@ -56,6 +70,8 @@ def _session_file_live(data: dict, pid) -> bool:
         started = float(data["startedAt"]) / 1000.0
     except Exception:
         return True
+    if started <= 0:
+        return True
     # startedAt is written 1.7-3.2 s after the process starts; 2 s covers
     # lstart's one-second resolution.
     return lstart <= started + 2
@@ -71,6 +87,41 @@ def _remove(path: str) -> None:
         pass
 
 
+def _key_files(session_file: str) -> list:
+    """The key files of the pid a sessions/<pid>.json is named after,
+    matched by name only; their content is never read."""
+    d, name = os.path.split(session_file)
+    stem = name[:-len(".json")]
+    if not stem.isdigit():
+        return []
+    try:
+        names = sorted(os.listdir(d))
+    except OSError:
+        return []
+    return [os.path.join(d, n) for n in names if re.fullmatch(rf"{stem}\.[0-9a-f]{{64}}\.key", n)]
+
+
+def _remove_job(job: str) -> None:
+    """state.json goes last: it is the only proof that the dir is this
+    session's, so a dir that could not be emptied keeps it."""
+    if os.path.islink(job):
+        _remove(job)
+        return
+    try:
+        names = os.listdir(job)
+    except OSError:
+        return
+    for n in names:
+        if n != "state.json":
+            _remove(os.path.join(job, n))
+    try:
+        left = os.listdir(job)
+    except OSError:
+        return
+    if left in ([], ["state.json"]):
+        _remove(job)
+
+
 def _label(path: str) -> str:
     """How a path is printed: a session key file is named by pid only, its
     hex part is not shown."""
@@ -211,23 +262,20 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
         if not _plausible_id(uuid):
             return []
         cfg = _config_dir(ctx)
-        jsonl = self.artifact_path(uuid, ctx)
-        paths = [p for p in (jsonl, jsonl[:-len(".jsonl")],
-                             os.path.join(cfg, "file-history", uuid),
-                             os.path.join(cfg, "session-env", uuid)) if os.path.lexists(p)]
+        paths = []
+        for pdir in _projects_dirs(ctx):
+            base = os.path.join(pdir, ctx.ws_key, uuid)
+            paths += [p for p in (base + ".jsonl", base) if os.path.lexists(p)]
+        paths += [p for p in (os.path.join(cfg, "file-history", uuid),
+                              os.path.join(cfg, "session-env", uuid)) if os.path.lexists(p)]
         job = self._owned_job(uuid, ctx)
         if job:
             paths.append(job)
         for path, pid, data in self._session_files(uuid, ctx):
-            if _session_file_live(data, pid):
-                continue
-            paths.append(path)
-            stem = os.path.basename(path)[:-len(".json")]
-            sessions = os.path.dirname(path)
-            if stem.isdigit():
-                # The key file is matched by name only; its content is never read.
-                paths.extend(os.path.join(sessions, n) for n in sorted(os.listdir(sessions))
-                             if re.fullmatch(rf"{stem}\.[0-9a-f]{{64}}\.key", n))
+            if not _session_file_live(data, pid):
+                # Keys before the session file: the file is what ties them to
+                # this session, so it must outlive them.
+                paths += _key_files(path) + [path]
         for root in _tmp_roots():
             try:
                 keys = sorted(os.listdir(root))
@@ -249,9 +297,17 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
         # Best effort: never raises (stop runs this inside its registry
         # transaction). What could not be removed is not listed here and
         # shows up in residual_artifacts instead.
+        # Ownership evidence (a job's state.json, a session file) is removed
+        # only after what it vouches for, so a retry can still find the rest.
         purged = []
+        job = self._owned_job(uuid, ctx)
         for path in self._owned_paths(uuid, ctx):
-            _remove(path)
+            if path == job:
+                _remove_job(path)
+            elif path.endswith(".json") and any(os.path.lexists(k) for k in _key_files(path)):
+                continue
+            else:
+                _remove(path)
             if not os.path.lexists(path) and _label(path) not in purged:
                 purged.append(_label(path))
         return purged
@@ -260,28 +316,33 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
         """What still carries this session: names holding the uuid under the
         config root, in projects (two levels: key, then entry) and in the tmp
         roots (two levels), plus the content-matched jobs dir and session
-        files. projects is not walked deeper; it can be very large."""
+        files with their (masked) key files. projects is not walked deeper;
+        it can be very large."""
         if not _plausible_id(uuid):
             return []
         cfg = _config_dir(ctx)
-        found = set(scan_names(cfg, uuid, skip=(os.path.join(cfg, "projects"), ctx.claude_dir)))
-        try:
-            keys = sorted(os.listdir(ctx.claude_dir))
-        except OSError:
-            keys = []
-        for key in keys:
-            kdir = os.path.join(ctx.claude_dir, key)
-            if uuid in key:
-                found.add(kdir)
-            elif os.path.isdir(kdir):
-                try:
-                    found.update(os.path.join(kdir, n) for n in os.listdir(kdir) if uuid in n)
-                except OSError:
-                    pass
+        pdirs = _projects_dirs(ctx)
+        found = set(scan_names(cfg, uuid, skip=pdirs))
+        for pdir in pdirs:
+            try:
+                keys = sorted(os.listdir(pdir))
+            except OSError:
+                keys = []
+            for key in keys:
+                kdir = os.path.join(pdir, key)
+                if uuid in key:
+                    found.add(kdir)
+                elif os.path.isdir(kdir):
+                    try:
+                        found.update(os.path.join(kdir, n) for n in os.listdir(kdir) if uuid in n)
+                    except OSError:
+                        pass
         job = self._owned_job(uuid, ctx)
         if job:
             found.add(job)
-        found.update(path for path, _, _ in self._session_files(uuid, ctx))
+        for path, _, _ in self._session_files(uuid, ctx):
+            found.add(path)
+            found.update(_label(k) for k in _key_files(path))
         for root in _tmp_roots():
             try:
                 tkeys = os.listdir(root)

.agents/skills/lib_py/agents/adapters/codex.py

@@ -1,6 +1,6 @@
 import os, json, glob, sqlite3, subprocess
 from typing import Optional, Any, List
-from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
+from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, scan_names, db_rows_naming
 from lib_py.verify_session import workspace_key
 
 class CodexAgentAdapter(BaseAgentAdapter):
@@ -158,6 +158,17 @@ class CodexAgentAdapter(BaseAgentAdapter):
                 pass
         return purged
 
+    def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]:
+        # Issue #31: names under CODEX_HOME plus thread rows. logs_*.sqlite is
+        # not scanned (no thread index: a full scan of a large file).
+        root = self._codex_home(ctx)
+        found = scan_names(root, uuid)
+        for db in sorted(glob.glob(f"{root}/state_*.sqlite")):
+            found += db_rows_naming(db, uuid, ("thread_id",), id_tables=(("threads", "id"),))
+        for db in sorted(glob.glob(f"{root}/thread_history_*.sqlite")):
+            found += db_rows_naming(db, uuid, ("thread_id",))
+        return found
+
     def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
         return f"{binary} --dangerously-bypass-approvals-and-sandbox"
 

.agents/skills/lib_py/agents/adapters/grok.py

@@ -1,7 +1,7 @@
 import os, json, glob, shutil
 from typing import Optional, Any
 from urllib.parse import quote
-from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
+from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, scan_names
 from lib_py.verify_session import workspace_key
 
 
@@ -108,6 +108,11 @@ class GrokAgentAdapter(BaseAgentAdapter):
             purged.append(self.artifact_path(uuid, ctx))
         return purged
 
+    def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
+        # Issue #31: reported, not removed - memory-v2 observations are the
+        # shared store of grok's memory pipeline.
+        return scan_names(f"{ctx.home_dir}/.grok", uuid)
+
     def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
         if session_uuid:
             return f"{binary} --session-id {session_uuid} --permission-mode bypassPermissions"

.agents/skills/lib_py/agents/adapters/opencode.py

@@ -1,6 +1,6 @@
 import os, sys, sqlite3, shutil
 from typing import Optional, Any, List
-from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
+from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext, scan_names, db_rows_naming
 from lib_py.verify_session import workspace_key
 
 _PROVIDER_ENV_VARS = (
@@ -194,6 +194,13 @@ class OpenCodeAgentAdapter(BaseAgentAdapter):
                 sys.stderr.write(f"WARN: purge opencode db records failed: {e}\n")
         return purged
 
+    def residual_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
+        # Issue #31: the purge leaves event-sourced rows (event, todo, ...);
+        # deleting those is not known to be safe, so they are reported.
+        found = scan_names(f"{ctx.home_dir}/.local/share/opencode", uuid)
+        return found + db_rows_naming(self._resolve_db(ctx), uuid, ("session_id", "aggregate_id"),
+                                      id_tables=(("session", "id"),))
+
     def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
         return f"{binary} --auto --agent build"
 

.agents/skills/lib_py/agents/base.py

@@ -2,6 +2,7 @@
 
 import os, json, sqlite3, shutil, subprocess
 from typing import Optional, Dict, Any, List
+from urllib.parse import quote
 from lib_py.paths import resolve_home
 from lib_py.verify_session import verify_session_uuid, workspace_key
 
@@ -107,6 +108,40 @@ def scan_names(root: str, uuid: str, skip=()) -> list:
     return sorted(found)
 
 
+def db_rows_naming(db: str, uuid: str, columns=(), id_tables=()) -> list:
+    """"<db> (<table>.<col>: <n> rows)" for every table holding rows of this
+    conversation in one of `columns`, or in the id column of an `id_tables`
+    (table, column) pair (Issue #31). Read-only; unreadable tables are skipped."""
+    if not _plausible_id(uuid) or not db or not os.path.isfile(db):
+        return []
+    try:
+        conn = sqlite3.connect(f"file:{quote(os.path.abspath(db))}?mode=ro", uri=True)
+    except Exception:
+        return []
+    found = []
+    try:
+        tables = [r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")]
+        for table in tables:
+            qt = table.replace('"', '""')
+            try:
+                cols = {r[1] for r in conn.execute(f'PRAGMA table_info("{qt}")')}
+            except Exception:
+                continue
+            wanted = [c for c in columns if c in cols]
+            wanted += [c for t, c in id_tables if t == table and c in cols and c not in wanted]
+            for col in wanted:
+                try:
+                    n = conn.execute(f'SELECT count(*) FROM "{qt}" WHERE "{col}"=?', (uuid,)).fetchone()[0]
+                except Exception:
+                    continue
+                if n:
+                    found.append(f"{db} ({table}.{col}: {n} rows)")
+    except Exception:
+        pass
+    finally:
+        conn.close()
+    return found
+
 class SpawnSpec:
     def __init__(self, agent_name: str, session_name: str, command: str, env: Optional[Dict[str, str]] = None):
         self.agent_name = agent_name
@@ -121,7 +156,9 @@ class DiscoveryContext:
         self.workspace = workspace
         self.agent_name = agent_name
         self.home_dir = resolve_home(home_dir)
-        self.claude_dir = claude_dir or os.environ.get("CLAUDE_PROJECT_DIR", f"{self.home_dir}/.claude/projects")
+        # projects/ lives under claude's config root, which CLAUDE_CONFIG_DIR moves.
+        config_root = os.environ.get("CLAUDE_CONFIG_DIR") or f"{self.home_dir}/.claude"
+        self.claude_dir = claude_dir or os.environ.get("CLAUDE_PROJECT_DIR", f"{config_root}/projects")
         self.epoch = epoch
         self.row = row or {}
         self.mode = mode

.agents/skills/multi-agent-mux-stop/SKILL.md

@@ -49,7 +49,7 @@ bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh \
   --session "$SESSION_NAME" --agent "$AGENT" --reason api_error
 
 # 3. Stop gracefully + clean up on-disk conversation (DANGEROUS)
-#    — this prevents any future resume (status=terminated, resumable=false).
+#    — this prevents any future resume and removes the session's registry row (YAML+DB).
 bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh \
   --session "$SESSION_NAME" --agent "$AGENT" --purge-conversation --yes
 ```
@@ -62,7 +62,7 @@ bash .agents/skills/multi-agent-mux-stop/scripts/stop_session.sh \
 | 1 | From the checks before the stop, nothing was changed: YAML not found, or the session is not registered. **The registry update after the stop also exits 1** (the row disappeared from the YAML during the run) |
 | 2 | Bad arguments (including the retired `--mode` / `--capture-id` / `--graceful`), or `--agent` omitted and not resolvable |
 | 3 | `--purge-conversation` without `--yes` |
-| 4 | Purge aborted: the herdr session survived the kill chain |
+| 4 | Purge aborted: the herdr session, or an agent process still holding the conversation, survived the kill chain (nothing was changed) |
 | other | The registry update after the stop (YAML / DB write) failed and its code is passed through |
 
 Exits 2 and 3, and exit 1 from the pre-stop checks, come before anything is stopped, so nothing was changed. On exit 1 from the registry update, or an `other` exit, the session may already be down while its row still says `running`: check with `multi-agent-mux-status`, then run the stop again (it is idempotent).
@@ -80,20 +80,20 @@ while its registry row survives.
 
 ```
 running ──(stop default / --reason)────────► stopped      (resumable:true, conv preserved)
-running ──(stop --purge-conversation --yes)► terminated   (resumable:false, conv deleted)
+running ──(stop --purge-conversation --yes)► row removed  (YAML+DB; conv deleted)
 stopped ──(stop default … again)───────────► stopped      (idempotent no-op)
 ```
 
 Fields written in STOP mode: `status: stopped`, `stopped_at`, `stopped_at_epoch`, `stop_reason`, `termination_mode: graceful`, `claude_session_id_own`/`agy_conversation_id_own` and `resumable: true`.
 
-If `--purge-conversation` is used: `status: terminated`, `terminated_at`, `terminated_at_epoch`, `termination_mode: purge` and `resumable: false`.
+If `--purge-conversation` is used: the row is removed from the YAML and the DB — no `terminated` row is left.
 
 The script:
 1. Verifies the session is in agent-sessions.yaml
 2. If `delegate_job_id` is set, automatically publishes a `progress --detail "terminating"` event to the multi-agent-mux-delegate-job registry
 3. Captures the `last_visible_status` from `herdr capture-pane` (so we have a final TUI snapshot for audit)
 4. Attempts graceful exit keys → SIGTERM kill-session → SIGKILL fallback
-5. For `purge-conversation`: deletes `~/.claude/projects/.../jsonl` (claude) or, for agy, `~/.gemini/antigravity-cli/conversations/<uuid>.db` with its `-wal`/`-shm`/`-journal` sidecars, `brain/<uuid>/`, `annotations/<uuid>.pbtxt` and `presence/<uuid>.lock` (the lock only once no live process references the conversation — Issue #10)
+5. For `purge-conversation`: deletes, for claude, every path the session owns — `projects/<key>/<uuid>.jsonl` and `<uuid>/`, `file-history/<uuid>/`, `session-env/<uuid>/`, `jobs/<uuid[:8]>/` (when its `state.json` names the session), stale `sessions/<pid>.json` (+ that pid's `.key`) and `<tmp>/claude-<uid>/*/<uuid>/`, then prints anything still carrying the UUID as `WARN residual:`; shared files such as `history.jsonl` are not rewritten (Issue #31). Or, for agy, `~/.gemini/antigravity-cli/conversations/<uuid>.db` with its `-wal`/`-shm`/`-journal` sidecars, `brain/<uuid>/`, `annotations/<uuid>.pbtxt` and `presence/<uuid>.lock` (the lock only once no live process references the conversation — Issue #10)
 6. Updates the YAML entry and SQLite database atomically
 7. If `delegate_job_id` is set, publishes a `completed` event to the multi-agent-mux-delegate-job registry
 
@@ -120,7 +120,7 @@ Exit codes: 0 removed, 1 stop/git failure, 2 not a worktree of the control root,
 - **Don't delete on-disk artifacts by default** — the agent's `*.jsonl` / `conversations/*.db` is the data that `multi-agent-mux-resume` needs. `--purge-conversation` is for when the user is genuinely done with the conversation and wants zero recovery chance.
 - **YAML is append-only until you write a stop** — if a previous run left the entry as `running` but herdr is actually dead (crash, host reboot), the YAML is stale. Running `multi-agent-mux-stop` will detect "herdr already dead, just update YAML" and proceed.
 - **Don't delete the `claude_session_id_own: null` placeholder** — when the user creates a fresh session with `multi-agent-mux-create` and never sent a message, the entry has `claude_session_id_own: null`. Stopping must preserve that field.
-- **Monitor skill may still be tracking** — if `multi-agent-mux-monitor` is running a heartbeat loop, stopping a session while it watches will trigger its `herdr ls != yaml` reconciliation. That's expected — let the monitor run, it will mark the entry as `terminated` on its own.
+- **Monitor skill may still be tracking** — if `multi-agent-mux-monitor` is running a heartbeat loop, stopping a session while it watches will trigger its `herdr ls != yaml` reconciliation. That's expected — let the monitor run: it leaves a `stopped` row as it is (drift A only marks a `running` row whose herdr is gone as `terminated`), and a purged row is gone.
 
 ## Verification
 

.agents/skills/multi-agent-mux-stop/scripts/stop_session.sh

@@ -18,8 +18,8 @@
 #                              셋 다 실패하면 exit 2)
 #   --herdr-session <name>  — isolated herdr session name (alias: --herdr-server)
 #   --reason <reason>       — 상태 전이 사유 (stop_reason). 기본값 manual_stop
-#   --purge-conversation    — 디스크의 conversation artifact 까지 삭제.
-#                             status=terminated, resumable=false 로 전이하며
+#   --purge-conversation    — 디스크의 conversation artifact 와 세션 부속 경로까지
+#                             삭제하고 레지스트리 행(YAML+DB)을 지우며
 #                             resume 불가. --yes 없이는 확인 프롬프트(exit 3)
 #   --yes                   — --purge-conversation 의 확인 프롬프트 생략
 #
@@ -30,7 +30,7 @@
 # Exit codes:
 #   0 = success (or already-stopped no-op) | 1 = YAML not found / not registered
 #   2 = invalid args | 3 = interactive confirmation required (--yes 누락)
-#   4 = purge aborted (herdr session survived the kill chain)
+#   4 = purge aborted (herdr session or an agent process holding the conversation survived the kill chain)
 set -euo pipefail
 
 # shellcheck disable=SC1091
@@ -56,7 +56,7 @@ Arguments:
                              read from its registry row, not from where you stand.
   --reason <reason>    — stop_reason field (default: manual_stop)
   --purge-conversation — also delete on-disk conversation artifacts;
-                         status becomes terminated and resume is impossible
+                         the registry row is removed and resume is impossible
   --yes                — skip the --purge-conversation confirmation prompt
 EOF
 }
@@ -282,6 +282,23 @@ if [ "$PURGE" = "1" ] && [ "$HERDR_ALIVE" = "1" ]; then
   fi
 fi
 
+# Purge process gate (Issue #31): a closed herdr session is not an exited
+# agent - the process may still be shutting down, or may have outlived a dead
+# herdr ("herdr already dead" above runs no gate at all). The purge asks the
+# adapter's own in-use verdict, the one the purge CLI uses, outside the
+# registry lock. Any answer but "free" (0) keeps everything.
+if [ "$PURGE" = "1" ] && [ -n "$PURGE_UUID" ]; then
+  inuse_rc=0
+  "$(_delegate_py_bin)" -m lib_py.agents in-use "$AGENT" "$PURGE_UUID" "$TARGET_CWD" --wait 5 >/dev/null || inuse_rc=$?
+  if [ "$inuse_rc" != "0" ]; then
+    echo "ERROR: conversation $PURGE_UUID is still held by a live agent process (rc=$inuse_rc)." >&2
+    echo "  Refusing purge — records preserved (no state was modified)." >&2
+    echo "  For claude, see the session files naming it under ~/.claude/sessions/." >&2
+    delegate_publish_event "$DELEGATE_JOB_ID" cancelled "purge aborted: conversation still in use; no state was modified" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT"
+    exit 4
+  fi
+fi
+
 # INVARIANT: Terminal events for a job must be published AFTER atomic_dump_yaml updates the session status
 atomic_dump_yaml "$AGENT_SESSIONS_YAML" \
   SESSION_NAME="$SESSION_NAME" AGENT="$AGENT" PURGE="$PURGE" \

tests/_agent_home_guard.py

@@ -18,7 +18,7 @@ import tempfile
 
 # The pane-inherited variables conftest._isolate_agent_homes drops for every
 # test, so HOME (which the tests set) governs, as it does in a plain shell.
-DROPPED_VARS = ("HOME_DIR", "CLAUDE_PROJECT_DIR", "CODEX_HOME", "LOCAL_BIN")
+DROPPED_VARS = ("HOME_DIR", "CLAUDE_PROJECT_DIR", "CLAUDE_CONFIG_DIR", "CODEX_HOME", "LOCAL_BIN")
 # Read by the code under test but left alone, on purpose: HOME is where the
 # real herdr tests and the shim bootstrap find the server socket, and it is
 # what the suite's own tests set for an isolated home; XDG_CONFIG_HOME and

tests/data/skill_code_blocks.json

@@ -413,7 +413,7 @@
    "start": 42,
    "end": 55,
    "lang": "bash",
-   "sha": "27ebf150b92a",
+   "sha": "301c347ff881",
    "class": "c",
    "kind": "invocation",
    "evidence": [
@@ -429,7 +429,7 @@
    "start": 81,
    "end": 85,
    "lang": "-",
-   "sha": "a2b56ac238ae",
+   "sha": "a973676cb2d1",
    "class": "c",
    "kind": "diagram",
    "evidence": [
@@ -599,7 +599,7 @@
     "other"
    ],
    "passthrough": {
-    "1": "multi-agent-mux-stop/scripts/stop_session.sh:310"
+    "1": "multi-agent-mux-stop/scripts/stop_session.sh:327"
    }
   },
   "multi-agent-mux-create/scripts/create_session.sh": {

tests/test_issue31_purge_residual.py

@@ -0,0 +1,711 @@
+#!/usr/bin/env python3
+"""Issue #31 — a claude purge removes every path the session owns and reports
+what still carries the uuid; stop and orc_onboard both go through it.
+
+Before #31 the claude adapter removed only `<uuid>.jsonl`. file-history,
+session-env, `projects/<key>/<uuid>/`, the daemon job dir, a stale
+`sessions/<pid>.json` and the tmp scratch dir survived, and with no
+`residual_artifacts` the purge printed `residual: none`. A live `claude -c`
+names no uuid on its command line, so only `sessions/<pid>.json` shows it is
+running: the purge CLI and stop both refuse while that process lives.
+codex, grok and opencode report their leftovers (their purge is unchanged).
+
+Every test makes fresh ids: conversation_in_use scans every process on the
+host, so a fixed id here is named by anything carrying this file's text.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import time
+import uuid as uuidlib
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+ONBOARD_SH = SKILLS / "multi-agent-mux-orc-onboard" / "scripts" / "orc_onboard.sh"
+
+sys.path.insert(0, str(SKILLS))
+from lib_py.agents.registry import get_adapter  # noqa: E402
+from lib_py.agents.base import DiscoveryContext  # noqa: E402
+from lib_py.verify_session import workspace_key  # noqa: E402
+
+SESSION_ENV = ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID",
+               "HERMES_SESSION_ID", "GROK_SESSION_ID", "OPENCODE_SESSION_ID")
+UID = os.getuid()
+
+
+def _fresh_id() -> str:
+    ident = str(uuidlib.uuid4())
+    out = subprocess.run(["ps", "-axo", "pid=,command="], capture_output=True, text=True, timeout=10).stdout
+    assert ident not in out, f"fresh id {ident} is already named by a live process"
+    return ident
+
+
+@pytest.fixture
+def conv() -> str:
+    return _fresh_id()
+
+
+@pytest.fixture
+def other() -> str:
+    return _fresh_id()
+
+
+@pytest.fixture
+def home(tmp_path, monkeypatch):
+    """A claude home of its own: HOME, projects and the claude tmp root."""
+    h = tmp_path / "home"
+    (h / ".claude" / "projects").mkdir(parents=True)
+    monkeypatch.setenv("HOME", str(h))
+    monkeypatch.setenv("HOME_DIR", str(h))
+    monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(h / ".claude" / "projects"))
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "ctmp"))
+    monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
+    for k in SESSION_ENV:
+        monkeypatch.delenv(k, raising=False)
+    return h
+
+
+def _dead_pid() -> int:
+    p = subprocess.Popen(["true"])
+    p.wait()
+    return p.pid
+
+
+@pytest.fixture
+def live():
+    """A running process standing in for claude; its session file must be
+    written after it started (startedAt >= its lstart)."""
+    p = subprocess.Popen(["sleep", "120"])
+    time.sleep(0.05)
+    yield p
+    p.kill()
+    p.wait()
+
+
+def _session_file(cfg: Path, pid: int, sid: str, started_ms=None, key=True) -> list:
+    d = cfg / "sessions"
+    d.mkdir(parents=True, exist_ok=True)
+    started_ms = int(time.time() * 1000) if started_ms is None else started_ms
+    f = d / f"{pid}.json"
+    f.write_text(json.dumps({"pid": pid, "sessionId": sid, "cwd": "/x", "startedAt": started_ms}))
+    made = [f]
+    if key:
+        k = d / f"{pid}.{uuidlib.uuid4().hex}{uuidlib.uuid4().hex}.key"
+        k.write_bytes(os.urandom(140))
+        made.append(k)
+    return made
+
+
+def _tree(h: Path, ws: Path, sid: str, tmp_root: Path, pid=None, cfg=None) -> dict:
+    """Everything a claude session `sid` owns, as claude 2.1 lays it out
+    (`cfg`: a CLAUDE_CONFIG_DIR root instead of <home>/.claude)."""
+    cfg = cfg or h / ".claude"
+    key = workspace_key(str(ws))
+    proj = cfg / "projects" / key
+    proj.mkdir(parents=True, exist_ok=True)
+    t = {}
+    t["jsonl"] = proj / f"{sid}.jsonl"
+    t["jsonl"].write_text(json.dumps({"sessionId": sid, "cwd": os.path.realpath(str(ws))}) + "\n")
+    t["proj_dir"] = proj / sid
+    (t["proj_dir"] / "tool-results").mkdir(parents=True)
+    (t["proj_dir"] / "tool-results" / "a.txt").write_text("r")
+    (t["proj_dir"] / "subagents").mkdir()
+    (t["proj_dir"] / "subagents" / "agent-1.jsonl").write_text("{}\n")
+    t["file_history"] = cfg / "file-history" / sid
+    t["file_history"].mkdir(parents=True)
+    (t["file_history"] / "x@v1").write_text("edited")
+    t["session_env"] = cfg / "session-env" / sid
+    t["session_env"].mkdir(parents=True)
+    t["job"] = cfg / "jobs" / sid[:8]
+    t["job"].mkdir(parents=True)
+    (t["job"] / "state.json").write_text(json.dumps({"sessionId": sid, "backend": "daemon"}))
+    (t["job"] / "timeline.jsonl").write_text("{}\n")
+    t["session_json"], t["session_key"] = _session_file(cfg, pid or _dead_pid(), sid)
+    t["tmp"] = tmp_root / f"claude-{UID}" / key / sid
+    (t["tmp"] / "scratchpad").mkdir(parents=True)
+    (t["tmp"] / "tasks").mkdir()
+    (t["tmp"] / "tasks" / "b.output").write_text("out")
+    return t
+
+
+def _ctx(ws: Path, h: Path) -> DiscoveryContext:
+    return DiscoveryContext(workspace=str(ws), agent_name="claude", home_dir=str(h))
+
+
+def _masked(p: Path) -> str:
+    return os.path.join(str(p.parent), p.name.split(".")[0] + ".*.key")
+
+
+def _env(extra=None) -> dict:
+    env = dict(os.environ)
+    for k in SESSION_ENV + ("MAM_ORCHESTRATOR_UUIDS",):
+        env.pop(k, None)
+    env["PYTHONPATH"] = f"{SKILLS}:{env.get('PYTHONPATH', '')}"
+    if extra:
+        env.update(extra)
+    return env
+
+
+def _agents_cli(args, cwd) -> subprocess.CompletedProcess:
+    return subprocess.run([sys.executable, "-m", "lib_py.agents"] + list(args), capture_output=True,
+                          text=True, env=_env(), cwd=str(cwd), timeout=60)
+
+
+# ---------------------------------------------------------------------------
+# claude adapter
+# ---------------------------------------------------------------------------
+def test_claude_purge_removes_every_owned_path(home, tmp_path, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+
+    purged = get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
+
+    assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
+    expected = {str(p) for k, p in t.items() if k != "session_key"} | {_masked(t["session_key"])}
+    assert set(purged) == expected
+    assert len(purged) == len(expected)
+    assert not any(t["session_key"].name.split(".")[1] in p for p in purged), "the key's hex is never printed"
+
+
+def test_claude_purge_leaves_other_sessions_and_shared_files(home, tmp_path, conv, other, live):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    cfg = home / ".claude"
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+    o = _tree(home, ws, other, tmp_path / "ctmp")
+    o_live = _session_file(cfg, live.pid, other)
+    history = cfg / "history.jsonl"
+    history.write_bytes((json.dumps({"sessionId": conv, "display": "hi"}) + "\n").encode())
+    before = history.read_bytes()
+    memory = cfg / "projects" / workspace_key(str(ws)) / "memory" / "MEMORY.md"
+    memory.parent.mkdir()
+    memory.write_text("m")
+
+    get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
+
+    assert all(p.exists() for p in o.values())
+    assert all(p.exists() for p in o_live)
+    assert history.read_bytes() == before
+    assert memory.exists()
+
+
+def test_claude_purge_keeps_a_job_dir_of_another_session_with_the_same_prefix(home, tmp_path, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+    twin = conv[:8] + str(uuidlib.uuid4())[8:]
+    (t["job"] / "state.json").write_text(json.dumps({"sessionId": twin}))
+    adapter = get_adapter("claude")
+
+    adapter.purge_artifacts(conv, _ctx(ws, home))
+
+    assert (t["job"] / "state.json").exists()
+    assert str(t["job"]) not in adapter.residual_artifacts(conv, _ctx(ws, home))
+
+
+def test_claude_residual_lists_leftovers_then_nothing(home, tmp_path, conv, other):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+    o = _tree(home, ws, other, tmp_path / "ctmp")
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    before = adapter.residual_artifacts(conv, ctx)
+
+    names = {str(t[k]) for k in ("jsonl", "proj_dir", "file_history", "session_env", "job", "session_json", "tmp")}
+    assert names <= set(before), names - set(before)
+    assert not any(other in p for p in before)
+    assert not any(str(o["job"]) == p or str(o["session_json"]) == p for p in before)
+
+    adapter.purge_artifacts(conv, ctx)
+    assert adapter.residual_artifacts(conv, ctx) == []
+
+    # A transcript under another workspace key is not this purge's to remove,
+    # but it still carries the uuid.
+    stray = home / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
+    stray.parent.mkdir()
+    stray.write_text("{}\n")
+    adapter.purge_artifacts(conv, ctx)
+    assert stray.exists()
+    assert adapter.residual_artifacts(conv, ctx) == [str(stray)]
+
+
+@pytest.mark.parametrize("bad", ["", ".", "..", "*", "../x", "a/b"])
+def test_claude_rejects_malformed_uuid(home, tmp_path, conv, bad):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    assert adapter.purge_artifacts(bad, ctx) == []
+    assert adapter.residual_artifacts(bad, ctx) == []
+    assert adapter.has_artifacts(bad, ctx) is False
+    assert all(p.exists() for p in t.values())
+
+
+def test_purge_cli_resolves_claude_from_leftovers_only(home, tmp_path, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    fh = home / ".claude" / "file-history" / conv
+    se = home / ".claude" / "session-env" / conv
+    fh.mkdir(parents=True)
+    se.mkdir(parents=True)
+
+    res = _agents_cli(["purge", "auto", conv, str(ws)], ws)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    out = json.loads(res.stdout)
+    assert out["agent"] == "claude"
+    assert set(out["purged"]) == {str(fh), str(se)}
+    assert out["residual"] == []
+    assert not fh.exists() and not se.exists()
+
+
+def test_purge_cli_refuses_a_live_claude_without_uuid_on_argv(home, tmp_path, conv, live):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp", pid=live.pid)
+
+    res = _agents_cli(["purge", "claude", conv, str(ws)], ws)
+
+    assert res.returncode == 4, res.stdout + res.stderr
+    assert json.loads(res.stdout)["in_use"] is True
+    assert all(p.exists() for p in t.values())
+
+
+def test_a_reused_pid_does_not_hold_the_conversation(home, tmp_path, conv, live):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    cfg = home / ".claude"
+    # The file was written an hour before the process now holding that pid
+    # started: the pid was reused, the session is gone.
+    made = _session_file(cfg, live.pid, conv, started_ms=int((time.time() - 3600) * 1000))
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    assert adapter.conversation_in_use(conv, ctx) is False
+    purged = adapter.purge_artifacts(conv, ctx)
+    assert not any(p.exists() for p in made)
+    assert set(purged) == {str(made[0]), _masked(made[1])}
+
+
+def test_session_file_liveness_fails_safe(monkeypatch, live):
+    from lib_py.agents.adapters import claude as claude_mod
+    now_ms = int(time.time() * 1000)
+    assert claude_mod._session_file_live({"startedAt": now_ms}, live.pid) is True
+    assert claude_mod._session_file_live({}, live.pid) is True, "no startedAt: cannot tell, so live"
+    assert claude_mod._session_file_live({"startedAt": now_ms}, _dead_pid()) is False
+    assert claude_mod._session_file_live({"startedAt": now_ms}, None) is True
+    assert claude_mod._session_file_live({"startedAt": now_ms}, 0) is True
+    assert claude_mod._session_file_live({"startedAt": 0}, live.pid) is True, "no usable startedAt: live"
+    hour_ago = int((time.time() - 3600) * 1000)
+    assert claude_mod._session_file_live({"startedAt": hour_ago}, live.pid) is False, "reused pid"
+    monkeypatch.setenv("PATH", "/nonexistent")
+    assert claude_mod._session_file_live({"startedAt": hour_ago}, live.pid) is True, "ps unavailable: live"
+
+
+def test_claude_purge_takes_the_stale_key_with_its_session_file(home, tmp_path, conv, other, live):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    cfg = home / ".claude"
+    stale = _session_file(cfg, _dead_pid(), conv)
+    held = _session_file(cfg, live.pid, other)
+    lone_key = cfg / "sessions" / f"{_dead_pid()}.{'a' * 64}.key"
+    lone_key.write_bytes(b"k")
+
+    purged = get_adapter("claude").purge_artifacts(conv, _ctx(ws, home))
+
+    assert not any(p.exists() for p in stale)
+    assert all(p.exists() for p in held)
+    assert lone_key.exists(), "a key with no session file of this conversation is not touched"
+    assert set(purged) == {str(stale[0]), _masked(stale[1])}
+
+
+@pytest.mark.parametrize("project_override", [False, True])
+def test_claude_config_dir_moves_the_config_root(home, tmp_path, monkeypatch, conv, live, project_override):
+    """CLAUDE_CONFIG_DIR moves sessions/, file-history/, ... and projects/:
+    a live session there is refused, a stopped one is cleaned up. With
+    CLAUDE_PROJECT_DIR naming the same projects/ the result is the same."""
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    cfg = tmp_path / "cfg"
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(cfg))
+    if project_override:
+        monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(cfg / "projects"))
+    else:
+        monkeypatch.delenv("CLAUDE_PROJECT_DIR")
+    t = _tree(home, ws, conv, tmp_path / "ctmp", pid=live.pid, cfg=cfg)
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+
+    assert adapter.conversation_in_use(conv, ctx) is True
+    res = _agents_cli(["purge", "claude", conv, str(ws)], ws)
+    assert res.returncode == 4, res.stdout + res.stderr
+    assert all(p.exists() for p in t.values())
+
+    live.kill()
+    live.wait()
+    purged = adapter.purge_artifacts(conv, ctx)
+    assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
+    assert str(t["file_history"]) in purged and str(t["jsonl"]) in purged
+    assert adapter.residual_artifacts(conv, ctx) == []
+    assert not (home / ".claude" / "file-history").exists(), "the default root is not used"
+
+
+def _failing_remove(monkeypatch, victim: Path):
+    from lib_py.agents.adapters import claude as claude_mod
+    real = os.remove
+
+    def remove(path, *a, **kw):
+        if os.path.realpath(path) == os.path.realpath(str(victim)):
+            raise PermissionError(13, "denied", str(path))
+        return real(path, *a, **kw)
+    monkeypatch.setattr(claude_mod.os, "remove", remove)
+
+
+def test_a_key_that_cannot_be_removed_keeps_its_session_file(home, tmp_path, monkeypatch, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    sfile, key = _session_file(home / ".claude", _dead_pid(), conv)
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+    with monkeypatch.context() as m:
+        _failing_remove(m, key)
+        purged = adapter.purge_artifacts(conv, ctx)
+
+    assert key.exists() and sfile.exists(), "the session file is what ties the key to this session"
+    assert purged == []
+    assert adapter.residual_artifacts(conv, ctx) == sorted([str(sfile), _masked(key)])
+
+    assert set(adapter.purge_artifacts(conv, ctx)) == {_masked(key), str(sfile)}
+    assert adapter.residual_artifacts(conv, ctx) == []
+
+
+def test_a_job_dir_that_cannot_be_emptied_keeps_its_state_json(home, tmp_path, monkeypatch, conv):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    t = _tree(home, ws, conv, tmp_path / "ctmp")
+    adapter = get_adapter("claude")
+    ctx = _ctx(ws, home)
+    with monkeypatch.context() as m:
+        _failing_remove(m, t["job"] / "timeline.jsonl")
+        purged = adapter.purge_artifacts(conv, ctx)
+
+    assert (t["job"] / "state.json").exists(), "state.json proves the dir is this session's"
+    assert str(t["job"]) not in purged
+    assert not t["file_history"].exists()
+    assert adapter.residual_artifacts(conv, ctx) == [str(t["job"])]
+
+    assert adapter.purge_artifacts(conv, ctx) == [str(t["job"])]
+    assert not t["job"].exists()
+    assert adapter.residual_artifacts(conv, ctx) == []
+
+
+def test_claude_tmp_root_rules(tmp_path, monkeypatch):
+    from lib_py.agents.adapters.claude import _tmp_roots
+    name = f"claude-{UID}"
+    x, y, z = tmp_path / "x", tmp_path / "y", tmp_path / "z"
+    for d in (x / name, y / name, z / name):
+        d.mkdir(parents=True)
+    real = lambda roots: [os.path.realpath(r) for r in roots]  # noqa: E731
+    sys_tmp = [os.path.realpath(f"/tmp/{name}")] if os.path.isdir(f"/tmp/{name}") else []
+
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(x))
+    monkeypatch.setenv("TMPDIR", str(z) + "/")
+    assert real(_tmp_roots()) == [str(x / name), str(z / name)] + sys_tmp
+
+    # A sandboxed tool shell: CLAUDE_CODE_TMPDIR is the per-uid root itself.
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(y / name))
+    assert real(_tmp_roots())[0] == str(y / name)
+
+    # Missing candidates are dropped, duplicates collapse.
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(z))
+    monkeypatch.delenv("TMPDIR", raising=False)
+    monkeypatch.delenv("TMP", raising=False)
+    monkeypatch.setenv("TEMP", str(z))
+    assert real(_tmp_roots()) == [str(z / name)] + sys_tmp
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "absent"))
+    assert real(_tmp_roots()) == [str(z / name)] + sys_tmp
+
+
+# ---------------------------------------------------------------------------
+# in-use CLI and the stop gate
+# ---------------------------------------------------------------------------
+def test_in_use_cli_waits_then_answers(home, tmp_path, conv, live):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+
+    t0 = time.monotonic()
+    res = _agents_cli(["in-use", "claude", conv, str(ws), "--wait", "1"], ws)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert json.loads(res.stdout) == {"agent": "claude", "in_use": False}
+    assert time.monotonic() - t0 < 5
+
+    _session_file(home / ".claude", live.pid, conv)
+    t0 = time.monotonic()
+    res = _agents_cli(["in-use", "claude", conv, str(ws), "--wait", "1"], ws)
+    assert res.returncode == 4, res.stdout + res.stderr
+    assert json.loads(res.stdout)["in_use"] is True
+    assert time.monotonic() - t0 >= 1
+
+    assert _agents_cli(["in-use", "nosuchagent", conv, str(ws)], ws).returncode == 1
+    assert _agents_cli(["in-use", "claude"], ws).returncode == 1
+
+
+def _stop_row(sandbox: Path, name: str, sid: str):
+    lib = sandbox / ".agents" / "skills" / "lib.sh"
+    yaml_path = sandbox / ".mam" / "agent-sessions.yaml"
+    mutation = (f"d['herdr_sessions'] = [{{'name': {name!r}, 'status': 'running', 'agent': 'claude', "
+                f"'claude_session_id_own': {sid!r}, 'pane': {{'cwd': {os.path.realpath(str(sandbox))!r}}}}}]\n")
+    res = subprocess.run(["bash", "-c", f"source {lib} && atomic_dump_yaml {yaml_path}"], input=mutation,
+                         capture_output=True, text=True, env=_env())
+    assert res.returncode == 0, res.stdout + res.stderr
+
+
+def _stop(sandbox: Path, name: str) -> subprocess.CompletedProcess:
+    script = sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
+    return subprocess.run(["bash", str(script), "--session", name, "--agent", "claude",
+                           "--purge-conversation", "--yes"], capture_output=True, text=True,
+                          env=_env(), cwd=str(sandbox), timeout=120)
+
+
+def _row_left(sandbox: Path, name: str) -> tuple:
+    in_yaml = name in (sandbox / ".mam" / "agent-sessions.yaml").read_text()
+    conn = sqlite3.connect(str(sandbox / ".mam" / "agent-sessions.db"))
+    try:
+        in_db = conn.execute("SELECT COUNT(*) FROM sessions WHERE name=?", (name,)).fetchone()[0] > 0
+    finally:
+        conn.close()
+    return in_yaml, in_db
+
+
+@pytest.fixture
+def stop_box(mam_sandbox, mock_herdr, monkeypatch):
+    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(mam_sandbox / "ctmp"))
+    return mam_sandbox
+
+
+def test_stop_purge_reports_residual_and_removes_the_row(stop_box, conv):
+    name = "issue31-creator-claude"
+    t = _tree(stop_box, stop_box, conv, stop_box / "ctmp")
+    _stop_row(stop_box, name, conv)
+    assert _row_left(stop_box, name) == (True, True)
+
+    res = _stop(stop_box, name)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert "herdr already dead" in res.stdout
+    assert res.stdout.count("purged: ") == 8, res.stdout
+    assert "residual: none (agent: claude)" in res.stdout
+    assert not any(p.exists() for p in t.values())
+    assert _row_left(stop_box, name) == (False, False)
+
+
+def test_stop_purge_warns_about_a_transcript_under_another_key(stop_box, conv):
+    name = "issue31-creator-claude"
+    _tree(stop_box, stop_box, conv, stop_box / "ctmp")
+    stray = stop_box / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
+    stray.parent.mkdir()
+    stray.write_text("{}\n")
+    _stop_row(stop_box, name, conv)
+
+    res = _stop(stop_box, name)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert f"WARN residual: {stray}" in res.stdout
+    assert "residual: none" not in res.stdout
+    assert _row_left(stop_box, name) == (False, False)
+
+
+def test_stop_purge_refuses_while_the_claude_process_lives(stop_box, conv):
+    name = "issue31-creator-claude"
+    holder = subprocess.Popen(["sleep", "120"])
+    try:
+        time.sleep(0.05)
+        t = _tree(stop_box, stop_box, conv, stop_box / "ctmp", pid=holder.pid)
+        _stop_row(stop_box, name, conv)
+
+        res = _stop(stop_box, name)
+
+        assert res.returncode == 4, res.stdout + res.stderr
+        assert "herdr already dead" in res.stdout
+        assert "records preserved" in res.stderr
+        assert "purged: " not in res.stdout
+        assert all(p.exists() for p in t.values())
+        assert _row_left(stop_box, name) == (True, True)
+    finally:
+        holder.kill()
+        holder.wait()
+
+    # Not a permanent block: once the process is gone the same stop purges.
+    res = _stop(stop_box, name)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert not any(p.exists() for p in t.values())
+    assert _row_left(stop_box, name) == (False, False)
+
+
+@pytest.mark.parametrize("project_override", [False, True])
+def test_stop_purge_follows_claude_config_dir(stop_box, conv, monkeypatch, project_override):
+    """stop resolves the uuid before it purges: with CLAUDE_CONFIG_DIR alone
+    the transcript under <cfg>/projects must be found too, or the row goes
+    and every artifact stays ("nothing purged")."""
+    cfg = stop_box / "custom-config"
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(cfg))
+    if project_override:
+        monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(cfg / "projects"))
+    else:
+        monkeypatch.delenv("CLAUDE_PROJECT_DIR", raising=False)
+    name = "issue31-creator-claude"
+    t = _tree(stop_box, stop_box, conv, stop_box / "ctmp", cfg=cfg)
+    _stop_row(stop_box, name, conv)
+
+    res = _stop(stop_box, name)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert "nothing purged" not in res.stdout
+    assert "residual: none (agent: claude)" in res.stdout
+    assert not any(p.exists() for p in t.values()), {k: str(p) for k, p in t.items() if p.exists()}
+    assert _row_left(stop_box, name) == (False, False)
+
+
+def test_discovery_context_projects_follow_claude_config_dir(tmp_path, monkeypatch):
+    from lib_py.agents.base import DiscoveryContext as Ctx
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cfg"))
+    assert Ctx(str(tmp_path), "claude", home_dir=str(tmp_path / "h")).claude_dir == str(tmp_path / "cfg" / "projects")
+    monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path / "p"))
+    assert Ctx(str(tmp_path), "claude", home_dir=str(tmp_path / "h")).claude_dir == str(tmp_path / "p")
+
+
+# ---------------------------------------------------------------------------
+# orc_onboard --remove --purge-conversation
+# ---------------------------------------------------------------------------
+def _run_onboard(args, sandbox: Path) -> subprocess.CompletedProcess:
+    env = _env({
+        "AGENT_SESSIONS_YAML": str(sandbox / ".mam" / "agent-sessions.yaml"),
+        "YAML_PATH": str(sandbox / ".mam" / "agent-sessions.yaml"),
+        "HOME_DIR": str(sandbox),
+        "AGENT_PYTHON_BIN": sys.executable,
+        "DELEGATE_JOB_PYTHON": sys.executable,
+    })
+    return subprocess.run(["bash", str(ONBOARD_SH)] + list(args), capture_output=True, text=True,
+                          env=env, cwd=str(sandbox), timeout=120)
+
+
+@pytest.mark.parametrize("stray", [False, True])
+def test_orc_remove_purge_reports_claude_leftovers(stop_box, conv, stray):
+    t = _tree(stop_box, stop_box, conv, stop_box / "ctmp")
+    other_key = stop_box / ".claude" / "projects" / "-elsewhere" / f"{conv}.jsonl"
+    if stray:
+        other_key.parent.mkdir()
+        other_key.write_text("{}\n")
+    assert _run_onboard(["--uuid", conv], stop_box).returncode == 0
+
+    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes", "--agent", "claude"], stop_box)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert not any(p.exists() for p in t.values())
+    if stray:
+        assert f"WARN residual: {other_key}" in res.stdout
+    else:
+        assert "residual: none (agent: claude)" in res.stdout
+
+
+# ---------------------------------------------------------------------------
+# codex, grok, opencode: residual only, purge unchanged
+# ---------------------------------------------------------------------------
+def _db(path: Path, ddl: list, rows: list):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(str(path))
+    for stmt in ddl:
+        conn.execute(stmt)
+    for stmt, args in rows:
+        conn.execute(stmt, args)
+    conn.commit()
+    conn.close()
+
+
+def test_opencode_residual_reports_event_rows(home, tmp_path, monkeypatch):
+    u, o = f"ses_{uuidlib.uuid4().hex}", f"ses_{uuidlib.uuid4().hex}"
+    db = home / ".local" / "share" / "opencode" / "opencode.db"
+    ddl = ["CREATE TABLE session (id TEXT PRIMARY KEY)",
+           "CREATE TABLE message (id TEXT, session_id TEXT)",
+           "CREATE TABLE part (id TEXT, session_id TEXT)",
+           "CREATE TABLE event (id TEXT, aggregate_id TEXT)",
+           "CREATE TABLE event_sequence (aggregate_id TEXT, seq INTEGER)",
+           "CREATE TABLE todo (id TEXT, session_id TEXT)"]
+    rows = []
+    for s in (u, o):
+        rows += [("INSERT INTO session VALUES (?)", (s,)),
+                 ("INSERT INTO message VALUES ('m', ?)", (s,)),
+                 ("INSERT INTO part VALUES ('p', ?)", (s,)),
+                 ("INSERT INTO event VALUES ('e1', ?)", (s,)),
+                 ("INSERT INTO event VALUES ('e2', ?)", (s,)),
+                 ("INSERT INTO event_sequence VALUES (?, 2)", (s,)),
+                 ("INSERT INTO todo VALUES ('t', ?)", (s,))]
+    _db(db, ddl, rows)
+    adapter = get_adapter("opencode")
+    monkeypatch.setattr(adapter, "_resolve_db", lambda ctx: str(db))  # never run a real opencode
+    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="opencode", home_dir=str(home))
+
+    adapter.purge_artifacts(u, ctx)
+
+    assert adapter.residual_artifacts(u, ctx) == [f"{db} (event.aggregate_id: 2 rows)",
+                                                  f"{db} (event_sequence.aggregate_id: 1 rows)",
+                                                  f"{db} (todo.session_id: 1 rows)"]
+
+
+def test_codex_residual_reports_name_and_thread_rows(home, tmp_path, monkeypatch, conv, other):
+    codex_home = tmp_path / "codex"
+    monkeypatch.setenv("CODEX_HOME", str(codex_home))
+    rollout = codex_home / "sessions" / "2026" / "10" / "15" / f"rollout-2026-10-15T00-00-00-{conv}.jsonl"
+    rollout.parent.mkdir(parents=True)
+    rollout.write_text("{}\n")
+    marker = codex_home / "tui-thread-reference-capabilities" / conv
+    marker.parent.mkdir()
+    marker.write_text("")
+    (codex_home / "tui-thread-reference-capabilities" / other).write_text("")
+    state = codex_home / "state_5.sqlite"
+    _db(state, ["CREATE TABLE threads (id TEXT PRIMARY KEY)"],
+        [("INSERT INTO threads VALUES (?)", (conv,)), ("INSERT INTO threads VALUES (?)", (other,))])
+    hist = codex_home / "thread_history_1.sqlite"
+    _db(hist, ["CREATE TABLE thread_items (thread_id TEXT, n INTEGER)"],
+        [("INSERT INTO thread_items VALUES (?, 1)", (conv,)), ("INSERT INTO thread_items VALUES (?, 2)", (conv,))])
+    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="codex", home_dir=str(home))
+
+    # purge is not run: it would execute the real `codex delete` on PATH.
+    residual = get_adapter("codex").residual_artifacts(conv, ctx)
+
+    assert residual == sorted([str(marker), str(rollout)]) + [
+        f"{state} (threads.id: 1 rows)", f"{hist} (thread_items.thread_id: 2 rows)"]
+
+
+def test_grok_residual_reports_memory_observations(home, tmp_path, conv, other):
+    from urllib.parse import quote
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    grok = home / ".grok"
+    sess = grok / "sessions" / quote(os.path.realpath(str(ws)), safe="") / conv
+    sess.mkdir(parents=True)
+    (sess / "chat_history.jsonl").write_text(json.dumps({"session_id": conv}) + "\n")
+    inbox = grok / "memory-v2" / "workspaces" / "w" / "observations" / "_inbox"
+    inbox.mkdir(parents=True)
+    mine = inbox / f"{conv}__t1.md"
+    mine.write_text("obs")
+    (inbox / f"{other}__t1.md").write_text("obs")
+    adapter = get_adapter("grok")
+    ctx = DiscoveryContext(workspace=str(ws), agent_name="grok", home_dir=str(home))
+
+    adapter.purge_artifacts(conv, ctx)
+
+    assert not sess.exists()
+    assert adapter.residual_artifacts(conv, ctx) == [str(mine)]