fix(gateway): a Group Chat whose members can't be woken says so once, with Retry

Review of #145981 (P2): after a Send is stored, a room preparation that keeps
failing (a member whose profile was deleted, a full room log) retried on every
driver pass forever, and the error only overwrote the process-wide driver
last_error that no client renders. The user saw "sent" and no member answered.

Now each room counts consecutive preparation failures (contention and
ownership moves never count). The third one appends ONE member.unavailable
event to the room ("Members could not be woken: <reason>"), which Desktop and
the Webapp already render, using the control headroom so a full room can still
say why. The room's status lists a Retry row (groups.retry, task_id
prepare:<room_id>) and a per-room last_error, and the driver stops preparing
that room until the next Send or Retry. A failed Retry stays stalled and is the
RPC's error, without a second notice.

Refs #145977

apps/shared/src/gateway-contract.generated.ts

@@ -1631,7 +1631,7 @@ export interface GroupsStateResult {
   room: Room
   driver_status?: RoomDriverStatus | null
 }
-/** ``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) or the driver's approval action (``kind: approval`` + run/session/approval context). ``member_activity`` lists live tasks in admission order (absent from older gateways). */
+/** ``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) or the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled room preparation's Retry (``task_id`` ``prepare:<room_id>``, ``member_id`` the members not woken). ``member_activity`` lists live tasks in admission order (absent from older gateways). ``last_error`` is this room's last preparation failure, null once one succeeds. */
 export interface RoomDriverStatus {
   running: boolean
   working: boolean
@@ -1639,6 +1639,7 @@ export interface RoomDriverStatus {
   counts: Record<string, number>
   pending_actions: Record<string, unknown>[]
   member_activity?: RoomMemberActivity[]
+  last_error?: string | null
   peer_routes: PeerRouteStatus[]
 }
 /** One live room task: ``status`` is ``queued``, ``running`` or ``stopping``. */

apps/shared/src/gateway-contract.openrpc.json

@@ -34015,7 +34015,7 @@
       },
       "RoomDriverStatus": {
         "additionalProperties": false,
-        "description": "``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt\n(``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)\nor the driver's approval action (``kind: approval`` + run/session/approval context).\n``member_activity`` lists live tasks in admission order (absent from older gateways).",
+        "description": "``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt\n(``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)\nor the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled\nroom preparation's Retry (``task_id`` ``prepare:<room_id>``, ``member_id`` the members not woken).\n``member_activity`` lists live tasks in admission order (absent from older gateways).\n``last_error`` is this room's last preparation failure, null once one succeeds.",
         "properties": {
           "running": {
             "title": "Running",
@@ -34051,6 +34051,18 @@
             "title": "Member Activity",
             "type": "array"
           },
+          "last_error": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "title": "Last Error"
+          },
           "peer_routes": {
             "items": {
               "$ref": "#/components/schemas/PeerRouteStatus"

gateway/hosted_rooms.py

@@ -956,9 +956,10 @@ def rename_room(db_path: DbPath, *, room_id: Any, event_id: Any, name: Any, now:
 def append_event(
     db_path: DbPath, *, room_id: Any, event_id: Any, kind: Any, actor: Any, payload: Any,
     authority_gateway_id: Any = None, authority_epoch: Any = None, now: float | None = None,
-    expected_latest_seq: int | None = None) -> dict[str, Any]:
+    expected_latest_seq: int | None = None, control_reserve: bool = False) -> dict[str, Any]:
     """Append one immutable event and allocate its per-room sequence atomically; repeating an ``event_id``
-    with identical content returns the original, different content fails closed."""
+    with identical content returns the original, different content fails closed. ``control_reserve`` lets a
+    bounded gateway notice use the control headroom, so a room whose log is full can still say why."""
     room_id = _room_id(room_id)
     event_id = _event_id(event_id)
     if expected_latest_seq is not None:
@@ -993,7 +994,7 @@ def append_event(
                                        manifest=json.loads(payload_json).get("attachments", []), now=now)
         event_bytes = _insert_event(
             conn, room, room_id, seq, event_id, kind, actor_json, authority_epoch, payload_json, now,
-            allow_control=kind in _CONTROL_EVENT_KINDS)
+            allow_control=kind in _CONTROL_EVENT_KINDS or (control_reserve and normalized_actor["kind"] == "gateway"))
         _fenced_update(conn, """UPDATE hosted_rooms SET next_seq=?, event_bytes=event_bytes+?, updated_at=?
                 WHERE room_id=? AND next_seq=?""", (seq + 1, event_bytes, now, room_id, seq),
             RuntimeError("hosted room sequence advance lost its write fence"))

tests/gateway/test_hosted_room_handoff_send.py

@@ -81,6 +81,11 @@ def test_a_room_handed_to_the_gateway_takes_sends_and_wakes_its_members(tmp_path
     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
+    # #145977: a Desktop that journaled the refused Send retries the exact same intent; it is
+    # acknowledged as the stored message (so the client retires it), never stored twice.
+    again = service.send(room_id="room-desk", event_id="user-1",
+                         payload={"text": "@ops inspect", "thread_id": "thread-1"})
+    assert again["seq"] == sent["seq"] and _kinds(target, "room-desk").count("message.user") == 1
     assert len(hosted_room_driver.list_tasks(target, room_id="room-desk", status="queued")) == 1
 
     service.start()
@@ -107,3 +112,56 @@ def test_a_send_already_stored_is_answered_as_sent_when_preparing_the_room_fails
 
     assert [sent["kind"], again["kind"]] == ["message.user", "message.user"]
     assert _kinds(db, "room-1").count("message.user") == 2
+
+
+def test_a_room_whose_members_cannot_be_woken_says_so_once_and_stops_retrying(tmp_path):
+    db = tmp_path / "state.db"
+    service, _ = _service(db)
+    service.create_room(room_id="room-1", name="Room", members=MEMBERS)
+    service.local_profiles = lambda: ("default", "dev", "ops")
+    service.create_room(room_id="room-2", name="Other", members=[
+        MEMBERS[0], {"member_id": "dev", "profile": "dev", "handle": "dev"}])
+    service.local_profiles = lambda: ("default", "dev")  # the ops member's profile was deleted
+    attempts: list[str] = []
+    real_prepare = service.prepare_room
+
+    def counted(binding):
+        attempts.append(binding.room_id)
+        return real_prepare(binding)
+
+    service.prepare_room = counted
+    sent = service.send(room_id="room-1", event_id="user-1", payload={"text": "@ops hi", "thread_id": "t"})
+    assert sent["kind"] == "message.user"
+
+    service.runtime.poll_interval_seconds = 0.02
+    service.start()
+    try:
+        wait_until(lambda: "member.unavailable" in _kinds(db, "room-1"), timeout=5, what="stall notice")
+        settled = attempts.count("room-1")
+        wait_until(lambda: service.runtime.status()["cycles"] > 30, timeout=5, what="more driver passes")
+    finally:
+        assert service.stop(timeout=5.0)
+    assert attempts.count("room-1") == settled == 3  # bounded: no retries after the stall
+    events = hosted_rooms.read_events(db, room_id="room-1", since_seq=0)["events"]
+    notices = [e for e in events if e["kind"] == "member.unavailable"]
+    assert len(notices) == 1 and "ops" in notices[0]["payload"]["error"]
+    status = service.status("room-1")
+    assert "ops" in status["last_error"] and service.status("room-2")["last_error"] is None
+    [retry] = [a for a in status["pending_actions"] if a["task_id"].startswith("prepare:")]
+
+    # Retry while the profile is still missing fails loudly and adds no second notice.
+    try:
+        service.retry_room_task("room-1", task_id=retry["task_id"], member_id=retry["member_id"],
+                                execution_generation=retry["execution_generation"])
+        raise AssertionError("retry of a still-broken room must fail")
+    except Exception as exc:
+        assert "ops" in str(exc)
+    assert _kinds(db, "room-1").count("member.unavailable") == 1
+
+    service.local_profiles = lambda: ("default", "dev", "ops")  # restored
+    receipt = service.retry_room_task("room-1", task_id=retry["task_id"], member_id=retry["member_id"],
+                                      execution_generation=1)
+    assert receipt["identity"].task_id == retry["task_id"]
+    assert service.status("room-1")["last_error"] is None
+    assert not [a for a in service.status("room-1")["pending_actions"] if a["task_id"].startswith("prepare:")]
+    assert len(hosted_room_driver.list_tasks(db, room_id="room-1", status="queued")) == 1

tui_gateway/contracts/groups_bot_relay.py

@@ -224,8 +224,10 @@ class RoomMemberActivity(Result):
 class RoomDriverStatus(Result):
     """``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt
     (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)
-    or the driver's approval action (``kind: approval`` + run/session/approval context).
-    ``member_activity`` lists live tasks in admission order (absent from older gateways)."""
+    or the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled
+    room preparation's Retry (``task_id`` ``prepare:<room_id>``, ``member_id`` the members not woken).
+    ``member_activity`` lists live tasks in admission order (absent from older gateways).
+    ``last_error`` is this room's last preparation failure, null once one succeeds."""
 
     running: bool
     working: bool
@@ -233,6 +235,7 @@ class RoomDriverStatus(Result):
     counts: dict[str, int]
     pending_actions: list[dict[str, JsonValue]]
     member_activity: list[RoomMemberActivity] = Field(default_factory=list)
+    last_error: str | None = None
     peer_routes: list[PeerRouteStatus]
 
 

tui_gateway/hosted_room_prepare_stall.py

@@ -0,0 +1,110 @@
+"""Bound the room driver's retries of a room preparation that keeps failing, and say so in the room.
+
+A Send returns the stored message even when preparing the room afterwards fails (#143239), and the
+driver retries the preparation on its next passes. A failure that never clears (a member whose
+profile was deleted, a full room log) would otherwise retry forever with the error only in the
+gateway log: the user sees "sent" and no member ever answers. After ``STALL_ATTEMPTS`` consecutive
+failures the room gets ONE ``member.unavailable`` event (Desktop and the Webapp render it), the
+room's status offers Retry, and the driver stops preparing that room until the next Send or Retry.
+"""
+
+from __future__ import annotations
+
+import logging
+import sqlite3
+import threading
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from gateway import hosted_room_driver as driver
+from gateway import hosted_rooms
+
+logger = logging.getLogger(__name__)
+
+STALL_ATTEMPTS = 3
+RETRY_TASK_PREFIX = "prepare:"
+_EVENT_PREFIX = "prepare-stalled:"
+_MAX_ERROR_CHARS = 300
+# Contention or an ownership move, not a broken room: never counted toward the stall.
+_TRANSIENT = (
+    sqlite3.OperationalError, hosted_rooms.AuthorityConflictError, hosted_rooms.EventCursorConflictError,
+    hosted_rooms.RoomNotFoundError, driver.LeaseHeldError, driver.StaleLeaseError, driver.RoomUnavailableError)
+
+
+@dataclass
+class _Streak:
+    failures: int = 0
+    error: str | None = None
+    stalled: bool = False
+
+
+def _describe(exc: BaseException) -> str:
+    text = str(exc).strip() or type(exc).__name__
+    return text if len(text) <= _MAX_ERROR_CHARS else text[:_MAX_ERROR_CHARS - 1] + "…"
+
+
+class PrepareStallTracker:
+    """Per-room consecutive preparation failures; thread-safe (driver room threads and RPC threads)."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._rooms: dict[str, _Streak] = {}
+
+    def stalled(self, room_id: str) -> bool:
+        with self._lock:
+            streak = self._rooms.get(room_id)
+            return bool(streak and streak.stalled)
+
+    def last_error(self, room_id: str) -> str | None:
+        with self._lock:
+            streak = self._rooms.get(room_id)
+            return streak.error if streak else None
+
+    def clear(self, room_id: str) -> None:
+        with self._lock:
+            self._rooms.pop(room_id, None)
+
+    def failed(self, room_id: str, exc: BaseException) -> bool:
+        """Record one failure; True exactly once, when this failure stalls the room."""
+        with self._lock:
+            streak = self._rooms.setdefault(room_id, _Streak())
+            streak.error = _describe(exc)
+            if isinstance(exc, _TRANSIENT) or streak.stalled:
+                return False
+            streak.failures += 1
+            streak.stalled = streak.failures >= STALL_ATTEMPTS
+            return streak.stalled
+
+    def stall(self, room_id: str, exc: BaseException) -> None:
+        """A failed explicit Retry: stalled again at once, without a second room event."""
+        with self._lock:
+            self._rooms[room_id] = _Streak(failures=STALL_ATTEMPTS, error=_describe(exc), stalled=True)
+
+
+
+def retry_action(room_id: str, room: Mapping[str, Any]) -> dict[str, Any]:
+    """A stalled room's Retry row in ``groups.state``; ``member_id`` names who was not woken."""
+    names = [str(m.get("handle") or m.get("member_id") or "") for m in room.get("members") or ()]
+    return {"kind": "retry", "member_id": ", ".join(n for n in names if n) or room_id,
+            "task_id": f"{RETRY_TASK_PREFIX}{room_id}", "execution_generation": 1}
+
+
+def report_stall(db_path: Any, room_id: str, exc: BaseException) -> None:
+    """Append the room's one stall notice; a stall already reported last in the log is not repeated."""
+    try:
+        room = hosted_rooms.room_state(db_path, room_id=room_id)
+        latest_seq = int(room["latest_seq"])
+        tail = hosted_rooms.read_events(
+            db_path, room_id=room_id, since_seq=max(0, latest_seq - 1), limit=1)["events"]
+        if tail and str(tail[-1]["event_id"]).startswith(_EVENT_PREFIX):
+            return
+        gateway_id, epoch = str(room["authority_gateway_id"]), int(room["authority_epoch"])
+        hosted_rooms.append_event(
+            db_path, room_id=room_id, event_id=f"{_EVENT_PREFIX}{latest_seq}", kind="member.unavailable",
+            actor={"kind": "gateway", "id": gateway_id},
+            payload={"error": f"Members could not be woken: {_describe(exc)}", "reason_code": "prepare_failed"},
+            authority_gateway_id=gateway_id, authority_epoch=epoch, control_reserve=True)
+    except Exception:
+        # The Retry row in the room's status still shows the stall.
+        logger.warning("hosted room %s: recording the stalled preparation failed", room_id, exc_info=True)

tui_gateway/hosted_room_service.py

@@ -21,6 +21,7 @@ from gateway import hosted_rooms
 from gateway.hosted_room_policy_checkpoint import HostedRoomPolicyCheckpoint, PolicySnapshot
 from gateway.hosted_room_peer import (
     GatewayRoomCatalog, HostedMemberDispatch, PROTOCOL_VERSION, room_grant_needs_dispatch_refresh)
+from tui_gateway import hosted_room_prepare_stall as prepare_stall
 from tui_gateway.hosted_room_driver import HostedRoomBinding, HostedRoomRuntime
 from tui_gateway.hosted_room_server_rpc import HostedRoomServerRPC
 from tui_gateway.hosted_room_peer_http import (
@@ -119,6 +120,7 @@ class HostedRoomService:
         self._peer_route_status: dict[tuple[str, str], str] = {}
         self.peer_routes: dict[tuple[str, str], PeerMemberRoute] = {}
         self.peer_clients: dict[tuple[str, str], Any] = {}
+        self._prepare_stalls = prepare_stall.PrepareStallTracker()
         try:
             self._load_stored_links()
         except Exception as exc:
@@ -132,7 +134,7 @@ class HostedRoomService:
         self.runtime = HostedRoomRuntime(
             db_path=self.db_path, rooms=self.bindings, rpc=self.rpc,
             transport_resolver=self._resolve_member_transport, turn_lock=self._turn_lock,
-            prepare_room=self.prepare_room, publish_terminal=self.publish_terminal,
+            prepare_room=self._prepare_tracked, publish_terminal=self.publish_terminal,
             pending_action=self._set_pending_action,
             poll_interval_seconds=_HOSTED_ROOM_IDLE_FALLBACK_SECONDS,
             active_poll_interval_seconds=_HOSTED_ROOM_ACTIVE_POLL_SECONDS,
@@ -511,8 +513,22 @@ class HostedRoomService:
             elif decision.status in {"settled", "bounded"}:
                 self._append_room_status(room, decision)
 
+    def _prepare_tracked(self, binding: HostedRoomBinding, *, retry_stalled: bool = False) -> None:
+        """``prepare_room`` with its failures counted per room: a stalled room is skipped (not retried)
+        until the next Send or Retry, and is reported once in the room (hosted_room_prepare_stall)."""
+        stalls = self._prepare_stalls
+        if stalls.stalled(binding.room_id) and not retry_stalled:
+            return
+        try:
+            self.prepare_room(binding)
+        except Exception as exc:
+            if stalls.failed(binding.room_id, exc):
+                prepare_stall.report_stall(self.db_path, binding.room_id, exc)
+            raise
+        stalls.clear(binding.room_id)
+
     def publish_terminal(self, binding: HostedRoomBinding, _task: Mapping[str, Any]) -> None:
-        self.prepare_room(binding)
+        self._prepare_tracked(binding)
         self.runtime.wakeup()
 
     def create_room(self, *, room_id: str, name: str, members: Any, admit: Any = None) -> dict[str, Any]:
@@ -537,12 +553,14 @@ class HostedRoomService:
             self, room_id=room_id, event_id=event_id, payload=normalized,
             gateway_id=gateway_id, epoch=epoch)
         # 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).
+        # which runs the same preparation on its next passes and, if it keeps failing, says so in
+        # the room once. Raising would tell the client a message it can already see in the room was
+        # not sent (#143239). A new message starts a fresh attempt streak for a stalled room.
+        self._prepare_stalls.clear(room_id)
         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)
+                self._prepare_tracked(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)
@@ -575,6 +593,8 @@ class HostedRoomService:
         self, room_id: str, *, task_id: str, member_id: str | None = None,
         execution_generation: int | None = None) -> dict[str, Any]:
         """Retry one uncertain or deferred task only after explicit user action."""
+        if task_id.startswith(prepare_stall.RETRY_TASK_PREFIX):
+            return self._retry_preparation(room_id, task_id)
         candidates = self._list_tasks(room_id, _RETRYABLE_STATUSES)
         task = next((c for c in candidates if c["identity"].task_id == task_id
                      and member_id in (None, task_member_id(c))
@@ -584,6 +604,21 @@ class HostedRoomService:
         return self.runtime.retry_indeterminate(
             task["identity"], execution_generation=task["execution_generation"])
 
+    def _retry_preparation(self, room_id: str, task_id: str) -> dict[str, Any]:
+        """The stalled room's Retry: prepare once now; a failure stays stalled and is the RPC's error."""
+        binding = next((b for b in self.bindings() if b.room_id == room_id), None)
+        if (binding is None or task_id != f"{prepare_stall.RETRY_TASK_PREFIX}{room_id}"
+                or not self._prepare_stalls.stalled(room_id)):
+            raise driver.InvalidTaskTransitionError("no retryable room task matches task_id")
+        try:
+            self._prepare_tracked(binding, retry_stalled=True)
+        except Exception as exc:
+            self._prepare_stalls.stall(room_id, exc)
+            raise
+        self.runtime.wakeup()
+        return {"identity": driver.TaskIdentity(room_id=room_id, task_id=task_id, thread_id=task_id, turn_id=task_id),
+                "status": "prepared", "execution_generation": 1, "cancel_generation": 0}
+
     def approve_room_task(
         self, room_id: str, *, member_id: str, task_id: str, execution_generation: int,
         choice: str, request_id: str | None = None) -> Mapping[str, Any]:
@@ -632,6 +667,8 @@ class HostedRoomService:
         counts = Counter(str(task["status"]) for task in tasks)
         pending_actions = [
             attempt_action("retry", task) for task in tasks if task["status"] in _RETRYABLE_STATUSES]
+        if self._prepare_stalls.stalled(room_id):
+            pending_actions.append(prepare_stall.retry_action(room_id, self._room(room_id)))
         with self._policy_lock:
             pending_actions.extend(
                 dict(action) for (action_room_id, _member_id), action
@@ -646,6 +683,7 @@ class HostedRoomService:
             or bool(counts.get("indeterminate") or counts.get("stopping")),
             "counts": dict(counts), "pending_actions": pending_actions,
             "member_activity": member_activity,
+            "last_error": self._prepare_stalls.last_error(room_id),
             "peer_routes": self._route_statuses(room_id)}