test(issue31): purge, residual and the stop process gate

17 of the 25 cases fail on 26e2c43; the other 8 pin invariants (other
sessions, shared files, malformed ids, a prefix-twin job dir).

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

tests/test_issue31_purge_residual.py

@@ -0,0 +1,595 @@
+#!/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) -> dict:
+    """Everything a claude session `sid` owns, as claude 2.1 lays it out."""
+    cfg = 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
+    monkeypatch.setenv("PATH", "/nonexistent")
+    assert claude_mod._session_file_live({"startedAt": 0}, 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])}
+
+
+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)
+
+
+# ---------------------------------------------------------------------------
+# 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)]