"""The WAL lock guard tracks the lifecycle of the handle it protects.

Three shapes the first cut got wrong (review on #110544): a writer reopened after ``close()``
raced a live caller came back unguarded; the periodic checkpoint's guard refresh could land after
``close()`` and pin an OFD lock with no connection behind it; and refcounts keyed on a reusable
descriptor NUMBER mistook a recycled fd for a surviving lock. Linux-only: OFD locks + ``/proc``.
"""

import os
import sqlite3
import subprocess
import sys

import pytest

import hermes_state_lockguard as lg
from hermes_state import SessionDB
from tests.hermes_state._wal_generation_harness import make_db, pin_wal, require_wal

pytestmark = pytest.mark.platforms("linux")


def _foreign_exclusive_ok(path: str) -> bool:
    """Another process tries the EXCLUSIVE a close-time WAL reset needs; True = nothing guards."""
    code = (
        "import fcntl, os, struct, sys\n"
        f"fd = os.open({path!r}, os.O_RDWR)\n"
        "lk = struct.pack('@hhqqi', fcntl.F_WRLCK, 0, 0x40000002, 510, 0)\n"
        "try:\n    fcntl.fcntl(fd, 37, lk); print('EXCLUSIVE_ACQUIRED')\n"
        "except BlockingIOError:\n    print('REFUSED')\n"
    )
    return "EXCLUSIVE_ACQUIRED" in subprocess.run(
        [sys.executable, "-c", code], capture_output=True, text=True, check=True).stdout


def test_reopened_writer_is_guarded_again(tmp_path, monkeypatch):
    pin_wal(monkeypatch)
    db = make_db(tmp_path / "state.db", "s", "seed")
    require_wal(db)
    db.close()
    db.append_message("s", role="user", content="after close")  # #94736 teardown/worker reopen
    try:
        assert db._conn is not None and db._wal_lock_guard, "reopen returned an unguarded writer"
        assert not _foreign_exclusive_ok(str(db.db_path))
    finally:
        db.close()
    assert _foreign_exclusive_ok(str(db.db_path))  # a true last close lifts the guard


def test_guard_never_outlives_the_handle_under_fd_reuse(tmp_path, monkeypatch):
    """A+B live -> close A (its fd number is recycled by C) -> close B: C must still be guarded,
    and once C closes nothing may be left locked."""
    pin_wal(monkeypatch)
    path = tmp_path / "state.db"
    a = make_db(path, "s", "seed")
    require_wal(a)
    b = SessionDB(db_path=path)
    a.close()
    c = SessionDB(db_path=path)
    b.close()
    try:
        assert not _foreign_exclusive_ok(str(path)), "C recorded as guarded while nothing locks"
        for name in ("state.db", "state.db-shm"):  # the stray close the guard exists for
            os.close(os.open(tmp_path / name, os.O_RDONLY))
        subprocess.run([sys.executable, "-c",
                        f"import sqlite3; c = sqlite3.connect({str(path)!r}); "
                        "c.execute('select count(*) from messages').fetchone(); c.close()"], check=True)
        c.append_message("s", role="user", content="still writes")
    finally:
        c.close()
    assert _foreign_exclusive_ok(str(path)), "a lock survived the last handle's close"
    assert not lg._HANDLES
    assert sqlite3.connect(path).execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 2


def test_closing_writer_stays_guarded_until_sqlite_closes(tmp_path, monkeypatch):
    """close() lifts the OFD guard before SQLite's own close so a true last close still ends the
    generation. A stray close had already cancelled this process's POSIX locks, so in that gap a
    sibling's close could take EXCLUSIVE and unlink -wal/-shm under the still-open connection
    (every concurrent opener then refuses with DeletedWalGenerationError)."""
    pin_wal(monkeypatch)
    path = tmp_path / "state.db"
    db = make_db(path, "s", "seed")
    require_wal(db)
    for name in ("state.db", "state.db-shm"):  # the stray close the guard exists for
        os.close(os.open(tmp_path / name, os.O_RDONLY))
    shm = os.stat(f"{path}-shm").st_ino
    in_gap: dict = {}
    real_close = db._close_connection_quietly

    def _sibling_closes_in_the_gap(conn):
        subprocess.run([sys.executable, "-c",
                        f"import sqlite3; c = sqlite3.connect({str(path)!r}); "
                        "c.execute('select count(*) from messages').fetchone(); c.close()"], check=True)
        in_gap["shm"] = os.stat(f"{path}-shm").st_ino if os.path.exists(f"{path}-shm") else None
        real_close(conn)

    monkeypatch.setattr(db, "_close_connection_quietly", _sibling_closes_in_the_gap)
    db.close()
    assert in_gap["shm"] == shm, "a sibling's close unlinked -shm under the closing writer"
    assert not os.path.exists(f"{path}-wal"), "the true last close must still end the generation"
    assert _foreign_exclusive_ok(str(path)) and not lg._HANDLES