multi-agent-mux/tests/test_issue10_13_purge.py

#!/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.
"""
from __future__ import annotations

import json
import os
import sqlite3
import subprocess
import sys
import time
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

UUID = "7d0b6f12-3a4c-4e5f-8a9b-0c1d2e3f4a5b"
OTHER = "11111111-2222-4333-8444-555555555555"


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 _run_onboard(args, sandbox: Path, extra_env=None) -> subprocess.CompletedProcess:
    env = dict(os.environ)
    for k in ("CLAUDE_CODE_SESSION_ID", "ANTIGRAVITY_CONVERSATION_ID", "HERMES_SESSION_ID", "MAM_ORCHESTRATOR_UUIDS"):
        env.pop(k, None)
    env.update({
        "PYTHONPATH": f"{SKILLS}:{env.get('PYTHONPATH', '')}",
        "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):
    paths = _agy_tree(tmp_path, UUID)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))

    purged = adapter.purge_artifacts(UUID, 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(UUID, ctx) == []


def test_agy_purge_keeps_presence_lock_while_in_use(tmp_path):
    paths = _agy_tree(tmp_path, UUID)
    adapter = get_adapter("agy")
    ctx = DiscoveryContext(workspace=str(tmp_path), agent_name="agy", home_dir=str(tmp_path))
    holder = _holder(UUID)
    try:
        purged = adapter.purge_artifacts(UUID, 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(UUID, ctx) == [str(paths["lock"])]


def test_agy_purge_reports_only_what_existed(tmp_path):
    """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"{UUID}.db").write_text("db")
    (root / "brain" / UUID).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(UUID, ctx)) == 2


def test_conversation_in_use_ignores_own_lineage():
    """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


# ---------------------------------------------------------------------------
# #13 — lib_py.agents purge CLI
# ---------------------------------------------------------------------------
def _purge_cli(agent: str, uuid: str, home: Path) -> subprocess.CompletedProcess:
    env = dict(os.environ, HOME_DIR=str(home), PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
    return subprocess.run([sys.executable, "-m", "lib_py.agents", "purge", agent, uuid, str(home)],
                          capture_output=True, text=True, env=env, cwd=str(home), timeout=60)


def test_purge_cli_auto_resolves_agy_from_disk(tmp_path):
    paths = _agy_tree(tmp_path, UUID)
    res = _purge_cli("auto", UUID, 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):
    paths = _agy_tree(tmp_path, UUID)
    holder = _holder(UUID)
    try:
        res = _purge_cli("agy", UUID, 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_unresolved_uuid(tmp_path):
    res = _purge_cli("auto", OTHER, tmp_path)
    assert res.returncode == 5, res.stdout + res.stderr
    assert json.loads(res.stdout)["agent"] is None


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

    res = _run_onboard(["--remove", UUID, "--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, UUID), "registration must be untouched without --yes"


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

    res = _run_onboard(["--remove", UUID, "--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, UUID)


def test_orc_remove_purge_refused_while_in_use(mam_sandbox):
    paths = _agy_tree(mam_sandbox, UUID)
    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
    holder = _holder(UUID)
    try:
        res = _run_onboard(["--remove", UUID, "--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, UUID), "a refused purge must leave the registration in place"


def test_orc_remove_purge_without_artifacts_still_unregisters(mam_sandbox):
    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
    res = _run_onboard(["--remove", UUID, "--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, UUID)


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


def test_orc_dry_run_purge_changes_nothing(mam_sandbox):
    paths = _agy_tree(mam_sandbox, UUID)
    assert _run_onboard(["--uuid", UUID], mam_sandbox).returncode == 0
    res = _run_onboard(["--remove", UUID, "--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, UUID)