HTTP fallback leg (docs/19): e2e scenario 13 + docs

- e2e.py: scenario 13 (http fallback) — drives a full turn over the
  HTTP leg (health + POST /v1/frame + SSE /v1/events, no WS) and
  asserts the user echo lands on the SSE stream in < 1.5 s.
- ws_probe.py --http: prints '== user echo in X.XXs' (the docs/19
  sendable-in-fallback timing assertion) alongside the existing
  health/POST/SSE output; same assertion flags as the WS leg.
- docs: 19 status flipped to implemented; 09-pairing-security §9.4
  cross-reference (second door, same lock: token + device allowlist,
  64 KiB cap, rate limit, optional TLS, unauthenticated /v1/health);
  13-testing manual scenario 15 + automated pointers.

gateway-plugin/tests/e2e.py

@@ -33,6 +33,7 @@ import sys
 import uuid
 import zlib
 from pathlib import Path
+from urllib.parse import urlparse
 
 HERE = Path(__file__).resolve().parent
 REPO = HERE.parent.parent
@@ -303,6 +304,33 @@ def s12_sync(env, url, token):
     return FAIL, f"probe rc={rc}"
 
 
+def s13_http_fallback(env, url, token):
+    """docs/19: the HTTP fallback leg. The probe drives a full turn over
+    health + POST /v1/frame + SSE /v1/events (no WS involved). The user echo
+    must land on the SSE stream promptly after the POST (< 1.5 s on LAN)."""
+    u = urlparse(url)
+    scheme = "https" if u.scheme == "wss" else "http"
+    http_port = os.getenv("ANDROID_HTTP_PORT", "8791")
+    http_url = f"{scheme}://{u.hostname or '127.0.0.1'}:{http_port}"
+    rc, out, _ = run_probe(env, url, token, "--http", "--http-url", http_url,
+                           "--send", "Reply with exactly: e2e http fallback OK",
+                           "--timeout", "120")
+    if rc == 0:
+        m = re.search(r"== user echo in ([\d.]+)s", out)
+        echo = float(m.group(1)) if m else None
+        if echo is not None and echo > 1.5:
+            return FAIL, f"user echo took {echo:.2f}s (> 1.5 s)"
+        return PASS, ("health + POST /v1/frame + SSE turn complete"
+                      + (f"; user echo in {echo:.2f}s" if echo is not None else ""))
+    if rc == 20:
+        return FAIL, "health check failed (HTTP leg not running?)"
+    if rc == 21:
+        return FAIL, "SSE open failed"
+    if rc == 22:
+        return FAIL, "POST /v1/frame rejected"
+    return FAIL, f"probe rc={rc}"
+
+
 SCENARIOS = [
     (1, "pair", s1_pair),
     (2, "text round-trip", s2_text),
@@ -316,6 +344,7 @@ SCENARIOS = [
     (10, "media out", s10_media_out),
     (11, "push", s11_push),
     (12, "reconnect/sync", s12_sync),
+    (13, "http fallback", s13_http_fallback),
 ]
 
 

gateway-plugin/tests/ws_probe.py

@@ -740,6 +740,7 @@ def run_http(args, base: str) -> int:
     print("== SSE open (/v1/events)")
 
     # 3. POST the message.send frame (accept-and-ack).
+    post_time: float | None = None
     if args.send:
         frame = {
             "v": 1, "id": 1, "type": "message.send",
@@ -753,6 +754,7 @@ def run_http(args, base: str) -> int:
         r = conn.getresponse()
         body = r.read()
         conn.close()
+        post_time = time.time()
         print(f"== POST /v1/frame -> {r.status} {body[:200]!r}")
         if r.status >= 400:
             print("!! POST /v1/frame rejected")
@@ -763,11 +765,12 @@ def run_http(args, base: str) -> int:
     st = _TurnState()
     got_final = False
     seen_final_frame = False
+    echo_logged = False
     deadline = time.time() + args.timeout
     cur_data: list[str] = []
 
     def feed(line: str) -> bool:
-        nonlocal cur_data, got_final, seen_final_frame
+        nonlocal cur_data, got_final, seen_final_frame, echo_logged
         line = line.rstrip("\r\n")
         if line == "":
             if cur_data:
@@ -776,6 +779,14 @@ def run_http(args, base: str) -> int:
                     ftype = data.get("type")
                     payload = data.get("payload") or {}
                     st.track(ftype, payload)
+                    # docs/19: the user echo must land on the SSE stream
+                    # promptly after the POST (the < 1 s sendable-in-fallback
+                    # UX assertion).
+                    if (not echo_logged and post_time is not None
+                            and ftype == "message"
+                            and payload.get("role") == "user"):
+                        echo_logged = True
+                        print(f"== user echo in {time.time() - post_time:.2f}s")
                     if ftype == "message" and payload.get("role") == "assistant":
                         got_final = True
                     if ftype == "message.stop":