"""SessionDB bootstrap must never run schema init on an event-loop thread.

Enterprise field report (2026-08-14): after an unclean shutdown plus a
double-instance startup race, the v25 schema migration inside
``SessionDB.__init__`` blocked the gateway's event-loop thread (reached via
``_post_turn_goal_continuation`` → ``GoalManager()`` → ``load_goal`` →
``_get_session_db``). The loop-liveness watchdog missed 3 probes and killed
the process with exit 75; the supervisor restarted into the same state —
an unbounded crash loop.

Contract fixed here, at the shared boundary (goals.py ``_get_session_db``,
which the heartbeat module reuses):

- Called on a thread with a RUNNING event loop and no cached DB, it must
  NOT construct SessionDB inline. It returns None immediately (callers
  already degrade gracefully on None) and populates the cache from a
  background thread.
- Called on a plain worker thread, it constructs inline as before.
- Once cached, every thread gets the cached instance.
"""

import asyncio
import threading
import time

import pytest

import hermes_cli.goals as goals


class _RecordingDB:
    """Stands in for SessionDB; records which thread constructed it."""

    constructed_on: list = []

    def __init__(self, db_path=None):
        _RecordingDB.constructed_on.append(threading.get_ident())

    def get_meta(self, key):
        return None


@pytest.fixture(autouse=True)
def _clean_cache(monkeypatch):
    _RecordingDB.constructed_on = []
    monkeypatch.setattr(goals, "_DB_CACHE", {})
    yield


def _patch_sessiondb(monkeypatch, cls=_RecordingDB):
    # goals.py acquires through hermes_state_registry (shared writer per home); patch its
    # construction seam, and keep every acquired fake out of the registry so tests don't share.
    import hermes_state_registry

    monkeypatch.setattr(hermes_state_registry, "_open_session_db", lambda path: cls(db_path=path))


def test_loop_thread_cache_miss_constructs_off_loop(monkeypatch):
    """On the event-loop thread with a cold cache: construction happens on a
    background thread (never the loop thread). A fast init is returned via
    the grace window; either way the loop thread never runs SessionDB()."""
    _patch_sessiondb(monkeypatch)
    loop_thread_id = None
    inline_result = "UNSET"

    async def main():
        nonlocal loop_thread_id, inline_result
        loop_thread_id = threading.get_ident()
        inline_result = goals._get_session_db()

    asyncio.run(main())

    # Fast init: the grace window returns the real instance (no silent
    # feature loss on the first call).
    assert isinstance(inline_result, _RecordingDB)
    # Cache populated for subsequent calls.
    deadline = time.monotonic() + 5
    while time.monotonic() < deadline and not goals._DB_CACHE:
        time.sleep(0.02)
    assert goals._DB_CACHE, "background bootstrap never populated the cache"
    assert _RecordingDB.constructed_on, "SessionDB never constructed"
    assert all(t != loop_thread_id for t in _RecordingDB.constructed_on), (
        "SessionDB was constructed on the event-loop thread"
    )


def test_loop_thread_cache_hit_returns_cached_instance(monkeypatch):
    """A warm cache is returned directly even on the loop thread."""
    _patch_sessiondb(monkeypatch)
    sentinel = _RecordingDB()
    from hermes_constants import get_hermes_home

    goals._DB_CACHE[str(get_hermes_home())] = sentinel
    result = "UNSET"

    async def main():
        nonlocal result
        result = goals._get_session_db()

    asyncio.run(main())
    assert result is sentinel


def test_worker_thread_constructs_inline(monkeypatch):
    """No running loop on the calling thread → construct inline, return it."""
    _patch_sessiondb(monkeypatch)
    db = goals._get_session_db()
    assert db is not None
    assert isinstance(db, _RecordingDB)
    assert _RecordingDB.constructed_on == [threading.get_ident()]


def test_slow_construction_does_not_block_the_loop(monkeypatch):
    """A SessionDB whose init blocks (locked-DB migration) must not
    stall the event loop. The kick call is bounded by the one-time init
    window. Every later call is bounded by the short per-call window.
    Both calls degrade to None.

    The windows are monkeypatched down so the test costs well under a
    second of wall time instead of sleeping through the production
    1.5s window plus margins; the two-window CONTRACT is what's under
    test, not the production constants.
    """

    monkeypatch.setattr(goals, "_DB_BOOTSTRAP_INIT_WAIT_S", 0.3)
    monkeypatch.setattr(goals, "_DB_BOOTSTRAP_LOOP_WAIT_S", 0.05)

    class _BlockingDB:
        def __init__(self, db_path=None):
            # Far past both (shrunk) wait windows. The margin keeps the
            # "still None" assertions from racing the bootstrap thread on
            # a loaded runner (negative-timing race, flake policy).
            time.sleep(1.5)  # simulated contended migration

        def get_meta(self, key):
            return None

    _patch_sessiondb(monkeypatch, _BlockingDB)
    elapsed = None
    elapsed2 = None
    result = "UNSET"
    result2 = "UNSET"

    async def main():
        nonlocal elapsed, elapsed2, result, result2
        t0 = time.monotonic()
        result = goals._get_session_db()
        elapsed = time.monotonic() - t0
        t0 = time.monotonic()
        result2 = goals._get_session_db()
        elapsed2 = time.monotonic() - t0

    asyncio.run(main())
    assert result is None, "contended init must degrade to None"
    # Each ceiling gets ~1s slack over its (shrunk) window; the 1.5s
    # blocking init exceeds both ceilings if it ever runs on-loop, and a
    # loaded runner does not cross either one.
    assert elapsed is not None and elapsed < 1.0, (
        f"kick call blocked for {elapsed:.2f}s — past the one-time init window"
    )
    # The in-flight call waits the short window, not the init window.
    # A contended migration must not stall the loop repeatedly.
    assert result2 is None, "contended init must keep degrading to None"
    assert elapsed2 is not None and elapsed2 < 1.0, (
        f"in-flight call blocked for {elapsed2:.2f}s — watchdog territory"
    )
    # The kick call must wait the LONGER window: otherwise a healthy cold
    # init loses its grace period and the first write drops again.
    assert elapsed > elapsed2, (
        "kick call should wait the one-time init window; in-flight calls the short one"
    )