"""Entity Dossier card tests — the explicit-ask intercept, the mention
auto-show, the /profile fetch + emit contract (voice/entity_show.py), and the
HUD→engine contract (control/entity_action.json, applied by
ControlWatcher._entity_action). The JSON payloads here are EXACTLY what the
Mac HUD's EntityPanel writes, so these tests pin the contract between the
Swift card and the Python engine. No network, no HUD, no audio (fakes).

Run:            .venv/bin/python -m pytest test_entity_card.py -q
Live smoke:     LIVE=1 .venv/bin/python -m pytest test_entity_card.py -q -k live
  (LIVE hits the real memory service /profile AND writes a real
   control/entity_action.json so a RUNNING Jarvis pops the card on screen —
   the eyes-on proof that the whole chain renders. Run it after any deploy
   that touches the card, the endpoint, or the intercept.)
"""
from __future__ import annotations

import json
import os
import time

import pytest

from voice import entity_show


# ---------------------------------------------------------------------------
# 1. Explicit-ask intercept — the DETERMINISTIC path ("pull up X" must fire
#    the card in code, before any brain/delegation can route it to a file).
# ---------------------------------------------------------------------------
EXPLICIT_CASES = [
    ("pull up everything we know about AIN", "AIN"),
    ("jarvis pull up AIN", "AIN"),
    ("show me Saud Altamimi", "Saud Altamimi"),
    ("who is Mohamed", "Mohamed"),
    ("who's Hisham", "Hisham"),
    ("tell me about Alrugaib", "Alrugaib"),
    ("tell me everything about Reval Studio", "Reval Studio"),
    ("what do we know about Dinasor", "Dinasor"),
    ("pull up the profile of Wafi Energy", "Wafi Energy"),
    ("من هو سعود", "سعود"),
    ("كل شي نعرفه عن عين", "عين"),
]
NEGATIVE_CASES = [
    "let's push the code now",
    "pull up",                       # no name
    "the show me nothing phrase isn't at the end so no tail",
]


@pytest.mark.parametrize("utterance,expected", EXPLICIT_CASES)
def test_explicit_ask_fires_show(monkeypatch, utterance, expected):
    """explicit_ask runs the show INLINE (truth over speed) and returns
    (status, name) so the brain note can never claim a card that isn't there."""
    fired: list[str] = []
    monkeypatch.setattr(entity_show, "_show",
                        lambda n: (fired.append(n) or ("shown", n, [])))
    monkeypatch.setattr(entity_show, "mentioned_entities", lambda t: [])
    entity_show._COOLDOWN.clear()
    res = entity_show.explicit_ask(utterance)
    assert res and res[0] == "shown" and expected.lower() in res[1].lower()
    assert fired and expected.lower() in fired[0].lower()


@pytest.mark.parametrize("utterance", NEGATIVE_CASES)
def test_explicit_ask_ignores_non_asks(monkeypatch, utterance):
    fired: list[str] = []
    monkeypatch.setattr(entity_show, "_show",
                        lambda n: (fired.append(n) or ("shown", n, [])))
    monkeypatch.setattr(entity_show, "mentioned_entities", lambda t: [])
    entity_show._COOLDOWN.clear()
    result = entity_show.explicit_ask(utterance)
    # A plain statement and a bare "pull up" (no name) must never fire — and
    # return None. The 3rd case is deliberately lenient: the greedy 'show me …$'
    # pattern legitimately matches its trailing clause (matches original intent).
    if utterance in (NEGATIVE_CASES[0], NEGATIVE_CASES[1]):
        assert result is None and not fired


def test_explicit_ask_truthful_on_unknown(monkeypatch):
    """'pull up reval clients' when no such entity exists → ('unknown', …) so
    the brain says "nothing found" instead of "on your HUD, sir" (the gaslight
    Ahmed hit)."""
    monkeypatch.setattr(entity_show, "_show", lambda n: ("unknown", n, []))
    monkeypatch.setattr(entity_show, "mentioned_entities", lambda t: [])
    entity_show._COOLDOWN.clear()
    res = entity_show.explicit_ask("pull up everything about reval clients")
    assert res and res[0] == "unknown"


def test_explicit_ask_prefers_known_entity(monkeypatch):
    """'show me saud' + known 'Saud Altamimi' → fires the canonical name."""
    fired: list[str] = []
    monkeypatch.setattr(entity_show, "_show",
                        lambda n: (fired.append(n) or ("shown", n, [])))
    monkeypatch.setattr(entity_show, "mentioned_entities",
                        lambda t: ["Saud", "Saud Altamimi"])
    entity_show._COOLDOWN.clear()
    assert entity_show.explicit_ask("show me saud") == ("shown", "Saud Altamimi")
    assert fired == ["Saud Altamimi"]     # longest known match wins


def test_explicit_ask_double_fire_guard(monkeypatch):
    fired: list[str] = []
    monkeypatch.setattr(entity_show, "_show",
                        lambda n: (fired.append(n) or ("shown", n, [])))
    monkeypatch.setattr(entity_show, "mentioned_entities", lambda t: [])
    entity_show._COOLDOWN.clear()
    # both calls return (status, name); the 2nd is guarded → _show runs ONCE
    assert entity_show.explicit_ask("who is Mohamed") == ("shown", "Mohamed")
    assert entity_show.explicit_ask("who is Mohamed") == ("shown", "Mohamed")
    assert len(fired) == 1


def test_handle_utterance_returns_status_on_explicit_ask(monkeypatch):
    """main.py's integration point: handle_utterance returns (status, name) on
    an explicit ask (driving a truthful brain note), None otherwise."""
    monkeypatch.setattr(entity_show, "_show", lambda n: ("shown", n, []))
    monkeypatch.setattr(entity_show, "mentioned_entities", lambda t: [])
    entity_show._COOLDOWN.clear()
    assert entity_show.handle_utterance("who is Mohamed") == ("shown", "Mohamed")
    entity_show._COOLDOWN.clear()
    # a passive mention (auto-show path) returns None — no brain note wanted
    assert entity_show.handle_utterance("let's push the code now") is None


# ---------------------------------------------------------------------------
# 2. show_entity — fetch → emit contract (the event the Swift card parses).
# ---------------------------------------------------------------------------
FAKE_PROFILE = {
    "entity": {"key": "ain", "name": "AIN", "etype": "thing",
               "aliases": [], "created_at": "2026-07-01T00:00:00+00:00"},
    "summary": "AIN is Reval Studio's client.",
    "read": "",
    "quick": {"phones": ["0544067114"], "emails": [], "urls": [], "amounts": []},
    "counts": {"facts": 27, "relations": 12, "last_touch": "2026-07-15"},
    "relations": [{"other": "Reval Studio", "otype": "company",
                   "rtype": "mentioned-with", "direction": "out",
                   "other_summary": "Ahmed's studio."}],
    "timeline": [{"id": "x1", "date": "2026-07-15", "kind": "fact",
                  "fact": "Ain lead — Mohammad.", "tone": ""}],
}
# The COMPOSED dossier the parallel server writes (POST /dossier → composed).
FAKE_COMPOSED = {
    "headline": "AIN — Reval Studio's client",
    "sections": [{"title": "Who", "body": "A lead handed to Reval Studio."}],
    "cached": False,
}


def _wait_for(pred, timeout=2.0):
    deadline = time.time() + timeout
    while not pred() and time.time() < deadline:
        time.sleep(0.02)


def test_show_entity_two_stage_emit(monkeypatch):
    """STAGE 1: card pops instantly (composing=True, composed=None). STAGE 2 (on
    a daemon thread): the composed dossier swaps in (composing=False,
    composed=<dict>). The spoken confirmation returns after STAGE 1 — it never
    waits for composition."""
    events: list[tuple] = []
    monkeypatch.setattr(entity_show, "fetch_profile", lambda n: dict(FAKE_PROFILE))
    monkeypatch.setattr(entity_show, "fetch_composed", lambda n: dict(FAKE_COMPOSED))
    monkeypatch.setattr(entity_show, "emit",
                        lambda etype, **kw: events.append((etype, kw)))
    msg = entity_show.show_entity("AIN")
    # confirmation comes back the moment the card is up (STAGE 1), not later
    assert "showing" in msg.lower() or "profile" in msg.lower()
    # STAGE 1 event — raw profile, spinner on, no composition yet
    assert events and events[0][0] == "entity_profile"
    e1 = events[0][1]
    assert e1["composing"] is True and e1["composed"] is None
    profile = e1["profile"]
    # the fields the Swift EntityProfile.from() parser REQUIRES
    assert profile["entity"]["name"] == "AIN"
    assert profile["entity"]["etype"] == "thing"
    assert isinstance(profile["quick"]["phones"], list)
    assert isinstance(profile["timeline"], list)
    # STAGE 2 event — composed dossier swaps in on the daemon thread
    _wait_for(lambda: len(events) >= 2)
    assert len(events) == 2 and events[1][0] == "entity_profile"
    e2 = events[1][1]
    assert e2["composing"] is False
    assert e2["composed"] == FAKE_COMPOSED
    assert e2["profile"]["entity"]["name"] == "AIN"   # same profile, composed added


def test_show_entity_compose_failure_degrades(monkeypatch):
    """POST /dossier 404s / times out / returns null → STAGE 2 still emits, with
    composed=None composing=False, so the HUD drops the spinner and renders the
    raw profile. STAGE 1 is unaffected."""
    events: list[tuple] = []
    monkeypatch.setattr(entity_show, "fetch_profile", lambda n: dict(FAKE_PROFILE))
    monkeypatch.setattr(entity_show, "fetch_composed", lambda n: None)  # 404/timeout
    monkeypatch.setattr(entity_show, "_COMPOSE_RETRY_S", 0.0)  # skip the live retry wait
    monkeypatch.setattr(entity_show, "emit",
                        lambda etype, **kw: events.append((etype, kw)))
    entity_show.show_entity("AIN")
    assert events[0][1]["composing"] is True and events[0][1]["composed"] is None
    _wait_for(lambda: len(events) >= 2)
    assert len(events) == 2
    e2 = events[1][1]
    assert e2["composed"] is None and e2["composing"] is False


def test_show_entity_unknown_is_silent(monkeypatch):
    events: list = []
    monkeypatch.setattr(entity_show, "fetch_profile",
                        lambda n: {"entity": None, "summary": "", "read": "",
                                   "quick": {}, "counts": {}})
    # fetch_composed must NEVER be reached for an unknown name (no STAGE 2)
    monkeypatch.setattr(entity_show, "fetch_composed",
                        lambda n: pytest.fail("composed unknown entity"))
    monkeypatch.setattr(entity_show, "emit",
                        lambda etype, **kw: events.append(etype))
    msg = entity_show.show_entity("Nobody Whatsoever")
    time.sleep(0.1)                       # give any stray thread a chance to err
    assert not events                     # NO card event for an unknown name
    assert "don't have" in msg.lower()


# ---------------------------------------------------------------------------
# 3. HUD → engine back-channel contract (what the Swift EntityPanel writes).
#    Pin that ControlWatcher knows entity_action.json and dispatches all 3 cmds.
# ---------------------------------------------------------------------------
def test_every_input_path_fires_the_intercept():
    """Ahmed's law: "pull up X" means the HUD card on EVERY input path. Pin
    that BOTH user-input entry points — the spoken listen loop and the typed
    HUD chat box (handle_text) — run entity_show.handle_utterance. A new input
    path added without the intercept should fail this test."""
    src = open(os.path.join(os.path.dirname(__file__), "main.py"),
               encoding="utf-8").read()
    typed = src.split("async def handle_text", 1)[1].split("\n    async def", 1)[0]
    assert "handle_utterance" in typed, \
        "typed HUD messages must fire the dossier intercept"
    assert src.count("handle_utterance(") >= 2, \
        "expected the intercept on both the spoken and typed paths"


def test_entity_note_is_a_curation_agent():
    """The card's note field must dispatch a pooled Claude worker that FIXES
    the graph (memory_update/memory_remember + card refresh), with the plain
    saved note only as the pool-down fallback — never the primary path."""
    import inspect

    import voice.control as control
    src = inspect.getsource(control)
    agent = src.split("def _entity_note_agent", 1)
    assert len(agent) == 2, "control.py lost the entity-note curation agent"
    body = agent[1].split("\n    async def", 1)[0]
    for needle in ("agent_pool", "memory_update", "memory_remember",
                   "show_entity", "fire_save"):
        assert needle in body, f"curation agent lost its {needle} wiring"


def test_control_watcher_handles_entity_action():
    import inspect

    import voice.control as control
    src = inspect.getsource(control)
    assert "entity_action.json" in src, \
        "control.py no longer watches the HUD card's back-channel file"
    assert "_entity_action" in src
    for cmd in ('"show_entity"', '"entity_note"', '"entity_summary"'):
        assert cmd.strip('"') in src, f"cmd {cmd} not dispatched in control.py"


# ---------------------------------------------------------------------------
# 4. LIVE smoke (LIVE=1) — the real chain: server /profile, then a real
#    control-file injection so a running Jarvis renders the card on screen.
# ---------------------------------------------------------------------------
@pytest.mark.skipif(os.environ.get("LIVE") != "1", reason="LIVE=1 only")
def test_live_profile_and_card():
    profile = entity_show.fetch_profile("AIN")
    assert profile and profile.get("entity"), "live /profile returned nothing"
    assert profile["counts"]["facts"] > 0
    ctrl = os.path.join(os.path.dirname(__file__), "control",
                        "entity_action.json")
    tmp = ctrl + ".tmp"
    with open(tmp, "w") as f:
        json.dump({"cmd": "show_entity", "name": "AIN"}, f)
    os.replace(tmp, ctrl)
    time.sleep(6)
    assert not os.path.exists(ctrl), \
        "engine did not consume entity_action.json — is Jarvis running?"
    print("\nLIVE: command consumed — the AIN card should be ON SCREEN now.")