"""Inactivity-watchdog test for ClaudeBrain.reply().

A hung SDK response stream must NOT block reply() forever (which would hold
the brain lock and freeze the HUD spinner). With a fake client whose stream
goes silent, reply() should trip _BRAIN_INACTIVITY_S, yield a recovery line,
release the lock, and leave the turn marked stale for the next drain. A
normal stream must be unaffected.

Run:  .venv/bin/python -m pytest test_brain_watchdog.py -q
"""
from __future__ import annotations

import asyncio

import voice.llm as llm
from voice.llm import ClaudeBrain


class _FakeStreamEvent:
    """Mimics the SDK StreamEvent shape reply() reads (type name matters)."""
    __name__ = "StreamEvent"

    def __init__(self, text: str) -> None:
        self.event = {"type": "content_block_delta",
                      "delta": {"type": "text_delta", "text": text}}


# reply() checks type(msg).__name__ == "StreamEvent"
_FakeStreamEvent.__qualname__ = "StreamEvent"
StreamEvent = type("StreamEvent", (), {
    "__init__": lambda self, text: setattr(
        self, "event",
        {"type": "content_block_delta",
         "delta": {"type": "text_delta", "text": text}})})


class FakeClient:
    def __init__(self, mode: str) -> None:
        self.mode = mode          # "normal" | "hang"
        self.interrupts = 0
        self.session_id = "sess-test"

    async def query(self, text: str) -> None:
        pass

    async def interrupt(self) -> None:
        self.interrupts += 1

    async def receive_response(self):
        if self.mode == "normal":
            yield StreamEvent("Right away, sir. ")
            yield StreamEvent("Done.")
            return
        # hang: emit nothing, then sleep far past the watchdog window
        await asyncio.sleep(60)
        yield StreamEvent("too late")


def _brain(mode: str) -> ClaudeBrain:
    b = ClaudeBrain.__new__(ClaudeBrain)   # skip __init__ (no SDK/env needed)
    b._client = FakeClient(mode)
    b._lock = asyncio.Lock()
    b._turn_open = False
    b._session_saved = True
    return b


def test_normal_stream_completes():
    async def go():
        b = _brain("normal")
        out = [s async for s in b.reply("hello")]
        assert "".join(out).strip() != ""
        assert b._turn_open is False       # clean close
    asyncio.run(go())


def test_hung_stream_recovers_fast():
    async def go():
        old = llm._BRAIN_INACTIVITY_S
        llm._BRAIN_INACTIVITY_S = 0.3      # trip quickly for the test
        try:
            b = _brain("hang")
            t0 = asyncio.get_event_loop().time()
            out = [s async for s in b.reply("do a thing")]
            took = asyncio.get_event_loop().time() - t0
            assert took < 3.0, "watchdog must not wait for the hung stream"
            assert any("hung" in s.lower() for s in out), \
                "should speak a recovery line"
            assert b._turn_open is True, \
                "turn left stale so the next reply drains it"
            # lock must be released — a second call can proceed
            assert not b._lock.locked()
        finally:
            llm._BRAIN_INACTIVITY_S = old
    asyncio.run(go())


def test_disabled_watchdog_setting_parses():
    # BRAIN_INACTIVITY_S=0 path: no wait_for wrapper (can't hang-test without
    # blocking, so just assert the constant is respected as a plain float)
    assert isinstance(llm._BRAIN_INACTIVITY_S, float)


if __name__ == "__main__":
    for name, fn in sorted(globals().items()):
        if name.startswith("test_") and callable(fn):
            fn()
            print(f"  ok  {name}")
    print("all green")