"""Inbound iMessage threaded (swipe) replies for PhotonAdapter (#100663).

spectrum-ts 12.x wraps a threaded reply as ``{type: "reply", content, target}``. The
sidecar normalises it to ``{type: "reply", content, targetMessageId, targetDirection,
targetText}``; the adapter must unwrap it instead of emitting
"[Photon content type not handled: reply]", and carry the quoted message as reply
context. spectrum usually can't hydrate the text of our own outbound bubbles, so the
adapter records what it sends in ``gateway.rich_sent_store`` and falls back to that.
"""
from __future__ import annotations

import os
from typing import Any, Dict, List, Tuple

import pytest

from gateway.config import PlatformConfig
from gateway.platforms.event import MessageEvent, MessageType
from plugins.platforms.photon import adapter as photon_adapter
from plugins.platforms.photon.adapter import PhotonAdapter

PHONE = "+15550001234"
DM = f"any;-;{PHONE}"


@pytest.fixture(autouse=True)
def _isolated_sent_index(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
    # rich_sent_store writes under HERMES_HOME/state; keep each test's index private.
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))


def _make_adapter(monkeypatch: pytest.MonkeyPatch, **extra: Any) -> PhotonAdapter:
    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
    return PhotonAdapter(PlatformConfig(enabled=True, token="", extra=dict(extra)))


def _capture_handled(adapter: PhotonAdapter, monkeypatch: pytest.MonkeyPatch) -> list[MessageEvent]:
    captured: list[MessageEvent] = []

    async def fake_handle(event: MessageEvent) -> None:
        captured.append(event)

    monkeypatch.setattr(adapter, "handle_message", fake_handle)
    return captured


def _capture_sidecar(adapter: PhotonAdapter, message_id: str = "out-1") -> list[tuple[str, dict[str, Any]]]:
    calls: list[tuple[str, dict[str, Any]]] = []

    async def fake_call(path: str, body: dict[str, Any]) -> dict[str, Any]:
        calls.append((path, body))
        return {"ok": True, "messageId": message_id}

    adapter._sidecar_call = fake_call  # type: ignore[assignment]
    return calls


def _reply_event(inner: dict[str, Any], *, message_id: str = "in-2", target_id: str = "out-0",
                 direction: str | None = "outbound", target_text: str | None = "earlier answer") -> dict[str, Any]:
    return {
        "messageId": message_id,
        "space": {"id": DM, "type": "dm", "phone": PHONE},
        "sender": {"id": PHONE},
        "timestamp": "2026-09-25T18:00:00Z",
        "content": {"type": "reply", "content": inner, "targetMessageId": target_id,
                    "targetDirection": direction, "targetText": target_text},
    }


@pytest.mark.asyncio
async def test_threaded_text_reply_reaches_agent_with_context(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)

    await adapter._dispatch_inbound(_reply_event({"type": "text", "text": "yes do it"}))

    assert len(handled) == 1
    event = handled[0]
    assert event.text == "yes do it"
    assert event.reply_to_message_id == "out-0"
    assert event.reply_to_text == "earlier answer"
    assert event.reply_to_is_own_message is True


# 1x1 transparent PNG (passes the base adapter's image magic check).
_PNG_1X1_B64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="


@pytest.mark.asyncio
async def test_threaded_photo_reply_keeps_the_photo(monkeypatch):
    """The inner content goes through the normal ladder, so a threaded photo is not reduced to text."""
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)

    inner = {"type": "attachment", "name": "pic.png", "mimeType": "image/png", "size": 68,
             "data": _PNG_1X1_B64, "encoding": "base64"}
    await adapter._dispatch_inbound(_reply_event(inner))

    event = handled[0]
    assert event.message_type == MessageType.PHOTO
    assert event.media_types == ["image/png"]
    assert len(event.media_urls) == 1
    assert event.reply_to_message_id == "out-0"


@pytest.mark.asyncio
async def test_reply_to_users_own_message_is_not_marked_own(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)

    await adapter._dispatch_inbound(_reply_event({"type": "text", "text": "ctx"}, direction="inbound"))

    assert handled[0].text == "ctx"
    assert handled[0].reply_to_is_own_message is False


@pytest.mark.asyncio
async def test_malformed_reply_envelope_keeps_fallback_marker(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)

    await adapter._dispatch_inbound(_reply_event({"type": "unknown"}))

    assert handled[0].text == "[Photon content type not handled: reply]"


@pytest.mark.asyncio
async def test_reply_to_our_message_hydrates_quoted_text_from_sent_index(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)
    calls = _capture_sidecar(adapter, message_id="out-1")

    # We send, the user later swipe-replies to that bubble; spectrum gives no target text.
    await adapter.send(DM, "the plan is A then B")
    assert calls[-1][0] == "/send"
    await adapter._dispatch_inbound(
        _reply_event({"type": "text", "text": "do B first"}, target_id="out-1", target_text=None))

    assert handled[-1].text == "do B first"
    assert handled[-1].reply_to_message_id == "out-1"
    assert handled[-1].reply_to_text == "the plan is A then B"
    assert handled[-1].reply_to_is_own_message is True


@pytest.mark.asyncio
async def test_reply_after_restart_is_still_marked_own(monkeypatch):
    """Cloud targets carry no direction; after a restart the index hit alone proves the bubble was ours."""
    sender = _make_adapter(monkeypatch)
    _capture_sidecar(sender, message_id="out-9")
    await sender.send(DM, "the plan is A then B")

    restarted = _make_adapter(monkeypatch)  # fresh in-memory _sent_message_ids
    handled = _capture_handled(restarted, monkeypatch)
    event = _reply_event({"type": "text", "text": "ok"}, target_id="out-9", direction=None, target_text=None)
    await restarted._dispatch_inbound(event)

    assert handled[-1].reply_to_text == "the plan is A then B"
    assert handled[-1].reply_to_is_own_message is True


@pytest.mark.asyncio
async def test_send_to_bare_phone_is_found_from_dm_guid(monkeypatch):
    """Cron/standalone sends often target the bare E.164; replies arrive in the DM GUID space."""
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)
    _capture_sidecar(adapter, message_id="out-7")

    await adapter.send(PHONE, "morning briefing")
    await adapter._dispatch_inbound(
        _reply_event({"type": "text", "text": "more on item 2"}, target_id="out-7", target_text=None))

    assert handled[-1].reply_to_text == "morning briefing"


@pytest.mark.asyncio
async def test_attachment_send_is_recorded_as_label(monkeypatch, tmp_path):
    monkeypatch.setattr(PhotonAdapter, "validate_media_delivery_path",
                        staticmethod(lambda p: p if os.path.exists(p) else None))
    img = tmp_path / "chart.png"
    img.write_bytes(b"\x89PNG fake")
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)
    _capture_sidecar(adapter, message_id="out-3")

    await adapter.send_image_file(DM, str(img))
    await adapter._dispatch_inbound(
        _reply_event({"type": "text", "text": "what is this"}, target_id="out-3", target_text=None))

    assert handled[-1].reply_to_text == "[attachment: chart.png]"


def _fake_standalone_sidecar(monkeypatch: pytest.MonkeyPatch) -> None:
    """Stand-in for the sidecar HTTP API behind ``_standalone_send``: ids are ``cron-<n>``."""
    monkeypatch.setenv("PHOTON_SIDECAR_TOKEN", "tok")
    counter = {"n": 0}

    class _Resp:
        status_code = 200

        def __init__(self, message_id: str):
            self._id = message_id

        def json(self) -> dict[str, Any]:
            return {"ok": True, "messageId": self._id}

    class _FakeClient:
        def __init__(self, *a, **k):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def post(self, url: str, json: dict[str, Any], headers=None):
            counter["n"] += 1
            return _Resp(f"cron-{counter['n']}")

    monkeypatch.setattr(photon_adapter.httpx, "AsyncClient", _FakeClient)


async def _reply_text_for(adapter: PhotonAdapter, monkeypatch: pytest.MonkeyPatch, target_id: str) -> str | None:
    handled = _capture_handled(adapter, monkeypatch)
    await adapter._dispatch_inbound(
        _reply_event({"type": "text", "text": "about this"}, target_id=target_id, target_text=None))
    assert handled[-1].text == "about this"
    return handled[-1].reply_to_text


@pytest.mark.asyncio
async def test_standalone_send_records_text_and_attachments(monkeypatch, tmp_path):
    monkeypatch.setattr(photon_adapter.BasePlatformAdapter, "validate_media_delivery_path",
                        staticmethod(lambda p: p if os.path.exists(p) else None))
    chart = tmp_path / "chart.png"
    chart.write_bytes(b"\x89PNG fake")
    _fake_standalone_sidecar(monkeypatch)
    cfg = PlatformConfig(enabled=True, token="", extra={})

    result = await photon_adapter._standalone_send(cfg, PHONE, "daily digest", media_files=[(str(chart), False)])
    assert result.get("success") is True

    adapter = _make_adapter(monkeypatch)
    assert await _reply_text_for(adapter, monkeypatch, "cron-1") == "daily digest"
    assert await _reply_text_for(adapter, monkeypatch, "cron-2") == "[attachment: chart.png]"


@pytest.mark.asyncio
async def test_poll_clarify_is_recorded_as_its_question(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    calls = _capture_sidecar(adapter, message_id="poll-1")

    await adapter.send_clarify(DM, "Which plan?", ["A", "B"], "clarify-1", "session-1")

    assert calls[-1][0] == "/send-poll"
    assert await _reply_text_for(adapter, monkeypatch, "poll-1") == "Which plan?"


@pytest.mark.asyncio
async def test_plain_fallback_resend_is_recorded(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    _capture_sidecar(adapter, message_id="fallback-1")

    result = await adapter._send_plain_fallback(DM, "plain retry", reply_to=None, metadata=None)

    assert result.success
    assert await _reply_text_for(adapter, monkeypatch, "fallback-1") == "plain retry"


@pytest.mark.asyncio
async def test_unknown_reply_target_leaves_text_empty(monkeypatch):
    adapter = _make_adapter(monkeypatch)
    handled = _capture_handled(adapter, monkeypatch)

    await adapter._dispatch_inbound(_reply_event({"type": "text", "text": "?"}, target_text=None))

    assert handled[-1].text == "?"
    assert handled[-1].reply_to_text is None