"""Issue #78: in_use_hints lists lsof holders, and codex purge removes
tui-thread-reference-capabilities/<uuid>.
"""
from __future__ import annotations

import json
import os
import sqlite3
import subprocess
import sys
import uuid as uuidlib
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
if str(SKILLS) not in sys.path:
    sys.path.insert(0, str(SKILLS))

from lib_py.agents.base import DiscoveryContext  # noqa: E402
from lib_py.agents.registry import get_adapter  # noqa: E402

pytestmark = pytest.mark.usefixtures("own_workspace_root")


def _clean_env(extra: dict | None = None) -> dict:
    env = {
        k: v
        for k, v in os.environ.items()
        if not k.startswith("XDG_")
        and not k.startswith("HERDR_")
        and k
        not in (
            "CLAUDE_CODE_SESSION_ID",
            "ANTIGRAVITY_CONVERSATION_ID",
            "HERMES_SESSION_ID",
            "GROK_SESSION_ID",
            "OPENCODE_SESSION_ID",
            "CLAUDE_CONFIG_DIR",
            "CLAUDE_PROJECT_DIR",
            "CODEX_HOME",
        )
    }
    env["PYTHONPATH"] = str(SKILLS)
    if extra:
        env.update(extra)
    return env


def _write_fake_lsof(bindir: Path, mapping: dict[str, list[int]]) -> Path:
    """Fake lsof that prints mapped PIDs when any argument matches a key substring."""
    bindir.mkdir(parents=True, exist_ok=True)
    cases = []
    for needle, pids in mapping.items():
        echoes = "\n".join(f"      echo {p}" for p in pids)
        cases.append(f"    *{needle}*)\n{echoes}\n      found=1 ;;")
    body = "\n".join(cases)
    exe = bindir / "lsof"
    exe.write_text(
        "#!/bin/sh\n"
        "found=0\n"
        'for a in "$@"; do\n'
        '  case "$a" in\n'
        f"{body}\n"
        "  esac\n"
        "done\n"
        '[ "$found" = "1" ] && exit 0\n'
        "exit 1\n",
        encoding="utf-8",
    )
    exe.chmod(0o755)
    return exe


def _materialize_artifact(agent: str, conv: str, ctx: DiscoveryContext) -> str:
    adapter = get_adapter(agent)
    if agent == "agy":
        # Hold the presence lock so agy's in_use_paths (_artifact_paths) is exercised too.
        paths = adapter._artifact_paths(conv, ctx)
        target = paths[-1]
    else:
        target = adapter.artifact_path(conv, ctx)
    Path(target).parent.mkdir(parents=True, exist_ok=True)
    Path(target).write_text("{}\n", encoding="utf-8")
    return target


@pytest.mark.parametrize("agent", ["claude", "agy", "codex", "grok", "hermes"])
def test_in_use_hints_list_lsof_holders(tmp_path, monkeypatch, agent):
    home = tmp_path / "home"
    ws = tmp_path / "ws"
    bindir = tmp_path / "bin"
    no_lsof_bin = tmp_path / "no-lsof-bin"
    for d in (home, ws, bindir, no_lsof_bin):
        d.mkdir(parents=True, exist_ok=True)

    # Populate no_lsof_bin with symlinks to basic tools (ps) but not lsof.
    for tool in ("ps", "sh", "bash"):
        for sysdir in ("/bin", "/usr/bin"):
            cand = Path(sysdir) / tool
            if cand.exists() and not (no_lsof_bin / tool).exists():
                (no_lsof_bin / tool).symlink_to(cand)

    conv = str(uuidlib.uuid4())
    monkeypatch.setenv("HOME", str(home))
    monkeypatch.setenv("HOME_DIR", str(home))
    monkeypatch.setenv("CODEX_HOME", str(home / ".codex"))
    ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home))
    target = _materialize_artifact(agent, conv, ctx)
    adapter = get_adapter(agent)

    holder = subprocess.Popen(["sleep", "300"])
    try:
        _write_fake_lsof(bindir, {conv: [holder.pid]})
        env = _clean_env({
            "HOME": str(home),
            "HOME_DIR": str(home),
            "CODEX_HOME": str(home / ".codex"),
            "PATH": f"{bindir}:{os.environ.get('PATH', '')}",
        })
        res = subprocess.run(
            [sys.executable, "-m", "lib_py.agents", "in-use", agent, conv, str(ws)],
            capture_output=True,
            text=True,
            env=env,
            timeout=30,
        )
        assert res.returncode == 4, res.stdout + res.stderr
        assert json.loads(res.stdout) == {"agent": agent, "in_use": True}
        assert f"ps -axww | grep {conv}" in res.stderr
        assert f"held open by pid {holder.pid}:" in res.stderr
        assert "sleep 300" in res.stderr
        assert "not stopped by MAM" in res.stderr

        monkeypatch.setenv("PATH", env["PATH"])
        hints = adapter.in_use_hints(conv, ctx)
        if agent == "claude":
            assert hints[0].startswith("a claude process whose session file under ")
            assert f"ps -axww | grep {conv}" in hints[1]
        else:
            assert f"ps -axww | grep {conv}" in hints[0]
        assert any(f"held open by pid {holder.pid}:" in h and "sleep 300" in h for h in hints)
        assert target in adapter.in_use_paths(conv, ctx)
    finally:
        holder.kill()
        holder.wait()


def test_in_use_hints_no_lsof_unchanged(tmp_path, monkeypatch):
    """§5.1 regression control: without lsof on PATH, hints match pre-#78 lines."""
    home = tmp_path / "home"
    ws = tmp_path / "ws"
    no_lsof_bin = tmp_path / "no-lsof-bin"
    for d in (home, ws, no_lsof_bin):
        d.mkdir(parents=True, exist_ok=True)
    monkeypatch.setenv("HOME", str(home))
    monkeypatch.setenv("HOME_DIR", str(home))
    monkeypatch.setenv("CODEX_HOME", str(home / ".codex"))
    monkeypatch.setenv("PATH", str(no_lsof_bin))

    for agent in ("claude", "agy", "codex", "grok", "hermes"):
        conv = str(uuidlib.uuid4())
        ctx = DiscoveryContext(workspace=str(ws), agent_name=agent, home_dir=str(home))
        _materialize_artifact(agent, conv, ctx)
        adapter = get_adapter(agent)
        hints = adapter.in_use_hints(conv, ctx)
        assert not any("held open by pid" in h for h in hints)
        assert any(f"ps -axww | grep {conv}" in h for h in hints)


def test_in_use_hints_opencode_skip_service(tmp_path, monkeypatch):
    home = tmp_path / "home"
    ws = tmp_path / "ws"
    bindir = tmp_path / "bin"
    for d in (home, ws, bindir):
        d.mkdir(parents=True, exist_ok=True)

    db = home / ".local" / "share" / "opencode" / "opencode.db"
    db.parent.mkdir(parents=True, exist_ok=True)
    db.write_text("", encoding="utf-8")

    adapter = get_adapter("opencode")
    monkeypatch.setattr(adapter, "_bin", lambda ctx: None)
    monkeypatch.setenv("HOME", str(home))
    monkeypatch.setenv("HOME_DIR", str(home))
    ctx = DiscoveryContext(workspace=str(ws), agent_name="opencode", home_dir=str(home))

    service_proc = subprocess.Popen(
        [sys.executable, "-c", "import time; time.sleep(300)", "serve", "--service"]
    )
    holder_proc = subprocess.Popen(["sleep", "300"])
    try:
        _write_fake_lsof(bindir, {"opencode.db": [service_proc.pid, holder_proc.pid]})
        monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")

        hints = adapter.in_use_hints("ses_test78", ctx)
        assert hints[0] == "any process with it on its command line (an editor, a grep, ...): ps -axww | grep ses_test78"
        assert any(f"held open by pid {holder_proc.pid}:" in h and "sleep 300" in h for h in hints)
        assert not any(f"pid {service_proc.pid}" in h or "serve --service" in h for h in hints)
        assert hints[-1] == f"another opencode session holding {db}: lsof {db}"
    finally:
        for p in (service_proc, holder_proc):
            p.kill()
            p.wait()


def test_stop_purge_refusal_prints_holders(mam_sandbox, mock_herdr, monkeypatch):
    conv = str(uuidlib.uuid4())
    name = "issue78-reviewer-claude"
    ws = mam_sandbox
    ctx = DiscoveryContext(workspace=str(ws), agent_name="claude", home_dir=str(mam_sandbox))
    transcript = Path(get_adapter("claude").artifact_path(conv, ctx))
    transcript.parent.mkdir(parents=True, exist_ok=True)
    transcript.write_text(json.dumps({"sessionId": conv}) + "\n", encoding="utf-8")

    lib = mam_sandbox / ".agents" / "skills" / "lib.sh"
    yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
    mutation = (
        f"d['herdr_sessions'] = [{{'name': {name!r}, 'status': 'running', 'agent': 'claude', "
        f"'role': 'reviewer', 'claude_session_id_own': {conv!r}, "
        f"'pane': {{'cwd': {os.path.realpath(str(ws))!r}}}}}]\n"
    )
    res_init = subprocess.run(
        ["bash", "-c", f'source "{lib}" && atomic_dump_yaml "{yaml_path}"'],
        input=mutation,
        capture_output=True,
        text=True,
        env=_clean_env({
            "HOME": str(mam_sandbox),
            "HOME_DIR": str(mam_sandbox),
            "WORKSPACE_ROOT": str(mam_sandbox),
            "AGENT_SESSIONS_YAML": str(yaml_path),
        }),
    )
    assert res_init.returncode == 0, res_init.stdout + res_init.stderr

    holder = subprocess.Popen(["sleep", "300"])
    try:
        bindir = mam_sandbox / ".local" / "bin"
        _write_fake_lsof(bindir, {conv: [holder.pid]})
        stop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-stop" / "scripts" / "stop_session.sh"
        env = _clean_env({
            "HOME": str(mam_sandbox),
            "HOME_DIR": str(mam_sandbox),
            "CLAUDE_PROJECT_DIR": str(mam_sandbox / ".claude" / "projects"),
            "WORKSPACE_ROOT": str(mam_sandbox),
            "AGENT_SESSIONS_YAML": str(yaml_path),
            "PATH": f"{bindir}:{os.environ.get('PATH', '')}",
        })
        res = subprocess.run(
            ["bash", str(stop_sh), "--session", name, "--agent", "claude", "--purge-conversation", "--yes"],
            capture_output=True,
            text=True,
            cwd=str(mam_sandbox),
            env=env,
            timeout=60,
        )
        assert res.returncode == 4, res.stdout + res.stderr
        assert f"held open by pid {holder.pid}:" in res.stderr
        assert "sleep 300" in res.stderr
        assert transcript.exists()
        assert name in yaml_path.read_text(encoding="utf-8")
    finally:
        holder.kill()
        holder.wait()


@pytest.mark.parametrize("delete_ok", [True, False], ids=["delete_ok", "delete_fails"])
def test_codex_purge_removes_tui_thread_capability(tmp_path, monkeypatch, delete_ok):
    home = tmp_path / "home"
    codex_home = tmp_path / "codex-home"
    ws = tmp_path / "ws"
    bindir = tmp_path / "bin"
    for d in (home, codex_home, ws, bindir):
        d.mkdir(parents=True, exist_ok=True)

    conv = str(uuidlib.uuid4())
    other = str(uuidlib.uuid4())

    rollout = codex_home / "sessions" / "2026" / "10" / "28" / f"rollout-2026-10-28T00-00-00-{conv}.jsonl"
    rollout.parent.mkdir(parents=True, exist_ok=True)
    rollout.write_text(json.dumps({"payload": {"id": conv, "cwd": str(ws)}}) + "\n", encoding="utf-8")

    cap_dir = codex_home / "tui-thread-reference-capabilities"
    cap_dir.mkdir(parents=True, exist_ok=True)
    cap_mine = cap_dir / conv
    cap_other = cap_dir / other
    cap_mine.write_text("cap-mine", encoding="utf-8")
    cap_other.write_text("cap-other", encoding="utf-8")

    state_db = codex_home / "state_5.sqlite"
    conn = sqlite3.connect(str(state_db))
    conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT, cwd TEXT)")
    conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (conv, str(rollout), str(ws)))
    conn.execute("INSERT INTO threads VALUES (?, ?, ?)", (other, "", str(ws)))
    conn.commit()
    conn.close()

    fake_codex = bindir / "codex"
    if delete_ok:
        # Simulate `codex delete <uuid> --force` removing rollout + DB row, leaving tui-thread-reference-capabilities.
        fake_codex.write_text(
            f"#!{sys.executable}\n"
            "import os, sqlite3, sys\n"
            "if len(sys.argv) >= 3 and sys.argv[1] == 'delete':\n"
            f"    p = {str(rollout)!r}\n"
            "    if os.path.exists(p):\n"
            "        os.remove(p)\n"
            f"    conn = sqlite3.connect({str(state_db)!r})\n"
            "    conn.execute('DELETE FROM threads WHERE id=?', (sys.argv[2],))\n"
            "    conn.commit()\n"
            "    conn.close()\n"
            "    sys.exit(0)\n"
            "sys.exit(1)\n",
            encoding="utf-8",
        )
    else:
        fake_codex.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
    fake_codex.chmod(0o755)

    monkeypatch.setenv("HOME", str(home))
    monkeypatch.setenv("HOME_DIR", str(home))
    monkeypatch.setenv("CODEX_HOME", str(codex_home))
    monkeypatch.setenv("PATH", f"{bindir}:{os.environ.get('PATH', '')}")

    adapter = get_adapter("codex")
    ctx = DiscoveryContext(workspace=str(ws), agent_name="codex", home_dir=str(home))
    assert str(cap_mine) in adapter.residual_artifacts(conv, ctx)

    purged = adapter.purge_artifacts(conv, ctx)

    assert str(cap_mine) in purged
    assert not cap_mine.exists()
    assert cap_other.exists()
    assert not rollout.exists()
    assert adapter.residual_artifacts(conv, ctx) == []
    assert str(cap_other) in adapter.residual_artifacts(other, ctx)