"""Tests for the Synapse reaction runner (dispatcher meta-pass → session run)."""

import asyncio
import json
from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest

import navi.api.deps as deps
import navi.synapse.reactions as reactions
from navi.core.registry import ProfileRegistry
from navi.profiles import ALL_PROFILES


def _registry() -> ProfileRegistry:
    reg = ProfileRegistry()
    for p in ALL_PROFILES:
        reg.register(p)
    return reg


# ── _parse_decision ─────────────────────────────────────────────────────────

def test_parse_decision_plain_and_fenced():
    plain = json.dumps({"profile_id": "developer", "task": "t", "understood": "u"})
    assert reactions._parse_decision(plain) == {
        "profile_id": "developer", "task": "t", "understood": "u"
    }

    fenced = "Sure:\n```json\n" + plain + "\n```"
    assert reactions._parse_decision(fenced) == {
        "profile_id": "developer", "task": "t", "understood": "u"
    }


def test_parse_decision_garbage_and_skip():
    assert reactions._parse_decision("no json here") is None
    assert reactions._parse_decision("") is None
    decision = reactions._parse_decision('{"skip": true, "reason": "junk delivery"}')
    assert decision is not None and decision["skip"] is True


# ── _resolve_profile ────────────────────────────────────────────────────────

def test_a_user_visible_profile_resolves():
    reg = _registry()
    assert reactions._resolve_profile(reg, "assistant").id == "assistant"


def test_an_admin_only_profile_is_not_a_reaction_target():
    """A reaction run inherits role "user" (see _spawn_run_task), so landing on an
    admin-only profile would hand the triggering user a wider tool surface than
    their own profile list offers them."""
    reg = _registry()
    assert reg.get("secretary").is_admin_only

    resolved = reactions._resolve_profile(reg, "secretary")

    assert resolved.id != "secretary"
    assert not resolved.is_admin_only


def test_hidden_profile_falls_back_to_a_user_visible_one():
    reg = _registry()
    # dispatcher exists but is hidden → must never host a session
    assert reg.get("dispatcher").is_hidden
    resolved = reactions._resolve_profile(reg, "dispatcher")
    assert not resolved.is_hidden
    assert not resolved.is_admin_only


def test_unknown_profile_falls_back_to_a_user_visible_one():
    reg = _registry()
    resolved = reactions._resolve_profile(reg, "astronaut")
    assert not resolved.is_hidden
    assert not resolved.is_admin_only


# ── the records the event leaves behind ──────────────────────────────────────

def test_event_card_is_a_system_record_carrying_the_request():
    card = reactions._event_card(
        {"event_id": "e1"}, "gntodo.task.created", "tg:42", "Напиши в чат",
    )
    assert card.role == "system"
    assert card.content == "Напиши в чат"
    assert card.is_display is True   # the reader sees it
    assert card.is_context is True   # and so does the model
    assert card.metadata["source"] == "synapse_event"
    assert card.metadata["event_type"] == "gntodo.task.created"
    assert card.metadata["session_key"] == "tg:42"


def test_context_message_frames_a_new_thread_and_not_a_continued_one():
    envelope = {"event_id": "e1", "payload": {"chat_id": 42}}

    fresh = reactions._reaction_context_message(envelope, fresh=True)
    assert "background reaction session" in fresh
    assert '"chat_id": 42' in fresh
    # the task travels in the visible card, never twice
    assert "## Task" not in fresh

    continued = reactions._reaction_context_message(envelope, fresh=False)
    assert "background reaction session" not in continued
    assert '"chat_id": 42' in continued


def test_record_event_keeps_the_latest_event_on_top_and_a_short_trail():
    class _S:
        session_metadata: dict = {}

    s = _S()
    for i in range(reactions._EVENTS_KEPT + 5):
        reactions._record_event(s, {"event_id": f"e{i}"}, "tg.message", "u", "tg:1")

    meta = s.session_metadata["synapse"]
    assert meta["event_id"] == f"e{reactions._EVENTS_KEPT + 4}"
    assert meta["session_key"] == "tg:1"
    assert len(meta["events"]) == reactions._EVENTS_KEPT
    assert meta["events"][0]["event_id"] == "e5"


# ── _dispatch ───────────────────────────────────────────────────────────────

async def test_dispatch_returns_decision(monkeypatch):
    decision = json.dumps({
        "profile_id": "developer", "task": "fix it", "understood": "ok",
        "session_key": "tg:42",
    })
    backend = MagicMock()
    backend.complete = AsyncMock(
        return_value=SimpleNamespace(content=f"```json\n{decision}\n```")
    )
    monkeypatch.setattr(
        deps, "get_profile_registry", lambda: _registry()
    )
    monkeypatch.setattr(
        deps, "get_backend_registry", lambda: SimpleNamespace(get=lambda key: backend)
    )

    profile_id, task, understood, session_key = await reactions._dispatch(
        {"event_id": "e1"}, "reaction rules", "route tg.* to assistant"
    )
    assert (profile_id, task, understood, session_key) == ("developer", "fix it", "ok", "tg:42")
    # the dispatcher pass sees both documents and the envelope — nothing else
    user_msg = backend.complete.await_args.args[0][1].content
    assert "reaction rules" in user_msg
    assert "route tg.* to assistant" in user_msg
    assert "event_id" in user_msg


async def test_dispatch_without_a_session_key_yields_none(monkeypatch):
    """A missing or non-string session_key must never become a thread key."""
    backend = MagicMock()
    backend.complete = AsyncMock(return_value=SimpleNamespace(content=json.dumps({
        "profile_id": "developer", "task": "t", "understood": "u", "session_key": None,
    })))
    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        deps, "get_backend_registry", lambda: SimpleNamespace(get=lambda key: backend)
    )

    assert (await reactions._dispatch({}, "i", ""))[3] is None

    backend.complete = AsyncMock(return_value=SimpleNamespace(content=json.dumps({
        "profile_id": "developer", "task": "t", "understood": "u", "session_key": "   ",
    })))
    assert (await reactions._dispatch({}, "i", ""))[3] is None


async def test_dispatch_backend_failure_yields_skip(monkeypatch):
    backend = MagicMock()
    backend.complete = AsyncMock(side_effect=RuntimeError("boom"))
    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        deps, "get_backend_registry", lambda: SimpleNamespace(get=lambda key: backend)
    )
    assert await reactions._dispatch({}, "", "") == (None, "", "", None)


# ── run_reaction ────────────────────────────────────────────────────────────

class _FakeSession:
    def __init__(self, session_id, profile_id="assistant"):
        self.id = session_id
        self.profile_id = profile_id
        self.special = False
        self.name = ""
        self.messages: list = []
        self.context: list = []
        self.session_metadata: dict = {}
        self.last_active = datetime.now(timezone.utc)


def _fake_store(saved, *, found=None):
    store = MagicMock()

    async def fake_create(profile_id, user_id=None):
        session = _FakeSession(f"sess-{len(saved) + 1}", profile_id)
        saved.append((profile_id, user_id, session))
        return session

    store.create = fake_create
    store.save = AsyncMock()
    store.set_profile = AsyncMock(return_value=True)
    # AsyncMock(return_value=None), never a bare MagicMock: a MagicMock returns a
    # truthy mock and every event would look like a continuation.
    store.find_reaction_session = AsyncMock(return_value=found)
    # PgSessionStore._get_pool is async; the settings store is mocked out in
    # these tests, so the pool itself is never used.
    store._get_pool = AsyncMock(return_value=None)
    return store


def _fake_orchestrator(queue, *, running=lambda sid: False):
    orchestrator = MagicMock()

    session_locks: dict = {}

    def fake_lock(session_id):
        # mirror the real orchestrator: session_lock returns an asyncio.Lock
        return session_locks.setdefault(session_id, asyncio.Lock())

    orchestrator.session_lock = fake_lock
    orchestrator.is_running = running

    run = MagicMock()
    run.subscribe = lambda: queue
    run.unsubscribe = lambda q: None
    orchestrator.create_run = MagicMock(side_effect=lambda sid: run)
    orchestrator.run_agent = AsyncMock()
    return orchestrator, run


def _settings(monkeypatch, **overrides):
    """Patch SynapseSettingsStore.get to return one settings row."""
    from navi.synapse.settings_store import SynapseSettings, SynapseSettingsStore

    current = SynapseSettings(user_id="u1", reactions_enabled=True, **overrides)

    async def fake_get(self, user_id):
        return current

    monkeypatch.setattr(SynapseSettingsStore, "get", fake_get)
    return current


async def test_run_reaction_disabled_gate(monkeypatch):
    """reactions_enabled=False → no session is created, nothing is dispatched."""
    from navi.synapse.settings_store import SynapseSettings

    saved: list = []
    store = _fake_store(saved)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    async def fake_get(self, user_id):
        return SynapseSettings(user_id=user_id)
    monkeypatch.setattr(
        "navi.synapse.settings_store.SynapseSettingsStore.get", fake_get
    )
    dispatched: list = []

    async def fake_dispatch(*args, **kwargs):
        dispatched.append(args)
        return "developer", "t", "u", None

    monkeypatch.setattr(reactions, "_dispatch", fake_dispatch)

    await reactions.run_reaction({"event_id": "e1"}, user_id="u1", event_type="todo.created")

    assert dispatched == []
    assert saved == []


async def test_run_reaction_happy_path(monkeypatch):
    """Gate passes → dispatcher picks a profile → special session is created and run."""
    saved: list = []
    store = _fake_store(saved)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch, instructions="watch gntodo")

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch",
        AsyncMock(return_value=("assistant", "fix it", "ok", "tg:42")),
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue)
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)

    # run_agent completes right away: drain must see "done" from queue.
    ran: list = []

    async def fake_run_agent(*args, **kwargs):
        ran.append((args, kwargs))
        await queue.put(("done", None))
    orchestrator.run_agent = fake_run_agent

    # No push service in deps, and no synapse source: finalise must be silent.
    monkeypatch.setattr(deps, "get_push_service", lambda: None)
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: False)

    await reactions.run_reaction(
        {"event_id": "e1", "subject": "gntodo", "action": "created"},
        user_id="u1", event_type="gntodo.task.created",
    )

    profile_id, user_id, session = saved[0]
    assert (profile_id, user_id) == ("assistant", "u1")
    assert session.special is True
    assert session.name == "Synapse: gntodo.task.created"
    assert session.session_metadata["synapse"]["event_id"] == "e1"
    assert session.session_metadata["synapse"]["session_key"] == "tg:42"
    assert len(session.session_metadata["synapse"]["events"]) == 1
    store.save.assert_awaited()

    # The visible record: the request as a system entry, and nothing else.
    card = session.messages[0]
    assert (card.role, card.content) == ("system", "fix it")
    assert card.metadata["source"] == "synapse_event"

    assert len(ran) == 1
    args, kwargs = ran[0]
    assert args[0] == session.id
    assert kwargs["hidden"] is True
    assert args[1].startswith("This is a background reaction session")
    orchestrator.broadcast_session_sync.assert_called_once_with(session.id)


async def test_run_reaction_continues_the_thread_within_the_ttl(monkeypatch):
    """A second event of one conversation joins the session the first one opened."""
    existing = _FakeSession("sess-existing", profile_id="assistant")
    existing.special = True
    existing.name = "Synapse: tg.message"
    existing.session_metadata["synapse"] = {"events": [{"event_id": "e0"}]}

    saved: list = []
    store = _fake_store(saved, found=existing)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch, reaction_session_ttl_minutes=1440)

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch",
        AsyncMock(return_value=("assistant", "answer him", "ok", "tg:42")),
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue)
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)
    ran: list = []

    async def fake_run_agent(*args, **kwargs):
        ran.append((args, kwargs))
        await queue.put(("done", None))
    orchestrator.run_agent = fake_run_agent
    monkeypatch.setattr(deps, "get_push_service", lambda: None)
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: False)

    await reactions.run_reaction({"event_id": "e2"}, user_id="u1", event_type="tg.message")

    assert saved == []  # no new session
    assert store.set_profile.await_count == 0  # profile already matches
    kwargs = store.find_reaction_session.await_args.kwargs
    assert kwargs["session_key"] == "tg:42"
    assert kwargs["user_id"] == "u1"
    assert isinstance(kwargs["not_active_before"], datetime)

    meta = existing.session_metadata["synapse"]
    assert meta["event_id"] == "e2"
    assert [e["event_id"] for e in meta["events"]] == ["e0", "e2"]
    # The thread was already framed when it was opened — do not say it again.
    assert "background reaction session" not in ran[0][0][1]
    assert '"event_id": "e2"' in ran[0][0][1]
    assert existing.messages[0].content == "answer him"


async def test_run_reaction_does_not_continue_when_the_ttl_is_zero(monkeypatch):
    """ttl=0 is the pre-continuation behaviour: never look for a thread."""
    saved: list = []
    store = _fake_store(saved)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch, reaction_session_ttl_minutes=0)

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch",
        AsyncMock(return_value=("assistant", "t", "u", "tg:42")),
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue)
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)

    async def fake_run_agent(*args, **kwargs):
        await queue.put(("done", None))
    orchestrator.run_agent = fake_run_agent
    monkeypatch.setattr(deps, "get_push_service", lambda: None)
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: False)

    await reactions.run_reaction({"event_id": "e3"}, user_id="u1", event_type="tg.message")

    assert store.find_reaction_session.await_count == 0
    assert len(saved) == 1


async def test_run_reaction_starts_a_new_session_when_the_thread_is_busy(monkeypatch):
    """A thread with a live turn must not get a second run over it — the event
    still runs, in a new session."""
    existing = _FakeSession("sess-busy", profile_id="assistant")
    existing.special = True

    saved: list = []
    store = _fake_store(saved, found=existing)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch)

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch",
        AsyncMock(return_value=("assistant", "t", "u", "tg:42")),
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue, running=lambda sid: sid == "sess-busy")
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)

    async def fake_run_agent(*args, **kwargs):
        await queue.put(("done", None))
    orchestrator.run_agent = fake_run_agent
    monkeypatch.setattr(deps, "get_push_service", lambda: None)
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: False)

    await reactions.run_reaction({"event_id": "e4"}, user_id="u1", event_type="tg.message")

    assert len(saved) == 1  # a fresh session was created
    assert saved[0][2].id != "sess-busy"
    assert orchestrator.create_run.call_args_list[-1].args[0] == saved[0][2].id


async def test_run_reaction_switches_the_profile_of_a_continued_session(monkeypatch):
    """The dispatcher may pick another profile for a later event of the same
    thread; the session follows it instead of forking."""
    existing = _FakeSession("sess-existing", profile_id="secretary")
    existing.special = True

    saved: list = []
    store = _fake_store(saved, found=existing)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch)

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch",
        AsyncMock(return_value=("assistant", "t", "u", "tg:42")),
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue)
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)

    async def fake_run_agent(*args, **kwargs):
        await queue.put(("done", None))
    orchestrator.run_agent = fake_run_agent
    monkeypatch.setattr(deps, "get_push_service", lambda: None)
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: False)

    await reactions.run_reaction({"event_id": "e5"}, user_id="u1", event_type="tg.message")

    store.set_profile.assert_awaited_once_with("sess-existing", "assistant")
    assert saved == []
    assert existing.profile_id == "assistant"


async def test_run_reaction_finalise_pushes_on_importance(monkeypatch):
    """completion_notify='important' + errors → app push is sent."""
    saved: list = []
    store = _fake_store(saved)
    monkeypatch.setattr(deps, "get_session_store", lambda: store)
    _settings(monkeypatch, push_target="app")

    monkeypatch.setattr(deps, "get_profile_registry", lambda: _registry())
    monkeypatch.setattr(
        reactions, "_dispatch", AsyncMock(return_value=("secretary", "t", "u", None))
    )

    queue: asyncio.Queue = asyncio.Queue()
    orchestrator, _run = _fake_orchestrator(queue)
    monkeypatch.setattr(deps, "get_orchestrator", lambda: orchestrator)

    async def failing_run_agent(*args, **kwargs):
        await queue.put(("error", "model exploded"))
        await queue.put(("done", None))
    orchestrator.run_agent = failing_run_agent

    pushes: list = []
    class _Push:
        async def notify_custom(self, session_id, user_id, title, body):
            pushes.append((session_id, user_id, title, body))
    monkeypatch.setattr(deps, "get_push_service", lambda: _Push())
    monkeypatch.setattr(reactions, "synapse_source_ready", lambda: True)
    emitted: list = []

    async def fake_emit_low_level(subject, action, payload, priority="low", dedup_key=None):
        emitted.append({"subject": subject, "action": action, "payload": payload})
    import navi.synapse.outbound as outbound_mod
    monkeypatch.setattr(outbound_mod, "emit_low_level", fake_emit_low_level)

    await reactions.run_reaction({"event_id": "e7"}, user_id="u1", event_type="gntodo.x")

    assert len(pushes) == 1
    title = pushes[0][2]
    assert "упала" in title
    assert "model exploded" in pushes[0][3]
    assert emitted[0]["action"] == "failed"
    assert emitted[0]["payload"]["event_id"] == "e7"


# ── real pool wiring ─────────────────────────────────────────────────────────

async def test_pool_of_awaits_the_async_session_pool():
    """Regression: `_pool_of` returned `_get_pool()`'s coroutine, so every
    reaction run died on `.fetchrow` before it could read the settings."""
    import inspect

    pool = object()

    class _SessionStore:
        async def _get_pool(self):
            return pool

    assert inspect.iscoroutinefunction(reactions._pool_of)
    assert await reactions._pool_of(_SessionStore()) is pool


async def test_run_reaction_reads_settings_through_the_real_pool(monkeypatch):
    """The disabled gate is decided by the row that comes out of the pool —
    SynapseSettingsStore and the pool are both real here."""
    from tests.conftest_factory import FakeConnection, FakePool, FakeRecord

    conn = FakeConnection()
    conn.enqueue(FakeRecord(
        user_id="u1", reactions_enabled=False, push_target="app",
        completion_notify="important", instructions="", dispatcher_instructions="",
        reaction_session_ttl_minutes=1440, updated_at=None,
    ))
    pool = FakePool(conn)

    class _SessionStore:
        async def _get_pool(self):
            return pool

    monkeypatch.setattr(deps, "get_session_store", lambda: _SessionStore())

    dispatched: list = []

    async def fake_dispatch(*args, **kwargs):
        dispatched.append(args)
        return "developer", "t", "u", None

    monkeypatch.setattr(reactions, "_dispatch", fake_dispatch)

    await reactions.run_reaction({"event_id": "e9"}, user_id="u1", event_type="todo.created")

    assert dispatched == []
    assert [c[0] for c in conn.calls] == ["fetchrow"]