"""Relay Slack live cards — gateway half (NS-658).

The relay adapter emits three new additive ops when the connector's
negotiated descriptor advertises them:

  {op: "draft", chat_id, draft_id, content, final, metadata}
  {op: "task_card", chat_id, card_id, chunks, metadata}
  {op: "task_card_stop", chat_id, card_id, metadata}

Gateway side is deliberately dumb: no Slack API knowledge, no new config
keys. Capability is descriptor-driven — an old connector that never
advertises "draft"/"task_card" gets byte-identical behavior to today.

Semantic bridge: the base send_draft contract is Telegram-shaped (draft
clears; final answer is a separate send). Slack native streaming makes the
stream THE message. The adapter tracks the open draft per chat; the
turn-final regular send() for that chat converts to draft(final=true) so
the connector seals the stream instead of posting a duplicate.
"""

from __future__ import annotations

import asyncio

import pytest

from gateway.config import PlatformConfig
from gateway.relay.adapter import RelayAdapter
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
from tests.gateway.relay.stub_connector import StubConnector


def make_desc(**kw) -> CapabilityDescriptor:
    base = dict(
        contract_version=CONTRACT_VERSION,
        platform="slack",
        label="Slack",
        max_message_length=39000,
        supports_draft_streaming=True,
        supports_edit=True,
        supports_threads=True,
        markdown_dialect="slack",
        len_unit="chars",
        emoji="\U0001f4ac",
        platform_hint="",
        pii_safe=False,
        supported_ops=("send", "edit", "typing", "draft", "task_card"),
    )
    base.update(kw)
    return CapabilityDescriptor(**base)


def _connected_adapter(**desc_kw):
    desc = make_desc(**desc_kw)
    stub = StubConnector(desc)
    adapter = RelayAdapter(PlatformConfig(), desc, transport=stub)
    return adapter, stub


def _run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)


@pytest.fixture()
def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()


# ---------------------------------------------------------------------------
# Draft streaming op
# ---------------------------------------------------------------------------


class TestDraftOp:
    def test_send_draft_emits_draft_op(self, loop):
        adapter, stub = _connected_adapter()
        result = loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=7, content="hel")
        )
        assert result.success
        frames = [s for s in stub.sent if s.get("op") == "draft"]
        assert len(frames) == 1
        frame = frames[0]
        assert frame["chat_id"] == "C1"
        assert frame["draft_id"] == 7
        assert frame["content"] == "hel"
        assert frame["final"] is False

    def test_send_draft_requires_descriptor_op(self, loop):
        # supports_draft_streaming True but op NOT advertised: old connector.
        adapter, stub = _connected_adapter(
            supported_ops=("send", "edit", "typing")
        )
        with pytest.raises(NotImplementedError):
            loop.run_until_complete(
                adapter.send_draft(chat_id="C1", draft_id=1, content="x")
            )
        assert not [s for s in stub.sent if s.get("op") == "draft"]

    def test_supports_draft_streaming_requires_both_flag_and_op(self):
        both, _ = _connected_adapter()
        assert both.supports_draft_streaming() is True
        flag_only, _ = _connected_adapter(
            supported_ops=("send", "edit", "typing")
        )
        assert flag_only.supports_draft_streaming() is False
        # Legacy connector with EMPTY supported_ops: fail-open elsewhere, but
        # draft must NOT fail open — the op did not exist pre-contract.
        legacy, _ = _connected_adapter(supported_ops=())
        assert legacy.supports_draft_streaming() is False

    def test_final_send_seals_open_draft(self, loop):
        adapter, stub = _connected_adapter()
        loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=7, content="hel")
        )
        loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=7, content="hello wor")
        )
        stub.next_draft_result = {"success": True, "message_id": "1723600000.1"}
        result = loop.run_until_complete(
            adapter.send("C1", "hello world", reply_to=None)
        )
        assert result.success
        # No regular send op — the final frame rides the draft stream.
        ops = [s["op"] for s in stub.sent]
        assert "send" not in ops, "final must seal the stream, not post anew"
        finals = [
            s for s in stub.sent if s.get("op") == "draft" and s.get("final")
        ]
        assert len(finals) == 1
        assert finals[0]["content"] == "hello world"
        # Stream ts comes back as the message identity.
        assert result.message_id == "1723600000.1"

    def test_send_without_open_draft_is_normal(self, loop):
        adapter, stub = _connected_adapter()
        loop.run_until_complete(adapter.send("C1", "plain", reply_to=None))
        assert [s["op"] for s in stub.sent] == ["send"]

    def test_draft_state_clears_after_seal(self, loop):
        adapter, stub = _connected_adapter()
        loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=7, content="a")
        )
        loop.run_until_complete(adapter.send("C1", "a!", reply_to=None))
        # Second send on the same chat is a NORMAL send again.
        loop.run_until_complete(adapter.send("C1", "followup", reply_to=None))
        ops = [s["op"] for s in stub.sent]
        assert ops == ["draft", "draft", "send"]

    def test_draft_failure_result_propagates(self, loop):
        adapter, stub = _connected_adapter()
        stub.next_draft_result = {"success": False, "error": "stream_gone"}
        result = loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=1, content="x")
        )
        assert not result.success
        # DEFINITE connector rejection disarms seal-interception: the
        # stream consumer falls back to edit-based streaming on this
        # failure, and its turn-final must go out as a REAL send — never
        # a seal on a stream the connector just rejected.
        stub.next_send_result = {"success": True, "message_id": "m2"}
        loop.run_until_complete(adapter.send("C1", "final", reply_to=None))
        assert [s["op"] for s in stub.sent][-1] == "send"

    def test_draft_transport_exception_keeps_interception_armed(self, loop):
        """Ambiguity contract (G-D1): a transport EXCEPTION — as opposed
        to an explicit rejection — may mean the frame was delivered, so
        interception stays armed and the turn-final still seals."""
        adapter, _ = _connected_adapter()

        class _FlakyOnce:
            def __init__(self):
                self.calls = 0

            async def send_outbound(self, payload, platform=None):
                self.calls += 1
                if self.calls == 1:
                    raise ConnectionError("mid-write drop")
                return {"success": True, "message_id": "ts.9"}

        adapter._transport = _FlakyOnce()
        result = loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=2, content="x")
        )
        assert not result.success
        # Still armed: the turn-final converts to the sealing frame.
        final = loop.run_until_complete(adapter.send("C1", "final", reply_to=None))
        assert final.success
        assert final.message_id == "ts.9"

    def test_drafts_are_per_chat(self, loop):
        adapter, stub = _connected_adapter()
        loop.run_until_complete(
            adapter.send_draft(chat_id="C1", draft_id=1, content="a")
        )
        # A send to a DIFFERENT chat is untouched.
        loop.run_until_complete(adapter.send("D9", "other", reply_to=None))
        by_op = [(s["op"], s["chat_id"]) for s in stub.sent]
        assert ("send", "D9") in by_op


# ---------------------------------------------------------------------------
# Task-card ops (#85476 TurnRunner seam, relay leg)
# ---------------------------------------------------------------------------


def _tasks():
    return [
        {
            "id": "call_1",
            "title": "terminal",
            "status": "in_progress",
            "details": "ls -la",
        },
        {
            "id": "call_2",
            "title": "web_search",
            "status": "complete",
            "details": "slack startStream",
        },
    ]


class TestTaskCardOps:
    def test_progress_emits_task_card_op(self, loop):
        adapter, stub = _connected_adapter()
        result = loop.run_until_complete(
            adapter.send_native_task_card_progress(
                chat_id="C1", tasks=_tasks(), reply_to="1700.100"
            )
        )
        assert result.success
        frames = [s for s in stub.sent if s.get("op") == "task_card"]
        assert len(frames) == 1
        assert frames[0]["card_id"] == "turn:1700.100"
        chunk_ids = [c["id"] for c in frames[0]["chunks"]]
        assert chunk_ids == ["call_1", "call_2"]
        statuses = {c["id"]: c["status"] for c in frames[0]["chunks"]}
        assert statuses["call_2"] == "complete"

    def test_stop_emits_task_card_stop(self, loop):
        adapter, stub = _connected_adapter()
        loop.run_until_complete(
            adapter.send_native_task_card_progress(
                chat_id="C1", tasks=_tasks(), reply_to="1700.100"
            )
        )
        loop.run_until_complete(
            adapter.stop_native_task_card_progress(chat_id="C1", reply_to="1700.100")
        )
        assert [s["op"] for s in stub.sent] == ["task_card", "task_card_stop"]

    def test_task_card_gated_on_descriptor(self, loop):
        adapter, stub = _connected_adapter(
            supported_ops=("send", "edit", "typing", "draft")
        )
        # No "task_card" in supported_ops: the TurnRunner's hasattr gate must
        # see NO capability — expose it via a probe method returning False,
        # and the send must be a clean no-op failure (never an exception on
        # the turn path).
        assert adapter.supports_native_task_cards() is False
        result = loop.run_until_complete(
            adapter.send_native_task_card_progress(
                chat_id="C1", tasks=_tasks(), reply_to="t"
            )
        )
        assert not result.success
        assert not stub.sent

    def test_task_card_legacy_empty_ops_not_fail_open(self):
        legacy, _ = _connected_adapter(supported_ops=())
        assert legacy.supports_native_task_cards() is False