iris_x_hermes/gateway-plugin/tests/test_android.py

"""Tests for the Iris x Hermes android gateway plugin (M4: media).

The plugin lives in the sibling ``iris_x_hermes`` checkout (installed into
``~/.hermes/plugins/iris`` as a symlink in production); tests load it
from the source tree directly so they never depend on that install.

Coverage (docs/13-testing.md §13.1, media bullets):
  * upload via ``POST /v1/media`` reassembles + sha256 verified
  * over-limit (Content-Length) -> ``media_too_large``
  * sha256 mismatch -> ``internal``
  * ``message.send`` with ``media_refs`` -> echo carries ``media[]`` and the
    ``MessageEvent`` carries ``media_urls``/``media_types``
  * ``send_*`` -> ``media.offer`` (fields + message association)
  * ``GET /v1/media/{id}`` serves only allowed paths (denied/unknown -> ``not_found``)
  * kind re-sniffing (don't trust the client)

Run via ``scripts/run_tests.sh tests/gateway/test_android.py``.
"""

from __future__ import annotations

import asyncio
import base64
import hashlib
import importlib.util
import json
import os
import sys
import socket
import threading
from http.client import HTTPConnection
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
import pytest_asyncio

# 1x1 red PNG (67 bytes) -- passes hermes' image magic-byte check.
PNG_1X1 = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJ"
    "AAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
)

TOKEN = "test-iris-token-0123456789"
DEVICE_ID = "test-device"


def _plugin_dir() -> Path:
    env = os.environ.get("IRIS_PLUGIN_DIR")
    if env:
        return Path(env)
    # Works from either copy of this file: gateway-plugin/tests/ (canonical,
    # plugin dir is parents[1]) or the hermes-agent/tests/gateway/ mirror
    # (repo root is parents[3]).
    here = Path(__file__).resolve()
    for candidate in (here.parents[1], here.parents[3] / "gateway-plugin"):
        if (candidate / "protocol.py").is_file():
            return candidate
    return here.parents[1]


def _load_plugin():
    """Load the gateway-plugin package under a unique module name.

    The plugin uses relative imports (``from . import protocol``), so it
    must be imported as a package (``submodule_search_locations``).
    """
    name = "iris_plugin_under_test"
    cached = sys.modules.get(name)
    if cached is not None:
        return cached
    pkg_dir = _plugin_dir()
    if not (pkg_dir / "__init__.py").is_file():
        pytest.fail(f"iris plugin not found at {pkg_dir}")
    spec = importlib.util.spec_from_file_location(
        name, pkg_dir / "__init__.py", submodule_search_locations=[str(pkg_dir)]
    )
    if spec is None or spec.loader is None:
        pytest.fail(f"could not build import spec for {pkg_dir}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    try:
        spec.loader.exec_module(module)
    except Exception:
        sys.modules.pop(name, None)
        raise
    return module


@pytest.fixture(scope="module")
def plugin():
    return _load_plugin()


@pytest.fixture
def adapter(plugin, monkeypatch):
    """A live IrisAdapter with an isolated HERMES_HOME (conftest)."""
    monkeypatch.setenv("IRIS_TOKEN", TOKEN)
    from gateway.platform_registry import PlatformEntry, platform_registry

    # Platform("iris") resolves only once the platform is registered
    # (the plugin's register(ctx) does this in production).
    if not platform_registry.is_registered("iris"):
        platform_registry.register(
            PlatformEntry(
                name="iris",
                label="Android",
                adapter_factory=lambda cfg: None,
                check_fn=lambda: True,
            )
        )

    config = SimpleNamespace(
        extra={
            "host": "127.0.0.1",
            "http_port": 0,  # ephemeral HTTP port
            "max_upload_bytes": 1024 * 1024,  # 1 MiB -- keeps over-limit tests fast
        },
        home_channel=None,
    )
    a = plugin.adapter.IrisAdapter(config)
    yield a
    try:
        a._devices.close()
    except Exception:
        pass
    try:
        a._outbox.close()
    except Exception:
        pass


class HttpTestClient:
    """Mimics the old WS client interface over the HTTP transport (docs/19).

    ``.send(json_str)`` -> ``POST /v1/frame``; ``.recv(timeout)`` -> the next
    frame from the SSE stream (a dict); ``.upload(...)`` -> ``POST /v1/media``
    returning the terminal frame (ack or error). The SSE stream is read in a
    background thread that feeds an asyncio queue, so ``recv`` is awaitable
    exactly like the old WS ``recv``.
    """

    def __init__(self, port: int, cursor: int = 0):
        self._port = port
        self._cursor = cursor
        self._queue: asyncio.Queue = asyncio.Queue()
        self._loop: asyncio.AbstractEventLoop | None = None
        self._thread: threading.Thread | None = None
        self._resp = None

    async def start(self) -> dict:
        """Open the SSE stream; returns the hello.ack frame. Any outbox
        catch-up frames replayed before the hello are buffered and re-enqueued
        so ``recv`` can still see them."""
        self._loop = asyncio.get_running_loop()
        self._thread = threading.Thread(target=self._sse_reader, daemon=True)
        self._thread.start()
        buffered: list = []
        hello = None
        while hello is None:
            frame = await asyncio.wait_for(self._queue.get(), timeout=5)
            if frame is None:
                raise AssertionError("SSE stream closed before hello")
            if frame.get("type") == "hello.ack":
                hello = frame
            else:
                buffered.append(frame)
        for f in buffered:
            self._queue.put_nowait(f)
        return hello

    def _sse_reader(self) -> None:
        conn = HTTPConnection("127.0.0.1", self._port, timeout=30)
        conn.request(
            "GET",
            f"/v1/events?cursor={self._cursor}",
            headers={"Authorization": f"Bearer {TOKEN}", "X-Iris-Device": DEVICE_ID},
        )
        resp = conn.getresponse()
        self._resp = resp
        cur_data: list[str] = []
        loop = self._loop
        assert loop is not None
        while True:
            line = resp.fp.readline()
            if not line:
                break
            line = line.decode("utf-8").rstrip("\r\n")
            if line == "":
                if cur_data:
                    try:
                        frame = json.loads("\n".join(cur_data))
                    except Exception:
                        frame = None
                    if frame is not None:
                        loop.call_soon_threadsafe(self._queue.put_nowait, frame)
                cur_data = []
            elif line.startswith(":"):
                continue  # heartbeat comment
            elif line.startswith("data:"):
                cur_data.append(line[5:].lstrip())
            # id:/event: fields are not needed for the test shim
        loop.call_soon_threadsafe(self._queue.put_nowait, None)  # EOF sentinel

    async def send(self, json_str: str) -> None:
        await asyncio.to_thread(self._post_frame, json_str)

    def _post_frame(self, json_str: str) -> None:
        conn = HTTPConnection("127.0.0.1", self._port, timeout=30)
        conn.request(
            "POST",
            "/v1/frame",
            body=json_str.encode("utf-8"),
            headers={
                "Authorization": f"Bearer {TOKEN}",
                "X-Iris-Device": DEVICE_ID,
                "Content-Type": "application/json",
            },
        )
        resp = conn.getresponse()
        body = resp.read()
        conn.close()
        # Fast responses (validation, channel ops, …) come back on the POST
        # response body via the reply sink, NOT the SSE stream. Enqueue any
        # protocol frame so recv_until sees it (the plain {"ok":true} ack is
        # not a frame and is skipped).
        if body:
            try:
                obj = json.loads(body)
            except Exception:
                obj = None
            if isinstance(obj, dict) and "type" in obj:
                assert self._loop is not None
                self._loop.call_soon_threadsafe(self._queue.put_nowait, obj)

    async def recv(self, timeout: float = 10.0) -> dict:
        frame = await asyncio.wait_for(self._queue.get(), timeout=timeout)
        if frame is None:
            raise ConnectionError("SSE stream closed")
        return frame

    async def upload(
        self,
        media_ref: str,
        data: bytes,
        *,
        kind: str = "image",
        mime: str = "image/png",
        filename: str = "t.png",
        sha256: str | None = None,
    ) -> dict:
        """Drive a media upload via ``POST /v1/media``; returns the terminal
        frame (``media.upload.ack`` or ``error``)."""

        def _do() -> dict:
            conn = HTTPConnection("127.0.0.1", self._port, timeout=60)
            conn.request(
                "POST",
                "/v1/media",
                body=data,
                headers={
                    "Authorization": f"Bearer {TOKEN}",
                    "X-Iris-Device": DEVICE_ID,
                    "Content-Type": mime,
                    "X-Iris-Media-Ref": media_ref,
                    "X-Iris-Media-Kind": kind,
                    "X-Iris-Media-Filename": filename,
                    "X-Iris-Media-Sha256": sha256
                    or hashlib.sha256(data).hexdigest(),
                },
            )
            resp = conn.getresponse()
            body = resp.read()
            conn.close()
            return json.loads(body)

        return await asyncio.to_thread(_do)

    async def pull(self, media_id: str) -> tuple[int, bytes]:
        """Drive a media pull via ``GET /v1/media/{id}``; returns (status, body)."""

        def _do() -> tuple[int, bytes]:
            conn = HTTPConnection("127.0.0.1", self._port, timeout=30)
            conn.request(
                "GET",
                f"/v1/media/{media_id}",
                headers={"Authorization": f"Bearer {TOKEN}", "X-Iris-Device": DEVICE_ID},
            )
            resp = conn.getresponse()
            body = resp.read()
            status = resp.status
            conn.close()
            return status, body

        return await asyncio.to_thread(_do)

    async def close(self) -> None:
        # Interrupt the reader thread's blocking readline() by shutting down
        # the socket first; otherwise resp.close() blocks until the in-flight
        # read returns (the file lock is held for the whole blocking read).
        if self._resp is not None:
            sock = getattr(self._resp.fp, "raw", None)
            sock = getattr(sock, "_sock", None) if sock is not None else None
            if sock is not None:
                try:
                    sock.shutdown(socket.SHUT_RDWR)
                except Exception:
                    pass
            try:
                self._resp.close()
            except Exception:
                pass


@pytest_asyncio.fixture
async def ws_client(adapter):
    """Connected + paired HTTP client; the adapter's server runs on an
    ephemeral port for the duration of the test. Yields ``(client, ack)`` to
    match the old WS fixture shape so test bodies need no signature change."""
    await adapter.connect()
    client = HttpTestClient(adapter._http_server.bound_port)
    try:
        ack = await client.start()
        yield client, ack
    finally:
        await client.close()
        await adapter.disconnect()


async def recv_until(ws, predicate, timeout: float = 10.0) -> list:
    """Collect frames (dicts) until *predicate* matches a frame. Returns all
    frames collected."""
    frames: list = []
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    while True:
        remaining = deadline - loop.time()
        if remaining <= 0:
            raise AssertionError(
                "timed out waiting for frame; got: "
                + ", ".join(f.get("type", "?") if isinstance(f, dict) else "?" for f in frames)
            )
        frame = await ws.recv(timeout=remaining)
        frames.append(frame)
        if predicate(frame):
            return frames


async def upload_file(ws, media_ref: str, data: bytes, *, kind: str = "image",
                      mime: str = "image/png", filename: str = "t.png",
                      request_id: int = 1, sha256: str | None = None) -> dict:
    """Drive a media upload via the HTTP leg; returns the terminal frame
    (ack or error)."""
    return await ws.upload(
        media_ref, data, kind=kind, mime=mime, filename=filename, sha256=sha256
    )


# ── Lifecycle ───────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_disconnect_broadcasts_status_restarting(adapter):
    """Teardown broadcasts ``status{state=restarting}`` before closing the
    streams, so the app can distinguish a clean gateway teardown (restart/
    stop) from a plain network drop — it shows the "Gateway restarting" chat
    notice only when this frame was received (docs/04 §status)."""
    await adapter.connect()
    client = HttpTestClient(adapter._http_server.bound_port)
    try:
        await client.start()
        await adapter.disconnect()
        frames = []
        while True:
            try:
                frames.append(await client.recv(timeout=5))
            except (ConnectionError, asyncio.TimeoutError):
                break
        statuses = [f for f in frames if f.get("type") == "status"]
        assert any(f["payload"]["state"] == "restarting" for f in statuses), (
            f"expected status{{restarting}} before close, got: {frames}"
        )
    finally:
        await client.close()
        # Idempotent: the second call is a no-op on the already-stopped server.
        await adapter.disconnect()


# ── Pure helpers ────────────────────────────────────────────────────────────


def test_reclassify_kind_does_not_trust_client(plugin):
    media = plugin.media
    # Image magic wins over a lying "document" claim.
    assert media.reclassify_kind(PNG_1X1, "document") == "image"
    # Ogg audio claimed as document.
    ogg = b"OggS" + b"\x00" * 16
    assert media.reclassify_kind(ogg, "document") == "audio"
    # Voice stays voice when the bytes are audio.
    assert media.reclassify_kind(ogg, "voice") == "voice"
    # m4a brand claimed as video -> audio.
    m4a = b"\x00\x00\x00\x20ftypm4a " + b"\x00" * 16
    assert media.reclassify_kind(m4a, "video") == "audio"
    # Generic mp4 (isom) claimed as audio -> video.
    mp4 = b"\x00\x00\x00\x20ftypisom" + b"\x00" * 16
    assert media.reclassify_kind(mp4, "audio") == "video"
    # No magic: keep declared video, else document.
    assert media.reclassify_kind(b"hello world", "video") == "video"
    assert media.reclassify_kind(b"hello world", "document") == "document"


def test_mime_and_ext_helpers(plugin):
    media = plugin.media
    assert media.mime_for_path("/x/a.PNG") == "image/png"
    assert media.mime_for_path("/x/clip.MP4") == "video/mp4"
    assert media.mime_for_path("/x/a.unknownext") == "application/octet-stream"
    assert media.ext_for_mime("image/png", "x.jpg", ".jpg") == ".png"
    assert media.ext_for_mime("application/pdf", "report.pdf", ".bin") == ".pdf"
    assert media.ext_for_mime("application/x-unknown", "noext", ".bin") == ".bin"


# ── Runtime-metadata footer (app-controlled display) ────────────────────────


def test_runtime_footer_helper_omits_absent_fields(plugin):
    pf = plugin.protocol.runtime_footer
    # All absent -> empty dict (the app shows nothing).
    assert pf() == {}
    # Only fields with data are included.
    d = pf(model="gpt-5.4", context_pct=38, cwd="~", latency=22.5, cost=0.0012)
    assert d == {
        "model": "gpt-5.4",
        "context_pct": 38,
        "cwd": "~",
        "latency": 22.5,
        "cost": 0.0012,
    }
    # context_pct clamped to 0-100.
    assert pf(context_pct=150)["context_pct"] == 100
    assert pf(context_pct=-5)["context_pct"] == 0
    # cost omitted when 0 (local/free models); latency omitted when negative.
    assert "cost" not in pf(cost=0.0)
    assert "latency" not in pf(latency=-1.0)


def test_home_relative_cwd_collapses_home(plugin):
    f = plugin.adapter._home_relative_cwd
    home = os.path.expanduser("~")
    assert f(home) == "~"
    assert f(os.path.join(home, "proj", "src")) == "~/proj/src"
    assert f("") == ""
    # A path outside the home is returned absolute (unmodified).
    assert f("/etc") == "/etc"


@pytest.mark.asyncio
async def test_final_message_carries_runtime_footer(plugin, adapter, ws_client, monkeypatch):
    ws, _ = ws_client
    # Simulate the post_api_request hook capturing the turn's model + tokens.
    plugin.adapter._on_post_api_request(
        platform="iris",
        model="openai/gpt-5.4",
        usage={"prompt_tokens": 12345},
    )
    # Stub context-length resolution (avoid network probing in tests).
    monkeypatch.setattr(plugin.adapter, "_resolve_context_length", lambda model: 32768)
    monkeypatch.setenv("TERMINAL_CWD", os.path.expanduser("~"))
    res = await adapter.send("default", "hello", metadata={"notify": True})
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    msg = frames[-1]
    runtime = msg["payload"].get("runtime")
    assert runtime is not None
    assert runtime["model"] == "gpt-5.4"  # vendor prefix dropped
    assert runtime["context_pct"] == round(12345 / 32768 * 100)  # 38
    assert runtime["cwd"] == "~"
    assert "latency" in runtime and runtime["latency"] >= 0
    # The turn buffer is drained: a second final send carries no stale model.
    res2 = await adapter.send("default", "again", metadata={"notify": True})
    assert res2.success
    frames2 = await recv_until(
        ws, lambda f: f.get("type") == "message" and f["payload"].get("text") == "again"
    )
    runtime2 = frames2[-1]["payload"].get("runtime")
    # cwd is env-derived (always present); model/tokens are turn-scoped (gone).
    assert runtime2 is not None
    assert "model" not in runtime2
    assert "context_pct" not in runtime2


@pytest.mark.asyncio
async def test_runtime_footer_ignores_other_platforms(plugin, adapter, ws_client, monkeypatch):
    ws, _ = ws_client
    # A non-iris turn must not pollute the iris runtime buffer.
    plugin.adapter._on_post_api_request(
        platform="telegram",
        model="openai/gpt-5.4",
        usage={"prompt_tokens": 999},
    )
    monkeypatch.setattr(plugin.adapter, "_resolve_context_length", lambda model: 32768)
    monkeypatch.setenv("TERMINAL_CWD", os.path.expanduser("~"))
    res = await adapter.send("default", "hi", metadata={"notify": True})
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    runtime = frames[-1]["payload"].get("runtime")
    assert runtime is not None
    assert "model" not in runtime  # the telegram turn's model was not captured


@pytest.mark.asyncio
async def test_history_preserves_runtime_footer(plugin, adapter, ws_client, monkeypatch):
    ws, _ = ws_client
    plugin.adapter._on_post_api_request(
        platform="iris",
        model="openai/gpt-5.4",
        usage={"prompt_tokens": 12345},
    )
    monkeypatch.setattr(plugin.adapter, "_resolve_context_length", lambda model: 32768)
    monkeypatch.setenv("TERMINAL_CWD", os.path.expanduser("~"))
    await adapter.send("default", "hello", metadata={"notify": True})
    # The outbox history reconstruction must carry the runtime object.
    page = adapter._outbox.history("default", limit=50)
    assert len(page["messages"]) == 1
    m = page["messages"][0]
    assert m.get("runtime", {}).get("model") == "gpt-5.4"
    assert m["runtime"]["context_pct"] == round(12345 / 32768 * 100)


# ── Inbound upload ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_upload_reassembles_verifies_and_caches(adapter, ws_client):
    ws, ack = ws_client
    assert ack["payload"]["server_caps"]["media"] is True

    terminal = await upload_file(ws, "mu_t1", PNG_1X1)
    assert terminal["type"] == "media.upload.ack", terminal
    assert terminal["payload"]["ok"] is True
    assert terminal["payload"]["media_ref"] == "mu_t1"

    entry = adapter._media.get_inbound("mu_t1")
    assert entry is not None
    assert entry.kind == "image"
    assert entry.mime == "image/png"
    assert entry.size == len(PNG_1X1)
    assert os.path.isfile(entry.path)
    assert Path(entry.path).read_bytes() == PNG_1X1
    # Cached under the hermes image cache root (profile-aware).
    from hermes_constants import get_hermes_home

    assert str(get_hermes_home() / "cache" / "images") in entry.path


@pytest.mark.asyncio
async def test_upload_declared_over_limit_rejected(adapter, ws_client):
    ws, _ = ws_client
    limit = adapter.max_upload_bytes
    # Over HTTP the server checks Content-Length before reading the body.
    err = await upload_file(
        ws, "mu_big", b"x" * (limit + 1), kind="document", mime="application/pdf",
        filename="big.pdf",
    )
    assert err["type"] == "error"
    assert err["payload"]["code"] == "media_too_large"


@pytest.mark.asyncio
async def test_upload_over_limit_ref_not_consumed(adapter, ws_client):
    """Over HTTP the over-limit check happens before the body is read, so a
    rejected upload must not consume its media_ref (a later valid upload with
    the same ref succeeds)."""
    ws, _ = ws_client
    limit = adapter.max_upload_bytes
    err = await upload_file(ws, "mu_reuse", b"x" * (limit + 1), kind="document")
    assert err["type"] == "error"
    assert err["payload"]["code"] == "media_too_large"
    # The ref is free: a valid upload with the same ref now succeeds.
    ok = await upload_file(ws, "mu_reuse", PNG_1X1)
    assert ok["type"] == "media.upload.ack", ok
    assert ok["payload"]["ok"] is True


@pytest.mark.asyncio
async def test_upload_sha256_mismatch_rejected(adapter, ws_client):
    ws, _ = ws_client
    err = await upload_file(ws, "mu_bad", PNG_1X1, sha256="0" * 64)
    assert err["type"] == "error"
    assert err["payload"]["code"] == "internal"
    assert adapter._media.get_inbound("mu_bad") is None


@pytest.mark.asyncio
async def test_upload_duplicate_ref_rejected(adapter, ws_client):
    ws, _ = ws_client
    first = await upload_file(ws, "mu_dup", PNG_1X1)
    assert first["type"] == "media.upload.ack"
    second = await upload_file(ws, "mu_dup", PNG_1X1, request_id=10)
    assert second["type"] == "error"
    assert second["payload"]["code"] == "unsupported"


@pytest.mark.asyncio
async def test_upload_non_image_bytes_reclassified_not_cached_as_image(adapter, ws_client):
    ws, _ = ws_client
    # Lies about being a PNG: the magic-byte re-sniff must keep it out of the
    # image cache (it lands as a document instead).
    payload = b"<html>not an image</html>"
    terminal = await upload_file(
        ws, "mu_liar", payload, kind="image", mime="image/png", filename="liar.html"
    )
    assert terminal["type"] == "media.upload.ack", terminal
    entry = adapter._media.get_inbound("mu_liar")
    assert entry is not None
    assert entry.kind == "document"
    from hermes_constants import get_hermes_home

    assert str(get_hermes_home() / "cache" / "images") not in entry.path


# ── Inbound message.send with media_refs ────────────────────────────────────


@pytest.mark.asyncio
async def test_message_send_with_media_refs(adapter, ws_client):
    ws, _ = ws_client
    from gateway.platforms.base import MessageType

    terminal = await upload_file(ws, "mu_msg", PNG_1X1)
    assert terminal["type"] == "media.upload.ack"

    captured: list = []
    adapter.handle_message = AsyncMock(side_effect=lambda event: captured.append(event))

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 5,
                "type": "message.send",
                "payload": {"text": "look at this", "media_refs": ["mu_msg"]},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    echo = frames[-1]
    media = echo["payload"]["media"]
    assert len(media) == 1
    assert media[0]["media_id"] == "mu_msg"
    assert media[0]["kind"] == "image"
    assert media[0]["mime"] == "image/png"
    assert media[0]["size"] == len(PNG_1X1)

    assert len(captured) == 1
    event = captured[0]
    assert event.message_type == MessageType.PHOTO
    assert len(event.media_urls) == 1
    assert os.path.isfile(event.media_urls[0])
    assert event.media_types == ["image/png"]

    # Refs are single-use: a second send with the same ref is rejected.
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 6,
                "type": "message.send",
                "payload": {"text": "again", "media_refs": ["mu_msg"]},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 6)
    assert frames[-1]["payload"]["code"] == "unsupported"


@pytest.mark.asyncio
async def test_message_send_media_only_allowed(adapter, ws_client):
    ws, _ = ws_client
    terminal = await upload_file(ws, "mu_only", PNG_1X1)
    assert terminal["type"] == "media.upload.ack"
    captured: list = []
    adapter.handle_message = AsyncMock(side_effect=lambda event: captured.append(event))
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 7,
                "type": "message.send",
                "payload": {"text": "", "media_refs": ["mu_only"]},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    assert frames[-1]["payload"]["text"] == ""
    assert len(captured) == 1
    assert captured[0].message_type.name == "PHOTO"


# ── Auto-threading (message.send auto_thread, docs/06 §6.3) ────────────────


@pytest.mark.asyncio
async def test_message_send_auto_thread_creates_named_thread(adapter, ws_client, monkeypatch):
    """auto_thread in a channel's flat lane mints a fresh thread named from
    the user's opening message (derived title); the echo and the agent event
    carry the new thread_id."""
    ws, _ = ws_client
    import agent.title_generator as tg

    monkeypatch.setattr(tg, "generate_title", lambda text, **kw: None)
    captured: list = []
    adapter.handle_message = AsyncMock(side_effect=lambda event: captured.append(event))

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 40,
                "type": "message.send",
                "chat_id": "default",
                "payload": {"text": "fix the login bug please", "auto_thread": True},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    created = next(f for f in frames if f.get("type") == "channel.created")
    assert created["payload"]["kind"] == "thread"
    assert created["payload"]["auto"] is True
    assert created["payload"]["parent_chat_id"] == "default"
    assert created["payload"]["name"] == "fix the login bug please"
    thread_id = created["payload"]["chat_id"]

    echo = frames[-1]
    assert echo["chat_id"] == "default"
    assert echo["thread_id"] == thread_id

    assert len(captured) == 1
    assert captured[0].source.chat_id == "default"
    assert captured[0].source.thread_id == thread_id


@pytest.mark.asyncio
async def test_message_send_auto_thread_llm_upgrade_renames(adapter, ws_client, monkeypatch):
    """Stage 2 of the two-stage titling: the background LLM call renames the
    auto-created thread and broadcasts channel.renamed."""
    ws, _ = ws_client
    import agent.title_generator as tg

    monkeypatch.setattr(tg, "generate_title", lambda text, **kw: "Fix login bug")
    adapter.handle_message = AsyncMock()

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 41,
                "type": "message.send",
                "chat_id": "default",
                "payload": {"text": "fix the login bug please", "auto_thread": True},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.renamed")
    renamed = frames[-1]
    assert renamed["payload"]["kind"] == "thread"
    assert renamed["payload"]["name"] == "Fix login bug"
    assert "auto" not in renamed["payload"]


@pytest.mark.asyncio
async def test_message_send_auto_thread_ignored_with_existing_thread(adapter, ws_client, monkeypatch):
    """auto_thread is a no-op when the message already targets a thread."""
    ws, _ = ws_client
    import agent.title_generator as tg

    monkeypatch.setattr(tg, "generate_title", lambda text, **kw: None)
    adapter.handle_message = AsyncMock()

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 42,
                "type": "message.send",
                "chat_id": "default",
                "thread_id": "t_9",
                "payload": {"text": "follow up", "auto_thread": True},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    assert not any(f.get("type") == "channel.created" for f in frames)
    assert frames[-1]["thread_id"] == "t_9"


@pytest.mark.asyncio
async def test_message_send_auto_thread_ignored_for_slash_command(adapter, ws_client, monkeypatch):
    """Slash commands are session-scoped, not conversation starters: no
    thread is minted for them."""
    ws, _ = ws_client
    import agent.title_generator as tg

    monkeypatch.setattr(tg, "generate_title", lambda text, **kw: None)
    adapter.handle_message = AsyncMock()

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 43,
                "type": "message.send",
                "chat_id": "default",
                "payload": {"text": "/new", "auto_thread": True},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    assert not any(f.get("type") == "channel.created" for f in frames)
    assert frames[-1].get("thread_id") is None


@pytest.mark.asyncio
async def test_message_send_auto_thread_media_only_stays_flat(adapter, ws_client, monkeypatch):
    """Media-only sends have no text to title from: they stay in the flat
    lane (no nameless threads)."""
    ws, _ = ws_client
    import agent.title_generator as tg

    monkeypatch.setattr(tg, "generate_title", lambda text, **kw: None)
    terminal = await upload_file(ws, "mu_flat", PNG_1X1)
    assert terminal["type"] == "media.upload.ack"
    adapter.handle_message = AsyncMock()

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 44,
                "type": "message.send",
                "chat_id": "default",
                "payload": {"text": "", "media_refs": ["mu_flat"], "auto_thread": True},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    assert not any(f.get("type") == "channel.created" for f in frames)
    assert frames[-1].get("thread_id") is None


# ── History / outbox persistence (restart restore) ─────────────────────────


@pytest.mark.asyncio
async def test_user_echo_parked_in_outbox(adapter, ws_client):
    """The user-message echo must be appended to the outbox (not just
    broadcast) so it survives a process death and is restored by sync /
    history. Regression: echoes were a bare broadcast, so user messages
    vanished on reconnect while bot messages survived."""
    ws, _ = ws_client
    adapter.handle_message = AsyncMock()
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 20,
                "type": "message.send",
                "chat_id": "default",
                "payload": {"text": "persist me"},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    echo = frames[-1]
    message_id = echo["payload"]["message_id"]
    # The echo is parked in the outbox under the home channel.
    page = adapter._outbox.history("default", limit=50)
    ids = [m["message_id"] for m in page["messages"]]
    assert message_id in ids
    parked = next(m for m in page["messages"] if m["message_id"] == message_id)
    assert parked["role"] == "user"
    assert parked["text"] == "persist me"


@pytest.mark.asyncio
async def test_history_returns_final_messages_oldest_first(plugin, adapter, ws_client):
    """A ``history`` request reconstructs the final messages for a chat
    (user echoes + assistant finals, streaming or not) in chronological
    order, skipping intermediate streaming/tool frames."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    chat_id = "default"
    # Two user turns, each with a (non-streaming) assistant final.
    for i, text in enumerate(["one", "two"]):
        await ws.send(
            json.dumps(
                {
                    "v": 1,
                    "id": 30 + i,
                    "type": "message.send",
                    "chat_id": chat_id,
                    "payload": {"text": text},
                }
            )
        )
        await recv_until(
            ws,
            lambda f, t=text: f.get("type") == "message"
            and f.get("payload", {}).get("role") == "user"
            and f.get("payload", {}).get("text") == t,
        )
        # Park a final assistant message for the turn.
        await adapter._broadcast_or_log(
            chat_id,
            protocol.message(
                chat_id=chat_id,
                message_id=f"asst_{i}",
                role=protocol.ROLE_ASSISTANT,
                text=f"reply {text}",
                ts=1_000_000_000_000 + i,
            ),
        )
    page = adapter._outbox.history(chat_id, limit=50)
    roles = [m["role"] for m in page["messages"]]
    texts = [m["text"] for m in page["messages"]]
    # Chronological: user one, asst one, user two, asst two.
    assert roles == ["user", "assistant", "user", "assistant"]
    assert texts == ["one", "reply one", "two", "reply two"]
    assert page["has_more"] is False
    assert page["oldest_message_id"] is not None


@pytest.mark.asyncio
async def test_history_paginates_older_pages(plugin, adapter, ws_client):
    """``before_message_id`` returns the page of messages older than the
    anchor; ``has_more`` signals whether older pages exist."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    chat_id = "default"
    for i in range(5):
        await adapter._broadcast_or_log(
            chat_id,
            protocol.message(
                chat_id=chat_id,
                message_id=f"m_{i}",
                role=protocol.ROLE_USER if i % 2 == 0 else protocol.ROLE_ASSISTANT,
                text=f"msg {i}",
                ts=1_000_000_000_000 + i,
            ),
        )
    newest = adapter._outbox.history(chat_id, limit=2)
    assert [m["message_id"] for m in newest["messages"]] == ["m_3", "m_4"]
    assert newest["has_more"] is True
    older = adapter._outbox.history(chat_id, before_message_id=newest["oldest_message_id"], limit=2)
    assert [m["message_id"] for m in older["messages"]] == ["m_1", "m_2"]
    assert older["has_more"] is True
    oldest = adapter._outbox.history(chat_id, before_message_id=older["oldest_message_id"], limit=2)
    assert [m["message_id"] for m in oldest["messages"]] == ["m_0"]
    assert oldest["has_more"] is False


@pytest.mark.asyncio
async def test_history_frame_roundtrip(plugin, adapter, ws_client):
    """The ``history`` request/response frame round-trips over the wire: the
    server answers a ``history`` request with a paged ``history`` frame."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    chat_id = "default"
    await adapter._broadcast_or_log(
        chat_id,
        protocol.message(
            chat_id=chat_id,
            message_id="wire_1",
            role=protocol.ROLE_USER,
            text="over the wire",
            ts=1_000_000_000_000,
        ),
    )
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 40,
                "type": "history",
                "chat_id": chat_id,
                "payload": {"limit": 10},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "history" and f.get("id") == 40)
    resp = frames[-1]
    assert resp["chat_id"] == chat_id
    messages = resp["payload"]["messages"]
    assert any(m["message_id"] == "wire_1" for m in messages)
    assert "has_more" in resp["payload"]


# ── Message deletion ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_message_delete_removes_from_outbox_and_broadcasts(plugin, adapter, ws_client):
    """A ``message.delete`` request removes the message from the outbox (so
    ``history`` no longer returns it) and broadcasts ``message.deleted`` to
    every device (the response carries the request id)."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    chat_id = "default"
    await adapter._broadcast_or_log(
        chat_id,
        protocol.message(
            chat_id=chat_id,
            message_id="del_1",
            role=protocol.ROLE_USER,
            text="delete me",
            ts=1_000_000_000_000,
        ),
    )
    page = adapter._outbox.history(chat_id, limit=50)
    assert any(m["message_id"] == "del_1" for m in page["messages"])
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 50,
                "type": "message.delete",
                "chat_id": chat_id,
                "payload": {"message_ids": ["del_1"]},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "message.deleted" and f.get("id") == 50)
    resp = frames[-1]
    assert resp["chat_id"] == chat_id
    assert resp["payload"]["message_ids"] == ["del_1"]
    # The message is gone from the outbox: history no longer returns it.
    page = adapter._outbox.history(chat_id, limit=50)
    assert not any(m["message_id"] == "del_1" for m in page["messages"])


@pytest.mark.asyncio
async def test_message_delete_idempotent(plugin, adapter, ws_client):
    """Deleting a message that is already gone (unknown / pruned) is a no-op
    on the outbox but still emits ``message.deleted`` so live caches drop it."""
    ws, _ = ws_client
    adapter.handle_message = AsyncMock()
    chat_id = "default"
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 51,
                "type": "message.delete",
                "chat_id": chat_id,
                "payload": {"message_ids": ["never_existed"]},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "message.deleted" and f.get("id") == 51)
    assert frames[-1]["payload"]["message_ids"] == ["never_existed"]


@pytest.mark.asyncio
async def test_message_delete_requires_message_ids(adapter, ws_client):
    """A ``message.delete`` with no message_ids is rejected with ``unsupported``."""
    ws, _ = ws_client
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 52,
                "type": "message.delete",
                "chat_id": "default",
                "payload": {},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 52)
    assert frames[-1]["payload"]["code"] == "unsupported"


def _make_state_db(path: Path) -> None:
    """Create a minimal hermes session store (sessions + messages) at *path*."""
    import sqlite3

    conn = sqlite3.connect(str(path))
    try:
        conn.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, chat_id TEXT, thread_id TEXT)"
        )
        conn.execute(
            "CREATE TABLE messages ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, "
            "role TEXT, content TEXT, timestamp REAL)"
        )
        conn.commit()
    finally:
        conn.close()


@pytest.mark.asyncio
async def test_message_delete_purges_session_store(plugin, adapter, ws_client):
    """A ``message.delete`` also removes the message from the hermes session
    store (state.db), so no search trace survives and it is not recoverable."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    from hermes_constants import get_hermes_home

    chat_id = "default"
    db = get_hermes_home() / "state.db"
    _make_state_db(db)
    import sqlite3

    conn = sqlite3.connect(str(db))
    try:
        conn.execute(
            "INSERT INTO sessions (id, chat_id, thread_id) VALUES ('s1', ?, NULL)",
            (chat_id,),
        )
        conn.execute(
            "INSERT INTO messages (session_id, role, content, timestamp) "
            "VALUES ('s1', 'user', 'delete me', 1000.0)"
        )
        conn.commit()
    finally:
        conn.close()
    # The outbox frame carries the same text + ts so the session row matches.
    await adapter._broadcast_or_log(
        chat_id,
        protocol.message(
            chat_id=chat_id,
            message_id="purge_1",
            role=protocol.ROLE_USER,
            text="delete me",
            ts=1_000_000,
        ),
    )
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 53,
                "type": "message.delete",
                "chat_id": chat_id,
                "payload": {"message_ids": ["purge_1"]},
            }
        )
    )
    await recv_until(ws, lambda f: f.get("type") == "message.deleted" and f.get("id") == 53)
    conn = sqlite3.connect(str(db))
    try:
        n = conn.execute("SELECT COUNT(*) FROM messages WHERE content = 'delete me'").fetchone()[0]
    finally:
        conn.close()
    assert n == 0  # purged from the session store (no search trace)


@pytest.mark.asyncio
async def test_channel_delete_wipes_session_store(plugin, adapter, ws_client):
    """A ``channel.delete`` wipes the lane's messages + session from the hermes
    session store (state.db) as well as the outbox, so nothing is recoverable."""
    ws, _ = ws_client
    protocol = plugin.protocol
    adapter.handle_message = AsyncMock()
    from hermes_constants import get_hermes_home

    # Create a channel via the wire.
    await ws.send(json.dumps({"v": 1, "id": 60, "type": "channel.create", "payload": {"name": "Secret"}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.created" and f.get("id") == 60)
    chat_id = frames[-1]["payload"]["chat_id"]
    db = get_hermes_home() / "state.db"
    _make_state_db(db)
    import sqlite3

    conn = sqlite3.connect(str(db))
    try:
        conn.execute(
            "INSERT INTO sessions (id, chat_id, thread_id) VALUES ('s9', ?, NULL)",
            (chat_id,),
        )
        conn.execute(
            "INSERT INTO messages (session_id, role, content, timestamp) "
            "VALUES ('s9', 'user', 'top secret', 2000.0)"
        )
        conn.commit()
    finally:
        conn.close()
    await adapter._broadcast_or_log(
        chat_id,
        protocol.message(
            chat_id=chat_id,
            message_id="chan_1",
            role=protocol.ROLE_USER,
            text="top secret",
            ts=2_000_000,
        ),
    )
    await ws.send(
        json.dumps({"v": 1, "id": 61, "type": "channel.delete", "chat_id": chat_id, "payload": {}})
    )
    await recv_until(ws, lambda f: f.get("type") == "notification" and f["payload"]["kind"] == "channel_deleted")
    # Outbox is wiped for the channel.
    page = adapter._outbox.history(chat_id, limit=50)
    assert not any(m["message_id"] == "chan_1" for m in page["messages"])
    # Session store is wiped: the message and its session are gone.
    conn = sqlite3.connect(str(db))
    try:
        n_msgs = conn.execute("SELECT COUNT(*) FROM messages WHERE content = 'top secret'").fetchone()[0]
        n_sess = conn.execute("SELECT COUNT(*) FROM sessions WHERE chat_id = ?", (chat_id,)).fetchone()[0]
    finally:
        conn.close()
    assert n_msgs == 0
    assert n_sess == 0


# ── Outbound offers + pull ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_send_video_emits_offer_with_message_association(adapter, ws_client):
    ws, _ = ws_client
    from gateway.platforms.base import get_video_cache_dir

    video = get_video_cache_dir() / "clip.mp4"
    video.write_bytes(b"fake-video-bytes")

    # A final message first, so the offer can associate with it.
    res = await adapter.send("default", "here you go", metadata={"notify": True})
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    msg_id = frames[-1]["payload"]["message_id"]

    res2 = await adapter.send_video("default", str(video))
    assert res2.success
    frames = await recv_until(ws, lambda f: f.get("type") == "media.offer")
    offer = frames[-1]["payload"]
    assert offer["kind"] == "video"
    assert offer["mime"] == "video/mp4"
    assert offer["size"] == len(b"fake-video-bytes")
    assert offer["filename"] == "clip.mp4"
    assert offer["media_id"] == res2.message_id
    assert offer["media_id"].startswith("md_")
    assert offer["message_id"] == msg_id


@pytest.mark.asyncio
async def test_send_document_uses_caller_filename(adapter, ws_client):
    ws, _ = ws_client
    from gateway.platforms.base import get_document_cache_dir

    doc = get_document_cache_dir() / "report.pdf"
    doc.write_bytes(b"%PDF-1.4 fake")
    res = await adapter.send_document(
        "default", str(doc), file_name="My Report.pdf"
    )
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "media.offer")
    offer = frames[-1]["payload"]
    assert offer["kind"] == "document"
    assert offer["filename"] == "My Report.pdf"
    assert offer["mime"] == "application/pdf"


@pytest.mark.asyncio
async def test_send_image_file_offers_image(adapter, ws_client):
    ws, _ = ws_client
    from gateway.platforms.base import get_image_cache_dir

    img = get_image_cache_dir() / "shot.png"
    img.write_bytes(PNG_1X1)
    res = await adapter.send_image_file("default", str(img))
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "media.offer")
    offer = frames[-1]["payload"]
    assert offer["kind"] == "image"
    assert offer["mime"] == "image/png"


@pytest.mark.asyncio
async def test_send_media_rejects_denied_path(adapter, ws_client):
    ws, _ = ws_client
    # /etc/passwd exists but is on hermes' delivery denylist.
    res = await adapter.send_document("default", "/etc/passwd")
    assert not res.success
    # Nothing was offered (a failed offer is silent, like other platforms).
    try:
        raw = await asyncio.wait_for(ws.recv(), timeout=1.0)
        frame = json.loads(raw) if isinstance(raw, str) else None
        assert frame is None or frame.get("type") != "media.offer"
    except asyncio.TimeoutError:
        pass  # silence is the expected outcome


@pytest.mark.asyncio
async def test_pull_serves_allowed_path(adapter, ws_client):
    ws, _ = ws_client
    from gateway.platforms.base import get_image_cache_dir

    img = get_image_cache_dir() / "pull_test.png"
    img.write_bytes(PNG_1X1)
    entry = adapter._media.register_outbound(
        str(img), "image", "image/png", "pull_test.png", len(PNG_1X1)
    )

    status, body = await ws.pull(entry.media_id)
    assert status == 200
    assert body == PNG_1X1


@pytest.mark.asyncio
async def test_pull_rejects_unknown_and_denied(adapter, ws_client):
    ws, _ = ws_client
    # Unknown media_id.
    status, body = await ws.pull("md_nope")
    assert status == 404
    assert json.loads(body)["payload"]["code"] == "not_found"

    # Known id, but the path fails delivery validation (denylist).
    entry = adapter._media.register_outbound(
        "/etc/passwd", "document", "text/plain", "passwd", 100
    )
    status, body = await ws.pull(entry.media_id)
    assert status == 404
    assert json.loads(body)["payload"]["code"] == "not_found"

    # Known id, file deleted since the offer.
    from gateway.platforms.base import get_image_cache_dir

    img = get_image_cache_dir() / "gone.png"
    img.write_bytes(PNG_1X1)
    entry2 = adapter._media.register_outbound(
        str(img), "image", "image/png", "gone.png", len(PNG_1X1)
    )
    img.unlink()
    status, body = await ws.pull(entry2.media_id)
    assert status == 404
    assert json.loads(body)["payload"]["code"] == "not_found"


# ── M5: push backends (pure) ───────────────────────────────────────────────


class _FakeResponse:
    def __init__(self, status=200, body="", json_data=None):
        self.status_code = status
        self.text = body
        self._json = json_data if json_data is not None else {}

    def json(self):
        return self._json


class _FakeHttpClient:
    """Stands in for ``httpx.AsyncClient`` (async CM + ``post``)."""

    def __init__(self, responder):
        self._responder = responder  # (url, kwargs) -> _FakeResponse
        self.calls = []

    async def __aenter__(self):
        return self

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

    async def post(self, url, **kwargs):
        self.calls.append((url, kwargs))
        return self._responder(url, kwargs)


def _patch_httpx(plugin, monkeypatch, responder):
    fake = _FakeHttpClient(responder)
    monkeypatch.setattr(plugin.push.httpx, "AsyncClient", lambda *a, **k: fake)
    return fake


def test_push_backend_selection(plugin):
    push = plugin.push
    assert isinstance(push.build_push_backend("fcm"), push.FcmBackend)
    assert isinstance(push.build_push_backend("ntfy"), push.NtfyBackend)
    assert isinstance(push.build_push_backend(None), push.FcmBackend)  # default
    assert isinstance(push.build_push_backend("  NTFY "), push.NtfyBackend)

    assert push.build_push_backend("ntfy", ntfy_topic="my-topic").configured() is True
    assert push.build_push_backend("ntfy").configured() is False
    assert push.build_push_backend("fcm", fcm_server_key="key-123").configured() is True
    assert push.build_push_backend("fcm").configured() is False


def test_push_backend_fcm_service_account_configured(plugin, tmp_path):
    push = plugin.push
    sa = tmp_path / "sa.json"
    sa.write_text(
        json.dumps(
            {
                "client_email": "x@y.iam.gserviceaccount.com",
                "private_key": "k",
                "project_id": "p",
            }
        )
    )
    assert (
        push.build_push_backend("fcm", fcm_service_account=str(sa)).configured()
        is True
    )
    assert (
        push.build_push_backend(
            "fcm", fcm_service_account=str(tmp_path / "missing.json")
        ).configured()
        is False
    )


@pytest.mark.asyncio
async def test_ntfy_backend_publishes_with_data_header(plugin, monkeypatch):
    push = plugin.push
    backend = push.NtfyBackend(topic="iris-topic", server_url="https://ntfy.example/")
    fake = _patch_httpx(
        plugin, monkeypatch, lambda url, kw: _FakeResponse(200, '{"id":"1"}')
    )
    ok = await backend.send(
        device_id="d1",
        chat_id="default",
        title="Iris",
        body="hello",
        data={"chat_id": "default", "kind": "message", "cursor": "7"},
        token="iris-topic",
    )
    assert ok is True
    url, kw = fake.calls[0]
    assert url == "https://ntfy.example/iris-topic"
    headers = kw["headers"]
    assert headers["X-Title"] == "Iris"
    assert headers["X-Priority"] == "3"
    assert json.loads(headers["X-Data"])["cursor"] == "7"
    assert kw["content"] == b"hello"

    fake2 = _patch_httpx(plugin, monkeypatch, lambda url, kw: _FakeResponse(200))
    await backend.send(
        device_id="d1", chat_id="c", title="t", body="b", data={},
        token="t", priority="high",
    )
    assert fake2.calls[0][1]["headers"]["X-Priority"] == "5"

    fake3 = _patch_httpx(plugin, monkeypatch, lambda url, kw: _FakeResponse(404, "nope"))
    assert (
        await backend.send(
            device_id="d1", chat_id="c", title="t", body="b", data={}, token="t"
        )
        is False
    )


@pytest.mark.asyncio
async def test_fcm_backend_legacy_server_key(plugin, monkeypatch):
    push = plugin.push
    backend = push.FcmBackend(server_key="legacy-key")
    fake = _patch_httpx(
        plugin, monkeypatch, lambda url, kw: _FakeResponse(200, '{"name":"m/1"}')
    )
    ok = await backend.send(
        device_id="d1",
        chat_id="default",
        title="Iris",
        body="hi",
        data={"chat_id": "default", "kind": "message", "cursor": "3"},
        token="fcm-token-1",
    )
    assert ok is True
    url, kw = fake.calls[0]
    assert url == push.FCM_LEGACY_SEND_URL
    assert kw["headers"]["Authorization"] == "Bearer legacy-key"
    assert kw["json"]["to"] == "fcm-token-1"
    assert kw["json"]["notification"] == {"title": "Iris", "body": "hi"}
    assert kw["json"]["data"]["cursor"] == "3"


@pytest.mark.asyncio
async def test_fcm_backend_service_account_v1(plugin, monkeypatch, tmp_path):
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import rsa

    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    pem = key.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    ).decode()
    sa = tmp_path / "sa.json"
    sa.write_text(
        json.dumps(
            {
                "project_id": "iris-test-project",
                "client_email": "svc@iris-test-project.iam.gserviceaccount.com",
                "private_key_id": "kid-1",
                "private_key": pem,
            }
        )
    )
    backend = plugin.push.FcmBackend(service_account=str(sa))
    assert backend.configured() is True

    def responder(url, kw):
        if url == plugin.push.FCM_TOKEN_URL:
            assert kw["data"]["grant_type"] == "urn:ietf:params:oauth:grant-type:jwt-bearer"
            return _FakeResponse(200, "", {"access_token": "at-123", "expires_in": 3600})
        return _FakeResponse(200, '{"name":"projects/iris-test-project/messages/9"}')

    fake = _patch_httpx(plugin, monkeypatch, responder)
    ok = await backend.send(
        device_id="d1",
        chat_id="default",
        title="Iris",
        body="hi",
        data={"chat_id": "default", "kind": "cron", "cursor": "4"},
        token="fcm-token-2",
        priority="high",
    )
    assert ok is True
    token_url, token_kw = fake.calls[0]
    send_url, send_kw = fake.calls[1]
    assert token_url == plugin.push.FCM_TOKEN_URL
    assert send_url == (
        "https://fcm.googleapis.com/v1/projects/iris-test-project/messages:send"
    )
    assert send_kw["headers"]["Authorization"] == "Bearer at-123"
    msg = send_kw["json"]["message"]
    assert msg["token"] == "fcm-token-2"
    assert msg["android"]["priority"] == "high"
    assert msg["data"]["kind"] == "cron"
    # The access token is cached: the next send does not re-exchange.
    await backend.send(
        device_id="d1", chat_id="c", title="t", body="b", data={}, token="fcm-token-2"
    )
    assert len(fake.calls) == 3


# ── M5: push firing (integration) ──────────────────────────────────────────


class _FakePush:
    name = "fake"
    token_field = "fcm_token"

    def __init__(self, configured=True):
        self.calls = []
        self._configured = configured
        self.fail_next = False

    def configured(self):
        return self._configured

    async def send(self, **kwargs):
        self.calls.append(kwargs)
        if self.fail_next:
            self.fail_next = False
            return False
        return True


@pytest.mark.asyncio
async def test_push_fires_when_no_live_subscriber(adapter):
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    res = await adapter.send(
        "default", "hello while offline", metadata={"notify": True}
    )
    assert res.success
    assert len(fake.calls) == 1
    call = fake.calls[0]
    assert call["token"] == "tok-1"
    assert call["chat_id"] == "default"
    assert call["data"]["kind"] == "message"
    assert call["data"]["chat_id"] == "default"
    assert call["data"]["cursor"] == "1"
    assert call["priority"] == "normal"
    # The frame is parked in the outbox for sync.
    assert adapter._outbox.latest_cursor() == 1


@pytest.mark.asyncio
async def test_push_not_fired_when_live(adapter, ws_client):
    ws, _ = ws_client
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    await adapter.send("default", "live reply", metadata={"notify": True})
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    assert frames[-1]["payload"]["text"] == "live reply"
    assert fake.calls == []


@pytest.mark.asyncio
async def test_live_delivered_frame_still_parked_for_sync(adapter, ws_client):
    """A frame delivered to a live subscriber is still appended to the outbox,
    so a reconnecting app (whose in-memory ChatStore was reset, e.g. after
    tapping a push notification) can catch up via sync. Regression: chat empty
    after tapping a message notification."""
    ws, _ = ws_client
    await adapter.send("default", "live reply", metadata={"notify": True})
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    assert frames[-1]["payload"]["text"] == "live reply"
    # The live-delivered frame is still parked in the outbox for sync.
    assert adapter._outbox.latest_cursor() == 1
    rows = adapter._outbox.replay(0)
    assert [r["frame"]["payload"]["text"] for r in rows] == ["live reply"]


@pytest.mark.asyncio
async def test_user_echo_parked_for_sync(adapter, ws_client):
    """The user-message echo is appended to the outbox so a reconnecting app
    (whose in-memory ChatStore was reset by a process death / activity
    recreation) restores the user's own message via the sync replay.
    Regression: user messages vanished after closing/multitasking while bot
    messages (already parked) survived."""
    ws, _ = ws_client
    adapter.handle_message = AsyncMock()

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 90,
                "type": "message.send",
                "payload": {"text": "remember me"},
            }
        )
    )
    frames = await recv_until(
        ws, lambda f: f.get("type") == "message" and f.get("payload", {}).get("role") == "user"
    )
    echo = frames[-1]
    assert echo["payload"]["text"] == "remember me"

    # The echo is parked in the outbox for sync catch-up, carrying the same
    # server-assigned id the live echo used (so the app dedupes on replay).
    rows = adapter._outbox.replay(0)
    user_rows = [
        r
        for r in rows
        if r["frame"].get("type") == "message"
        and r["frame"].get("payload", {}).get("role") == "user"
    ]
    assert len(user_rows) == 1
    assert user_rows[0]["frame"]["payload"]["text"] == "remember me"
    assert user_rows[0]["frame"]["payload"]["message_id"] == echo["payload"]["message_id"]


@pytest.mark.asyncio
async def test_intermediate_frames_park_without_push(adapter):
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    await adapter.send("default", "seg", metadata={"expect_edits": True})
    stream_id = adapter._turns["default"].stream_id
    await adapter.edit_message("default", stream_id, "seg more")
    assert adapter._outbox.latest_cursor() == 2
    assert fake.calls == []


@pytest.mark.asyncio
async def test_high_priority_notification_pushes_even_when_live(plugin, adapter, ws_client):
    ws, _ = ws_client
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    await adapter._broadcast_or_log(
        "default",
        plugin.protocol.notification(
            "default", plugin.protocol.NOTIF_CRON, "Cron: Job", "body"
        ),
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "notification")
    assert frames[-1]["payload"]["kind"] == "cron"
    assert len(fake.calls) == 1
    assert fake.calls[0]["priority"] == "high"
    assert fake.calls[0]["data"]["kind"] == "cron"


@pytest.mark.asyncio
async def test_push_skipped_when_device_has_no_token(adapter):
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {})  # no push token

    await adapter.send("default", "no token", metadata={"notify": True})
    assert fake.calls == []
    assert adapter._outbox.latest_cursor() == 1  # still parked for sync


@pytest.mark.asyncio
async def test_push_skipped_when_backend_unconfigured(adapter):
    fake = _FakePush(configured=False)
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {})

    await adapter.send("default", "unconfigured", metadata={"notify": True})
    assert fake.calls == []


# ── M5: fcm.register ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_fcm_register_updates_registry(adapter, ws_client):
    ws, _ = ws_client
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "type": "fcm.register",
                "payload": {"fcm_token": "rotated-token", "ntfy_topic": "dev-topic"},
            }
        )
    )
    # Ordering barrier: frames are processed in order, so by the time the
    # channel.list response arrives, fcm.register has been handled.
    await ws.send(
        json.dumps({"v": 1, "id": 1, "type": "channel.list", "payload": {}})
    )
    await recv_until(ws, lambda f: f.get("type") == "channel.list")
    dev = adapter._devices.get(DEVICE_ID)
    assert dev["fcm_token"] == "rotated-token"
    assert dev["ntfy_topic"] == "dev-topic"

    # The next offline push targets the rotated token. Close the stream and
    # force-remove the (possibly stale) subscriber so the push isn't counted
    # as delivered to a dead connection.
    await ws.close()
    with adapter._http_server._subs_lock:
        adapter._http_server._subs.clear()
    fake = _FakePush()
    adapter._push = fake
    await adapter.send("default", "after rotation", metadata={"notify": True})
    assert len(fake.calls) == 1
    assert fake.calls[0]["token"] == "rotated-token"


# ── M5: notification banners ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_channel_events_emit_notification_banners(adapter, ws_client):
    ws, _ = ws_client
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 1,
                "type": "channel.create",
                "payload": {"name": "Cron Reports"},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "notification")
    notif = frames[-1]
    assert notif["payload"]["kind"] == "channel_created"
    assert "Cron Reports" in notif["payload"]["body"]
    chat_id = notif["chat_id"]

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 2,
                "type": "channel.rename",
                "chat_id": chat_id,
                "payload": {"name": "Reports"},
            }
        )
    )
    frames = await recv_until(
        ws,
        lambda f: f.get("type") == "notification"
        and f["payload"]["kind"] == "channel_renamed",
    )
    assert frames[-1]["payload"]["body"] == "Renamed to Reports"

    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 3,
                "type": "channel.delete",
                "chat_id": chat_id,
                "payload": {},
            }
        )
    )
    frames = await recv_until(
        ws,
        lambda f: f.get("type") == "notification"
        and f["payload"]["kind"] == "channel_deleted",
    )
    assert "Reports" in frames[-1]["payload"]["body"]


# ── M3: channel favorite / icon (cosmetic, synced) ─────────────────────────


async def _create_channel(ws, name: str, req_id: int) -> str:
    await ws.send(
        json.dumps({"v": 1, "id": req_id, "type": "channel.create", "payload": {"name": name}})
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.created" and f.get("id") == req_id)
    return frames[-1]["payload"]["chat_id"]


@pytest.mark.asyncio
async def test_channel_favorite_toggle(adapter, ws_client):
    ws, _ = ws_client
    chat_id = await _create_channel(ws, "Work", 1)
    # Fresh channel is not a favorite.
    await ws.send(json.dumps({"v": 1, "id": 2, "type": "channel.list", "payload": {}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.list" and f.get("id") == 2)
    entry = next(c for c in frames[-1]["payload"]["channels"] if c["chat_id"] == chat_id)
    assert "favorite" not in entry

    # Favorite it -> the renamed-shaped response carries favorite=true.
    await ws.send(json.dumps({"v": 1, "id": 3, "type": "channel.favorite", "chat_id": chat_id, "payload": {"on": True}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.renamed" and f.get("id") == 3)
    assert frames[-1]["payload"]["favorite"] is True

    # Unfavorite it -> the flag is cleared (absent on the wire).
    await ws.send(json.dumps({"v": 1, "id": 4, "type": "channel.favorite", "chat_id": chat_id, "payload": {"on": False}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.renamed" and f.get("id") == 4)
    assert "favorite" not in frames[-1]["payload"]


@pytest.mark.asyncio
async def test_channel_favorite_unknown_id_rejected(adapter, ws_client):
    ws, _ = ws_client
    await ws.send(json.dumps({"v": 1, "id": 1, "type": "channel.favorite", "chat_id": "chan_999", "payload": {"on": True}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 1)
    assert frames[-1]["payload"]["code"] == "not_found"


@pytest.mark.asyncio
async def test_channel_icon_set_and_clear(adapter, ws_client):
    ws, _ = ws_client
    chat_id = await _create_channel(ws, "Bugs", 1)
    icon = "aGVsbG8="  # base64 of "hello"
    await ws.send(json.dumps({"v": 1, "id": 2, "type": "channel.icon", "chat_id": chat_id, "payload": {"icon": icon, "color": "#FF0000"}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.renamed" and f.get("id") == 2)
    assert frames[-1]["payload"]["icon"] == icon
    assert frames[-1]["payload"]["color"] == "#FF0000"

    # Clear both (empty payload) -> icon/color absent on the wire.
    await ws.send(json.dumps({"v": 1, "id": 3, "type": "channel.icon", "chat_id": chat_id, "payload": {}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.renamed" and f.get("id") == 3)
    assert "icon" not in frames[-1]["payload"]
    assert "color" not in frames[-1]["payload"]


@pytest.mark.asyncio
async def test_channel_icon_too_large_rejected(adapter, ws_client):
    ws, _ = ws_client
    chat_id = await _create_channel(ws, "Big", 1)
    await ws.send(json.dumps({"v": 1, "id": 2, "type": "channel.icon", "chat_id": chat_id, "payload": {"icon": "x" * (512 * 1024 + 1)}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 2)
    assert frames[-1]["payload"]["code"] == "unsupported"


@pytest.mark.asyncio
async def test_channel_list_orders_favorites_first(adapter, ws_client):
    ws, _ = ws_client
    alpha = await _create_channel(ws, "Alpha", 1)
    beta = await _create_channel(ws, "Beta", 2)
    # Favorite the later-created channel; it must sort ahead of Alpha.
    await ws.send(json.dumps({"v": 1, "id": 3, "type": "channel.favorite", "chat_id": beta, "payload": {"on": True}}))
    await recv_until(ws, lambda f: f.get("type") == "channel.renamed" and f.get("id") == 3)
    await ws.send(json.dumps({"v": 1, "id": 4, "type": "channel.list", "payload": {}}))
    frames = await recv_until(ws, lambda f: f.get("type") == "channel.list" and f.get("id") == 4)
    names = [c["name"] for c in frames[-1]["payload"]["channels"]]
    assert names.index("Beta") < names.index("Alpha")


@pytest.mark.asyncio
async def test_cron_delivery_emits_banner_and_message(adapter, ws_client):
    ws, _ = ws_client
    wrapped = (
        "Cronjob Response: My Job\n"
        "(job_id: abc123)\n"
        "-------------\n\n"
        "hello from cron\n\n"
        "To stop or manage this job, send me a new message (e.g. \"stop reminder My Job\")."
    )
    res = await adapter.send("default", wrapped, metadata={"job_id": "abc123"})
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    msg = frames[-1]
    assert msg["payload"]["text"] == "hello from cron"
    assert msg["payload"]["role"] == "assistant"
    notif = [f for f in frames if f.get("type") == "notification"]
    assert len(notif) == 1
    assert notif[0]["payload"]["kind"] == "cron"
    assert notif[0]["payload"]["title"] == "Cron: My Job"
    assert notif[0]["payload"]["body"] == "hello from cron"

    # wrap_response: false -- raw content, job id as the name.
    res2 = await adapter.send("default", "raw cron output", metadata={"job_id": "j2"})
    assert res2.success
    frames = await recv_until(
        ws,
        lambda f: f.get("type") == "message"
        and f["payload"].get("text") == "raw cron output",
    )
    notif2 = [f for f in frames if f.get("type") == "notification"]
    assert notif2[0]["payload"]["title"] == "Cron: j2"


@pytest.mark.asyncio
async def test_clarify_emits_banner_and_message(adapter, ws_client):
    ws, _ = ws_client
    from tools import clarify_gateway as cg

    cg.register("cl_1", "sk", "Which one?", ["A", "B"])
    try:
        res = await adapter.send_clarify(
            "default", "Which one?", ["A", "B"], "cl_1", "sk"
        )
        assert res.success
        frames = await recv_until(ws, lambda f: f.get("type") == "message")
        msg = frames[-1]
        assert "❓ Which one?" in msg["payload"]["text"]
        assert "1. A" in msg["payload"]["text"]
        notif = [f for f in frames if f.get("type") == "notification"]
        assert notif[0]["payload"]["kind"] == "clarify"
        assert notif[0]["payload"]["body"] == "Which one?"
        entry = cg._entries.get("cl_1")
        assert entry is not None and entry.awaiting_text is True
    finally:
        cg._entries.pop("cl_1", None)
        cg._session_index.pop("sk", None)


# ── M5: sync catch-up + outbox row cap ─────────────────────────────────────


@pytest.mark.asyncio
async def test_sync_replays_parked_frames_and_done_cursor(adapter):
    # Park two frames while offline.
    await adapter.send("default", "one", metadata={"notify": True})
    await adapter.send("default", "two", metadata={"notify": True})
    assert adapter._outbox.latest_cursor() == 2

    await adapter.connect()
    # Open at the latest cursor so the SSE catch-up doesn't replay the parked
    # frames (the sync request below is what we're testing).
    ws = HttpTestClient(adapter._http_server.bound_port, cursor=adapter._outbox.latest_cursor())
    try:
        ack = await ws.start()
        assert ack["payload"]["sync_cursor"] == 2
        await ws.send(
            json.dumps({"v": 1, "id": 10, "type": "sync", "payload": {"cursor": 0}})
        )
        frames = await recv_until(ws, lambda f: f.get("type") == "sync.done")
        texts = [f["payload"]["text"] for f in frames if f.get("type") == "message"]
        assert texts == ["one", "two"]
        done = frames[-1]
        assert done["id"] == 10
        assert done["payload"]["cursor"] == 2
        # A sync at the current cursor replays nothing.
        await ws.send(
            json.dumps({"v": 1, "id": 11, "type": "sync", "payload": {"cursor": 2}})
        )
        frames = await recv_until(
            ws, lambda f: f.get("type") == "sync.done" and f.get("id") == 11
        )
        assert len(frames) == 1
    finally:
        await ws.close()
        await adapter.disconnect()


@pytest.mark.asyncio
async def test_hello_ack_last_pushed_cursor_default_zero(adapter):
    """A device that never received a push reports last_pushed_cursor=0 in
    hello.ack (docs/08 §8.7 dedupe watermark)."""
    await adapter.connect()
    ws = HttpTestClient(adapter._http_server.bound_port)
    try:
        ack = await ws.start()
        assert ack["payload"]["last_pushed_cursor"] == 0
    finally:
        await ws.close()
        await adapter.disconnect()


@pytest.mark.asyncio
async def test_push_success_advances_last_pushed_cursor(adapter):
    """A successful push advances the device's last_pushed_cursor, and the
    next hello.ack reports it — the app uses it to skip re-notifying
    sync-replayed frames (docs/08 §8.7). Back-to-back frames for the same
    chat coalesce into one push; a failed push does not advance the cursor."""
    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    await adapter.send("default", "one", metadata={"notify": True})
    assert len(fake.calls) == 1
    assert adapter._devices.last_pushed_cursor(DEVICE_ID) == 1

    # Immediate second frame (cron's message frame) coalesces — no second
    # push, cursor unchanged.
    await adapter.send("default", "two", metadata={"notify": True})
    assert len(fake.calls) == 1
    assert adapter._devices.last_pushed_cursor(DEVICE_ID) == 1

    # Simulate the coalesce window elapsing, then push again.
    adapter._last_push_at["default"] = 0.0
    await adapter.send("default", "three", metadata={"notify": True})
    assert len(fake.calls) == 2
    assert adapter._devices.last_pushed_cursor(DEVICE_ID) == 3

    # A failed push must NOT advance the cursor (the device never woke).
    fake.fail_next = True
    adapter._last_push_at["default"] = 0.0
    await adapter.send("default", "four", metadata={"notify": True})
    assert adapter._devices.last_pushed_cursor(DEVICE_ID) == 3

    await adapter.connect()
    ws = HttpTestClient(adapter._http_server.bound_port, cursor=adapter._outbox.latest_cursor())
    try:
        ack = await ws.start()
        assert ack["payload"]["last_pushed_cursor"] == 3
    finally:
        await ws.close()
        await adapter.disconnect()


@pytest.mark.asyncio
async def test_sync_replay_frames_carry_outbox_cursor(adapter):
    """Frames replayed by sync carry their outbox cursor in the envelope so
    the app can compare it against last_pushed_cursor (docs/08 §8.7). Live
    frames carry no cursor."""
    await adapter.send("default", "one", metadata={"notify": True})
    await adapter.send("default", "two", metadata={"notify": True})

    await adapter.connect()
    ws = HttpTestClient(adapter._http_server.bound_port, cursor=adapter._outbox.latest_cursor())
    try:
        await ws.start()
        await ws.send(
            json.dumps({"v": 1, "id": 20, "type": "sync", "payload": {"cursor": 0}})
        )
        frames = await recv_until(ws, lambda f: f.get("type") == "sync.done")
        msgs = [f for f in frames if f.get("type") == "message"]
        assert [f["cursor"] for f in msgs] == [1, 2]
        # sync.done itself carries no envelope cursor.
        assert "cursor" not in frames[-1]
    finally:
        await ws.close()
        await adapter.disconnect()


@pytest.mark.asyncio
async def test_live_frames_carry_no_cursor(adapter, ws_client):
    """Live (non-replay) frames must not carry a cursor — the app only
    suppresses notifications for replayed frames (docs/08 §8.7)."""
    ws, _ = ws_client
    await adapter.send("default", "live", metadata={"notify": True})
    frames = await recv_until(ws, lambda f: f.get("type") == "message")
    assert "cursor" not in frames[-1]


def test_outbox_row_cap_prunes_oldest(plugin, tmp_path):
    outbox = plugin.outbox.Outbox(tmp_path / "ob.db", max_rows=5)
    try:
        for i in range(7):
            outbox.append(
                "default",
                json.dumps({"v": 1, "type": "message", "payload": {"n": i}}),
            )
        assert outbox.latest_cursor() == 7  # cursor stays monotonic
        assert outbox.take_overflow_pruned() == 2
        rows = outbox.replay(0)
        assert [r["frame"]["payload"]["n"] for r in rows] == [2, 3, 4, 5, 6]
        assert outbox.take_overflow_pruned() == 0
    finally:
        outbox.close()


def test_outbox_delete_message_removes_all_frames_for_id(plugin, tmp_path):
    """Outbox.delete_message removes every frame for a message_id (any type)
    in the chat, leaving other messages intact; a thread_id scopes the delete."""
    outbox = plugin.outbox.Outbox(tmp_path / "ob.db")
    try:
        chat = "default"
        # A streaming message spans start/update/stop; a standalone message is
        # one frame. Plus an unrelated message that must survive.
        outbox.append(chat, json.dumps({"v": 1, "type": "message.start", "payload": {"message_id": "m1", "role": "assistant"}}))
        outbox.append(chat, json.dumps({"v": 1, "type": "message.update", "payload": {"message_id": "m1", "text": "hi"}}))
        outbox.append(chat, json.dumps({"v": 1, "type": "message.stop", "payload": {"message_id": "m1", "final_text": "hi"}}))
        outbox.append(chat, json.dumps({"v": 1, "type": "message", "payload": {"message_id": "m2", "role": "user", "text": "keep"}}))
        # A same-id message in a different thread must survive a flat-lane delete.
        outbox.append(chat, json.dumps({"v": 1, "type": "message", "thread_id": "t_1", "payload": {"message_id": "m1", "role": "user", "text": "thread"}}))
        removed = outbox.delete_message(chat, "m1")
        assert removed == 3  # the three flat-lane frames; the t_1 one is scoped out
        rows = outbox.replay(0)
        ids = [r["frame"].get("payload", {}).get("message_id") for r in rows]
        assert "m2" in ids
        assert ids.count("m1") == 1  # only the t_1 frame remains
        # Deleting it with the thread scope removes the last one.
        assert outbox.delete_message(chat, "m1", thread_id="t_1") == 1
        # Idempotent: deleting again removes nothing.
        assert outbox.delete_message(chat, "m1") == 0
    finally:
        outbox.close()


def test_outbox_delete_lane_removes_channel_and_thread_frames(plugin, tmp_path):
    """Outbox.delete_lane wipes a whole lane: a channel delete (thread_id=None)
    removes every frame for the chat (flat + all threads); a thread delete is
    scoped to that thread's frames only."""
    outbox = plugin.outbox.Outbox(tmp_path / "ob.db")
    try:
        chan = "chan_9"
        # Flat-lane frames + two threads' frames, plus an unrelated channel.
        outbox.append(chan, json.dumps({"v": 1, "type": "message", "payload": {"message_id": "a", "role": "user", "text": "flat"}}))
        outbox.append(chan, json.dumps({"v": 1, "type": "message", "thread_id": "t_1", "payload": {"message_id": "b", "role": "user", "text": "t1"}}))
        outbox.append(chan, json.dumps({"v": 1, "type": "message", "thread_id": "t_2", "payload": {"message_id": "c", "role": "user", "text": "t2"}}))
        outbox.append("chan_8", json.dumps({"v": 1, "type": "message", "payload": {"message_id": "z", "role": "user", "text": "other"}}))
        # Thread delete: only t_1's frame goes.
        assert outbox.delete_lane(chan, thread_id="t_1") == 1
        rows = outbox.replay(0)
        mids = [r["frame"].get("payload", {}).get("message_id") for r in rows]
        assert "b" not in mids and "a" in mids and "c" in mids and "z" in mids
        # Channel delete: the rest of the channel (flat + t_2) goes, chan_8 stays.
        assert outbox.delete_lane(chan) == 2
        rows = outbox.replay(0)
        mids = [r["frame"].get("payload", {}).get("message_id") for r in rows]
        assert mids == ["z"]
    finally:
        outbox.close()


def test_channels_delete_hard_deletes_row_and_child_threads(plugin, tmp_path):
    """ChannelDirectory.delete removes the row entirely (not archived) and, for
    a channel, its threads; the default channel cannot be deleted."""
    d = plugin.channels.ChannelDirectory(tmp_path / "ch.db")
    try:
        d.ensure_default("default", "Default")
        chan = d.create("Work", kind="channel")
        t1 = d.create("Topic", kind="thread", parent_chat_id=chan["chat_id"])
        # Deleting the channel removes it AND its thread from the directory.
        entry = d.delete(chan["chat_id"])
        assert entry is not None and entry["chat_id"] == chan["chat_id"]
        assert d.get(chan["chat_id"]) is None  # hard-deleted, not archived
        assert d.get(t1["chat_id"]) is None  # child thread gone too
        assert chan["chat_id"] not in [c["chat_id"] for c in d.list()]
        # Deleting a thread removes just that thread.
        chan2 = d.create("Ops", kind="channel")
        t2 = d.create("Bug", kind="thread", parent_chat_id=chan2["chat_id"])
        assert d.delete(t2["chat_id"]) is not None
        assert d.get(t2["chat_id"]) is None
        assert d.get(chan2["chat_id"]) is not None  # parent channel survives
        # The default channel cannot be deleted.
        assert d.delete("default") is None
        assert d.get("default") is not None
        # Unknown id -> None.
        assert d.delete("chan_nope") is None
    finally:
        d.close()


# ── Pairing sanity (M1 regression guard for the media-capable server) ──────


@pytest.mark.asyncio
async def test_wrong_token_rejected(adapter):
    """A request with a wrong Bearer token is rejected with 401 (the HTTP
    equivalent of the WS hello auth rejection)."""
    await adapter.connect()
    try:
        port = adapter._http_server.bound_port

        def _req() -> int:
            conn = HTTPConnection("127.0.0.1", port, timeout=5)
            conn.request(
                "GET",
                "/v1/events",
                headers={"Authorization": "Bearer wrong-token", "X-Iris-Device": DEVICE_ID},
            )
            resp = conn.getresponse()
            resp.read()
            status = resp.status
            conn.close()
            return status

        status = await asyncio.to_thread(_req)
        assert status == 401
    finally:
        await adapter.disconnect()


# ── M2: tool-detail capture (verbose args + post_tool_call output) ─────────


def test_extract_verbose_args(plugin):
    a = plugin.adapter
    content = '🔍 web_search(["query"])\n{"query": "hermes agent"}'
    line = '🔍 web_search(["query"])'
    assert a._extract_verbose_args(line, content) == {"query": "hermes agent"}
    # Non-verbose line (no "(keys)" header) -> None.
    assert a._extract_verbose_args('🔍 web_search: "x"', content) is None
    # Header with no JSON body on the following line -> None.
    assert a._extract_verbose_args(line, line) is None


def test_short_preview_from_args(plugin):
    a = plugin.adapter
    assert a._short_preview_from_args({"query": "hermes agent"}) == "hermes agent"
    # First non-empty string value wins (skips non-strings).
    assert a._short_preview_from_args({"n": 5, "path": "/tmp/x"}) == "/tmp/x"
    assert a._short_preview_from_args({}) is None
    prev = a._short_preview_from_args({"cmd": "a" * 100})
    assert prev is not None and prev.endswith("…") and len(prev) <= 60


def test_tool_end_fields_from_hook(plugin):
    a = plugin.adapter
    a._reset_tool_results()
    try:
        # No captured record yet -> empty (caller keeps ok=True default).
        assert a._tool_end_fields("terminal") == {}
        # Capture a completed call, then consume it.
        a._on_post_tool_call(
            tool_name="terminal",
            result='{"ok": true, "stdout": "hello"}',
            duration_ms=1234,
            status="ok",
        )
        fields = a._tool_end_fields("terminal")
        assert fields["ok"] is True
        assert fields["output_preview"] == '{"ok": true, "stdout": "hello"}'
        assert fields["duration"] == 1.234
        # Consumed -> a second lookup is empty.
        assert a._tool_end_fields("terminal") == {}
        # Error status -> ok False; zero duration omitted.
        a._on_post_tool_call(tool_name="terminal", result="boom", duration_ms=0, status="error")
        f2 = a._tool_end_fields("terminal")
        assert f2["ok"] is False and "duration" not in f2
    finally:
        a._reset_tool_results()


def test_parse_tool_line_or_block_verbose(plugin):
    a = plugin.adapter
    content = '🔍 web_search(["query"])\n{"query": "hermes agent"}'
    name, preview, args = a.IrisAdapter._parse_tool_line_or_block(
        '🔍 web_search(["query"])', content
    )
    assert name == "web_search"
    assert args == {"query": "hermes agent"}
    assert preview == "hermes agent"  # derived short preview
    # Non-verbose line -> args None, preview from the line.
    name2, preview2, args2 = a.IrisAdapter._parse_tool_line_or_block(
        '🔍 web_search: "x"', '🔍 web_search: "x"'
    )
    assert name2 == "web_search" and preview2 == "x" and args2 is None


def test_tool_start_frame_emoji_field(plugin):
    """``tool.start`` carries the cosmetic emoji when given, omits it when
    None (the app then falls back to its own default glyph)."""
    pf = plugin.protocol.tool_start
    f = pf("default", 3, "terminal", emoji="💻")
    assert f.payload["emoji"] == "💻"
    f2 = pf("default", 3, "terminal")
    assert "emoji" not in f2.payload
    f3 = pf("default", 3, "terminal", emoji=None)
    assert "emoji" not in f3.payload


def test_tool_emoji_resolves_via_hermes_display(plugin, monkeypatch):
    """``_tool_emoji`` delegates to hermes' ``get_tool_emoji`` (skin overrides
    -> registry) and maps unknown tools / failures to None (field omitted)."""
    a = plugin.adapter
    seen = {}

    def fake_get_tool_emoji(name, default="⚡"):
        seen[name] = default
        return {"terminal": "💻", "read_file": "📖"}.get(name, default)

    monkeypatch.setattr("agent.display.get_tool_emoji", fake_get_tool_emoji)
    assert a._tool_emoji("terminal") == "💻"
    assert a._tool_emoji("read_file") == "📖"
    # Unknown tool: the adapter asks for an empty default and drops it.
    assert a._tool_emoji("no_such_tool") is None
    assert seen["no_such_tool"] == ""
    # A display-layer failure degrades to None, never a crash.
    def boom(name, default="⚡"):
        raise RuntimeError("no display")

    monkeypatch.setattr("agent.display.get_tool_emoji", boom)
    assert a._tool_emoji("terminal") is None


@pytest.mark.asyncio
async def test_tool_start_frame_carries_emoji(plugin, adapter, ws_client, monkeypatch):
    """A tool-progress line emits ``tool.start`` with the per-tool emoji
    resolved from hermes' display layer."""
    ws, _ = ws_client
    monkeypatch.setattr(
        "agent.display.get_tool_emoji",
        lambda name, default="⚡": "💻" if name == "terminal" else default,
    )
    res = await adapter.send('default', '💻 terminal: "ls -la"')
    assert res.success
    frames = await recv_until(ws, lambda f: f.get("type") == "tool.start")
    payload = frames[-1]["payload"]
    assert payload["name"] == "terminal"
    assert payload["emoji"] == "💻"
    # Unknown tool -> field omitted (app falls back to its default glyph).
    monkeypatch.setattr("agent.display.get_tool_emoji", lambda name, default="⚡": default)
    res2 = await adapter.send('default', '🔧 patch: "x"')
    assert res2.success
    frames2 = await recv_until(
        ws, lambda f: f.get("type") == "tool.start" and f["payload"]["name"] == "patch"
    )
    assert "emoji" not in frames2[-1]["payload"]


# ── Slash-command catalog (the app's "/" drawer) ──────────────────────────


def test_slash_catalog_shape_and_gateway_filtering(plugin):
    """The catalog is the gateway-available COMMAND_REGISTRY subset: every
    entry carries the wire shape, cli_only commands are excluded, aliases
    keep the leading slash."""
    a = plugin.adapter
    catalog = a._slash_command_catalog()
    assert catalog, "expected a non-empty slash catalog"
    names = {e["name"] for e in catalog}
    # A core gateway command is present with its alias.
    assert "/new" in names
    new = next(e for e in catalog if e["name"] == "/new")
    assert "/reset" in new["aliases"]
    # A cli_only command is not surfaced to the gateway platform.
    assert "/clear" not in names
    for e in catalog:
        assert e["name"].startswith("/")
        assert isinstance(e["description"], str) and e["description"]
        assert isinstance(e["args_hint"], str)
        assert isinstance(e["category"], str) and e["category"]
        assert all(alias.startswith("/") for alias in e["aliases"])


@pytest.mark.asyncio
async def test_commands_catalog_frame_roundtrip(plugin, adapter, ws_client):
    """A ``commands.catalog`` request is answered with the catalog on the
    same id; the response carries the wire shape the app's drawer consumes."""
    ws, _ = ws_client
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 42,
                "type": "commands.catalog",
                "payload": {},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "commands.catalog")
    resp = frames[-1]
    assert resp["id"] == 42
    commands = resp["payload"]["commands"]
    assert commands, "expected a non-empty catalog"
    names = {c["name"] for c in commands}
    assert "/new" in names
    assert "/clear" not in names  # cli_only
    for c in commands:
        assert c["name"].startswith("/")
        assert "description" in c and "args_hint" in c and "category" in c
        assert isinstance(c["aliases"], list)