feat(bot-mode): replay pages carry authority lineage; pin byte-bounded replay

Follow-ups on top of the salvaged #97712 foundation:

- read_events() pages now include the room's authority stamp
  (authority.gateway_id + authority.epoch), so a replicating participant
  can persist lineage with every page and a future takeover layer can
  fence stale authorities from replayed state alone.
- New regression test proves the replay page bound counts UTF-8 BYTES,
  not characters: sabotaging LENGTH(CAST(.. AS BLOB)) back to
  LENGTH(TEXT) and dropping the .encode('utf-8') guard fails the test;
  the pre-existing multibyte test passed under that sabotage.
- _raise_room_not_found typed NoReturn so narrowing survives closures.

gateway/hosted_rooms.py

@@ -19,7 +19,7 @@ import sqlite3
 import time
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Iterator
+from typing import Any, Iterator, NoReturn
 
 
 PROTOCOL_VERSION = 2
@@ -461,7 +461,7 @@ def _transaction(
         conn.close()
 
 
-def _raise_room_not_found(conn: sqlite3.Connection, room_id: str) -> None:
+def _raise_room_not_found(conn: sqlite3.Connection, room_id: str) -> NoReturn:
     retained = conn.execute(
         "SELECT 1 FROM hosted_rooms WHERE room_id=?",
         (room_id,),
@@ -1390,13 +1390,16 @@ def read_events(
 
     with _transaction(db_path) as conn:
         room = conn.execute(
-            """SELECT next_seq FROM hosted_rooms
+            """SELECT next_seq, authority_gateway_id, authority_epoch
+               FROM hosted_rooms
                WHERE room_id=? AND (disbanded_at IS NULL OR ?)""",
             (room_id, int(include_disbanded)),
         ).fetchone()
         if room is None:
             _raise_room_not_found(conn, room_id)
         latest_seq = int(room["next_seq"]) - 1
+        authority_gateway = str(room["authority_gateway_id"])
+        authority_epoch = int(room["authority_epoch"])
         if since_seq > latest_seq:
             raise HostedRoomError("since_seq is ahead of the hosted room log")
         rows = conn.execute(
@@ -1429,6 +1432,10 @@ def read_events(
             "cursor": cursor,
             "latest_seq": latest_seq,
             "has_more": cursor < latest_seq,
+            "authority": {
+                "gateway_id": authority_gateway,
+                "epoch": authority_epoch,
+            },
         }
 
     def page_bytes(page: dict[str, Any]) -> int:

tests/gateway/test_hosted_rooms.py

@@ -497,6 +497,7 @@ def test_since_seq_returns_ordered_deltas_and_stable_cursor(tmp_path):
         "cursor": 2,
         "latest_seq": 4,
         "has_more": True,
+        "authority": {"gateway_id": "gateway-a", "epoch": 1},
     }
 
     second = rooms.read_events(
@@ -1016,3 +1017,45 @@ def test_gateway_event_budget_leaves_pre_update_snapshot_headroom():
         rooms.MAX_GATEWAY_EVENT_BYTES * 32
         <= update_cmd._PRE_UPDATE_SNAPSHOT_MAX_FILE_SIZE
     )
+
+
+def test_room_log_page_bound_counts_bytes_not_characters(tmp_path, monkeypatch):
+    """Char-based accounting (SQLite LENGTH(TEXT) / len(str)) must not let a
+    multibyte page exceed the byte ceiling.
+
+    Budget sits between the character length and the byte length of a
+    two-event page: character accounting would admit both events, byte
+    accounting admits exactly one.
+    """
+    db = tmp_path / "state.db"
+    _create(db)
+    for index in range(2):
+        _append(
+            db,
+            room_id="room-1",
+            event_id=f"bytes-{index}",
+            kind="message.user",
+            actor=USER,
+            payload={"text": "😀" * 200, "index": index},
+        )
+
+    def page_json(page):
+        return json.dumps(page, ensure_ascii=False, separators=(",", ":"))
+
+    def page_bytes(page):
+        return len(page_json(page).encode("utf-8"))
+
+    one_event = rooms.read_events(db, room_id="room-1", since_seq=0, limit=1)
+    unbounded_two = rooms.read_events(db, room_id="room-1", since_seq=0, limit=2)
+    budget = page_bytes(one_event) + 64
+
+    # Precondition: the budget discriminates — char accounting would accept
+    # the two-event page, byte accounting must reject it.
+    assert len(page_json(unbounded_two)) <= budget
+    assert page_bytes(unbounded_two) > budget
+
+    monkeypatch.setattr(rooms, "MAX_LOG_PAGE_BYTES", budget)
+    page = rooms.read_events(db, room_id="room-1", since_seq=0, limit=2)
+    assert [event["seq"] for event in page["events"]] == [1]
+    assert page_bytes(page) <= budget
+    assert page["has_more"] is True