HTTP transport: drop WS server, offline send queue + dead-stream watchdog

Gateway (docs/19):
- Remove ws_server.py; frame dispatch factored into dispatch.py
- http_server: media upload/pull, pairing over HTTP
- protocol: media frames mirrored; tests + ws_probe updated for HTTP

App:
- HttpGateway: postFrame/uploadMedia/pullMedia no longer throw on
  network failure (PostResult ok=false / Result.failure) — uncaught
  SocketTimeoutException on Dispatchers.Default crashed the app
- GatewayClient: dead-stream watchdog (health probe every 10s, 2
  failures -> redial in ~20s instead of the 45s SSE read timeout);
  state flips to Reconnecting when the stream dies, restored from the
  last hello.ack on long-poll success; poke() + backoff reset on app
  resume (MainActivity.onResume)
- Offline sends: composer enabled while disconnected; a send with no
  response (status 0) stays queued (Pending) and is auto-resent on the
  next (re)connect after a 2s outbox-replay grace; gateway 4xx
  rejections fail the bubble (tap to retry, no auto-loop)
- ChatStore: echo-replace and thread-relocate also match Failed
  bubbles (POST response lost in a network drop); loadHistory dedupes
  local failed bubbles the server already has; failMessage()
- MainActivity: poke() on resume so a backgrounded app reconnects
  promptly instead of waiting out the backoff

gateway-plugin/tests/test_android.py

@@ -5,13 +5,13 @@ The plugin lives in the sibling ``iris_x_hermes`` checkout (installed into
 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``
+  * 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)
-  * ``media.pull`` serves only allowed paths (denied/unknown -> ``not_found``)
+  * ``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``.
@@ -26,6 +26,9 @@ 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
@@ -47,8 +50,14 @@ 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"
+    # 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():
@@ -105,7 +114,7 @@ def adapter(plugin, monkeypatch):
     config = SimpleNamespace(
         extra={
             "host": "127.0.0.1",
-            "port": 0,  # ephemeral port
+            "http_port": 0,  # ephemeral HTTP port
             "max_upload_bytes": 1024 * 1024,  # 1 MiB -- keeps over-limit tests fast
         },
         home_channel=None,
@@ -122,48 +131,206 @@ def adapter(plugin, monkeypatch):
         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": {},
-                },
-            }
+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},
         )
-    )
-    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
+        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 WS client; the adapter's server runs on an
-    ephemeral port for the duration of the test."""
-    from websockets.asyncio.client import connect
-
+    """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:
-        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
+        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; binary frames as ("binary", bytes)) until
-    *predicate* matches a JSON frame. Returns all frames collected."""
+    """Collect frames (dicts) until *predicate* matches a frame. Returns all
+    frames collected."""
     frames: list = []
     loop = asyncio.get_running_loop()
     deadline = loop.time() + timeout
@@ -172,13 +339,9 @@ async def recv_until(ws, predicate, timeout: float = 10.0) -> list:
         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)
+                + ", ".join(f.get("type", "?") if isinstance(f, dict) else "?" 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)
+        frame = await ws.recv(timeout=remaining)
         frames.append(frame)
         if predicate(frame):
             return frames
@@ -186,43 +349,42 @@ async def recv_until(ws, predicate, timeout: float = 10.0) -> list:
 
 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,
-                },
-            }
-        )
+                      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
     )
-    # 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(),
-                },
-            }
+
+
+# ── 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}"
         )
-    )
-    frames = await recv_until(ws, lambda f: f.get("type") in ("media.upload.ack", "error"))
-    return frames[-1]
+    finally:
+        await client.close()
+        # Idempotent: the second call is a no-op on the already-stopped server.
+        await adapter.disconnect()
 
 
 # ── Pure helpers ────────────────────────────────────────────────────────────
@@ -375,7 +537,6 @@ async def test_upload_reassembles_verifies_and_caches(adapter, ws_client):
 
     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"
 
@@ -396,100 +557,37 @@ async def test_upload_reassembles_verifies_and_caches(adapter, ws_client):
 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",
-                },
-            }
-        )
+    # 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",
     )
-    frames = await recv_until(ws, lambda f: f.get("type") == "error")
-    err = frames[-1]
+    assert err["type"] == "error"
     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):
+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
-    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"
+    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
-    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"
+    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
 
 
@@ -1223,47 +1321,26 @@ async def test_pull_serves_allowed_path(adapter, ws_client):
         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
+    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.
-    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"
+    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
     )
-    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"
+    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
@@ -1274,13 +1351,9 @@ async def test_pull_rejects_unknown_and_denied(adapter, ws_client):
         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"
+    status, body = await ws.pull(entry2.media_id)
+    assert status == 404
+    assert json.loads(body)["payload"]["code"] == "not_found"
 
 
 # ── M5: push backends (pure) ───────────────────────────────────────────────
@@ -1653,7 +1726,7 @@ async def test_push_skipped_when_backend_unconfigured(adapter):
 
 
 @pytest.mark.asyncio
-async def test_fcm_register_updates_registry_and_live_conn(adapter, ws_client):
+async def test_fcm_register_updates_registry(adapter, ws_client):
     ws, _ = ws_client
     await ws.send(
         json.dumps(
@@ -1664,7 +1737,7 @@ async def test_fcm_register_updates_registry_and_live_conn(adapter, ws_client):
             }
         )
     )
-    # Ordering barrier: WS frames are processed in order, so by the time the
+    # 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": {}})
@@ -1673,16 +1746,13 @@ async def test_fcm_register_updates_registry_and_live_conn(adapter, ws_client):
     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.
+    # 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()
-    for _ in range(100):
-        if not adapter._ws_server.has_devices():
-            break
-        await asyncio.sleep(0.01)
+    with adapter._http_server._subs_lock:
+        adapter._http_server._subs.clear()
     fake = _FakePush()
     adapter._push = fake
     await adapter.send("android:default", "after rotation", metadata={"notify": True})
@@ -1898,32 +1968,32 @@ async def test_sync_replays_parked_frames_and_done_cursor(adapter):
     await adapter.send("android:default", "two", metadata={"notify": True})
     assert adapter._outbox.latest_cursor() == 2
 
-    from websockets.asyncio.client import connect
-
     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:
-        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
+        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()
 
 
@@ -1931,15 +2001,13 @@ async def test_sync_replays_parked_frames_and_done_cursor(adapter):
 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()
+    ws = HttpTestClient(adapter._http_server.bound_port)
     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
+        ack = await ws.start()
+        assert ack["payload"]["last_pushed_cursor"] == 0
     finally:
+        await ws.close()
         await adapter.disconnect()
 
 
@@ -1949,8 +2017,6 @@ async def test_push_success_advances_last_pushed_cursor(adapter):
     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")
@@ -1978,12 +2044,12 @@ async def test_push_success_advances_last_pushed_cursor(adapter):
     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:
-        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
+        ack = await ws.start()
+        assert ack["payload"]["last_pushed_cursor"] == 3
     finally:
+        await ws.close()
         await adapter.disconnect()
 
 
@@ -1992,25 +2058,23 @@ 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()
+    ws = HttpTestClient(adapter._http_server.bound_port, cursor=adapter._outbox.latest_cursor())
     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]
+        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()
 
 
@@ -2129,30 +2193,27 @@ def test_channels_delete_hard_deletes_row_and_child_threads(plugin, tmp_path):
 
 @pytest.mark.asyncio
 async def test_wrong_token_rejected(adapter):
-    from websockets.asyncio.client import connect
-
+    """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._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": {},
-                        },
-                    }
-                )
+        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},
             )
-            raw = await asyncio.wait_for(ws.recv(), timeout=5)
-            err = json.loads(raw)
-            assert err["type"] == "error"
-            assert err["payload"]["code"] == "auth"
+            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()
 

gateway-plugin/tests/test_android_http.py

@@ -0,0 +1,826 @@
+"""Tests for the android plugin's HTTP fallback transport (docs/19).
+
+The plugin lives in the sibling ``iris_x_hermes`` checkout; tests load it
+from the source tree directly (same pattern as ``test_android.py``).
+
+Coverage (docs/19 §19.12):
+  * auth: bad/missing token -> 401; missing device header -> 401;
+    allowlist rejection -> 401
+  * ``POST /v1/frame``: valid ``message.send`` dispatches (202 + echo on
+    the SSE stream); empty text -> 400 error frame; automation channel ->
+    400; bad JSON -> 400; wrong content-type -> 400; oversize body -> 413;
+    media frames -> 400 (WS-only in v1); rate limit -> 429
+  * SSE: catch-up rows carry correct ``id``s + cursor envelope;
+    ``event: hello`` present; a live frame appended after connect arrives
+    on the stream; ``Last-Event-ID`` resume replays exactly the delta;
+    heartbeat observed
+  * long-poll: returns on new frame; empty 200 at timeout with advanced
+    cursor
+  * **delivery counting (docs/19 §19.8)**: a frame with only an SSE
+    subscriber is ``delivered >= 1`` -> NO push fired (the critical
+    regression test)
+
+Run via ``scripts/run_tests.sh tests/gateway/test_android_http.py``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import contextlib
+import importlib.util
+import json
+import os
+import sys
+import time
+from http.client import HTTPConnection
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_asyncio
+
+# Test-only token (not a credential; the adapter is built with it via
+# monkeypatch in the fixture below).
+# pi-lens-ignore: S105
+TOKEN = "test-android-http-token-0123456789"
+DEVICE_ID = "test-http-device"
+CHAT_ID = "android:default"
+
+# 1x1 PNG (same fixture as test_android.py).
+PNG_1X1 = base64.b64decode(
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJ"
+    "AAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
+)
+
+
+def _plugin_dir() -> Path:
+    env = os.environ.get("ANDROID_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 (same
+    pattern as test_android.py)."""
+    name = "android_plugin_http_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
+
+    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 WS port
+            "http_port": 0,  # ephemeral HTTP port
+            "max_upload_bytes": 1024 * 1024,
+        },
+        home_channel=None,
+    )
+    a = plugin.adapter.AndroidAdapter(config)
+    yield a
+    with contextlib.suppress(Exception):
+        a._devices.close()
+    with contextlib.suppress(Exception):
+        a._outbox.close()
+
+
+@pytest_asyncio.fixture
+async def gw(adapter):
+    """Connected adapter (WS + HTTP legs up); the HTTP port is ephemeral."""
+    await adapter.connect()
+    try:
+        yield adapter
+    finally:
+        await adapter.disconnect()
+
+
+def http_port(adapter) -> int:
+    assert adapter._http_server.enabled, "HTTP leg should be enabled after connect()"
+    return adapter._http_server.bound_port
+
+
+# ── Blocking HTTP helpers (run via asyncio.to_thread) ──────────────────────
+
+
+def _request(
+    port: int,
+    method: str,
+    path: str,
+    *,
+    token: str | None = TOKEN,
+    device: str | None = DEVICE_ID,
+    body: bytes | str | None = None,
+    content_type: str = "application/json",
+    timeout: float = 10.0,
+    extra_headers: dict | None = None,
+) -> tuple[int, bytes]:
+    conn = HTTPConnection("127.0.0.1", port, timeout=timeout)
+    headers = {}
+    if token is not None:
+        headers["Authorization"] = f"Bearer {token}"
+    if device is not None:
+        headers["X-Iris-Device"] = device
+    if extra_headers:
+        headers.update(extra_headers)
+    if body is not None:
+        data = body if isinstance(body, bytes) else body.encode("utf-8")
+        headers["Content-Type"] = content_type
+        conn.request(method, path, body=data, headers=headers)
+    else:
+        conn.request(method, path, headers=headers)
+    resp = conn.getresponse()
+    payload = resp.read()
+    status = resp.status
+    conn.close()
+    return status, payload
+
+
+def _post_frame(port: int, frame: dict, **kw) -> tuple[int, dict]:
+    status, payload = _request(port, "POST", "/v1/frame", body=json.dumps(frame), **kw)
+    return status, json.loads(payload)
+
+
+def _frame_json(frame: dict) -> dict:
+    return {"v": 1, **frame}
+
+
+def _parse_sse(lines: list[str]) -> tuple[list[tuple[str | None, str | None, str]], int]:
+    """Parse raw SSE lines into ``[(event, id, data), ...]`` + comment count."""
+    events: list[tuple[str | None, str | None, str]] = []
+    comments = 0
+    cur_event: str | None = None
+    cur_id: str | None = None
+    cur_data: list[str] = []
+    for raw in lines:
+        line = raw.rstrip("\r\n")
+        if line == "":
+            if cur_data:
+                events.append((cur_event, cur_id, "\n".join(cur_data)))
+            cur_event, cur_id, cur_data = None, None, []
+        elif line.startswith(":"):
+            comments += 1
+        else:
+            field, _, value = line.partition(":")
+            if value.startswith(" "):
+                value = value[1:]
+            if field == "event":
+                cur_event = value
+            elif field == "id":
+                cur_id = value
+            elif field == "data":
+                cur_data.append(value)
+    return events, comments
+
+
+def _sse_open(port: int, *, cursor: int | None = None, last_event_id: str | None = None):
+    """Open an SSE connection (blocking); returns the HTTPResponse (read
+    lines via ``_sse_read_lines``; close with ``resp.close()``)."""
+    conn = HTTPConnection("127.0.0.1", port, timeout=30)
+    path = "/v1/events" + (f"?cursor={cursor}" if cursor is not None else "")
+    headers = {
+        "Authorization": f"Bearer {TOKEN}",
+        "X-Iris-Device": DEVICE_ID,
+    }
+    if last_event_id is not None:
+        headers["Last-Event-ID"] = last_event_id
+    conn.request("GET", path, headers=headers)
+    resp = conn.getresponse()
+    assert resp.status == 200, f"SSE open failed: {resp.status}"
+    assert resp.getheader("Content-Type", "").startswith("text/event-stream")
+    return resp
+
+
+def _sse_read_lines(resp, n: int, timeout: float = 10.0) -> list[str]:
+    """Read up to n lines from the SSE stream (blocking)."""
+    raw = resp.fp.raw
+    sock = getattr(raw, "_sock", None)
+    if sock is not None:
+        sock.settimeout(timeout)
+    lines: list[str] = []
+    while len(lines) < n:
+        line = resp.fp.readline()
+        if not line:
+            break
+        lines.append(line.decode("utf-8"))
+    return lines
+
+
+# ── /v1/health ──────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_health_no_auth(gw):
+    status, payload = await asyncio.to_thread(
+        _request, http_port(gw), "GET", "/v1/health", token=None, device=None
+    )
+    assert status == 200
+    assert json.loads(payload) == {"ok": True}
+
+
+@pytest.mark.asyncio
+async def test_unknown_path_404(gw):
+    status, _ = await asyncio.to_thread(_request, http_port(gw), "GET", "/v1/nope")
+    assert status == 404
+
+
+# ── Auth ────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_post_bad_token_401(gw):
+    status, _ = await asyncio.to_thread(
+        _request,
+        http_port(gw),
+        "POST",
+        "/v1/frame",
+        token="wrong-token",
+        body=json.dumps(_frame_json({"type": "ping", "payload": {}})),
+    )
+    assert status == 401
+
+
+@pytest.mark.asyncio
+async def test_post_missing_token_401(gw):
+    status, _ = await asyncio.to_thread(
+        _request,
+        http_port(gw),
+        "POST",
+        "/v1/frame",
+        token=None,
+        body=json.dumps(_frame_json({"type": "ping", "payload": {}})),
+    )
+    assert status == 401
+
+
+@pytest.mark.asyncio
+async def test_post_missing_device_401(gw):
+    status, _ = await asyncio.to_thread(
+        _request,
+        http_port(gw),
+        "POST",
+        "/v1/frame",
+        device=None,
+        body=json.dumps(_frame_json({"type": "ping", "payload": {}})),
+    )
+    assert status == 401
+
+
+@pytest.mark.asyncio
+async def test_post_allowlist_rejection_401(gw):
+    gw.allowed_users = ["some-other-device"]
+    gw.allow_all = False
+    status, _ = await asyncio.to_thread(
+        _request,
+        http_port(gw),
+        "POST",
+        "/v1/frame",
+        body=json.dumps(_frame_json({"type": "ping", "payload": {}})),
+    )
+    assert status == 401
+
+
+# ── POST /v1/frame: validation ─────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_post_bad_json_400(gw):
+    status, payload = await asyncio.to_thread(
+        _request, http_port(gw), "POST", "/v1/frame", body=b"not json"
+    )
+    assert status == 400
+    frame = json.loads(payload)
+    assert frame["type"] == "error"
+
+
+@pytest.mark.asyncio
+async def test_post_wrong_content_type_400(gw):
+    status, _ = await asyncio.to_thread(
+        _request,
+        http_port(gw),
+        "POST",
+        "/v1/frame",
+        body=json.dumps(_frame_json({"type": "ping", "payload": {}})),
+        content_type="text/plain",
+    )
+    assert status == 400
+
+
+@pytest.mark.asyncio
+async def test_post_oversize_body_413(gw):
+    big = json.dumps(_frame_json({"type": "ping", "payload": {"pad": "x" * (1024 * 1024 + 1)}}))
+    status, _ = await asyncio.to_thread(_request, http_port(gw), "POST", "/v1/frame", body=big)
+    assert status == 413
+
+
+@pytest.mark.asyncio
+async def test_post_empty_message_400(gw):
+    gw.handle_message = AsyncMock()
+    status, payload = await asyncio.to_thread(
+        _post_frame,
+        http_port(gw),
+        _frame_json(
+            {"id": 7, "type": "message.send", "chat_id": CHAT_ID, "payload": {"text": "  "}}
+        ),
+    )
+    assert status == 400
+    frame = payload
+    assert frame["type"] == "error"
+    assert frame["id"] == 7
+    assert frame["payload"]["code"] == "unsupported"
+    gw.handle_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_post_automation_channel_400(gw):
+    gw.handle_message = AsyncMock()
+    entry = gw._channels.create(name="Cron")
+    gw._channels.set_automation(entry["chat_id"], True)
+    status, payload = await asyncio.to_thread(
+        _post_frame,
+        http_port(gw),
+        _frame_json(
+            {
+                "id": 8,
+                "type": "message.send",
+                "chat_id": entry["chat_id"],
+                "payload": {"text": "hi"},
+            }
+        ),
+    )
+    assert status == 400
+    assert payload["type"] == "error"
+    gw.handle_message.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_post_rate_limit_429(gw):
+    gw.handle_message = AsyncMock()
+    port = http_port(gw)
+    # Exhaust the per-device bucket (INBOUND_BURST = 40) then expect 429.
+    got_429 = False
+    for i in range(60):
+        status, _ = await asyncio.to_thread(
+            _post_frame,
+            port,
+            _frame_json({"id": i, "type": "ping", "payload": {}}),
+        )
+        if status == 429:
+            got_429 = True
+            break
+    assert got_429, "expected a 429 within 60 rapid frames"
+
+
+# ── POST /v1/frame: dispatch ────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_post_message_send_dispatches(gw):
+    """202 ack; the user echo + read receipt arrive on the SSE stream; the
+    agent turn fires (docs/19 §19.7: async responses on the event stream)."""
+    gw.handle_message = AsyncMock()
+    port = http_port(gw)
+    conn = _sse_open(port)
+    try:
+        # Consume the open sequence (hello + status = 6 lines) first.
+        lines = _sse_read_lines(conn, 6, timeout=5)
+        events, _ = _parse_sse(lines)
+        assert events[0][0] == "hello"
+
+        status, payload = await asyncio.to_thread(
+            _post_frame,
+            port,
+            _frame_json(
+                {
+                    "id": 42,
+                    "type": "message.send",
+                    "chat_id": CHAT_ID,
+                    "payload": {"text": "hi there"},
+                }
+            ),
+        )
+        # The read receipt (sent when the turn is handed to the agent) is
+        # the handler's single point-to-point reply -> 200 with the frame
+        # as the body (a plain 202 {"ok": true} is also valid when no
+        # synchronous reply exists).
+        assert status in (200, 202)
+        if status == 200:
+            assert payload["type"] == "read.receipt"
+        else:
+            assert payload == {"ok": True}
+
+        # The echo must arrive on the stream, tagged with its outbox
+        # cursor as the SSE id (live frames carry no cursor in the
+        # envelope, same as the WS path).
+        deadline = time.monotonic() + 10
+        echo = None
+        while time.monotonic() < deadline and echo is None:
+            lines = _sse_read_lines(conn, 4, timeout=5)
+            for _event, sse_id, data in _parse_sse(lines)[0]:
+                frame = json.loads(data)
+                if (
+                    frame.get("type") == "message"
+                    and frame.get("payload", {}).get("text") == "hi there"
+                ):
+                    echo = (frame, sse_id)
+        assert echo is not None, "user echo did not arrive on the SSE stream"
+        assert echo[1] is not None  # SSE id = outbox cursor
+        await asyncio.sleep(0.2)
+        gw.handle_message.assert_called_once()
+    finally:
+        conn.close()
+
+
+# ── SSE: catch-up, hello, live, resume, heartbeat ──────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sse_catchup_and_hello(gw):
+    """Catch-up rows carry correct ids + cursor envelope; hello present."""
+    port = http_port(gw)
+    # Park two frames through the real outbound path (no live devices).
+    push_calls: list = []
+    gw._maybe_push = AsyncMock(side_effect=lambda *a, **k: push_calls.append(a))
+    for text in ("one", "two"):
+        await _park_frame(gw, text)
+    cursors = [1, 2]
+
+    conn = _sse_open(port, cursor=0)
+    try:
+        # 2 replayed frames (id + event + data + blank = 4 lines each) +
+        # hello (3 lines) + status (3 lines) = 14 lines.
+        lines = _sse_read_lines(conn, 14, timeout=5)
+        events, _ = _parse_sse(lines)
+        assert events[0][0] == "frame"
+        assert events[0][1] == str(cursors[0])
+        f0 = json.loads(events[0][2])
+        assert f0["payload"]["text"] == "one"
+        assert f0["cursor"] == cursors[0]
+        assert events[1][0] == "frame"
+        assert events[1][1] == str(cursors[1])
+        assert json.loads(events[1][2])["payload"]["text"] == "two"
+        assert events[2][0] == "hello"
+        hello = json.loads(events[2][2])
+        assert hello["type"] == "hello.ack"
+        assert hello["payload"]["sync_cursor"] == 2
+        assert events[3][0] == "frame"
+        assert json.loads(events[3][2])["type"] == "status"
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_sse_live_frame_after_connect(gw):
+    port = http_port(gw)
+    conn = _sse_open(port)
+    try:
+        # Consume the open sequence (hello + status = 6 lines).
+        _sse_read_lines(conn, 6, timeout=5)
+        await _park_frame(gw, "live!")
+        deadline = time.monotonic() + 10
+        got = None
+        while time.monotonic() < deadline and got is None:
+            lines = _sse_read_lines(conn, 4, timeout=5)
+            for _event, sse_id, data in _parse_sse(lines)[0]:
+                frame = json.loads(data)
+                if frame.get("payload", {}).get("text") == "live!":
+                    got = (frame, sse_id)
+        assert got is not None, "live frame did not arrive on the SSE stream"
+        assert got[1] is not None  # SSE id = outbox cursor
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_sse_last_event_id_resume(gw):
+    """Resume with Last-Event-ID replays exactly the delta."""
+    port = http_port(gw)
+    for text in ("a", "b", "c"):
+        await _park_frame(gw, text)
+
+    conn = _sse_open(port, last_event_id="1")
+    try:
+        # Frames 2 and 3 replayed (8 lines) + hello (3) + status (3) = 14.
+        lines = _sse_read_lines(conn, 14, timeout=5)
+        events, _ = _parse_sse(lines)
+        replayed = [e for e in events if e[0] == "frame" and e[1] is not None]
+        assert [e[1] for e in replayed] == ["2", "3"]
+        assert json.loads(replayed[0][2])["payload"]["text"] == "b"
+        assert json.loads(replayed[1][2])["payload"]["text"] == "c"
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_sse_heartbeat(gw, monkeypatch, plugin):
+    """A comment heartbeat is written when the stream is idle."""
+    monkeypatch.setattr(plugin.http_server, "SSE_HEARTBEAT_S", 1.0)
+    port = http_port(gw)
+    conn = _sse_open(port)
+    try:
+        # Consume the open sequence (6 lines), then wait for the heartbeat.
+        _sse_read_lines(conn, 6, timeout=5)
+        lines = _sse_read_lines(conn, 2, timeout=5)
+        events, comments = _parse_sse(lines)
+        assert comments >= 1, f"no heartbeat comment in {lines!r}"
+        assert events == []
+    finally:
+        conn.close()
+
+
+# ── Long-poll ───────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_poll_returns_on_new_frame(gw):
+    port = http_port(gw)
+    await _park_frame(gw, "pre")  # cursor 1: returned immediately (catch-up)
+
+    # A second poll at the high-water mark blocks until a new frame lands.
+    def poll_and_broadcast():
+        status, payload = _request(port, "GET", "/v1/poll?cursor=1", timeout=30)
+        return status, json.loads(payload)
+
+    async def late_frame():
+        await asyncio.sleep(0.5)
+        await _park_frame(gw, "late")
+
+    poll_task = asyncio.create_task(asyncio.to_thread(poll_and_broadcast))
+    late_task = asyncio.create_task(late_frame())
+    status, body = await asyncio.wait_for(poll_task, timeout=15)
+    await late_task
+    assert status == 200
+    assert body["cursor"] >= 2
+    assert len(body["frames"]) == 1
+    assert json.loads(body["frames"][0])["payload"]["text"] == "late"
+
+
+@pytest.mark.asyncio
+async def test_poll_timeout_empty(gw, monkeypatch, plugin):
+    monkeypatch.setattr(plugin.http_server, "POLL_TIMEOUT_S", 1.0)
+    port = http_port(gw)
+    await _park_frame(gw, "x")
+    hwm = gw._outbox.latest_cursor()
+    status, payload = await asyncio.to_thread(
+        _request, port, "GET", f"/v1/poll?cursor={hwm}", timeout=15
+    )
+    body = json.loads(payload)
+    assert status == 200
+    assert body["frames"] == []
+    assert body["cursor"] == hwm
+
+
+# ── Delivery counting (docs/19 §19.8 — the critical regression) ────────────
+
+
+@pytest.mark.asyncio
+async def test_sse_subscriber_counts_as_delivered_no_push(gw):
+    """A frame with only an SSE subscriber is delivered >= 1 -> NO push."""
+    port = http_port(gw)
+    conn = _sse_open(port)
+    try:
+        _sse_read_lines(conn, 6, timeout=5)  # open sequence
+        push = AsyncMock()
+        gw._maybe_push = push
+        await _park_frame(gw, "no push for me")
+        await asyncio.sleep(0.2)
+        push.assert_not_called()
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+async def test_no_subscribers_still_pushes(gw):
+    """Control: with no live devices at all, the push path still fires."""
+    push = AsyncMock()
+    gw._maybe_push = push
+    await _park_frame(gw, "wake me up")
+    await asyncio.sleep(0.2)
+    push.assert_called_once()
+
+
+# ── Media over HTTP (docs/19 §19.15, v2) ──────────────────────────────────
+
+
+def _upload(
+    port: int,
+    data: bytes,
+    *,
+    media_ref: str = "mu_http1",
+    kind: str = "image",
+    mime: str = "image/png",
+    filename: str = "t.png",
+    sha256: str | None = None,
+    **kw,
+) -> tuple[int, dict]:
+    import hashlib
+
+    headers = {
+        "X-Iris-Media-Ref": media_ref,
+        "X-Iris-Media-Kind": kind,
+        "X-Iris-Media-Filename": filename,
+        "X-Iris-Media-Sha256": sha256 if sha256 is not None else hashlib.sha256(data).hexdigest(),
+    }
+    status, payload = _request(
+        port,
+        "POST",
+        "/v1/media",
+        body=data,
+        content_type=mime,
+        extra_headers=headers,
+        **kw,
+    )
+    return status, json.loads(payload)
+
+
+@pytest.mark.asyncio
+async def test_media_upload_ok(gw):
+    port = http_port(gw)
+    status, body = await asyncio.to_thread(_upload, port, PNG_1X1)
+    assert status == 201, body
+    assert body["type"] == "media.upload.ack"
+    assert body["payload"]["ok"] is True
+    assert body["payload"]["media_ref"] == "mu_http1"
+    entry = gw._media.get_inbound("mu_http1")
+    assert entry is not None
+    assert entry.kind == "image"
+    assert entry.size == len(PNG_1X1)
+
+
+@pytest.mark.asyncio
+async def test_media_upload_sha_mismatch(gw):
+    port = http_port(gw)
+    status, body = await asyncio.to_thread(
+        _upload, port, PNG_1X1, media_ref="mu_badsha", sha256="0" * 64
+    )
+    assert status == 500, body  # internal: digest mismatch
+    assert body["type"] == "error"
+    assert body["payload"]["code"] == "internal"
+    assert gw._media.get_inbound("mu_badsha") is None
+
+
+@pytest.mark.asyncio
+async def test_media_upload_oversize_413(gw):
+    port = http_port(gw)
+    oversize = b"x" * (gw.max_upload_bytes + 1)
+    status, body = await asyncio.to_thread(_upload, port, oversize, media_ref="mu_big")
+    assert status == 413, body
+    assert body["payload"]["code"] == "media_too_large"
+
+
+@pytest.mark.asyncio
+async def test_media_upload_missing_ref_400(gw):
+    port = http_port(gw)
+    status, payload = await asyncio.to_thread(
+        _request,
+        port,
+        "POST",
+        "/v1/media",
+        body=PNG_1X1,
+        content_type="image/png",
+        extra_headers={"X-Iris-Media-Kind": "image"},
+    )
+    body = json.loads(payload)
+    assert status == 400
+    assert body["payload"]["code"] == "unsupported"
+
+
+@pytest.mark.asyncio
+async def test_media_upload_bad_kind_400(gw):
+    port = http_port(gw)
+    status, body = await asyncio.to_thread(_upload, port, PNG_1X1, kind="hologram")
+    assert status == 400
+    assert body["payload"]["code"] == "unsupported"
+
+
+@pytest.mark.asyncio
+async def test_media_upload_auth_401(gw):
+    port = http_port(gw)
+    status, _ = await asyncio.to_thread(_upload, port, PNG_1X1, token="wrong-token")
+    assert status == 401
+
+
+@pytest.mark.asyncio
+async def test_media_upload_liar_reclassified(gw):
+    """Lies about being a PNG: magic-byte re-sniff keeps it out of the image
+    cache (lands as a document) — same contract as the WS path."""
+    port = http_port(gw)
+    payload = b"<html>not an image</html>"
+    status, body = await asyncio.to_thread(
+        _upload, port, payload, media_ref="mu_liar", filename="liar.html"
+    )
+    assert status == 201, body
+    entry = gw._media.get_inbound("mu_liar")
+    assert entry is not None
+    assert entry.kind == "document"
+
+
+@pytest.mark.asyncio
+async def test_media_pull_ok(gw):
+    from gateway.platforms.base import get_image_cache_dir
+
+    img = get_image_cache_dir() / "http_pull_test.png"
+    img.write_bytes(PNG_1X1)
+    entry = gw._media.register_outbound(
+        str(img), "image", "image/png", "http_pull_test.png", len(PNG_1X1)
+    )
+    port = http_port(gw)
+    status, payload = await asyncio.to_thread(_request, port, "GET", f"/v1/media/{entry.media_id}")
+    assert status == 200
+    assert payload == PNG_1X1
+    conn = HTTPConnection("127.0.0.1", port, timeout=10)
+    conn.request(
+        "GET",
+        f"/v1/media/{entry.media_id}",
+        headers={"Authorization": f"Bearer {TOKEN}", "X-Iris-Device": DEVICE_ID},
+    )
+    resp = conn.getresponse()
+    resp.read()
+    assert resp.getheader("Content-Type") == "image/png"
+    assert resp.getheader("Content-Length") == str(len(PNG_1X1))
+    conn.close()
+
+
+@pytest.mark.asyncio
+async def test_media_pull_unknown_404(gw):
+    port = http_port(gw)
+    status, payload = await asyncio.to_thread(_request, port, "GET", "/v1/media/md_nope")
+    body = json.loads(payload)
+    assert status == 404
+    assert body["payload"]["code"] == "not_found"
+
+
+@pytest.mark.asyncio
+async def test_media_pull_denied_path_404(gw):
+    """Known id, but the path fails delivery validation (denylist) — same
+    re-check at pull time as the WS path."""
+    entry = gw._media.register_outbound("/etc/passwd", "document", "text/plain", "passwd", 100)
+    port = http_port(gw)
+    status, payload = await asyncio.to_thread(_request, port, "GET", f"/v1/media/{entry.media_id}")
+    body = json.loads(payload)
+    assert status == 404
+    assert body["payload"]["code"] == "not_found"
+
+
+# ── Helpers ─────────────────────────────────────────────────────────────────
+
+
+async def _park_frame(adapter, text: str) -> int:
+    """Emit one message frame through ``_broadcast_or_log`` (the real
+    outbound path); returns the outbox cursor."""
+    plugin = _load_plugin()
+    frame = plugin.protocol.message(
+        chat_id=CHAT_ID,
+        message_id=f"m_{abs(hash(text)) % 10**8:08x}",
+        role="assistant",
+        text=text,
+    )
+    await adapter._broadcast_or_log(CHAT_ID, frame)
+    return adapter._outbox.latest_cursor()