#!/usr/bin/env python3
"""Issues #10 and #13 — purging a conversation must leave nothing resumable
behind, and an orchestrator conversation (no herdr_sessions row) must be
purgeable through orc_onboard.sh.

#10: the agy adapter deleted only the conversation DB and the brain dir; the
     SQLite sidecars (-wal/-shm/-journal), the title annotation and the presence
     lock survived. The lock is removed only once nothing references the
     conversation any more.
#13: `orc_onboard.sh --remove <uuid> --purge-conversation --yes` resolves the
     adapter from what is on disk (orchestrator_uuids stores bare UUIDs), purges
     through it, and refuses while a live process still names the UUID -
     including the orchestrator's own session running the command.

Every test gets fresh conversation ids (the `conv` / `other` fixtures):
conversation_in_use scans every process on the host, so a fixed id in this
file is named by anything that carries the file's text on its command line.
"""
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, conversation_in_use  # noqa: E402

# The session-id variables an agent CLI exports to its tool shells. Scrubbed
# from every subprocess env below so the caller's own session never leaks in.
SESSION_ENV = ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID",
               "HERMES_SESSION_ID", "GROK_SESSION_ID", "OPENCODE_SESSION_ID")


def _processes_naming(ident: str) -> list:
    out = subprocess.run(["ps", "-axo", "pid=,command="], capture_output=True, text=True, timeout=10).stdout
    return [line.strip()[:120] for line in out.splitlines() if ident in line]


def _fresh_id() -> str:
    """A conversation id no other process can already be naming. With fixed
    ids, the delegation wrapper of a review job - whose --prompt carried this
    file's diff - named them, and five tests read "in use" during review."""
    ident = str(uuidlib.uuid4())
    named = _processes_naming(ident)
    assert not named, f"fresh id {ident} is already named by a live process: {named}"
    return ident


@pytest.fixture
def conv() -> str:
    return _fresh_id()


@pytest.fixture
def other() -> str:
    return _fresh_id()


def _clean_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 _agy_tree(home: Path, uuid: str) -> dict:
    root = home / ".gemini" / "antigravity-cli"
    paths = {
        "db": root / "conversations" / f"{uuid}.db",
        "wal": root / "conversations" / f"{uuid}.db-wal",
        "shm": root / "conversations" / f"{uuid}.db-shm",
        "brain": root / "brain" / uuid,
        "annotation": root / "annotations" / f"{uuid}.pbtxt",
        "lock": root / "presence" / f"{uuid}.lock",
    }
    for key, p in paths.items():
        if key == "brain":
            p.mkdir(parents=True, exist_ok=True)
            (p / "note.txt").write_text("brain")
        elif key == "db":
            # A real conversation DB: the adapter's verify_artifact opens it
            # with sqlite3, and sqlite discards a garbage -wal next to a
            # garbage main file, which would hide the sidecar from the purge.
            p.parent.mkdir(parents=True, exist_ok=True)
            conn = sqlite3.connect(p)
            conn.execute("CREATE TABLE steps (id INTEGER)")
            conn.execute("INSERT INTO steps VALUES (1)")
            conn.commit()
            conn.close()
        else:
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_text("x" if key != "lock" else "")
    return paths


def _holder(uuid: str) -> subprocess.Popen:
    """A live process whose command line names the conversation, the way a
    resumed `agy --conversation <uuid>` or `claude -r <uuid>` does. (Not
    `sh -c 'sleep …'`: a shell execs its last simple command and the uuid
    would vanish from the process list.)"""
    proc = subprocess.Popen([sys.executable, "-c", "import sys, time; time.sleep(60)", "--conversation", uuid])
    time.sleep(0.3)
    return proc


def _stop(proc: subprocess.Popen) -> None:
    proc.kill()
    proc.wait(timeout=10)


def _fake_agent_cli(tmp_path: Path, name: str) -> Path:
    """An executable named like an agent CLI. `<name> <flag> <uuid> -- cmd…`
    runs cmd as its child (not exec: this process stays in the tree with the
    uuid on its command line) - the agent session a skill runs under."""
    bindir = tmp_path / "agent-bin"
    bindir.mkdir(exist_ok=True)
    cli = bindir / name
    cli.write_text('#!/bin/sh\nshift 3\n"$@"\nexit $?\n')
    cli.chmod(0o755)
    return cli


def _run_onboard(args, sandbox: Path, extra_env=None) -> subprocess.CompletedProcess:
    env = _clean_env()
    env.update({
        "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,
    })
    if extra_env:
        env.update(extra_env)
    return subprocess.run(["bash", str(ONBOARD_SH)] + list(args), capture_output=True, text=True,
                          env=env, cwd=str(sandbox), timeout=120)


def _registered(sandbox: Path, uuid: str) -> bool:
    return uuid in _run_onboard(["--list"], sandbox).stdout


# ---------------------------------------------------------------------------
# #10 — agy adapter
# ---------------------------------------------------------------------------
def test_agy_purge_removes_sidecars_annotation_and_lock(tmp_path, conv):
    paths = _agy_tree(tmp_path, conv)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))

    purged = adapter.purge_artifacts(conv, ctx)

    assert len(purged) == 6, purged
    for key, p in paths.items():
        assert not p.exists(), f"{key} survived the purge: {p}"
    assert adapter.residual_artifacts(conv, ctx) == []


def test_agy_purge_keeps_presence_lock_while_in_use(tmp_path, conv):
    paths = _agy_tree(tmp_path, conv)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
    holder = _holder(conv)
    try:
        purged = adapter.purge_artifacts(conv, ctx)
    finally:
        _stop(holder)

    assert paths["lock"].exists(), "presence lock must survive while a process references the conversation"
    assert str(paths["lock"]) not in purged
    for key in ("db", "wal", "shm", "brain", "annotation"):
        assert not paths[key].exists(), key
    assert adapter.residual_artifacts(conv, ctx) == [str(paths["lock"])]


def test_agy_purge_reports_only_what_existed(tmp_path, conv):
    """The historical 2-item contract (db + brain) still holds when no sidecar
    exists (test_a4_adapter_contract pins len == 2)."""
    root = tmp_path / ".gemini" / "antigravity-cli"
    (root / "conversations").mkdir(parents=True)
    (root / "conversations" / f"{conv}.db").write_text("db")
    (root / "brain" / conv).mkdir(parents=True)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
    assert len(adapter.purge_artifacts(conv, ctx)) == 2


def test_conversation_in_use_ignores_own_lineage(other):
    """The purge command names the uuid on its own command line (and its
    parent shell does); only *other* processes count as holders."""
    assert conversation_in_use(other) is False
    holder = _holder(other)
    try:
        assert conversation_in_use(other) is True
    finally:
        _stop(holder)
    assert conversation_in_use(other) is False


def test_conversation_in_use_excludes_only_the_non_agent_helper_chain(tmp_path, other):
    """The purge's own helper chain names the uuid too - the python argv and
    the script that launched it - and must not count. The exclusion stops at
    the nearest agent CLI ancestor, though: that process is the session the
    purge runs in, and when it names the uuid the conversation is live."""
    probe = ("import sys; sys.path.insert(0, %r); "
             "from lib_py.agents.base import conversation_in_use; "
             "print(conversation_in_use(sys.argv[1]))") % str(SKILLS)
    helper = tmp_path / "helper.sh"
    helper.write_text('#!/bin/sh\n"$@"\nexit $?\n')
    helper.chmod(0o755)

    alone = subprocess.run([str(helper), sys.executable, "-c", probe, other],
                           capture_output=True, text=True, env=_clean_env(), timeout=60)
    assert alone.stdout.strip() == "False", alone.stdout + alone.stderr

    cli = _fake_agent_cli(tmp_path, "agy")
    under_agent = subprocess.run([str(cli), "--conversation", other, "--",
                                  str(helper), sys.executable, "-c", probe, other],
                                 capture_output=True, text=True, env=_clean_env(), timeout=60)
    assert under_agent.stdout.strip() == "True", under_agent.stdout + under_agent.stderr


# ---------------------------------------------------------------------------
# #13 — lib_py.agents purge CLI
# ---------------------------------------------------------------------------
def _purge_cli(agent: str, uuid: str, home: Path, extra_env=None, prefix=()) -> subprocess.CompletedProcess:
    env = _clean_env(extra_env)
    env["HOME_DIR"] = str(home)
    cmd = list(prefix) + [sys.executable, "-m", "lib_py.agents", "purge", agent, uuid, str(home)]
    return subprocess.run(cmd, capture_output=True, text=True, env=env, cwd=str(home), timeout=60)


def test_purge_cli_auto_resolves_agy_from_disk(tmp_path, conv):
    paths = _agy_tree(tmp_path, conv)
    res = _purge_cli("auto", conv, tmp_path)
    assert res.returncode == 0, res.stderr
    out = json.loads(res.stdout)
    assert out["agent"] == "agy"
    assert len(out["purged"]) == 6 and out["residual"] == [] and out["in_use"] is False
    assert not any(p.exists() for p in paths.values())


def test_purge_cli_refuses_while_in_use(tmp_path, conv):
    paths = _agy_tree(tmp_path, conv)
    holder = _holder(conv)
    try:
        res = _purge_cli("agy", conv, tmp_path)
    finally:
        _stop(holder)
    assert res.returncode == 4, res.stdout + res.stderr
    assert json.loads(res.stdout)["in_use"] is True
    assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"


def test_purge_cli_refuses_inside_the_agent_session_that_owns_the_uuid(tmp_path, conv):
    """An orchestrator purging its own conversation from its own session: the
    purge runs as a descendant of the agent CLI whose command line names the
    uuid. That ancestor is the conversation's live process, not part of the
    purge's helper chain, so the purge must be refused and delete nothing."""
    paths = _agy_tree(tmp_path, conv)
    cli = _fake_agent_cli(tmp_path, "claude")
    res = _purge_cli("agy", conv, tmp_path, prefix=(str(cli), "--session-id", conv, "--"))
    assert res.returncode == 4, res.stdout + res.stderr
    assert json.loads(res.stdout)["in_use"] is True
    assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"


@pytest.mark.parametrize("var", SESSION_ENV)
def test_purge_cli_refuses_from_the_sessions_own_shell(tmp_path, var, conv):
    """The session a tool shell belongs to is in its environment even when no
    process carries the uuid in argv and nothing holds the artifact open
    (claude keeps no transcript open): a purge run from there is refused."""
    paths = _agy_tree(tmp_path, conv)
    res = _purge_cli("agy", conv, tmp_path, extra_env={var: conv})
    assert res.returncode == 4, res.stdout + res.stderr
    assert json.loads(res.stdout)["in_use"] is True
    assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"


def test_purge_cli_unresolved_uuid(tmp_path, other):
    res = _purge_cli("auto", other, tmp_path)
    assert res.returncode == 5, res.stdout + res.stderr
    assert json.loads(res.stdout)["agent"] is None


def _lock_holder(lock: Path) -> subprocess.Popen:
    """A process that holds the presence lock open (flock) but names the
    conversation nowhere - not on its command line (the path travels through
    the environment) and not in a session variable."""
    code = ("import fcntl, os, time; f = open(os.environ['MAM_TEST_LOCK'], 'a'); "
            "fcntl.flock(f, fcntl.LOCK_EX); time.sleep(60)")
    proc = subprocess.Popen([sys.executable, "-c", code], env=dict(os.environ, MAM_TEST_LOCK=str(lock)))
    time.sleep(0.5)
    return proc


def test_purge_cli_refuses_when_only_the_presence_lock_is_held_open(tmp_path, conv):
    """codex R1: the in-use check must also ask lsof about the presence lock
    and the sidecars, not just the DB - a holder can keep the lock open with
    the uuid on no command line and in no session variable."""
    paths = _agy_tree(tmp_path, conv)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
    holder = _lock_holder(paths["lock"])
    try:
        assert adapter.conversation_in_use(conv, ctx) is True
        res = _purge_cli("agy", conv, tmp_path)
    finally:
        _stop(holder)
    assert res.returncode == 4, res.stdout + res.stderr
    assert json.loads(res.stdout)["in_use"] is True
    assert all(p.exists() for p in paths.values()), "a refused purge must remove nothing"
    assert adapter.conversation_in_use(conv, ctx) is False


def test_purge_cli_resolves_agy_from_orphaned_leftovers(tmp_path, conv):
    """codex R2: after the DB is gone (an interrupted or pre-#10 purge) the
    WAL, annotation and lock must still resolve to agy and be purged, instead
    of 'no artifact' leaving them behind forever."""
    root = tmp_path / ".gemini" / "antigravity-cli"
    leftovers = [root / "conversations" / f"{conv}.db-wal",
                 root / "annotations" / f"{conv}.pbtxt",
                 root / "presence" / f"{conv}.lock"]
    for p in leftovers:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text("x")

    res = _purge_cli("auto", conv, tmp_path)

    assert res.returncode == 0, res.stdout + res.stderr
    out = json.loads(res.stdout)
    assert out["agent"] == "agy"
    assert sorted(out["purged"]) == sorted(str(p) for p in leftovers)
    assert out["residual"] == []
    assert not any(p.exists() for p in leftovers)


# ---------------------------------------------------------------------------
# #13 — orc_onboard.sh --remove --purge-conversation
# ---------------------------------------------------------------------------
def test_orc_remove_purge_requires_yes(mam_sandbox, conv):
    paths = _agy_tree(mam_sandbox, conv)
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0

    res = _run_onboard(["--remove", conv, "--purge-conversation"], mam_sandbox)

    assert res.returncode == 4, res.stdout + res.stderr
    assert "Re-run with --yes" in res.stderr
    assert all(p.exists() for p in paths.values())
    assert _registered(mam_sandbox, conv), "registration must be untouched without --yes"


def test_orc_remove_purge_deletes_and_unregisters(mam_sandbox, conv):
    paths = _agy_tree(mam_sandbox, conv)
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0

    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)

    assert res.returncode == 0, res.stdout + res.stderr
    assert "Successfully removed orchestrator UUID" in res.stdout
    assert res.stdout.count("purged: ") == 6, res.stdout
    assert "residual: none (agent: agy)" in res.stdout
    assert not any(p.exists() for p in paths.values())
    assert not _registered(mam_sandbox, conv)


def test_orc_remove_purge_refused_while_in_use(mam_sandbox, conv):
    paths = _agy_tree(mam_sandbox, conv)
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
    holder = _holder(conv)
    try:
        res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)
    finally:
        _stop(holder)

    assert res.returncode == 5, res.stdout + res.stderr
    assert "still in use" in res.stderr
    assert all(p.exists() for p in paths.values())
    assert _registered(mam_sandbox, conv), "a refused purge must leave the registration in place"


def test_orc_remove_purge_refused_from_the_orchestrators_own_session(mam_sandbox, conv):
    """The case the skill documents: the orchestrator removes and purges its
    own UUID from inside its own session (its tool shell carries the session
    id). The purge must be refused (exit 5) and the registration kept."""
    paths = _agy_tree(mam_sandbox, conv)
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0

    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox,
                       extra_env={"CLAUDE_CODE_SESSION_ID": conv})

    assert res.returncode == 5, res.stdout + res.stderr
    assert "still in use" in res.stderr
    assert all(p.exists() for p in paths.values())
    assert _registered(mam_sandbox, conv), "a refused purge must leave the registration in place"


def test_orc_remove_purge_without_artifacts_still_unregisters(mam_sandbox, conv):
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes"], mam_sandbox)
    assert res.returncode == 0, res.stdout + res.stderr
    assert "nothing to purge" in res.stderr
    assert not _registered(mam_sandbox, conv)


def test_orc_purge_flag_needs_remove(mam_sandbox, conv):
    res = _run_onboard(["--uuid", conv, "--purge-conversation", "--yes"], mam_sandbox)
    assert res.returncode == 2
    assert not _registered(mam_sandbox, conv)


def test_orc_dry_run_purge_changes_nothing(mam_sandbox, conv):
    paths = _agy_tree(mam_sandbox, conv)
    assert _run_onboard(["--uuid", conv], mam_sandbox).returncode == 0
    res = _run_onboard(["--remove", conv, "--purge-conversation", "--yes", "--dry-run"], mam_sandbox)
    assert res.returncode == 0
    assert "purge its conversation artifacts" in res.stdout
    assert all(p.exists() for p in paths.values())
    assert _registered(mam_sandbox, conv)