refactor(tui_gateway): table-driven groups passthrough handlers, pid/authority helpers, docstring compaction (group E)

tui_gateway/compute_host.py

@@ -1,8 +1,5 @@
-"""Persistent dashboard compute-host process.
-
-The long-lived child that owns live AIAgent objects when ``dashboard.turn_isolation``
-is enabled; frames are line-JSON over stdin/stdout.
-"""
+"""Persistent dashboard compute-host child: owns live AIAgent objects when
+``dashboard.turn_isolation`` is enabled; frames are line-JSON over stdin/stdout."""
 
 from __future__ import annotations
 
@@ -101,17 +98,15 @@ class ComputeHost:
     def shutdown(self, *, reason: str = "shutdown", wait: float = 10.0) -> None:
         """Drain in-flight turns, then finalize every session.
 
-        Order matters: ``_finalize_session`` is a one-shot latch, so finalizing before
-        the drain would spend the flush's single chance mid-turn, fire
-        ``on_session_end(interrupted=True)`` against a running session and release the
-        active-session lease under a live turn. ``_FLUSH_RESERVE_SECS`` (never more than
-        half of ``wait``) is withheld from the drain so the flush still runs when turns
-        outlast the window. Sessions whose turn is *still running* at the deadline are
-        excluded from the flush (``_executor.shutdown`` does not join them): finalizing
-        one mid-turn would leave it un-finalizable with its lease released, whereas
-        leaving it unfinalized keeps it recoverable. ``server._shutdown_sessions``
-        (atexit) may still re-finalize skipped sessions on the SIGTERM / stdin_closed
-        paths; the orphan path (``os._exit``) bypasses atexit.
+        Order matters: ``_finalize_session`` is a one-shot latch, so finalizing before the
+        drain would spend it mid-turn, fire ``on_session_end(interrupted=True)`` on a running
+        session and release its lease. ``_FLUSH_RESERVE_SECS`` (at most half of ``wait``) is
+        withheld from the drain so the flush still runs when turns outlast the window.
+        Sessions still running at the deadline are skipped (``_executor.shutdown`` does not
+        join them): finalizing mid-turn would leave them un-finalizable with the lease
+        released; unfinalized keeps them recoverable. ``server._shutdown_sessions`` (atexit)
+        may re-finalize skipped sessions on SIGTERM / stdin_closed; ``os._exit`` (orphan)
+        bypasses atexit.
         """
         self._closed.set()
         budget = max(0.0, wait)
@@ -130,7 +125,7 @@ class ComputeHost:
         with self._turn_futures_lock:
             live_sids = {sid for f, sid in self._turn_futures.items() if sid and not f.done()}
         self.flush_all_sessions(reason=reason, skip_sids=live_sids)
-        self._executor.shutdown(wait=False, cancel_futures=True)
+        self.close()
 
     def flush_all_sessions(
         self, *, reason: str = "shutdown", skip_sids: Collection[str] | None = None) -> None:
@@ -160,12 +155,10 @@ class ComputeHost:
         self.emit({"type": "shutdown.ack", "request_id": frame.get("request_id")})
         # Explicit supervisor/test shutdown is a clean child-process close;
         # SIGTERM and orphan paths are the durability flush paths.
-        self._closed.set()
-        self._executor.shutdown(wait=False, cancel_futures=True)
+        self.close()
 
     def _track_turn_future(self, future: concurrent.futures.Future, sid: str) -> None:
-        """Register an in-flight turn against its session; the done callback must pop
-        under the lock or the mapping grows for the host's life."""
+        """Track an in-flight turn; the done callback pops it or the map grows forever."""
         with self._turn_futures_lock:
             self._turn_futures[future] = sid
         future.add_done_callback(self._untrack_turn_future)
@@ -252,16 +245,13 @@ class ComputeHost:
             if run_thread is not None and hasattr(run_thread, "join"):
                 run_thread.join()
             with session["history_lock"]:
-                history_version = int(session.get("history_version", 0))
-                message_count = len(session.get("history") or [])
+                meta = _history_meta(session)
                 interrupted = bool(session.get("_turn_cancel_requested"))
-                session_key = str(session.get("session_key") or "")
             session_info = server._session_info(session.get("agent"), session)
             self._bump_progress()
             self._reply(
-                "turn.end", sid, request_id, history_version=history_version,
-                session_key=session_key, message_count=message_count, interrupted=interrupted,
-                ended_ns=now_ns(), session_info=session_info, session_info_emitted=True)
+                "turn.end", sid, request_id, **meta, interrupted=interrupted, ended_ns=now_ns(),
+                session_info=session_info, session_info_emitted=True)
         except Exception as exc:
             with contextlib.suppress(Exception):
                 from tui_gateway import server
@@ -392,13 +382,6 @@ class ComputeHost:
                 _error(str(response["error"].get("message") or failure))
                 return None
             return response
-
-        def _history_meta() -> dict[str, Any]:
-            """Ack metadata read under ``history_lock`` (caller holds it)."""
-            return {
-                "session_key": str(session.get("session_key") or ""),
-                "history_version": int(session.get("history_version", 0)),
-                "message_count": len(session.get("history") or [])}
         try:
             from tui_gateway import server
             route = MUTATOR_ROUTE_TABLE.get(route_name)
@@ -429,7 +412,7 @@ class ComputeHost:
                 if response is None:
                     return
                 with session["history_lock"]:
-                    meta = _history_meta()
+                    meta = _history_meta(session)
                 _ack(
                     result=response.get("result") or {}, **meta,
                     session_info=server._session_info(session.get("agent"), session))
@@ -438,11 +421,10 @@ class ComputeHost:
             output = server._mirror_slash_side_effects(sid, session, command) if command else ""
             with session["history_lock"]:
                 messages = server._history_to_messages(list(session.get("history") or []))
-                meta = _history_meta()
+                meta = _history_meta(session)
             _ack(
-                output=output, session_key=meta["session_key"],
-                history_version=meta["history_version"], message_count=meta["message_count"],
-                messages=messages, session_info=server._session_info(session.get("agent"), session))
+                output=output, **meta, messages=messages,
+                session_info=server._session_info(session.get("agent"), session))
         except Exception as exc:
             if route_name in {"session.compress", "slash.compress"}:
                 # The compress mirror defers the context-engine boundary notification until
@@ -482,6 +464,14 @@ class ComputeHost:
                 os._exit(0)
 
 
+def _history_meta(session: dict) -> dict[str, Any]:
+    """Transcript identity for turn.end / control.ack frames; caller holds history_lock."""
+    return {
+        "session_key": str(session.get("session_key") or ""),
+        "history_version": int(session.get("history_version", 0)),
+        "message_count": len(session.get("history") or [])}
+
+
 def _rss_mb(pid: int) -> float:
     try:
         out = subprocess.check_output(
@@ -547,8 +537,7 @@ def run_host(stdin: Any = None, stdout: Any = None) -> None:
 
 
 def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description="Dashboard compute-host process")
-    parser.parse_args(argv)
+    argparse.ArgumentParser(description="Dashboard compute-host process").parse_args(argv)
     run_host()
     return 0
 

tui_gateway/compute_host_bridge.py

@@ -1,9 +1,6 @@
-"""Compute-host (turn isolation) bridge: relay prompts/controls to the child process
-and mirror its metadata/clarify/compress acks back into the session.
-
-Bodies are rebound onto server.py's globals at install time (see
-method_ctx.bind_module), so they reference server.py globals bare.
-"""
+"""Compute-host (turn isolation) bridge: relay prompts/controls to the child process and
+mirror its metadata/clarify/compress acks back into the session. Bodies are rebound onto
+server.py's globals at install time (method_ctx.bind_module), so they use them bare."""
 
 from __future__ import annotations
 

tui_gateway/host_supervisor.py

@@ -1,9 +1,6 @@
-"""Supervisor for the dashboard compute-host child process.
-
-When ``dashboard.turn_isolation`` is enabled, agent turns move behind one persistent
-``python -m tui_gateway.compute_host`` child so compute-heavy agent threads do not
-contend with the serving process' event loop for the same GIL.
-"""
+"""Supervisor for the dashboard compute-host child: with ``dashboard.turn_isolation``
+agent turns run in one persistent ``python -m tui_gateway.compute_host`` child so heavy
+agent threads do not contend with the serving process' event loop for the GIL."""
 
 from __future__ import annotations
 
@@ -52,10 +49,9 @@ _CONTROL_REPLY_TYPES = frozenset({
 
 def append_log_record(path: str | Path, record: str) -> None:
     """Append one log record using O_APPEND and exactly one os.write call."""
-    p = Path(path)
-    p.parent.mkdir(parents=True, exist_ok=True)
+    Path(path).parent.mkdir(parents=True, exist_ok=True)
     text = record if record.endswith("\n") else f"{record}\n"
-    fd = os.open(str(p), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
+    fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
     try:
         os.write(fd, text.encode("utf-8", errors="replace"))
     finally:
@@ -87,13 +83,21 @@ def _pid_alive(pid: int) -> bool:
         return False
     try:
         os.kill(pid, 0)
-        return True
+    except Exception as exc:
+        return isinstance(exc, PermissionError)
+    return True
+
+
+def _signal_pid(pid: int, sig: int, label: str) -> bool:
+    """Send ``sig``; False when the pid is gone or the signal failed (logged)."""
+    try:
+        os.kill(pid, sig)
     except ProcessLookupError:
         return False
-    except PermissionError:
-        return True
     except Exception:
+        logger.debug("failed to %s compute host pid=%s", label, pid, exc_info=True)
         return False
+    return True
 
 
 def _pid_command(pid: int) -> str:
@@ -155,10 +159,6 @@ class HostSupervisor:
         proc = self._proc
         return int(proc.pid or 0) if proc is not None else 0
 
-    @property
-    def hello(self) -> dict[str, Any]:
-        return dict(self._hello)
-
     def is_running(self) -> bool:
         proc = self._proc
         return proc is not None and proc.poll() is None and not self._stopped_respawning
@@ -266,10 +266,9 @@ class HostSupervisor:
     ) -> dict:
         """Send a control frame; with ``wait`` block up to ``timeout`` for its ack.
 
-        ``on_late_ack`` (only with ``wait``) keeps the request adoptable after the
-        waiter gives up: the host's eventual ``control.ack``/``control.error``/``error``
-        for this ``request_id`` fires the handler once instead of being dropped.
-        Bounded by ``_LATE_CONTROL_TTL_SECS`` / ``_LATE_CONTROL_MAX``.
+        ``on_late_ack`` (only with ``wait``) keeps the request adoptable after the waiter
+        gives up: the host's eventual ``control.ack``/``control.error``/``error`` fires it
+        once instead of being dropped (bounded by ``_LATE_CONTROL_TTL_SECS``/``_MAX``).
         """
         if route_name not in MUTATOR_ROUTE_TABLE:
             raise ValueError(f"unclassified host mutator route: {route_name}")
@@ -459,18 +458,14 @@ class HostSupervisor:
         with self._lock:
             pending = self._pending_turns
             self._pending_turns = {}
+        failure = {"reason": reason, "message": message}
         for request_id, (sid, cb) in pending.items():
             self.rpc_sink({
-                "jsonrpc": "2.0",
-                "method": "event",
-                "params": {
-                    "type": "error", "session_id": sid,
-                    "payload": {"message": message, "reason": reason}}})
+                "jsonrpc": "2.0", "method": "event",
+                "params": {"type": "error", "session_id": sid, "payload": dict(failure)}})
             if cb is not None:
                 try:
-                    cb({
-                        "type": "turn.error", "sid": sid, "request_id": request_id,
-                        "reason": reason, "message": message})
+                    cb({"type": "turn.error", "sid": sid, "request_id": request_id, **failure})
                 except Exception:
                     logger.exception("compute host error callback failed")
         # A crashed host never emits the late acks timed-out control waiters still
@@ -480,9 +475,7 @@ class HostSupervisor:
             self._late_control_handlers = {}
         for request_id, (_registered_at, handler) in late.items():
             try:
-                handler({
-                    "type": "control.error", "request_id": request_id, "reason": reason,
-                    "message": message})
+                handler({"type": "control.error", "request_id": request_id, **failure})
             except Exception:
                 logger.exception("compute host late control error handler failed")
 
@@ -512,34 +505,22 @@ class HostSupervisor:
     _pid_matches_compute_host = staticmethod(is_compute_host_identity)
 
     def _terminate_pid(self, pid: int, *, timeout: float = _SHUTDOWN_TIMEOUT_SECS) -> None:
-        try:
-            os.kill(pid, signal.SIGTERM)
-        except ProcessLookupError:
-            return
-        except Exception:
-            logger.debug("failed to SIGTERM compute host pid=%s", pid, exc_info=True)
+        if not _signal_pid(pid, signal.SIGTERM, "SIGTERM"):
             return
         deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             if not _pid_alive(pid):
                 return
             time.sleep(0.05)
-        try:
-            os.kill(pid, signal.SIGKILL)
-        except ProcessLookupError:
-            return
-        except Exception:
-            logger.debug("failed to SIGKILL compute host pid=%s", pid, exc_info=True)
+        _signal_pid(pid, signal.SIGKILL, "SIGKILL")
 
     def _terminate_process(self, proc: subprocess.Popen[str]) -> None:
         if proc.poll() is not None:
             return
-        try:
+        with contextlib.suppress(Exception):
             proc.terminate()
             proc.wait(timeout=_SHUTDOWN_TIMEOUT_SECS)
             return
-        except Exception:
-            pass
         with contextlib.suppress(Exception):
             proc.kill()
         with contextlib.suppress(Exception):

tui_gateway/hosted_room_service.py

@@ -35,6 +35,10 @@ _TERMINAL_STATUSES = ("deferred", "settled", "failed", "cancelled")
 _LIVE_STATUSES = ("queued", "running", "stopping")
 _STOPPABLE_STATUSES = ("queued", "running", "indeterminate", "deferred", "stopping")
 _RETRYABLE_STATUSES = ("indeterminate", "deferred")
+# (message, error_code) for a refreshed peer catalog whose digests drift from the dispatch.
+_POLICY_DRIFT = (
+    "peer room execution policy needs reauthorization", "room_execution_policy_changed")
+_CATALOG_DRIFT = ("peer room capabilities need reauthorization", "room_capability_catalog_changed")
 
 
 def _hosted_room_turn_timeout_seconds() -> float:
@@ -85,10 +89,9 @@ class HostedRoomService:
             self._load_stored_links()
         except Exception as exc:
             self._link_load_error = str(exc)
-        supplied_routes = dict(peer_routes or {})
         supplied_clients = dict(peer_clients or {})
-        self.peer_routes.update(supplied_routes)
-        for key, route in supplied_routes.items():
+        for key, route in dict(peer_routes or {}).items():
+            self.peer_routes[key] = route
             client = supplied_clients.get(key)
             if client is None:
                 client = supplied_clients.get(route.target_install_id)
@@ -149,12 +152,13 @@ class HostedRoomService:
     def _room(self, room_id: str) -> dict[str, Any]:
         return hosted_rooms.room_state(self.db_path, room_id=room_id)
 
-    def _owned_room(self, room_id: str) -> dict[str, Any]:
-        room = self._room(room_id)
-        if str(room["authority_gateway_id"]) != hosted_rooms.local_authority_gateway_id():
+    def _owned_authority(self, room_id: str) -> tuple[str, int]:
+        """(gateway_id, epoch) of a room this gateway owns; conflict error otherwise."""
+        gateway_id, epoch = _authority(self._room(room_id))
+        if gateway_id != hosted_rooms.local_authority_gateway_id():
             raise hosted_rooms.AuthorityConflictError(
                 "This Group Chat is managed by another gateway.")
-        return room
+        return gateway_id, epoch
 
     @contextlib.contextmanager
     def _turn_lock(self, profile: str) -> Iterator[None]:
@@ -215,11 +219,9 @@ class HostedRoomService:
         self.runtime.wakeup()
 
     def revoke_room_routes(self, room_id: str) -> int:
-        """Revoke and forget every scoped peer route for one room.
-
-        Remote revocation is the boundary: an unreachable target leaves the room
-        intact for retry rather than reporting a false disband with a live grant.
-        """
+        """Revoke and forget every scoped peer route for one room. Remote revocation is the
+        boundary: an unreachable target leaves the room intact for retry rather than
+        reporting a false disband with a live grant."""
         with self._policy_lock:
             routes = [(key, route) for key, route in self.peer_routes.items() if key[0] == room_id]
         for key, route in routes:
@@ -335,11 +337,8 @@ class HostedRoomService:
         route = self.peer_routes.get(key)
         if route is None:
             raise RuntimeError("peer room route is unavailable")
-        stored = next(
-            (
-                link for link in hosted_room_links.load_room_links(self.db_path)
-                if (link.room_id, link.member_id) == key),
-            None)
+        links = hosted_room_links.load_room_links(self.db_path)
+        stored = next((l for l in links if (l.room_id, l.member_id) == key), None)
         if stored is None:
             raise RuntimeError("peer room route cannot be renewed before persistence")
         digests = {}
@@ -390,10 +389,6 @@ class HostedRoomService:
                 raise RuntimeError("hosted room replay cursor did not advance")
             cursor = next_cursor
 
-    def _append_plan(self, room_id: str, plan: discussion.PublicationPlan) -> None:
-        for event in plan.events:
-            hosted_rooms.append_event(self.db_path, **event.append_kwargs(room_id))
-
     def _policy_snapshot(self, room: Mapping[str, Any]) -> PolicySnapshot:
         return self.policy_checkpoint.snapshot(
             room_id=str(room["room_id"]), latest_seq=int(room["latest_seq"]))
@@ -417,7 +412,8 @@ class HostedRoomService:
                     room, task_events, plan, status=status, result=task.get("result"),
                     execution_generation=execution_generation if status == "deferred" else None,
                     local_profiles=local_profiles)
-                self._append_plan(room_id, publication)
+                for event in publication.events:
+                    hosted_rooms.append_event(self.db_path, **event.append_kwargs(room_id))
                 changed = True
         return changed
 
@@ -442,7 +438,7 @@ class HostedRoomService:
     def prepare_room(self, binding: HostedRoomBinding) -> None:
         with self._policy_lock:
             room = self._room(binding.room_id)
-            snapshot = self._policy_snapshot(room)
+            snapshot = self._policy_snapshot(room)  # sync() side effect feeds the publish below
             if self._publish_terminal_tasks(room):
                 room = self._room(binding.room_id)
                 snapshot = self._policy_snapshot(room)
@@ -494,7 +490,7 @@ class HostedRoomService:
 
     def send(self, *, room_id: str, event_id: str, payload: Any) -> dict[str, Any]:
         normalized = discussion.validate_user_payload(payload)
-        gateway_id, epoch = _authority(self._owned_room(room_id))
+        gateway_id, epoch = self._owned_authority(room_id)
         event = hosted_rooms.append_event(
             self.db_path, room_id=room_id, event_id=event_id, kind="message.user",
             actor={"kind": "user", "id": "desktop"}, payload=normalized,
@@ -508,11 +504,10 @@ class HostedRoomService:
 
     def stop_room(
         self, room_id: str, *, cancel_id: str, require_acknowledged: bool = False) -> int:
-        gateway_id, epoch = _authority(self._owned_room(room_id))
+        gateway_id, epoch = self._owned_authority(room_id)
         hosted_rooms.request_room_stop(
             self.db_path, room_id=room_id, cancel_id=cancel_id, expected_gateway_id=gateway_id,
             expected_epoch=epoch)
-        cancelled = 0
         pending = 0
         with self._policy_lock:
             tasks = {
@@ -523,9 +518,9 @@ class HostedRoomService:
                     str(task.get("cancel_id") or "") if task.get("status") == "stopping" else "")
                 result = self.runtime.cancel(
                     task["identity"], cancel_id=task_cancel_id or cancel_id)
-                cancelled += 1
                 if result["status"] == "stopping":
                     pending += 1
+        cancelled = len(tasks)
         if require_acknowledged and pending:
             raise RuntimeError("room work is still stopping; retry deletion after Stop completes")
         self.runtime.wakeup()
@@ -533,11 +528,8 @@ class HostedRoomService:
 
     def retry_room_task(self, room_id: str, *, task_id: str) -> dict[str, Any]:
         """Retry one uncertain or deferred task only after explicit user action."""
-        task = next(
-            (
-                candidate for candidate in self._list_tasks(room_id, _RETRYABLE_STATUSES)
-                if candidate["identity"].task_id == task_id),
-            None)
+        candidates = self._list_tasks(room_id, _RETRYABLE_STATUSES)
+        task = next((c for c in candidates if c["identity"].task_id == task_id), None)
         if task is None:
             raise driver.InvalidTaskTransitionError("no retryable room task matches task_id")
         return self.runtime.retry_indeterminate(task["identity"])
@@ -601,8 +593,7 @@ class HostedRoomService:
                 if action_room_id == room_id)
         return {
             "running": runtime["running"],
-            "working": bool(
-                counts.get("running") or counts.get("queued") or counts.get("stopping")),
+            "working": any(counts.get(s) for s in _LIVE_STATUSES),
             "blocked": room_id in runtime["blocked_rooms"]
             or bool(counts.get("indeterminate") or counts.get("stopping")),
             "counts": dict(counts),
@@ -622,13 +613,10 @@ class _RouteStatusPeerClient:
         self._on_refreshed = on_refreshed
 
     def _refresh_grant(self, kwargs: dict) -> dict:
-        """Rotate an expiring grant before dispatch; return the kwargs to send.
-
-        Refresh failures only escalate to reauthorization when the peer says so or the
-        grant is already past its hard expiry; otherwise the original grant is tried
-        as-is. A refreshed catalog whose digests drift from the dispatch is a policy
-        change and is refused before any dispatch.
-        """
+        """Rotate an expiring grant before dispatch; return the kwargs to send. Refresh
+        failures escalate to reauthorization only when the peer says so or the grant is
+        past its hard expiry; otherwise the original grant is tried as-is. A refreshed
+        catalog whose digests drift from the dispatch is a policy change: refused."""
         grant = kwargs["grant"]
         if not room_grant_needs_dispatch_refresh(grant):
             return kwargs
@@ -654,13 +642,9 @@ class _RouteStatusPeerClient:
             refreshed_catalog = GatewayRoomCatalog.from_mapping(refreshed.get("catalog"))
             drift = None
             if refreshed_catalog.execution_policy.policy_digest != checked.execution_policy_digest:
-                drift = (
-                    "peer room execution policy needs reauthorization",
-                    "room_execution_policy_changed")
+                drift = _POLICY_DRIFT
             elif refreshed_catalog.catalog_digest != checked.capability_digest:
-                drift = (
-                    "peer room capabilities need reauthorization", "room_capability_catalog_changed"
-                )
+                drift = _CATALOG_DRIFT
             if drift is not None:
                 self._on_reauthorization()
                 raise PeerRunsHTTPError(

tui_gateway/methods_groups.py

@@ -1,17 +1,14 @@
-"""Hosted-room JSON-RPC contract.
-
-These methods expose durable room identity, replay, and the process-owned
+"""Hosted-room JSON-RPC contract: durable room identity, replay, and the process-owned
 same-gateway Discussion driver. ``groups.capabilities`` keeps that boundary
 machine-readable so older clients stay on the renderer-owned room path.
 
-Handlers are rebound onto server.py's globals at install (see method_ctx.py), so
-bodies see only server globals plus the names methods_bot_relay.register publishes;
-module-private helpers reach them through keyword defaults. ``_room_method`` wraps
-each handler with the shared service-lookup / error-code envelope.
-"""
+Handlers are rebound onto server.py's globals at install (method_ctx.py), so bodies see
+only server globals plus what methods_bot_relay.register publishes; module-private
+helpers reach them through keyword defaults. ``_room_method`` is the shared envelope."""
 
 from .method_ctx import HandlerRegistry
 
+import importlib
 import os
 import threading
 
@@ -181,11 +178,10 @@ def _room_method(
     service_message: str = _DRIVER_UNAVAILABLE, db: bool = False):
     """Register ``fn`` under ``name`` with the shared hosted-room error envelope.
 
-    ``service_code`` set: the live service is required and passed as a third argument;
-    when absent the handler fails with that code. ``db``: the default room db path is
-    passed as the next argument. ``room_code`` maps ``HostedRoomError`` (or only
-    ``ReplicaError`` when ``replica_only``) to a 4xxx client error, attaching
-    ``{"reason"}`` data when ``with_reason``; any other exception maps to ``code``.
+    ``service_code``: the live service is required (else fail with that code) and passed
+    as a third argument; ``db``: the default room db path follows. ``room_code`` maps
+    ``HostedRoomError`` (only ``ReplicaError`` when ``replica_only``) to a 4xxx client
+    error with ``{"reason"}`` data when ``with_reason``; anything else maps to ``code``.
     """
 
     def dec(fn):
@@ -376,11 +372,8 @@ def _(rid, params: dict, db_path) -> dict:
     "groups.create", code=5111, room_code=4110, service_code=4123,
     service_message=_WORKER_UNAVAILABLE)
 def _(rid, params: dict, service) -> dict:
-    """Create a hosted room idempotently.
-
-    Required params: ``room_id``, ``name``, and ``members``. Authority is
-    derived from this gateway's stable install identity, never from the client.
-    """
+    """Create a hosted room idempotently; authority comes from this gateway's stable
+    install identity, never from the client."""
     room = service.create_room(
         room_id=params.get("room_id"), name=params.get("name"), members=params.get("members"))
     return _ok(rid, {"room": room})
@@ -404,12 +397,8 @@ def _(rid, params: dict, db_path) -> dict:
     "groups.send", code=5112, room_code=4111, service_code=4123, service_message=_WORKER_UNAVAILABLE
 )
 def _(rid, params: dict, service) -> dict:
-    """Append one typed event to a hosted room idempotently.
-
-    Required params: ``room_id``, ``event_id``, and object ``payload``. Only
-    inert ``message.user`` events are accepted through this client-facing
-    method; the actor is server-owned rather than trusted from params.
-    """
+    """Append one typed event idempotently. Only inert ``message.user`` events are
+    accepted from clients; the actor is server-owned rather than trusted from params."""
     from gateway.hosted_rooms import user_event_id
     client_event_id = params.get("event_id")
     event = service.send(
@@ -420,16 +409,6 @@ def _(rid, params: dict, service) -> dict:
     })
 
 
-@_room_method("groups.rename", code=5117, room_code=4117, db=True)
-def _(rid, params: dict, db_path) -> dict:
-    """Rename one hosted room atomically with its replay event."""
-    from gateway.hosted_rooms import rename_room
-    renamed = rename_room(
-        db_path, room_id=params.get("room_id"), event_id=params.get("event_id"),
-        name=params.get("name"))
-    return _ok(rid, {"room": renamed})
-
-
 @_room_method(
     "groups.disband", code=5114, room_code=4113, service_code=4123,
     service_message=_WORKER_UNAVAILABLE)
@@ -500,70 +479,76 @@ def _(rid, params: dict, service) -> dict:
     return _ok(rid, {"retried": True, "task": receipt})
 
 
-@_room_method("groups.log", code=5113, room_code=4112, db=True)
-def _(rid, params: dict, db_path) -> dict:
-    """Return a monotonic room-log delta after ``since_seq``."""
-    from gateway.hosted_rooms import read_events
-    delta = read_events(
-        db_path, room_id=params.get("room_id"), since_seq=params.get("since_seq", 0),
-        limit=params.get("limit", 100), include_disbanded=params.get("include_disbanded") is True)
-    return _ok(rid, delta)
+def _passthrough(
+    name: str, module: str, fn_name: str, doc: str, *, code: int, room_code: int,
+    params: tuple, replica_only: bool = False, wrap: str | None = None,
+) -> None:
+    """Register a method whose result is ``module.fn(db_path, **params)`` verbatim (or under
+    key ``wrap``). ``params`` items are ``key`` (-> ``params.get(key)``) or
+    ``(key, extractor(params))``."""
+
+    @_room_method(
+        name, code=code, room_code=room_code, replica_only=replica_only,
+        with_reason=not replica_only, db=True)
+    def handler(rid, params_in: dict, db_path) -> dict:
+        fn = getattr(importlib.import_module(module), fn_name)
+        kwargs = {}
+        for spec in params:
+            if isinstance(spec, str):
+                kwargs[spec] = params_in.get(spec)
+            else:
+                kwargs[spec[0]] = spec[1](params_in)
+        result = fn(db_path, **kwargs)
+        return _ok(rid, {wrap: result} if wrap else result)
+    handler.__doc__ = doc
 
 
-@_room_method(
-    "groups.replicate", code=5116, room_code=4116, replica_only=True, with_reason=False, db=True)
-def _(rid, params: dict, db_path) -> dict:
-    """Persist one authority-stamped replay page into the local replica store.
-
-    ``page`` is the verbatim ``groups.log`` result read from the room's
-    authority gateway; ingest is idempotent and refuses sequence gaps and
-    authority-epoch regressions.
-    """
-    from gateway.hosted_room_replicas import ingest_page
-    result = ingest_page(
-        db_path, room_id=params.get("room_id"), room_name=params.get("room_name"),
-        members=params.get("members"), page=params.get("page"))
-    return _ok(rid, result)
+def _include_disbanded(params: dict) -> bool:
+    return params.get("include_disbanded") is True
 
 
-@_room_method(
-    "groups.replica_state", code=5117, room_code=4117, replica_only=True, with_reason=False, db=True
-)
-def _(rid, params: dict, db_path) -> dict:
-    """Report the local replica's coverage and authority lineage."""
-    from gateway.hosted_room_replicas import replica_state
-    return _ok(rid, replica_state(db_path, room_id=params.get("room_id")))
+_passthrough(
+    "groups.rename", "gateway.hosted_rooms", "rename_room",
+    """Rename one hosted room atomically with its replay event.""",
+    code=5117, room_code=4117, params=("room_id", "event_id", "name"), wrap="room")
+_passthrough(
+    "groups.log", "gateway.hosted_rooms", "read_events",
+    """Return a monotonic room-log delta after ``since_seq``.""",
+    code=5113, room_code=4112,
+    params=(
+        "room_id", ("since_seq", lambda p: p.get("since_seq", 0)),
+        ("limit", lambda p: p.get("limit", 100)), ("include_disbanded", _include_disbanded)))
+_passthrough(
+    "groups.replicate", "gateway.hosted_room_replicas", "ingest_page",
+    """Persist one authority-stamped replay page (a verbatim ``groups.log`` result) into
+    the local replica store; idempotent, refuses sequence gaps and epoch regressions.""",
+    code=5116, room_code=4116, params=("room_id", "room_name", "members", "page"),
+    replica_only=True)
+_passthrough(
+    "groups.replica_state", "gateway.hosted_room_replicas", "replica_state",
+    """Report the local replica's coverage and authority lineage.""",
+    code=5117, room_code=4117, params=("room_id",), replica_only=True)
 
 
 @_room_method("groups.promote", code=5118, room_code=4118, with_reason=False, db=True)
 def _(rid, params: dict, db_path) -> dict:
-    """Continue a replicated room on THIS gateway at ``epoch + 1``.
-
-    Requires ``confirm: true`` — the caller asserts the previous authority can
-    no longer commit (explicit user action; a lease/quorum driver later).
-    """
+    """Continue a replicated room on THIS gateway at ``epoch + 1``. Requires ``confirm:
+    true`` — the caller asserts the previous authority can no longer commit."""
     from gateway.hosted_room_replicas import promote_replica
     if params.get("confirm") is not True:
         return _err(
             rid, 4118,
             "promotion requires confirm=true acknowledging the previous "
             "authority can no longer commit")
-    result = promote_replica(
-        db_path, room_id=params.get("room_id"), reason=params.get("reason", "authority-unreachable")
-    )
-    return _ok(rid, result)
+    reason = params.get("reason", "authority-unreachable")
+    return _ok(rid, promote_replica(db_path, room_id=params.get("room_id"), reason=reason))
 
 
-@_room_method(
-    "groups.demote", code=5119, room_code=4119, replica_only=True, with_reason=False, db=True)
-def _(rid, params: dict, db_path) -> dict:
-    """Fence this gateway's stale room authority against a proven newer epoch."""
-    from gateway.hosted_room_replicas import demote_room
-    result = demote_room(
-        db_path, room_id=params.get("room_id"),
-        observed_gateway_id=params.get("observed_gateway_id"),
-        observed_epoch=params.get("observed_epoch"))
-    return _ok(rid, result)
+_passthrough(
+    "groups.demote", "gateway.hosted_room_replicas", "demote_room",
+    """Fence this gateway's stale room authority against a proven newer epoch.""",
+    code=5119, room_code=4119, params=("room_id", "observed_gateway_id", "observed_epoch"),
+    replica_only=True)
 
 
 def register(server) -> None: