fix(gateway): Group Chats handed from the standalone backend take Sends again

A room the gateway took over from the standalone backend
(hosted_rooms_gateway_import) refused every groups.send with
invalid_params after storing the message, and no member was woken.

The handoff commits two authority transfers: the standalone copy's fence
(authority.lost, install -> install:gateway-store, epoch 2) and the
gateway store's claim (authority.claimed, gateway-store -> install,
epoch 3). The history reader and the work-record lineage took span
boundaries from authority.claimed rows only, so the chain started at the
gateway store and the epoch-3 claim did not extend it: "authority claim
does not extend its predecessor" from every prepare_room, on Send and on
each driver pass.

authority.lost is the outgoing owner's committed record of the same
transfer (same system actor, same previous/next/epoch payload), so both
readers now count it as a span boundary under the same checks. The chain
reads install@1 -> gateway-store@2 -> install@3. Nothing is rewritten:
rooms users already imported (state.db epoch 3, shared-state.db epoch 2)
work as they are. A fork (a claim whose previous owner is not the
recorded holder, a skipped epoch) is still refused.

Send also stops reporting a stored message as failed: once the user
event commits it is the Send's result, and a failure preparing the room
afterwards is logged and left to the room driver, which runs the same
preparation on its next pass (#143239).

gateway/hosted_room_authority_history.py

@@ -14,6 +14,13 @@ from typing import Any
 from gateway.hosted_rooms_common import identifier
 
 MAX_AUTHORITY_SPANS = 1024
+#: Committed authority transfers, each the first event of its span. ``authority.claimed`` is the new
+#: owner's record; ``authority.lost`` is the record the outgoing owner commits when it is fenced to
+#: a proven newer epoch (``hosted_room_replicas.demote_room``; the standalone -> gateway handoff in
+#: ``hosted_rooms_gateway_import``). Both carry the same previous/next/epoch payload and are
+#: checked alike, so a store whose log holds its own fence followed by a later claim (a handed-off
+#: room, a returning gateway that reclaims) reads as one unbroken chain.
+TRANSFER_KINDS = ("authority.claimed", "authority.lost")
 
 
 class AuthorityHistoryError(ValueError):
@@ -77,14 +84,14 @@ def origin_gateway_id(room: dict[str, Any]) -> str:
 
 
 def read_history_locked(conn, room_id: str, *, gateway_id: str, epoch: int) -> list[dict[str, Any]] | None:
-    """Read committed claims within the caller's room-state transaction.
+    """Read committed authority transfers within the caller's room-state transaction.
 
     Legacy adoption is left on its existing path; it is not retroactively
     certified as a history of physical-host transfers.
     """
     rows = conn.execute(
-        """SELECT seq,authority_epoch,actor_json,payload_json FROM hosted_room_events
-           WHERE room_id=? AND kind='authority.claimed' ORDER BY seq LIMIT ?""",
+        f"""SELECT seq,authority_epoch,actor_json,payload_json FROM hosted_room_events
+           WHERE room_id=? AND kind IN {TRANSFER_KINDS} ORDER BY seq LIMIT ?""",
         (room_id, MAX_AUTHORITY_SPANS),
     ).fetchall()
     if len(rows) >= MAX_AUTHORITY_SPANS:

gateway/hosted_room_work_lineage.py

@@ -1,5 +1,5 @@
 """Validate passive evidence against committed authority spans, never elect an owner."""
-from gateway.hosted_room_authority_history import at_sequence, read_history_locked, validate_history
+from gateway.hosted_room_authority_history import TRANSFER_KINDS, at_sequence, read_history_locked, validate_history
 
 
 def source_prefix_locked(conn, room_id, authority, seq):
@@ -17,7 +17,7 @@ def source_prefix_locked(conn, room_id, authority, seq):
         boundary = span.epoch > 1 and count == span.from_seq
         if (event["seq"] != count or event["authority_epoch"] != span.epoch
                 or (actor.get("kind") == "gateway" and actor.get("id") != span.gateway_id)
-                or (event["kind"] == "authority.claimed") != boundary):
+                or (event["kind"] in TRANSFER_KINDS) != boundary):
             raise WorkRecordPrefixError("work record history prefix is not verified")
     if count != seq:
         raise WorkRecordPrefixError("work record history prefix is unavailable")

tests/gateway/test_hosted_room_handoff_send.py

@@ -0,0 +1,111 @@
+"""A Group Chat the gateway took over keeps answering, and a stored Send is never reported as failed.
+
+#145960: rooms handed from the standalone backend to the gateway (``hosted_rooms_gateway_import``)
+refused every Send with ``invalid_params`` after storing the message, and no member was woken.
+#143239: any failure after the user event committed came back to the client as a failed Send.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+from gateway import hosted_room_driver, hosted_rooms
+from gateway.hosted_rooms_gateway_import import handoff_gateway_id, import_standalone_rooms
+from tui_gateway.hosted_room_service import HostedRoomService
+
+MEMBERS = [{"member_id": "default", "profile": "default", "handle": "hermes"},
+           {"member_id": "ops", "profile": "ops", "handle": "ops"}]
+
+
+class _RPC:
+    def __init__(self) -> None:
+        self.sessions, self.submits = {}, []
+
+    def resolve_exact(self, *, profile, title, source):
+        return self.sessions.get((profile, title))
+
+    def create(self, *, profile, title, source):
+        self.sessions[(profile, title)] = session = {"session_id": f"{profile}-session", "title": title}
+        return session
+
+    def resume(self, *, profile, session_id, source):
+        return {"session_id": session_id}
+
+    def submit(self, **kwargs):
+        self.submits.append(kwargs["profile"])
+        kwargs["on_terminal"]({"status": "settled", "text": f"reply from {kwargs['profile']}"})
+        return {"accepted": True}
+
+    def history(self, **kwargs):
+        return []
+
+    def info(self, **kwargs):
+        return {"active": False, "task_id": None}
+
+
+def _service(db_path: Path) -> tuple[HostedRoomService, _RPC]:
+    service = HostedRoomService(
+        SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock()), db_path=db_path)
+    service.rpc = service.runtime.rpc = rpc = _RPC()
+    service.local_profiles = lambda: ("default", "ops")
+    return service, rpc
+
+
+def _kinds(db: Path, room_id: str) -> list[str]:
+    return [e["kind"] for e in hosted_rooms.read_events(db, room_id=room_id, since_seq=0)["events"]]
+
+
+def _authority(db: Path, room_id: str) -> tuple[str, int]:
+    with sqlite3.connect(db) as conn:
+        return conn.execute("SELECT authority_gateway_id, authority_epoch FROM hosted_rooms WHERE room_id=?",
+                            (room_id,)).fetchone()
+
+
+def test_a_room_handed_to_the_gateway_takes_sends_and_wakes_its_members(tmp_path):
+    source, target = tmp_path / "shared-state.db", tmp_path / "state.db"
+    with sqlite3.connect(target) as conn:  # the gateway's SessionDB table the owner claim lives in
+        conn.execute("CREATE TABLE state_meta (key TEXT PRIMARY KEY, value TEXT)")
+    local = hosted_rooms.local_authority_gateway_id()
+    hosted_rooms.create_room(source, room_id="room-desk", name="Desk", members=MEMBERS,
+                             authority_gateway_id=local)
+    assert import_standalone_rooms(target) == 1
+    # The stores every room the shipped import handed over is in (#145960's report): nothing to migrate.
+    assert _authority(source, "room-desk") == (handoff_gateway_id(), 2)
+    assert _authority(target, "room-desk") == (local, 3)
+
+    service, rpc = _service(target)
+    sent = service.send(room_id="room-desk", event_id="user-1",
+                        payload={"text": "@ops inspect", "thread_id": "thread-1"})
+    assert sent["kind"] == "message.user" and sent["authority_epoch"] == 3
+    assert len(hosted_room_driver.list_tasks(target, room_id="room-desk", status="queued")) == 1
+
+    service.start()
+    try:
+        deadline = time.monotonic() + 5
+        while "message.member" not in _kinds(target, "room-desk") and time.monotonic() < deadline:
+            time.sleep(0.01)
+    finally:
+        assert service.stop(timeout=5.0)
+    assert rpc.submits == ["ops"]
+    assert _kinds(target, "room-desk").count("message.member") == 1
+
+
+def test_a_send_already_stored_is_answered_as_sent_when_preparing_the_room_fails(tmp_path):
+    db = tmp_path / "state.db"
+    service, _ = _service(db)
+    service.create_room(room_id="room-1", name="Room", members=MEMBERS)
+
+    def storage_full(binding):
+        raise hosted_rooms.HostedRoomError("Group Chat storage is full on this host.")
+
+    service.prepare_room = storage_full
+    sent = service.send(room_id="room-1", event_id="user-1", payload={"text": "hello", "thread_id": "t"})
+    service.bindings = lambda: ()  # the binding list has not caught up with the room yet
+    again = service.send(room_id="room-1", event_id="user-2", payload={"text": "again", "thread_id": "t"})
+
+    assert [sent["kind"], again["kind"]] == ["message.user", "message.user"]
+    assert _kinds(db, "room-1").count("message.user") == 2

tui_gateway/hosted_room_service.py

@@ -536,10 +536,16 @@ class HostedRoomService:
         event = append_user_event(
             self, room_id=room_id, event_id=event_id, payload=normalized,
             gateway_id=gateway_id, epoch=epoch)
-        binding = next((b for b in self.bindings() if b.room_id == room_id), None)
-        if binding is None:
-            raise hosted_rooms.RoomNotFoundError("hosted room not found")
-        self.prepare_room(binding)
+        # The committed event is the Send's result: a failure from here on is the room driver's,
+        # which runs the same preparation on its next pass and records the error there. Raising
+        # would tell the client a message it can already see in the room was not sent (#143239).
+        try:
+            binding = next((b for b in self.bindings() if b.room_id == room_id), None)
+            if binding is not None:
+                self.prepare_room(binding)
+        except Exception:
+            logger.warning("hosted room %s: preparing the room after a stored Send failed; "
+                           "the room driver retries it", room_id, exc_info=True)
         self.runtime.wakeup()
         return event