"""Kanban wake events must key to the same session as inbound messages.

Slack session keys include the workspace id, so the wake source the notifier
rebuilds from a subscription row must carry it too. The contract asserted here:
the key built from the wake source byte-matches the key built from an inbound
source for the same conversation, a scope-less key does not, and platforms
without tenant scoping keep their exact key shape.
"""

import asyncio
from dataclasses import replace
from unittest.mock import AsyncMock, MagicMock

from gateway.config import Platform, PlatformConfig
from gateway.kanban_watchers import _wake_scope_id
from gateway.run import GatewayRunner
from gateway.session import build_session_key
from hermes_cli import kanban_db as kb
from plugins.platforms.slack.adapter import SlackAdapter

TEAM = "T0B8U2M6NRE"
CHANNEL = "C0BCDG3H66P"
THREAD = "1720000000.000100"
USER = "U0BCE4NRVKN"


class UnscopedAdapter:
    """Push-capable adapter for a platform without tenant scoping."""

    def __init__(self):
        self.sent = []
        self.handled = []

    async def send(self, chat_id, text, metadata=None):
        self.sent.append({"chat_id": chat_id, "text": text, "metadata": metadata or {}})

    async def handle_message(self, event):
        self.handled.append(event)


def _slack_adapter(channel_team=None):
    """Real SlackAdapter with only its I/O stubbed."""
    adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake-token"))
    adapter._app = MagicMock()
    adapter._app.client = AsyncMock()
    adapter._running = True
    adapter.send = AsyncMock()
    adapter.handle_message = AsyncMock()
    if channel_team:
        adapter._channel_team.update(channel_team)
    return adapter


def _runner(adapter, platform=Platform.SLACK):
    runner = GatewayRunner.__new__(GatewayRunner)
    runner._running = True
    runner.adapters = {platform: adapter}
    runner._kanban_sub_fail_counts = {}
    # A gateway whose dispatcher owns the singleton lock.
    runner._kanban_dispatcher_lock_handle = object()
    return runner


async def _one_notifier_tick(monkeypatch, runner):
    real_sleep = asyncio.sleep

    async def fake_sleep(delay):
        if delay == 5:
            return None
        runner._running = False
        await real_sleep(0)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    await runner._kanban_notifier_watcher(interval=1)


def _completed_subscription(**sub_kwargs):
    conn = kb.connect()
    try:
        tid = kb.create_task(
            conn,
            title="wake scope",
            assignee="worker",
            session_id="origin-session",
        )
        # Push-adapter wake injection is gated on the subscription's
        # delivery_mode ("notify+wake"/"wake") on current main; the plain
        # "notify" default would never reach the wake path under test.
        sub_kwargs.setdefault("delivery_mode", "notify+wake")
        kb.add_notify_sub(conn, task_id=tid, **sub_kwargs)
        kb.complete_task(conn, tid, summary="done")
        return tid
    finally:
        conn.close()


def _wake_source_from(adapter):
    assert adapter.handle_message.await_count == 1, (
        f"expected exactly one wake injection, got {adapter.handle_message.await_count}"
    )
    return adapter.handle_message.await_args.args[0].source


def test_slack_wake_resumes_the_creators_workspace_scoped_session(tmp_path, monkeypatch):
    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope.db"))
    kb.init_db()
    _completed_subscription(
        platform="slack",
        chat_id=CHANNEL,
        chat_type="group",
        thread_id=THREAD,
        user_id=USER,
        # Slack sources are stamped with slack_team_id when subscribing.
        delivery_metadata={"slack_team_id": TEAM, "thread_id": THREAD},
    )

    adapter = _slack_adapter()
    asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter)))

    wake = _wake_source_from(adapter)
    assert wake.scope_id == TEAM

    inbound = adapter.build_source(
        chat_id=CHANNEL,
        chat_type="group",
        user_id=USER,
        thread_id=THREAD,
        scope_id=TEAM,
    )
    wake_key = build_session_key(wake)
    assert wake_key == build_session_key(inbound)
    assert TEAM in wake_key
    # A scope-less source keys to a different session for the same chat.
    assert build_session_key(replace(wake, scope_id=None, guild_id=None)) != wake_key


def test_slack_wake_falls_back_to_the_adapter_channel_workspace_map(tmp_path, monkeypatch):
    """Subscriptions that stored no workspace resolve it from the adapter."""
    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-fallback.db"))
    kb.init_db()
    _completed_subscription(
        platform="slack",
        chat_id=CHANNEL,
        chat_type="group",
        thread_id=THREAD,
        delivery_metadata={"thread_id": THREAD, "chat_type": "group"},
    )

    adapter = _slack_adapter(channel_team={CHANNEL: TEAM})
    asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter)))

    wake = _wake_source_from(adapter)
    assert wake.scope_id == TEAM
    assert TEAM in build_session_key(wake)


def test_unknown_channel_keeps_the_previous_unscoped_wake(tmp_path, monkeypatch):
    """An unresolvable workspace yields an unscoped key, not a wrong scope."""
    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-unknown.db"))
    kb.init_db()
    _completed_subscription(
        platform="slack",
        chat_id=CHANNEL,
        chat_type="group",
    )

    adapter = _slack_adapter()
    asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter)))

    wake = _wake_source_from(adapter)
    assert wake.scope_id is None
    assert build_session_key(wake) == f"agent:main:slack:group:{CHANNEL}"


def test_unscoped_platform_wake_key_is_byte_identical(tmp_path, monkeypatch):
    """Platforms without tenant scoping must keep their exact key shape."""
    monkeypatch.setenv("HERMES_KANBAN_DB", str(tmp_path / "wake-scope-telegram.db"))
    kb.init_db()
    _completed_subscription(
        platform="telegram",
        chat_id="chat-dm",
        chat_type="dm",
    )

    adapter = UnscopedAdapter()
    asyncio.run(_one_notifier_tick(monkeypatch, _runner(adapter, Platform.TELEGRAM)))

    assert len(adapter.handled) == 1
    wake = adapter.handled[0].source
    assert wake.scope_id is None
    assert build_session_key(wake) == "agent:main:telegram:dm:chat-dm"


def test_wake_scope_id_prefers_persisted_metadata_over_the_adapter_map():
    """Persisted metadata wins; the adapter map is only a fallback."""
    adapter = SlackAdapter.__new__(SlackAdapter)
    adapter._channel_team = {CHANNEL: "T_STALE"}

    assert _wake_scope_id(
        adapter, {"chat_id": CHANNEL, "delivery_metadata": {"slack_team_id": TEAM}}
    ) == TEAM
    assert _wake_scope_id(adapter, {"chat_id": CHANNEL}) == "T_STALE"
    assert _wake_scope_id(adapter, {"chat_id": "C_OTHER"}) is None


def test_wake_scope_id_degrades_when_the_adapter_lookup_raises():
    class Exploding:
        def scope_id_for_chat(self, chat_id):
            raise RuntimeError("adapter state gone")

    assert _wake_scope_id(Exploding(), {"chat_id": CHANNEL}) is None


def test_wake_scope_id_is_none_for_adapters_without_the_hook():
    """Adapters that don't resolve scopes leave the wake unscoped."""
    assert _wake_scope_id(UnscopedAdapter(), {"chat_id": CHANNEL}) is None


def test_slack_adapter_reports_the_channel_workspace():
    adapter = SlackAdapter.__new__(SlackAdapter)
    adapter._channel_team = {CHANNEL: TEAM}

    assert adapter.scope_id_for_chat(CHANNEL) == TEAM
    assert adapter.scope_id_for_chat("C_UNKNOWN") is None
    assert adapter.scope_id_for_chat("") is None


def test_slack_adapter_reports_no_scope_for_ambiguous_channels():
    """A channel claimed by two workspaces resolves to no scope."""
    adapter = _slack_adapter()
    adapter._remember_channel_team("D_SHARED", "T_ONE")
    adapter._remember_channel_team("D_SHARED", "T_TWO")

    assert adapter.scope_id_for_chat("D_SHARED") is None