fix(gateway): the gateway route lists and runs a stalled room's Retry too

CanonicalHostedRoomService's status dropped every service retry row and its
retry_room_task only knew deferred attempts, so on the gateway route the
stalled room showed no Retry and groups.retry answered stale_generation. Keep
the prepare:<room_id> row and hand that task id to the shared service, so both
routes answer the same. The stall test now runs on both routes.

gateway/session_hosted_controls.py

@@ -2,6 +2,7 @@
 from gateway import hosted_room_driver as tasks
 from hermes_state_runtime import RuntimeStoreError
 from tui_gateway.hosted_room_driver import HostedRoomBinding
+from tui_gateway.hosted_room_prepare_stall import RETRY_TASK_PREFIX
 from tui_gateway.hosted_room_service import attempt_action, task_member_id
 
 
@@ -47,6 +48,10 @@ class HostedControls:
             return result
 
     def retry_room_task(self, room_id, *, member_id, task_id, execution_generation):
+        if isinstance(task_id, str) and task_id.startswith(RETRY_TASK_PREFIX):
+            # A stalled room preparation, not an attempt: the shared service retries it.
+            return super().retry_room_task(
+                room_id, task_id=task_id, member_id=member_id, execution_generation=execution_generation)
         with self._policy_lock:
             task, binding = self._control_task(room_id, member_id, task_id, execution_generation)
             # Unknown is not non-admission. Never advance its hosted generation
@@ -69,7 +74,8 @@ class HostedControls:
         result = super().status(room_id)
         if room_id is None:
             return result
-        actions = [a for a in result['pending_actions'] if a['kind'] != 'retry']
+        actions = [a for a in result['pending_actions']
+                   if a['kind'] != 'retry' or str(a.get('task_id', '')).startswith(RETRY_TASK_PREFIX)]
         for task in tasks.list_tasks(self.db_path, room_id=room_id):
             if task['status'] not in {'indeterminate', 'deferred'}:
                 continue

tests/gateway/test_hosted_room_handoff_send.py

@@ -9,12 +9,15 @@ from __future__ import annotations
 
 import sqlite3
 import threading
+
+import pytest
 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 tests.waiting import wait_until
+from gateway.session_hosted_controls import HostedControls
 from tui_gateway.hosted_room_service import HostedRoomService
 
 MEMBERS = [{"member_id": "default", "profile": "default", "handle": "hermes"},
@@ -47,8 +50,12 @@ class _RPC:
         return {"active": False, "task_id": None}
 
 
-def _service(db_path: Path) -> tuple[HostedRoomService, _RPC]:
-    service = HostedRoomService(
+class _GatewayRouteService(HostedControls, HostedRoomService):
+    """The gateway route's control overrides (CanonicalHostedRoomService) over the shared service."""
+
+
+def _service(db_path: Path, cls: type[HostedRoomService] = HostedRoomService) -> tuple[HostedRoomService, _RPC]:
+    service = cls(
         SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock()), db_path=db_path)
     service.rpc = service.runtime.rpc = rpc = _RPC()
     service.local_profiles = lambda: ("default", "ops")
@@ -114,9 +121,10 @@ def test_a_send_already_stored_is_answered_as_sent_when_preparing_the_room_fails
     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):
+@pytest.mark.parametrize("route", [HostedRoomService, _GatewayRouteService], ids=["tui_gateway", "gateway"])
+def test_a_room_whose_members_cannot_be_woken_says_so_once_and_stops_retrying(tmp_path, route):
     db = tmp_path / "state.db"
-    service, _ = _service(db)
+    service, _ = _service(db, route)
     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=[