"""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/android`` 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 start -> binary chunks -> end reassembles + sha256 verified
  * over-limit (declared and mid-stream) -> ``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)
  * ``media.pull`` 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
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-android-token-0123456789"
DEVICE_ID = "test-device"


def _plugin_dir() -> Path:
    env = os.environ.get("ANDROID_PLUGIN_DIR")
    if env:
        return Path(env)
    # hermes-agent/tests/gateway/test_android.py -> repo root is parents[3].
    return Path(__file__).resolve().parents[3] / "gateway-plugin"


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 = "android_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"android 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 AndroidAdapter with an isolated HERMES_HOME (conftest)."""
    monkeypatch.setenv("ANDROID_TOKEN", TOKEN)
    from gateway.platform_registry import PlatformEntry, platform_registry

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

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


async def _hello(ws) -> dict:
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "type": "hello",
                "payload": {
                    "token": TOKEN,
                    "device_id": DEVICE_ID,
                    "device_name": "Test Device",
                    "caps": {},
                },
            }
        )
    )
    raw = await asyncio.wait_for(ws.recv(), timeout=5)
    ack = json.loads(raw)
    assert ack["type"] == "hello.ack", f"expected hello.ack, got {ack}"
    return ack


@pytest_asyncio.fixture
async def ws_client(adapter):
    """Connected + paired WS client; the adapter's server runs on an
    ephemeral port for the duration of the test."""
    from websockets.asyncio.client import connect

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(
            f"ws://127.0.0.1:{port}/ws", max_size=4 * 1024 * 1024
        ) as ws:
            ack = await _hello(ws)
            yield ws, ack
    finally:
        await adapter.disconnect()


async def recv_until(ws, predicate, timeout: float = 10.0) -> list:
    """Collect frames (dicts; binary frames as ("binary", bytes)) until
    *predicate* matches a JSON 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 "binary" for f in frames)
            )
        raw = await asyncio.wait_for(ws.recv(), timeout=remaining)
        if isinstance(raw, (bytes, bytearray)):
            frames.append(("binary", bytes(raw)))
            continue
        frame = json.loads(raw)
        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) -> dict:
    """Drive a full media.upload flow; returns the terminal frame (ack or error)."""
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": request_id,
                "type": "media.upload.start",
                "payload": {
                    "media_ref": media_ref,
                    "kind": kind,
                    "mime": mime,
                    "size": len(data),
                    "filename": filename,
                },
            }
        )
    )
    # Two chunks to exercise reassembly.
    half = len(data) // 2
    await ws.send(data[:half])
    await ws.send(data[half:])
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": request_id + 1,
                "type": "media.upload.end",
                "payload": {
                    "media_ref": media_ref,
                    "sha256": hashlib.sha256(data).hexdigest(),
                },
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") in ("media.upload.ack", "error"))
    return frames[-1]


# ── 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="android",
        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("android: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("android: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-android turn must not pollute the android 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("android: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="android",
        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("android:default", "hello", metadata={"notify": True})
    # The outbox history reconstruction must carry the runtime object.
    page = adapter._outbox.history("android: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["id"] == 2
    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
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 1,
                "type": "media.upload.start",
                "payload": {
                    "media_ref": "mu_big",
                    "kind": "document",
                    "mime": "application/pdf",
                    "size": limit + 1,
                    "filename": "big.pdf",
                },
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error")
    err = frames[-1]
    assert err["payload"]["code"] == "media_too_large"
    assert err["id"] == 1


@pytest.mark.asyncio
async def test_upload_midstream_over_limit_rejected(adapter, ws_client):
    ws, _ = ws_client
    limit = adapter.max_upload_bytes
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 1,
                "type": "media.upload.start",
                "payload": {
                    "media_ref": "mu_mid",
                    "kind": "document",
                    "mime": "application/octet-stream",
                    "size": limit,
                    "filename": "mid.bin",
                },
            }
        )
    )
    # Declared size passes the start check; the stream exceeds it.
    await ws.send(b"x" * (limit // 2))
    await ws.send(b"x" * (limit // 2 + 1))
    frames = await recv_until(ws, lambda f: f.get("type") == "error")
    assert frames[-1]["payload"]["code"] == "media_too_large"
    # The session is discarded: a late end cannot complete it.
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 2,
                "type": "media.upload.end",
                "payload": {"media_ref": "mu_mid", "sha256": "0" * 64},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 2)
    assert frames[-1]["payload"]["code"] == "not_found"


@pytest.mark.asyncio
async def test_upload_sha256_mismatch_rejected(adapter, ws_client):
    ws, _ = ws_client
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 1,
                "type": "media.upload.start",
                "payload": {
                    "media_ref": "mu_bad",
                    "kind": "image",
                    "mime": "image/png",
                    "size": len(PNG_1X1),
                    "filename": "bad.png",
                },
            }
        )
    )
    await ws.send(PNG_1X1)
    await ws.send(
        json.dumps(
            {
                "v": 1,
                "id": 2,
                "type": "media.upload.end",
                "payload": {"media_ref": "mu_bad", "sha256": "0" * 64},
            }
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error")
    assert frames[-1]["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": "android: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"] == "android:default"
    assert created["payload"]["name"] == "fix the login bug please"
    thread_id = created["payload"]["chat_id"]

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

    assert len(captured) == 1
    assert captured[0].source.chat_id == "android: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": "android: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": "android: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": "android: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": "android: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": "android: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("android: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 = "android: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 = "android: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 = "android: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 = "android: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 = "android: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": "android: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 = "android: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("android: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("android: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(
        "android: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("android: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("android: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)
    )

    await ws.send(
        json.dumps(
            {"v": 1, "id": 9, "type": "media.pull", "payload": {"media_id": entry.media_id}}
        )
    )
    chunks: list[bytes] = []
    terminal = None
    while terminal is None:
        raw = await asyncio.wait_for(ws.recv(), timeout=10)
        if isinstance(raw, (bytes, bytearray)):
            chunks.append(bytes(raw))
            continue
        frame = json.loads(raw)
        if frame.get("type") == "media.pull.end":
            terminal = frame
    assert terminal["id"] == 9
    assert terminal["payload"]["ok"] is True
    assert b"".join(chunks) == PNG_1X1


@pytest.mark.asyncio
async def test_pull_rejects_unknown_and_denied(adapter, ws_client):
    ws, _ = ws_client
    # Unknown media_id.
    await ws.send(
        json.dumps({"v": 1, "id": 1, "type": "media.pull", "payload": {"media_id": "md_nope"}})
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 1)
    assert frames[-1]["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
    )
    await ws.send(
        json.dumps(
            {"v": 1, "id": 2, "type": "media.pull", "payload": {"media_id": entry.media_id}}
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 2)
    assert frames[-1]["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()
    await ws.send(
        json.dumps(
            {"v": 1, "id": 3, "type": "media.pull", "payload": {"media_id": entry2.media_id}}
        )
    )
    frames = await recv_until(ws, lambda f: f.get("type") == "error" and f.get("id") == 3)
    assert frames[-1]["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="android:default",
        title="Iris",
        body="hello",
        data={"chat_id": "android: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="android:default",
        title="Iris",
        body="hi",
        data={"chat_id": "android: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="android:default",
        title="Iris",
        body="hi",
        data={"chat_id": "android: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(
        "android: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"] == "android:default"
    assert call["data"]["kind"] == "message"
    assert call["data"]["chat_id"] == "android: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("android: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("android: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("android:default", "seg", metadata={"expect_edits": True})
    stream_id = adapter._turns["android:default"].stream_id
    await adapter.edit_message("android: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(
        "android:default",
        plugin.protocol.notification(
            "android: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("android: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("android:default", "unconfigured", metadata={"notify": True})
    assert fake.calls == []


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


@pytest.mark.asyncio
async def test_fcm_register_updates_registry_and_live_conn(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: WS 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"
    conn = adapter._ws_server.connection(DEVICE_ID)
    assert conn.fcm_token == "rotated-token"
    assert conn.ntfy_topic == "dev-topic"

    # The next offline push targets the rotated token.
    await ws.close()
    for _ in range(100):
        if not adapter._ws_server.has_devices():
            break
        await asyncio.sleep(0.01)
    fake = _FakePush()
    adapter._push = fake
    await adapter.send("android: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": "android: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("android: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("android: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(
            "android: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("android:default", "one", metadata={"notify": True})
    await adapter.send("android:default", "two", metadata={"notify": True})
    assert adapter._outbox.latest_cursor() == 2

    from websockets.asyncio.client import connect

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(f"ws://127.0.0.1:{port}/ws") as ws:
            ack = await _hello(ws)
            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 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)."""
    from websockets.asyncio.client import connect

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(f"ws://127.0.0.1:{port}/ws") as ws:
            ack = await _hello(ws)
            assert ack["payload"]["last_pushed_cursor"] == 0
    finally:
        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."""
    from websockets.asyncio.client import connect

    fake = _FakePush()
    adapter._push = fake
    adapter._devices.upsert(DEVICE_ID, "Test", {}, fcm_token="tok-1")

    await adapter.send("android: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("android: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["android:default"] = 0.0
    await adapter.send("android: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["android:default"] = 0.0
    await adapter.send("android:default", "four", metadata={"notify": True})
    assert adapter._devices.last_pushed_cursor(DEVICE_ID) == 3

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(f"ws://127.0.0.1:{port}/ws") as ws:
            ack = await _hello(ws)
            assert ack["payload"]["last_pushed_cursor"] == 3
    finally:
        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."""
    from websockets.asyncio.client import connect

    await adapter.send("android:default", "one", metadata={"notify": True})
    await adapter.send("android:default", "two", metadata={"notify": True})

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(f"ws://127.0.0.1:{port}/ws") as ws:
            await _hello(ws)
            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 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("android: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(
                "android: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 = "android: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 = "android: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("android: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("android: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("android:default") is None
        assert d.get("android:default") is not None
        # Unknown id -> None.
        assert d.delete("android: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):
    from websockets.asyncio.client import connect

    await adapter.connect()
    try:
        port = adapter._ws_server._server.sockets[0].getsockname()[1]
        async with connect(f"ws://127.0.0.1:{port}/ws") as ws:
            await ws.send(
                json.dumps(
                    {
                        "v": 1,
                        "type": "hello",
                        "payload": {
                            "token": "wrong-token",
                            "device_id": DEVICE_ID,
                            "device_name": "Bad",
                            "caps": {},
                        },
                    }
                )
            )
            raw = await asyncio.wait_for(ws.recv(), timeout=5)
            err = json.loads(raw)
            assert err["type"] == "error"
            assert err["payload"]["code"] == "auth"
    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.AndroidAdapter._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.AndroidAdapter._parse_tool_line_or_block(
        '🔍 web_search: "x"', '🔍 web_search: "x"'
    )
    assert name2 == "web_search" and preview2 == "x" and args2 is None


# ── 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)