#!/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(autouse=True)
def own_tmp_roots(tmp_path, monkeypatch):
    """Every tmp root the claude adapter may look in stays inside the test:
    TMP/TEMP/TMPDIR and the fixed /tmp fallback, here and in every python a
    test starts (a sitecustomize on PYTHONPATH sets the fallback there)."""
    from lib_py.agents.adapters import claude as claude_mod
    sys_tmp, tmpdir, shim = tmp_path / "systmp", tmp_path / "tmpdir", tmp_path / "pyshim"
    for d in (sys_tmp, tmpdir, shim):
        d.mkdir()
    monkeypatch.setattr(claude_mod, "_SYSTEM_TMP", str(sys_tmp))
    monkeypatch.setenv("TMPDIR", str(tmpdir))
    monkeypatch.delenv("TMP", raising=False)
    monkeypatch.delenv("TEMP", raising=False)
    (shim / "sitecustomize.py").write_text(
        "try:\n"
        "    from lib_py.agents.adapters import claude\n"
        f"    claude._SYSTEM_TMP = {str(sys_tmp)!r}\n"
        "except Exception:\n"
        "    pass\n")
    monkeypatch.setenv("PYTHONPATH", f"{shim}:{os.environ.get('PYTHONPATH', '')}")
    return sys_tmp


@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", "memory", " ", "notes",
                                 "12345678-1234-1234-1234-12345678901g", "{conv} ", "x{conv}"])
def test_claude_rejects_malformed_uuid(home, tmp_path, conv, bad):
    bad = bad.format(conv=conv)
    ws = tmp_path / "ws"
    ws.mkdir()
    t = _tree(home, ws, conv, tmp_path / "ctmp")
    # Claude Code's auto memory sits beside the session dirs, under the one
    # non-uuid name there.
    memory = home / ".claude" / "projects" / workspace_key(str(ws)) / "memory" / "MEMORY.md"
    memory.parent.mkdir()
    memory.write_text("m")
    named = [home / ".claude" / d / bad / "x" for d in ("file-history", "session-env")
             if bad.strip(". ") and "/" not in bad and "*" not in bad]
    for p in named:
        p.parent.mkdir(parents=True)
        p.write_text("x")
    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 adapter.conversation_in_use(bad, ctx) is False
    assert all(p.exists() for p in t.values())
    assert memory.exists() and all(p.exists() for p in named)


def test_claude_purge_never_follows_a_linked_dir_below_a_root(home, tmp_path, conv):
    """A symlinked <key> (tmp, projects) or file-history/ below its root is
    not followed: what lies behind it stays and is reported as residual."""
    ws = tmp_path / "ws"
    ws.mkdir()
    cfg = home / ".claude"
    key = workspace_key(str(ws))
    out = tmp_path / "outside"
    victims = [out / "tmp" / conv / "keep.txt", out / "proj" / f"{conv}.jsonl", out / "proj" / conv / "keep.txt",
               out / "fh" / conv / "keep.txt"]
    for v in victims:
        v.parent.mkdir(parents=True, exist_ok=True)
        v.write_text("keep")
    troot = tmp_path / "ctmp" / f"claude-{UID}"
    troot.mkdir(parents=True)
    links = [troot / "linked-workspace", cfg / "projects" / key, cfg / "file-history"]
    for link, target in zip(links, (out / "tmp", out / "proj", out / "fh")):
        link.symlink_to(target)
    adapter = get_adapter("claude")
    ctx = _ctx(ws, home)

    assert adapter.purge_artifacts(conv, ctx) == []
    assert all(v.exists() for v in victims) and all(link.is_symlink() for link in links)
    assert adapter.residual_artifacts(conv, ctx) == sorted(
        [str(troot / "linked-workspace" / conv), str(cfg / "projects" / key / f"{conv}.jsonl"),
         str(cfg / "projects" / key / conv), str(cfg / "file-history" / conv)])


def test_claude_purge_removes_only_the_link_at_the_leaf(home, tmp_path, conv):
    ws = tmp_path / "ws"
    ws.mkdir()
    key = workspace_key(str(ws))
    out = tmp_path / "outside"
    keep = [out / "tmp" / "keep.txt", out / "proj" / "keep.txt"]
    for k in keep:
        k.parent.mkdir(parents=True)
        k.write_text("keep")
    leaves = [tmp_path / "ctmp" / f"claude-{UID}" / key / conv, home / ".claude" / "projects" / key / conv]
    for leaf, target in zip(leaves, (out / "tmp", out / "proj")):
        leaf.parent.mkdir(parents=True)
        leaf.symlink_to(target)
    adapter = get_adapter("claude")
    ctx = _ctx(ws, home)

    assert set(adapter.purge_artifacts(conv, ctx)) == {str(p) for p in leaves}
    assert not any(os.path.lexists(p) for p in leaves)
    assert all(k.exists() for k in keep)
    assert adapter.residual_artifacts(conv, ctx) == []


def test_claude_roots_may_be_links(home, tmp_path, monkeypatch, conv):
    """A root that is itself a link (/tmp -> /private/tmp on macOS, a linked
    ~/.claude) is the boundary, not a way out of it."""
    ws = tmp_path / "ws"
    ws.mkdir()
    real_cfg, real_tmp = tmp_path / "real-cfg", tmp_path / "real-tmp"
    real_cfg.mkdir()
    real_tmp.mkdir()
    (tmp_path / "cfg-link").symlink_to(real_cfg)
    (tmp_path / "tmp-link").symlink_to(real_tmp)
    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cfg-link"))
    monkeypatch.delenv("CLAUDE_PROJECT_DIR")
    monkeypatch.setenv("CLAUDE_CODE_TMPDIR", str(tmp_path / "tmp-link"))
    t = _tree(home, ws, conv, tmp_path / "tmp-link", cfg=tmp_path / "cfg-link")
    adapter = get_adapter("claude")
    ctx = _ctx(ws, home)

    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 adapter.residual_artifacts(conv, ctx) == []


def test_a_non_uuid_id_takes_only_its_transcript_file(home, tmp_path):
    """The pre-#31 reach, kept for ids that are not uuids (test_a4 purges
    "uuid-c"): the `<id>.jsonl` file, never a directory of that name."""
    ws = tmp_path / "ws"
    ws.mkdir()
    proj = home / ".claude" / "projects" / workspace_key(str(ws))
    proj.mkdir()
    jsonl = proj / "uuid-c.jsonl"
    jsonl.write_text("{}\n")
    kept = [proj / "uuid-c" / "a", home / ".claude" / "file-history" / "uuid-c" / "b"]
    for p in kept:
        p.parent.mkdir(parents=True)
        p.write_text("x")
    adapter = get_adapter("claude")
    ctx = _ctx(ws, home)

    assert adapter.has_artifacts("uuid-c", ctx) is True
    assert adapter.residual_artifacts("uuid-c", ctx) == [str(jsonl)]
    assert adapter.purge_artifacts("uuid-c", ctx) == [str(jsonl)]
    assert not jsonl.exists() and all(p.exists() for p in kept)
    assert adapter.residual_artifacts("uuid-c", ctx) == []


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"


@pytest.mark.parametrize("started", ["NaN", float("nan"), float("inf"), float("-inf"), "inf", "1760000000000",
                                     "soon", True, False, None, [1], "<missing>"])
def test_an_unusable_started_at_means_live(live, started):
    from lib_py.agents.adapters import claude as claude_mod
    data = {} if started == "<missing>" else {"startedAt": started}
    assert claude_mod._session_file_live(data, live.pid) is True
    assert claude_mod._session_file_live(data, _dead_pid()) is False, "a dead pid holds nothing"


@pytest.mark.parametrize("raw", ["NaN", '"NaN"', "Infinity", "true"])
def test_a_live_holder_with_an_unusable_started_at_blocks_the_purge(home, tmp_path, conv, live, raw):
    """json.load turns a bare NaN or Infinity into a float that every
    comparison fails, which once made a live holder look stale."""
    ws = tmp_path / "ws"
    ws.mkdir()
    sfile, key = _session_file(home / ".claude", live.pid, conv)
    sfile.write_text(f'{{"pid": {live.pid}, "sessionId": "{conv}", "cwd": "/x", "startedAt": {raw}}}')
    adapter = get_adapter("claude")
    ctx = _ctx(ws, home)

    assert adapter.conversation_in_use(conv, ctx) is True
    res = _agents_cli(["in-use", "claude", conv, str(ws)], ws)
    assert res.returncode == 4, res.stdout + res.stderr
    assert adapter.purge_artifacts(conv, ctx) == []
    assert sfile.exists() and key.exists()


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, own_tmp_roots):
    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, own_tmp_roots / name):
        d.mkdir(parents=True)
    real = lambda roots: [os.path.realpath(r) for r in roots]  # noqa: E731
    sys_tmp = [os.path.realpath(str(own_tmp_roots / name))]

    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


def test_tests_never_reach_the_real_tmp_root(tmp_path, own_tmp_roots):
    from lib_py.agents.adapters.claude import _tmp_roots
    assert all(os.path.realpath(r).startswith(os.path.realpath(str(tmp_path))) for r in _tmp_roots())
    res = subprocess.run([sys.executable, "-c", "from lib_py.agents.adapters import claude; print(claude._SYSTEM_TMP)"],
                         capture_output=True, text=True, env=_env(), timeout=30)
    assert res.stdout.strip() == str(own_tmp_roots), res.stdout + res.stderr


# ---------------------------------------------------------------------------
# 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)]