"""Issue #64: read-only SQLite connections (`_ro_connect`) for agy, codex and
hermes conversation/state databases across adapters, status.sh and reconcile.sh.

1. `_ro_connect` lives in `lib_py.agents.base` and `opencode` reuses it.
2. Read-only queries on `agy` (`conversations/<uuid>.db`), `codex`
   (`state_*.sqlite`), `hermes` (`state.db`), and `db_rows_naming` return the
   committed rows and leave the directory listing untouched across all four WAL
   states (`clean`, `leftover`, `walonly`, `writer`).
3. `status.sh` (JSON and text modes) and `reconcile.sh` (drift C for agy,
   hermes, codex) open agent DBs via `_ro_connect` and never create or remove
   `-wal` / `-shm` sidecars, including on another workspace's `agy` DB when an
   unpinned `agy` session triggers drift-C discovery.
"""

from __future__ import annotations

import json
import os
import shutil
import sqlite3
import subprocess
import sys
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, db_rows_naming  # noqa: E402
from lib_py.agents.registry import get_adapter  # noqa: E402
from lib_py.verify_session import verify_session_uuid  # noqa: E402


def _make_wal_state(db: Path, state: str, setup_fn, tmp_root: Path, opened: list) -> list:
    """Build `db` in one of four WAL states (`clean`, `leftover`, `walonly`,
    `writer`) using `setup_fn(conn)` to create tables and insert rows, then
    return `sorted(os.listdir(db.parent))`."""
    db.parent.mkdir(parents=True, exist_ok=True)
    for name in list(os.listdir(db.parent)):
        if name.startswith(db.name):
            os.remove(db.parent / name)

    def _open_and_seed(target: Path) -> sqlite3.Connection:
        conn = sqlite3.connect(str(target))
        conn.execute("PRAGMA journal_mode=wal")
        setup_fn(conn)
        conn.commit()
        return conn

    if state == "clean":
        _open_and_seed(db).close()
        # macOS persistent WAL keeps -wal and -shm on close; remove them after
        # the close checkpointed the rows into the main DB file.
        for suffix in ("-wal", "-shm"):
            p = Path(str(db) + suffix)
            if p.exists():
                p.unlink()
    elif state in ("leftover", "walonly"):
        src = tmp_root / f"src-{state}-{db.name}"
        if src.parent != tmp_root:
            src.parent.mkdir(parents=True, exist_ok=True)
        conn = _open_and_seed(src)
        suffixes = ("", "-wal", "-shm") if state == "leftover" else ("", "-wal")
        for suffix in suffixes:
            shutil.copyfile(str(src) + suffix, str(db) + suffix)
        conn.close()
        for suffix in ("", "-wal", "-shm"):
            p = Path(str(src) + suffix)
            if p.exists():
                p.unlink()
    elif state == "writer":
        opened.append(_open_and_seed(db))
    else:
        raise ValueError(f"unknown state: {state}")

    return sorted(os.listdir(db.parent))


def _expected_listing(db_name: str, state: str, extra: list | None = None) -> list:
    base = [db_name]
    if state in ("leftover", "writer"):
        base += [f"{db_name}-shm", f"{db_name}-wal"]
    elif state == "walonly":
        base += [f"{db_name}-wal"]
    if extra:
        base += extra
    return sorted(base)


# ---------------------------------------------------------------------------
# 1. Shared helper in lib_py.agents.base
# ---------------------------------------------------------------------------
def test_ro_connect_lives_in_base_and_opencode_reuses_it():
    from lib_py.agents import base as base_mod
    from lib_py.agents.adapters import opencode as opencode_mod

    assert hasattr(base_mod, "_ro_connect"), "_ro_connect must be defined in lib_py.agents.base"
    assert opencode_mod._ro_connect is base_mod._ro_connect


# ---------------------------------------------------------------------------
# 2. Adapter queries across all four WAL states (agy, codex, hermes, base)
# ---------------------------------------------------------------------------
STATES = ("clean", "leftover", "walonly", "writer")


@pytest.mark.parametrize("state", STATES)
@pytest.mark.parametrize("query", ["verify_artifact", "discover"])
def test_agy_queries_read_right_and_leave_sidecars_alone(tmp_path, state, query):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    uuid = "a6400001-0000-4000-8000-000000000001"
    db = home / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid}.db"
    lc = home / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
    lc.parent.mkdir(parents=True, exist_ok=True)
    lc.write_text(json.dumps({str(ws): uuid}))

    opened = []
    try:
        listing = _make_wal_state(
            db, state,
            lambda c: (c.execute("CREATE TABLE steps (id INTEGER)"),
                       c.execute("INSERT INTO steps VALUES (1)")),
            tmp_path, opened,
        )
        assert listing == _expected_listing(db.name, state)

        adapter = get_adapter("agy")
        ctx = DiscoveryContext(workspace=str(ws), agent_name="agy", home_dir=str(home))
        if query == "verify_artifact":
            assert adapter.verify_artifact(uuid, ctx) is True
        else:
            assert adapter.discover(ctx) == [uuid]

        assert sorted(os.listdir(db.parent)) == listing
    finally:
        for c in opened:
            c.close()


@pytest.mark.parametrize("state", STATES)
@pytest.mark.parametrize("query", ["_find_rollout_via_db", "verify_artifact", "discover", "residual_artifacts", "db_rows_naming"])
def test_codex_and_db_rows_naming_queries_read_right_and_leave_sidecars_alone(tmp_path, state, query):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    uuid = "c6400001-0000-4000-8000-000000000001"
    codex_home = home / ".codex"
    rollout = codex_home / "sessions" / "2026" / "10" / "19" / f"rollout-001-{uuid}.jsonl"
    rollout.parent.mkdir(parents=True, exist_ok=True)
    rollout.write_text(json.dumps({"payload": {"id": uuid, "cwd": str(ws)}}) + "\n")
    db = codex_home / "state_5.sqlite"

    opened = []
    try:
        listing = _make_wal_state(
            db, state,
            lambda c: (
                c.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, updated_at INTEGER)"),
                c.execute("INSERT INTO threads VALUES (?, ?, ?, 2000000)", (uuid, str(ws), str(rollout))),
            ),
            tmp_path, opened,
        )
        expected = _expected_listing(db.name, state, extra=["sessions"])
        assert listing == expected

        adapter = get_adapter("codex")
        ctx = DiscoveryContext(workspace=str(ws), agent_name="codex", home_dir=str(home))
        if query == "_find_rollout_via_db":
            assert adapter._find_rollout_via_db(uuid, ctx) == (str(rollout), str(ws))
        elif query == "verify_artifact":
            assert adapter.verify_artifact(uuid, ctx) is True
        elif query == "discover":
            assert adapter.discover(ctx) == [uuid]
        elif query == "residual_artifacts":
            res = adapter.residual_artifacts(uuid, ctx)
            assert any("threads.id: 1 rows" in r for r in res), res
        elif query == "db_rows_naming":
            res = db_rows_naming(str(db), uuid, ("thread_id",), id_tables=(("threads", "id"),))
            assert res == [f"{db} (threads.id: 1 rows)"]

        assert sorted(os.listdir(db.parent)) == expected
    finally:
        for c in opened:
            c.close()


@pytest.mark.parametrize("state", STATES)
@pytest.mark.parametrize("query", ["verify_artifact", "discover"])
def test_hermes_queries_read_right_and_leave_sidecars_alone(tmp_path, state, query):
    home, ws = tmp_path / "home", tmp_path / "ws"
    ws.mkdir()
    uuid = "h6400001-0000-4000-8000-000000000001"
    db = home / ".hermes" / "state.db"

    opened = []
    try:
        listing = _make_wal_state(
            db, state,
            lambda c: (
                c.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at REAL)"),
                c.execute("INSERT INTO sessions VALUES (?, ?, 2000000.0)", (uuid, str(ws))),
            ),
            tmp_path, opened,
        )
        assert listing == _expected_listing(db.name, state)

        adapter = get_adapter("hermes")
        ctx = DiscoveryContext(workspace=str(ws), agent_name="hermes", home_dir=str(home), epoch=1500000)
        if query == "verify_artifact":
            assert adapter.verify_artifact(uuid, ctx) is True
        else:
            assert adapter.discover(ctx) == [uuid]

        assert sorted(os.listdir(db.parent)) == listing
    finally:
        for c in opened:
            c.close()


# ---------------------------------------------------------------------------
# 3. Static + runtime checks for status.sh and reconcile.sh
# ---------------------------------------------------------------------------
def test_status_and_reconcile_static_ro_connect_usage():
    status_text = (SKILLS / "multi-agent-mux-status" / "scripts" / "status.sh").read_text()
    assert status_text.count("_ro_connect(hdb)") == 2
    assert status_text.count("_ro_connect(odb)") == 2
    assert status_text.count("_ro_connect(db)") == 2
    assert "sqlite3.connect" not in status_text

    reconcile_text = (SKILLS / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh").read_text()
    assert reconcile_text.count("_ro_connect(hdb)") == 1
    assert reconcile_text.count("_ro_connect(odb)") == 1
    assert reconcile_text.count("_ro_connect(state_db)") == 1
    assert "sqlite3.connect" not in reconcile_text


@pytest.mark.parametrize("state", STATES)
def test_status_and_reconcile_runtime_leave_sidecars_alone(mam_sandbox, mock_herdr, state):
    home = mam_sandbox / "isolated_home"
    ws = mam_sandbox / "ws"
    ws.mkdir(exist_ok=True)

    u_agy = "a6400002-0000-4000-8000-000000000002"
    u_hermes = "h6400002-0000-4000-8000-000000000002"
    u_codex = "c6400002-0000-4000-8000-000000000002"

    agy_db = home / ".gemini" / "antigravity-cli" / "conversations" / f"{u_agy}.db"
    lc = home / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
    lc.parent.mkdir(parents=True, exist_ok=True)
    lc.write_text(json.dumps({str(ws): u_agy}))

    hermes_db = home / ".hermes" / "state.db"

    codex_home = home / ".codex"
    rollout = codex_home / "sessions" / "2026" / "10" / "19" / f"rollout-001-{u_codex}.jsonl"
    rollout.parent.mkdir(parents=True, exist_ok=True)
    rollout.write_text(json.dumps({"payload": {"id": u_codex, "cwd": str(ws)}}) + "\n")
    codex_db = codex_home / "state_5.sqlite"

    opened = []
    try:
        agy_list = _make_wal_state(
            agy_db, state,
            lambda c: (c.execute("CREATE TABLE steps (id INTEGER)"), c.execute("INSERT INTO steps VALUES (1)")),
            mam_sandbox, opened,
        )
        hermes_list = _make_wal_state(
            hermes_db, state,
            lambda c: (
                c.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, cwd TEXT, started_at REAL)"),
                c.execute("INSERT INTO sessions VALUES (?, ?, 2000000000.0)", (u_hermes, str(ws))),
            ),
            mam_sandbox, opened,
        )
        codex_list = _make_wal_state(
            codex_db, state,
            lambda c: (
                c.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT, rollout_path TEXT, updated_at INTEGER)"),
                c.execute("INSERT INTO threads VALUES (?, ?, ?, 2000000000)", (u_codex, str(ws), str(rollout))),
            ),
            mam_sandbox, opened,
        )

        yaml_path = mam_sandbox / ".mam" / "agent-sessions.yaml"
        reconcile_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
        status_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"
        lib_sh = mam_sandbox / ".agents" / "skills" / "lib.sh"

        env = {
            **os.environ,
            "HOME": str(home),
            "HOME_DIR": str(home),
            "CODEX_HOME": str(codex_home),
            "WORKSPACE_ROOT": str(mam_sandbox),
            "AGENT_SESSIONS_YAML": str(yaml_path),
        }

        # 1) Unpinned running sessions -> reconcile.sh --once --emit-diff --dry-run (drift C queries)
        mutate_unpinned = f"""
d['herdr_sessions'] = [
    {{'name': 's-creator-agy', 'status': 'running', 'role': 'creator', 'herdr_session_epoch': 1000000001,
      'agy_conversation_id_own': None, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'agy'}}}},
    {{'name': 's-creator-hermes', 'status': 'running', 'role': 'creator', 'herdr_session_epoch': 1000000001,
      'hermes_conversation_id_own': None, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'hermes'}}}},
    {{'name': 's-creator-codex', 'status': 'running', 'role': 'creator', 'herdr_session_epoch': 1000000001,
      'codex_session_id_own': None, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'codex'}}}},
]
"""
        subprocess.run(
            ["bash", "-c", f"source '{lib_sh}' && atomic_dump_yaml '{yaml_path}'"],
            input=mutate_unpinned, text=True, env=env, check=True, capture_output=True,
        )
        res_rec = subprocess.run(
            ["bash", str(reconcile_sh), "--once", "--emit-diff", "--dry-run"],
            env=env, capture_output=True, text=True, timeout=60,
        )
        assert res_rec.returncode == 0, res_rec.stderr
        assert sorted(os.listdir(agy_db.parent)) == agy_list
        assert sorted(os.listdir(hermes_db.parent)) == hermes_list
        assert sorted(os.listdir(codex_db.parent)) == codex_list

        # 2) Pinned sessions -> status.sh --json and text mode (resume_on_disk queries)
        mutate_pinned = f"""
d['herdr_sessions'] = [
    {{'name': 's-creator-agy', 'status': 'stopped', 'role': 'creator',
      'agy_conversation_id_own': {u_agy!r}, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'agy'}}}},
    {{'name': 's-creator-hermes', 'status': 'stopped', 'role': 'creator',
      'hermes_conversation_id_own': {u_hermes!r}, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'hermes'}}}},
    {{'name': 's-creator-codex', 'status': 'stopped', 'role': 'creator',
      'codex_session_id_own': {u_codex!r}, 'pane': {{'cwd': {str(ws)!r}, 'cmd': 'codex'}}}},
]
"""
        subprocess.run(
            ["bash", "-c", f"source '{lib_sh}' && atomic_dump_yaml '{yaml_path}'"],
            input=mutate_pinned, text=True, env=env, check=True, capture_output=True,
        )
        res_json = subprocess.run(
            ["bash", str(status_sh), "--json"],
            env=env, capture_output=True, text=True, timeout=60,
        )
        assert res_json.returncode == 0, res_json.stderr
        details = {row["name"]: row["resume_state"] for row in json.loads(res_json.stdout)["sessions_detail"]}
        assert details == {
            "s-creator-agy": "yes",
            "s-creator-hermes": "yes",
            "s-creator-codex": "yes",
        }

        res_txt = subprocess.run(
            ["bash", str(status_sh)],
            env=env, capture_output=True, text=True, timeout=60,
        )
        assert res_txt.returncode == 0, res_txt.stderr
        for sname in ("s-creator-agy", "s-creator-hermes", "s-creator-codex"):
            (line,) = [l for l in res_txt.stdout.splitlines() if l.startswith(sname)]
            assert " yes " in line, line

        assert sorted(os.listdir(agy_db.parent)) == agy_list
        assert sorted(os.listdir(hermes_db.parent)) == hermes_list
        assert sorted(os.listdir(codex_db.parent)) == codex_list
    finally:
        for c in opened:
            c.close()


# ---------------------------------------------------------------------------
# 4. Cross-workspace agy DB sidecar regression test (Task 1 root-cause)
# ---------------------------------------------------------------------------
def test_unpinned_agy_reconcile_leaves_other_workspace_clean_db_without_sidecars(tmp_path):
    """When workspace A has an unpinned running agy session, reconcile.sh's
    drift-C scan iterates every `*.db` in `~/.gemini/antigravity-cli/conversations/`
    and calls `verify_session_uuid(ws_a, 'agy', uuid, ..., mode='discover')`.
    Even when `last_conversations.json` maps `uuid_b` to another workspace `ws_b`,
    `verify_artifact` opens `uuid_b.db` to query `SELECT count(*) FROM steps`;
    with `_ro_connect`, no `-wal` or `-shm` sidecar is created beside `uuid_b.db`."""
    home = tmp_path / "home"
    ws_a = tmp_path / "ws_a"
    ws_b = tmp_path / "ws_b"
    ws_a.mkdir()
    ws_b.mkdir()

    uuid_b = "b6400003-0000-4000-8000-000000000003"
    db_b = home / ".gemini" / "antigravity-cli" / "conversations" / f"{uuid_b}.db"
    lc = home / ".gemini" / "antigravity-cli" / "cache" / "last_conversations.json"
    lc.parent.mkdir(parents=True, exist_ok=True)
    lc.write_text(json.dumps({str(ws_b): uuid_b}))

    opened = []
    listing = _make_wal_state(
        db_b, "clean",
        lambda c: (c.execute("CREATE TABLE steps (id INTEGER)"), c.execute("INSERT INTO steps VALUES (1)")),
        tmp_path, opened,
    )
    assert listing == [f"{uuid_b}.db"]

    row_a = {
        "name": "ws-a-creator-agy",
        "status": "running",
        "agy_conversation_id_own": None,
        "herdr_session_epoch": 1000000001,
        "pane": {"cwd": str(ws_a), "cmd": "agy"},
        "_sibling_claimed_uuids": [],
    }
    assert verify_session_uuid(str(ws_a), "agy", uuid_b, row=row_a, home_dir=str(home), mode="discover") is True
    assert sorted(os.listdir(db_b.parent)) == [f"{uuid_b}.db"]