From ae94886814beaf9101d2090ba8f4a2eae2d692cd Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 00:17:58 -0700 Subject: [PATCH] refactor(tui_gateway): table-driven groups passthrough handlers, pid/authority helpers, docstring compaction (group E) --- tui_gateway/compute_host.py | 71 ++++++-------- tui_gateway/compute_host_bridge.py | 9 +- tui_gateway/host_supervisor.py | 73 ++++++-------- tui_gateway/hosted_room_service.py | 78 ++++++--------- tui_gateway/methods_groups.py | 149 +++++++++++++---------------- 5 files changed, 158 insertions(+), 222 deletions(-) diff --git a/tui_gateway/compute_host.py b/tui_gateway/compute_host.py index 0f3d49cf40..379d12b1c9 100644 --- a/tui_gateway/compute_host.py +++ b/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 diff --git a/tui_gateway/compute_host_bridge.py b/tui_gateway/compute_host_bridge.py index 4754a14efd..8ae3628b92 100644 --- a/tui_gateway/compute_host_bridge.py +++ b/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 diff --git a/tui_gateway/host_supervisor.py b/tui_gateway/host_supervisor.py index dddb4401b9..925bac33b6 100644 --- a/tui_gateway/host_supervisor.py +++ b/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): diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 7303058c45..1c5086310b 100644 --- a/tui_gateway/hosted_room_service.py +++ b/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( diff --git a/tui_gateway/methods_groups.py b/tui_gateway/methods_groups.py index e286266174..7eac62cb1c 100644 --- a/tui_gateway/methods_groups.py +++ b/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: