From 31184bdc0fc52c33a77ebee862dc677ac6d167dc Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:15:46 -0700 Subject: [PATCH 1/2] refactor(tui_gateway): unify hosted-room driver/peer plumbing into fenced-op, session-probe and error-table helpers --- tui_gateway/hosted_room_driver.py | 909 +++++++--------------- tui_gateway/hosted_room_peer_http.py | 550 ++++--------- tui_gateway/hosted_room_peer_transport.py | 123 +-- tui_gateway/hosted_room_server_rpc.py | 55 +- 4 files changed, 467 insertions(+), 1170 deletions(-) diff --git a/tui_gateway/hosted_room_driver.py b/tui_gateway/hosted_room_driver.py index 345b6c4ff1..526bb3e5bf 100644 --- a/tui_gateway/hosted_room_driver.py +++ b/tui_gateway/hosted_room_driver.py @@ -1,15 +1,12 @@ """Runtime adapter for gateway-owned hosted room turns. -The durable state machine lives in :mod:`gateway.hosted_room_driver`. This -module owns the process-local worker and a deliberately small, injected session -adapter (seven methods). It does not import the gateway server, construct -agents, or depend on any client transport; tests use deterministic fakes. - -One bounded supervisor schedules independent room workers. Profile turn locks -still serialize Bots that share one profile, while a room waiting for approval -cannot stop unrelated rooms from progressing. Hosted member sessions -intentionally reuse ``Group: `` so a local-to-hosted migration -preserves the same canonical transcript instead of forking a second conversation. +The durable state machine lives in :mod:`gateway.hosted_room_driver`. This module +owns the process-local worker and a small injected session adapter (seven methods); +it never imports the gateway server or constructs agents. One bounded supervisor +schedules independent room workers: profile turn locks still serialize Bots sharing +a profile, while a room waiting for approval cannot stall unrelated rooms. Hosted +member sessions reuse ``Group: `` so a local-to-hosted migration keeps one +canonical transcript instead of forking a second conversation. """ from __future__ import annotations @@ -18,7 +15,8 @@ import threading import time import uuid from collections.abc import Callable, Iterable, Mapping, Sequence -from dataclasses import dataclass +from contextlib import suppress +from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, ContextManager, Protocol, cast @@ -30,8 +28,8 @@ _STOP_ACK_STATUSES = {"cancelled", "interrupted"} ROOM_SESSION_SOURCE = "bot_room" MAX_TERMINAL_TEXT_BYTES = 64 * 1024 _TERMINAL_TRUNCATION_NOTICE = ( - "\n\n[Reply truncated. Ask the Bot to share the full result as a file.]" -) + "\n\n[Reply truncated. Ask the Bot to share the full result as a file.]") +_STOP_PENDING = "stop retry remains pending: {exc}" class InternalSessionRPC(Protocol): @@ -47,15 +45,8 @@ class InternalSessionRPC(Protocol): """Resume the canonical room session.""" def submit( - self, - *, - profile: str, - session_id: str, - prompt: str, - source: str, - task: state.TaskIdentity, - execution_generation: int, - on_terminal: Callable[[Mapping[str, Any]], None], + self, *, profile: str, session_id: str, prompt: str, source: str, task: state.TaskIdentity, + execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None], ) -> Mapping[str, Any]: """Submit one fenced room turn and durably report its terminal result.""" @@ -71,13 +62,8 @@ class InternalSessionRPC(Protocol): """Interrupt only when the current turn still matches the expected task.""" -class MemberTransportResolver(Protocol): - """Resolve the session transport for one durable room task.""" - - def __call__( - self, binding: "HostedRoomBinding", task: Mapping[str, Any] - ) -> InternalSessionRPC: - """Return a local or peer transport without changing task identity.""" +# Resolve the local or peer session transport for one durable room task. +MemberTransportResolver = Callable[["HostedRoomBinding", Mapping[str, Any]], InternalSessionRPC] @dataclass(frozen=True) @@ -106,52 +92,49 @@ class _RecoveryInspection: _NO_INSPECTION = _RecoveryInspection(terminal=None, active=False, status=None) +def _session_kw(profile: str, session_id: str) -> dict[str, str]: + return {"profile": profile, "session_id": session_id, "source": ROOM_SESSION_SOURCE} + + +def _fences(task: Mapping[str, Any]) -> dict[str, int]: + return { + "expected_execution_generation": int(task["execution_generation"]), + "expected_cancel_generation": int(task["cancel_generation"])} + + class HostedRoomRuntime: """Run queued hosted-room tasks independently of Desktop connections.""" def __init__( - self, - *, - db_path: Path | str, + self, *, db_path: Path | str, rooms: Iterable[HostedRoomBinding] | Callable[[], Iterable[HostedRoomBinding]], - turn_lock: Callable[[str], ContextManager[Any]], - rpc: InternalSessionRPC | None = None, + turn_lock: Callable[[str], ContextManager[Any]], rpc: InternalSessionRPC | None = None, transport_resolver: MemberTransportResolver | None = None, prepare_room: Callable[[HostedRoomBinding], None] | None = None, - publish_terminal: Callable[[HostedRoomBinding, Mapping[str, Any]], None] - | None = None, - pending_action: Callable[[str, str, Mapping[str, Any] | None], None] - | None = None, + publish_terminal: Callable[[HostedRoomBinding, Mapping[str, Any]], None] | None = None, + pending_action: Callable[[str, str, Mapping[str, Any] | None], None] | None = None, clock: Callable[[], float] = time.time, - lease_ttl_seconds: float = 30.0, - poll_interval_seconds: float = 5.0, - active_poll_interval_seconds: float = 0.25, - turn_timeout_seconds: float = 1830.0, - indeterminate_defer_seconds: float = 60.0, - max_concurrent_rooms: int = 4, - unavailable_retry_min_seconds: float = 1.0, - unavailable_retry_max_seconds: float = 30.0, - process_generation: str | None = None, - ) -> None: + lease_ttl_seconds: float = 30.0, poll_interval_seconds: float = 5.0, + active_poll_interval_seconds: float = 0.25, turn_timeout_seconds: float = 1830.0, + indeterminate_defer_seconds: float = 60.0, max_concurrent_rooms: int = 4, + unavailable_retry_min_seconds: float = 1.0, unavailable_retry_max_seconds: float = 30.0, + process_generation: str | None = None) -> None: for name, value in ( ("lease_ttl_seconds", lease_ttl_seconds), ("poll_interval_seconds", poll_interval_seconds), ("active_poll_interval_seconds", active_poll_interval_seconds), ("turn_timeout_seconds", turn_timeout_seconds), - ("indeterminate_defer_seconds", indeterminate_defer_seconds), - ): + ("indeterminate_defer_seconds", indeterminate_defer_seconds)): if value <= 0: raise ValueError(f"{name} must be positive") if ( isinstance(max_concurrent_rooms, bool) or not isinstance(max_concurrent_rooms, int) - or max_concurrent_rooms < 1 - ): + or max_concurrent_rooms < 1): raise ValueError("max_concurrent_rooms must be a positive integer") if not ( unavailable_retry_min_seconds > 0 - and unavailable_retry_max_seconds >= unavailable_retry_min_seconds - ): + and unavailable_retry_max_seconds >= unavailable_retry_min_seconds): raise ValueError("unavailable retry bounds are invalid") if rpc is None and transport_resolver is None: raise ValueError("rpc or transport_resolver is required") @@ -174,9 +157,7 @@ class HostedRoomRuntime: self.process_generation = process_generation or uuid.uuid4().hex self._rooms_provider: Callable[[], Iterable[HostedRoomBinding]] if callable(rooms): - self._rooms_provider = cast( - Callable[[], Iterable[HostedRoomBinding]], rooms - ) + self._rooms_provider = cast(Callable[[], Iterable[HostedRoomBinding]], rooms) else: room_bindings = tuple(rooms) self._rooms_provider = lambda: room_bindings @@ -190,9 +171,7 @@ class HostedRoomRuntime: self._recovered_leases: set[tuple[str, int]] = set() self._inspected_indeterminate_attempts: set[tuple[str, str, int]] = set() self._ambiguous_rooms: dict[str, float] = {} - self._unavailable_route_retries: dict[ - tuple[str, str], dict[str, float] - ] = {} + self._unavailable_route_retries: dict[tuple[str, str], dict[str, float]] = {} self._blocked_rooms: set[str] = set() self._status_lock = threading.Lock() self._current_tasks: dict[str, state.TaskIdentity] = {} @@ -201,7 +180,6 @@ class HostedRoomRuntime: self._cycles = 0 # ------------------------------------------------------------------ lifecycle - def start(self) -> None: """Start the bounded room-worker supervisor idempotently.""" with self._status_lock: @@ -210,10 +188,7 @@ class HostedRoomRuntime: self._stop.clear() self._wake.set() self._thread = threading.Thread( - target=self._worker_loop, - name="hosted-room-driver-supervisor", - daemon=True, - ) + target=self._worker_loop, name="hosted-room-driver-supervisor", daemon=True) self._thread.start() def stop(self, *, timeout: float = 5.0) -> bool: @@ -230,9 +205,7 @@ class HostedRoomRuntime: room_threads = tuple(self._room_threads.values()) for room_thread in room_threads: room_thread.join(max(0.0, deadline - time.monotonic())) - return not thread.is_alive() and all( - not room_thread.is_alive() for room_thread in room_threads - ) + return not thread.is_alive() and all(not t.is_alive() for t in room_threads) def wakeup(self) -> None: """Wake the worker after task admission or a room-state change.""" @@ -257,17 +230,10 @@ class HostedRoomRuntime: "leased_rooms": tuple(sorted(self._leases)), "blocked_rooms": tuple(sorted(self._blocked_rooms)), "last_error": self._last_error, - "cycles": self._cycles, - } + "cycles": self._cycles} # ------------------------------------------------------------------ public ops - - def cancel( - self, - identity: state.TaskIdentity, - *, - cancel_id: str, - ) -> dict[str, Any]: + def cancel(self, identity: state.TaskIdentity, *, cancel_id: str) -> dict[str, Any]: """Persist a stop intent, then commit cancellation after acknowledgement. The worker thread transitions tasks concurrently (queued -> running -> @@ -281,40 +247,28 @@ class HostedRoomRuntime: return before if before["status"] in state.TERMINAL_STATUSES: raise state.InvalidTaskTransitionError( - f"cannot cancel task in state '{before['status']}'" - ) + f"cannot cancel task in state '{before['status']}'") + fenced = dict( + cancel_id=cancel_id, expected_cancel_generation=before["cancel_generation"], + clock=self.clock) if before["status"] in {"queued", "deferred"}: try: - cancelled = state.cancel_task( - self.db_path, - identity, - cancel_id=cancel_id, - expected_cancel_generation=before["cancel_generation"], - clock=self.clock, - ) + cancelled = state.cancel_task(self.db_path, identity, **fenced) except (state.InvalidTaskTransitionError, state.StaleTaskError): continue # lost the race with the worker; re-route self.wakeup() return cancelled try: - stopping = state.begin_task_cancel( - self.db_path, - identity, - cancel_id=cancel_id, - expected_cancel_generation=before["cancel_generation"], - clock=self.clock, - ) + stopping = state.begin_task_cancel(self.db_path, identity, **fenced) except (state.InvalidTaskTransitionError, state.StaleTaskError): continue # settled or re-queued mid-flight; re-route binding = self._binding_for_room(identity.room_id) try: if binding is not None: lease = self._ensure_lease(binding) - if self._peer_stop_acknowledged(binding, stopping): - self._complete_cancel(stopping, cancel_id=cancel_id) - elif not self._settle_stopping_completion( - binding, stopping, lease - ) and self._interrupt_stopping_task(binding, stopping): + if self._peer_stop_acknowledged(binding, stopping) or ( + not self._settle_stopping_completion(binding, stopping, lease) + and self._interrupt_stopping_task(binding, stopping)): self._complete_cancel(stopping, cancel_id=cancel_id) except Exception as exc: self._record_error(f"stop remains pending: {exc}") @@ -326,25 +280,20 @@ class HostedRoomRuntime: if final["status"] == "cancelled": return final raise state.InvalidTaskTransitionError( - f"cancel kept losing races with task transitions " - f"(last observed state '{final['status']}')" - ) + "cancel kept losing races with task transitions " + f"(last observed state '{final['status']}')") def retry_indeterminate(self, identity: state.TaskIdentity) -> dict[str, Any]: """Explicitly retry one uncertain attempt under the current room lease.""" task = state.get_task(self.db_path, identity) if task["status"] not in {"indeterminate", "deferred"}: - raise state.InvalidTaskTransitionError( - f"cannot retry task in state '{task['status']}'" - ) + raise state.InvalidTaskTransitionError(f"cannot retry task in state '{task['status']}'") binding = self._binding_for_room(identity.room_id) if binding is None: raise state.RoomUnavailableError("hosted room is unavailable") lease = self._ensure_lease(binding) if task["status"] == "deferred": - return self._requeue( - state.requeue_deferred_task, task, lease, identity.room_id - ) + return self._requeue(state.requeue_deferred_task, task, lease, identity.room_id) # Explicit Retry may resume the exact stored session; an automatic # abandoned-attempt scan remains non-resuming for local sessions. inspection = self._inspect_recovery_session(binding, task) @@ -355,14 +304,10 @@ class HostedRoomRuntime: if inspection.active: self._set_blocked(identity.room_id, True) raise state.InvalidTaskTransitionError( - "cannot retry while the original task attempt is still active" - ) - return self._requeue( - state.requeue_indeterminate_task, task, lease, identity.room_id - ) + "cannot retry while the original task attempt is still active") + return self._requeue(state.requeue_indeterminate_task, task, lease, identity.room_id) # ------------------------------------------------------------------ durable-state helpers - def _publish(self, binding: HostedRoomBinding, task: dict[str, Any]) -> dict[str, Any]: if self.publish_terminal is not None: self.publish_terminal(binding, task) @@ -370,183 +315,124 @@ class HostedRoomRuntime: def _set_blocked(self, room_id: str, blocked: bool) -> None: with self._status_lock: - if blocked: - self._blocked_rooms.add(room_id) - else: - self._blocked_rooms.discard(room_id) + (self._blocked_rooms.add if blocked else self._blocked_rooms.discard)(room_id) + + def _fenced( + self, op: Callable[..., dict[str, Any]], binding: HostedRoomBinding | None, + task: Mapping[str, Any], lease: state.DriverLease, *, publish: bool = True, **extra: Any, + ) -> dict[str, Any]: + """Run one lease-fenced state transition on ``task``; ``extra`` may override fences.""" + result = op( + self.db_path, task["identity"], lease, **{**_fences(task), "clock": self.clock, **extra} + ) + return self._publish(binding, result) if publish and binding is not None else result def _requeue( - self, - requeue: Callable[..., dict[str, Any]], - task: Mapping[str, Any], - lease: state.DriverLease, - room_id: str, - ) -> dict[str, Any]: - retried = requeue( - self.db_path, - task["identity"], - lease, - expected_execution_generation=task["execution_generation"], - expected_cancel_generation=task["cancel_generation"], - clock=self.clock, - ) + self, requeue: Callable[..., dict[str, Any]], task: Mapping[str, Any], + lease: state.DriverLease, room_id: str) -> dict[str, Any]: + retried = self._fenced(requeue, None, task, lease) self._set_blocked(room_id, False) self.wakeup() return retried def _complete_cancel( - self, task: Mapping[str, Any], *, cancel_id: str | None = None - ) -> dict[str, Any]: + self, task: Mapping[str, Any], *, cancel_id: str | None = None) -> dict[str, Any]: return state.complete_task_cancel( - self.db_path, - task["identity"], + self.db_path, task["identity"], cancel_id=task["cancel_id"] if cancel_id is None else cancel_id, - expected_cancel_generation=task["cancel_generation"], - clock=self.clock, - ) + expected_cancel_generation=task["cancel_generation"], clock=self.clock) def _resolve_indeterminate( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, - terminal: _TerminalReceipt, - *, - publish: bool = True, - ) -> dict[str, Any]: - resolved = state.resolve_indeterminate_task( - self.db_path, - task["identity"], - lease, - expected_execution_generation=int(task["execution_generation"]), - expected_cancel_generation=task["cancel_generation"], - settlement_id=terminal.settlement_id, - status=terminal.status, - result=terminal.result, - clock=self.clock, - ) - return self._publish(binding, resolved) if publish else resolved + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease, + terminal: _TerminalReceipt, *, publish: bool = True) -> dict[str, Any]: + return self._fenced( + state.resolve_indeterminate_task, binding, task, lease, publish=publish, + **asdict(terminal)) def _resolve_remote_cancel( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, - *, - publish: bool = True, - ) -> dict[str, Any]: - resolved = state.resolve_indeterminate_cancellation( - self.db_path, - task["identity"], - lease, - expected_execution_generation=task["execution_generation"], - expected_cancel_generation=task["cancel_generation"], - cancel_id=f"remote-cancel:{task['execution_generation']}", - clock=self.clock, - ) - return self._publish(binding, resolved) if publish else resolved + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease, + *, publish: bool = True) -> dict[str, Any]: + return self._fenced( + state.resolve_indeterminate_cancellation, binding, task, lease, publish=publish, + cancel_id=f"remote-cancel:{task['execution_generation']}") def _settle_stopping( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, - *, - settlement_id: str, - status: state.TerminalStatus, - result: Mapping[str, Any], - expected_execution_generation: int | None = None, - ) -> dict[str, Any]: - settled = state.settle_stopping_task( - self.db_path, - task["identity"], - lease, - expected_execution_generation=( - int(task["execution_generation"]) - if expected_execution_generation is None - else expected_execution_generation - ), - expected_cancel_generation=int(task["cancel_generation"]), - settlement_id=settlement_id, - status=status, - result=result, - clock=self.clock, - ) - return self._publish(binding, settled) + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease, + **terminal: Any) -> dict[str, Any]: + """Settle a stopping task from ``terminal`` (settlement_id/status/result + fences).""" + return self._fenced(state.settle_stopping_task, binding, task, lease, **terminal) + + def _finish_stop( + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease + ) -> bool: + """Terminalize a stopping task from its receipt or an acknowledged interrupt.""" + if self._settle_stopping_completion(binding, task, lease): + return True + if self._interrupt_stopping_task(binding, task): + self._complete_acknowledged_stop(binding, task, lease) + return True + return False # ------------------------------------------------------------------ session probes - def _resume_exact( - self, transport: InternalSessionRPC, room_id: str, profile: str - ) -> str | None: + self, transport: InternalSessionRPC, room_id: str, profile: str) -> str | None: """Resolve the canonical room session and return its resumed runtime id. Callers must use the returned id (not the stored one) for every subsequent history/info probe; resume may hand back a different id. """ session = transport.resolve_exact( - profile=profile, - title=room_session_title(room_id), - source=ROOM_SESSION_SOURCE, - ) + profile=profile, title=room_session_title(room_id), source=ROOM_SESSION_SOURCE) if session is None: return None - resumed = transport.resume( - profile=profile, - session_id=_session_id(session), - source=ROOM_SESSION_SOURCE, - ) - return _session_id(resumed) + return _session_id(transport.resume(**_session_kw(profile, _session_id(session)))) + + def _open_session( + self, binding: HostedRoomBinding, task: Mapping[str, Any], *, peer_only: bool = False + ) -> tuple[InternalSessionRPC | None, str | None, str | None]: + """Return ``(transport, profile, resumed session id)``; the id is None when unusable. + + A missing transport (or the local one when ``peer_only``) yields no session id. + """ + transport = self._transport_for(binding, task) + if transport is None or (peer_only and transport is self.rpc): + return transport, None, None + profile = task["payload"]["target_profile"] + return transport, profile, self._resume_exact(transport, binding.room_id, profile) @staticmethod - def _info_acknowledges_peer_cancel( - info: Mapping[str, Any], task: Mapping[str, Any] - ) -> bool: + def _terminal_from_history( + transport: InternalSessionRPC, profile: str, session_id: str, task: Mapping[str, Any] + ) -> _TerminalReceipt | None: + return _find_terminal_receipt( + transport.history(**_session_kw(profile, session_id)), + task["identity"], int(task["execution_generation"])) + + @staticmethod + def _info_acknowledges_peer_cancel(info: Mapping[str, Any], task: Mapping[str, Any]) -> bool: """Accept only one exact peer task attempt's terminal Stop receipt.""" return ( not _info_active(info) and str(info.get("status") or "") in _STOP_ACK_STATUSES and str(info.get("task_id") or "") == task["identity"].task_id - and int(info.get("execution_generation") or 0) - == int(task["execution_generation"]) - ) + and int(info.get("execution_generation") or 0) == int(task["execution_generation"])) - def _peer_stop_acknowledged( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - ) -> bool: + def _peer_stop_acknowledged(self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> bool: """Probe a peer's exact durable terminal status before reading history.""" - transport = self._transport_for(binding, task) - if transport is None or transport is self.rpc: - return False - profile = task["payload"]["target_profile"] - session_id = self._resume_exact(transport, binding.room_id, profile) + transport, profile, session_id = self._open_session(binding, task, peer_only=True) if session_id is None: return False - info = transport.info( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ) - return self._info_acknowledges_peer_cancel(info, task) + return self._info_acknowledges_peer_cancel( + transport.info(**_session_kw(profile, session_id)), task) - def _interrupt_stopping_task( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - ) -> bool: - transport = self._transport_for(binding, task) - if transport is None: - return False - profile = task["payload"]["target_profile"] - session_id = self._resume_exact(transport, binding.room_id, profile) + def _interrupt_stopping_task(self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> bool: + transport, profile, session_id = self._open_session(binding, task) if session_id is None: # A local accepted turn cannot survive without its canonical # session: an authoritative absence is a safe Stop acknowledgement # (resolution errors raise). A peer remains uncertain instead. - return transport is self.rpc - info = transport.info( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ) + return transport is not None and transport is self.rpc + info = transport.info(**_session_kw(profile, session_id)) if not _info_active(info): # History was checked immediately before this probe. An exact # session that is no longer active cannot keep executing, and after @@ -555,69 +441,35 @@ class HostedRoomRuntime: if not _info_is_active_for(info, task["identity"], require_exact=True): return False result = transport.interrupt( - profile=profile, - session_id=session_id, - source=ROOM_SESSION_SOURCE, - expected_task_id=task["identity"].task_id, - ) + **_session_kw(profile, session_id), expected_task_id=task["identity"].task_id) if result is None: return False return ( result.get("interrupted") is True - or str(result.get("status") or "") in _STOP_ACK_STATUSES - ) + or str(result.get("status") or "") in _STOP_ACK_STATUSES) def _settle_stopping_completion( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease ) -> bool: """Publish a terminal receipt that arrived before Stop was acknowledged.""" - transport = self._transport_for(binding, task) - if transport is None: - return False - profile = task["payload"]["target_profile"] - session_id = self._resume_exact(transport, binding.room_id, profile) + transport, profile, session_id = self._open_session(binding, task) if session_id is None: return False - receipt = _find_terminal_receipt( - transport.history( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ), - task["identity"], - int(task["execution_generation"]), - ) + receipt = self._terminal_from_history(transport, profile, session_id, task) if receipt is None: return False - self._settle_stopping( - binding, - task, - lease, - settlement_id=receipt.settlement_id, - status=receipt.status, - result=receipt.result, - ) + self._settle_stopping(binding, task, lease, **asdict(receipt)) return True def _report_pending_action( - self, - task: Mapping[str, Any], - *, - session_id: str, - info: Mapping[str, Any], - ) -> None: + self, task: Mapping[str, Any], *, session_id: str, info: Mapping[str, Any]) -> None: if self.pending_action is None: return approval = info.get("pending_approval") or info.get("approval") action = None if isinstance(approval, Mapping): safe_approval = dict(approval) - choices = [ - choice - for choice in safe_approval.get("choices") or () - if choice in {"once", "deny"} - ] + choices = [c for c in safe_approval.get("choices") or () if c in {"once", "deny"}] safe_approval["choices"] = choices or ["once", "deny"] action = { "kind": "approval", @@ -626,34 +478,24 @@ class HostedRoomRuntime: "run_id": info.get("run_id"), "session_id": session_id, "request_id": safe_approval.get("request_id"), - "approval": safe_approval, - } + "approval": safe_approval} self.pending_action(task["identity"].room_id, _member_id(task), action) - def _retry_stopping_tasks( - self, binding: HostedRoomBinding, lease: state.DriverLease - ) -> bool: - pending = state.list_tasks( - self.db_path, room_id=binding.room_id, status="stopping" - ) - for task in pending: + def _retry_stopping_tasks(self, binding: HostedRoomBinding, lease: state.DriverLease) -> bool: + for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="stopping"): try: lease = self._renew_lease_if_needed(binding, lease) if self._peer_stop_acknowledged(binding, task): self._complete_cancel(task) continue - if self._settle_stopping_completion(binding, task, lease): - continue - if not self._interrupt_stopping_task(binding, task): + if not self._finish_stop(binding, task, lease): return True - self._complete_acknowledged_stop(binding, task, lease) except Exception as exc: - self._record_error(f"stop retry remains pending: {exc}") + self._record_error(_STOP_PENDING.format(exc=exc)) return True return False # ------------------------------------------------------------------ scheduling - def _worker_loop(self) -> None: try: while not self._stop.is_set(): @@ -670,11 +512,7 @@ class HostedRoomRuntime: finally: while True: with self._status_lock: - room_threads = tuple( - thread - for thread in self._room_threads.values() - if thread.is_alive() - ) + room_threads = tuple(t for t in self._room_threads.values() if t.is_alive()) if not room_threads: break for room_thread in room_threads: @@ -695,8 +533,7 @@ class HostedRoomRuntime: self._room_threads = { room_id: thread for room_id, thread in self._room_threads.items() - if thread.is_alive() - } + if thread.is_alive()} available = self.max_concurrent_rooms - len(self._room_threads) active_rooms = set(self._room_threads) if available <= 0: @@ -706,20 +543,16 @@ class HostedRoomRuntime: if not bindings: return start = self._room_schedule_cursor % len(bindings) - ordered_bindings = bindings[start:] + bindings[:start] self._room_schedule_cursor = (start + 1) % len(bindings) - for binding in ordered_bindings: + for binding in bindings[start:] + bindings[:start]: if self._stop.is_set() or available <= 0: return if binding.room_id in active_rooms: continue room_thread = threading.Thread( - target=self._run_room_once, - args=(binding,), - name=f"hosted-room-{binding.room_id[:24]}", - daemon=True, - ) + target=self._run_room_once, args=(binding,), + name=f"hosted-room-{binding.room_id[:24]}", daemon=True) with self._status_lock: self._room_threads[binding.room_id] = room_thread active_rooms.add(binding.room_id) @@ -753,9 +586,7 @@ class HostedRoomRuntime: self._inspect_abandoned_attempts(binding) deferred_until = self._ambiguous_rooms.get(binding.room_id) if deferred_until is not None: - running = state.list_tasks( - self.db_path, room_id=binding.room_id, status="running" - ) + running = state.list_tasks(self.db_path, room_id=binding.room_id, status="running") if running and self.clock() < deferred_until: return self._ambiguous_rooms.pop(binding.room_id, None) @@ -770,27 +601,19 @@ class HostedRoomRuntime: if self._reconcile_indeterminate(binding, lease): return - queued = state.list_tasks( - self.db_path, room_id=binding.room_id, status="queued" - ) - for task in queued: + for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="queued"): if self._stop.is_set() or self._route_retry_is_deferred(task): return lease = self._renew_lease_if_needed(binding, lease) attempt = state.start_task( - self.db_path, - task["identity"], - lease, - expected_cancel_generation=task["cancel_generation"], - clock=self.clock, - ) + self.db_path, task["identity"], lease, + expected_cancel_generation=task["cancel_generation"], clock=self.clock) self._execute_attempt(binding, task, attempt) current = state.get_task(self.db_path, task["identity"]) if current["status"] not in state.TERMINAL_STATUSES: return # ------------------------------------------------------------------ route retry backoff - @staticmethod def _route_retry_key(task: Mapping[str, Any]) -> tuple[str, str]: return task["identity"].room_id, _member_id(task) @@ -807,20 +630,15 @@ class HostedRoomRuntime: if previous is None else min( self.unavailable_retry_max_seconds, - max(self.unavailable_retry_min_seconds, previous["delay"] * 2), - ) - ) + max(self.unavailable_retry_min_seconds, previous["delay"] * 2))) self._unavailable_route_retries[key] = { - "delay": delay, - "next_attempt_at": self.clock() + delay, - } + "delay": delay, "next_attempt_at": self.clock() + delay} return delay def _clear_unavailable_route_retry(self, task: Mapping[str, Any]) -> None: self._unavailable_route_retries.pop(self._route_retry_key(task), None) # ------------------------------------------------------------------ leases - def _ensure_lease(self, binding: HostedRoomBinding) -> state.DriverLease: with self._status_lock: current = self._leases.get(binding.room_id) @@ -831,35 +649,24 @@ class HostedRoomRuntime: self._drop_lease(binding.room_id) lease = state.acquire_lease( - self.db_path, - room_id=binding.room_id, - gateway_id=binding.gateway_id, - authority_epoch=binding.authority_epoch, - process_generation=self.process_generation, - ttl_seconds=self.lease_ttl_seconds, - clock=self.clock, - ) + self.db_path, room_id=binding.room_id, gateway_id=binding.gateway_id, + authority_epoch=binding.authority_epoch, process_generation=self.process_generation, + ttl_seconds=self.lease_ttl_seconds, clock=self.clock) with self._status_lock: self._leases[binding.room_id] = lease self._recovered_leases = { - key for key in self._recovered_leases if key[0] != binding.room_id - } + key for key in self._recovered_leases if key[0] != binding.room_id} return lease def _renew_lease_if_needed( - self, - binding: HostedRoomBinding, - lease: state.DriverLease, - *, - force: bool = False, + self, binding: HostedRoomBinding, lease: state.DriverLease, *, force: bool = False ) -> state.DriverLease: del binding renew_at = lease.expires_at - (self.lease_ttl_seconds / 2) if not force and self.clock() < renew_at: return lease renewed = state.renew_lease( - self.db_path, lease, ttl_seconds=self.lease_ttl_seconds, clock=self.clock - ) + self.db_path, lease, ttl_seconds=self.lease_ttl_seconds, clock=self.clock) with self._status_lock: self._leases[lease.room_id] = renewed return renewed @@ -877,12 +684,8 @@ class HostedRoomRuntime: self._drop_lease(room_id) # ------------------------------------------------------------------ attempt execution - def _execute_attempt( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - attempt: state.TaskAttempt, + self, binding: HostedRoomBinding, task: Mapping[str, Any], attempt: state.TaskAttempt ) -> None: profile = task["payload"]["target_profile"] transport = self._transport_for(binding, task) @@ -896,63 +699,40 @@ class HostedRoomRuntime: # after it, but an unexpected exception at that boundary is # still ambiguous. Never terminalize it as a proven failure. submit_attempted = True + session_id = _session_id(session) deadline_monotonic = time.monotonic() + self.turn_timeout_seconds transport.submit( - profile=profile, - session_id=_session_id(session), - prompt=task["payload"]["prompt"], - source=ROOM_SESSION_SOURCE, - task=attempt.identity, - execution_generation=attempt.execution_generation, - on_terminal=lambda receipt: self._on_terminal( - binding, attempt, receipt - ), - ) + **_session_kw(profile, session_id), prompt=task["payload"]["prompt"], + task=attempt.identity, execution_generation=attempt.execution_generation, + on_terminal=lambda receipt: self._on_terminal(binding, attempt, receipt)) self._clear_unavailable_route_retry(task) receipt = self._wait_for_terminal( - binding, - profile=profile, - session_id=_session_id(session), - attempt=attempt, - transport=transport, - deadline_monotonic=deadline_monotonic, - ) + binding, profile=profile, session_id=session_id, attempt=attempt, + transport=transport, deadline_monotonic=deadline_monotonic) if receipt is None: return - state.settle_task( - self.db_path, - attempt, - settlement_id=receipt.settlement_id, - status=receipt.status, - result=receipt.result, - clock=self.clock, - ) + state.settle_task(self.db_path, attempt, **asdict(receipt), clock=self.clock) except (state.StaleLeaseError, state.StaleTaskError) as exc: self._drop_lease(binding.room_id) self._record_error(f"task {attempt.identity.task_id} fenced: {exc}") except Exception as exc: if submit_attempted and bool(getattr(exc, "not_admitted", False)): try: - state.requeue_not_admitted_task( - self.db_path, attempt, clock=self.clock - ) + state.requeue_not_admitted_task(self.db_path, attempt, clock=self.clock) except (state.StaleLeaseError, state.StaleTaskError) as fence_exc: self._mark_ambiguous(binding, attempt) self._record_error( f"task {attempt.identity.task_id} not-admitted proof lost " - f"its fence: {fence_exc}" - ) + f"its fence: {fence_exc}") else: delay = self._defer_unavailable_route(task) self._record_error( f"task {attempt.identity.task_id} was not admitted; " - f"queued for retry in {delay:g}s" - ) + f"queued for retry in {delay:g}s") elif submit_attempted: self._mark_ambiguous(binding, attempt) self._record_error( - f"task {attempt.identity.task_id} observation failed after submit: {exc}" - ) + f"task {attempt.identity.task_id} observation failed after submit: {exc}") else: self._settle_failure_if_current(attempt, exc) finally: @@ -964,56 +744,33 @@ class HostedRoomRuntime: # thread leaves; idle room scans never set this marker. self._rooms_needing_reschedule.add(binding.room_id) - def _mark_ambiguous( - self, binding: HostedRoomBinding, attempt: state.TaskAttempt - ) -> None: + def _mark_ambiguous(self, binding: HostedRoomBinding, attempt: state.TaskAttempt) -> None: self._drop_lease(binding.room_id) self._ambiguous_rooms[binding.room_id] = attempt.lease.expires_at def _on_terminal( - self, - binding: HostedRoomBinding, - attempt: state.TaskAttempt, - receipt: Mapping[str, Any], + self, binding: HostedRoomBinding, attempt: state.TaskAttempt, receipt: Mapping[str, Any] ) -> None: """Durably commit one in-process terminal receipt for ``attempt``.""" status = receipt.get("status") if status == "cancelled": self.wakeup() return - terminal_status: state.TerminalStatus = ( - "settled" if status == "settled" else "failed" - ) - settlement_id = ( - receipt.get("settlement_id") - or f"reply:{attempt.identity.task_id}:{attempt.execution_generation}" - ) - result = _bounded_terminal_result(receipt) + terminal = _TerminalReceipt( + status="settled" if status == "settled" else "failed", + settlement_id=receipt.get("settlement_id") + or f"reply:{attempt.identity.task_id}:{attempt.execution_generation}", + result=_bounded_terminal_result(receipt)) try: - settled = state.settle_task( - self.db_path, - attempt, - settlement_id=settlement_id, - status=terminal_status, - result=result, - clock=self.clock, - ) + settled = state.settle_task(self.db_path, attempt, **asdict(terminal), clock=self.clock) self._publish(binding, settled) except state.StaleTaskError: - try: + with suppress(state.StaleLeaseError, state.StaleTaskError): current = state.get_task(self.db_path, attempt.identity) if current["status"] == "stopping": self._settle_stopping( - binding, - current, - attempt.lease, - settlement_id=settlement_id, - status=terminal_status, - result=result, - expected_execution_generation=attempt.execution_generation, - ) - except (state.StaleLeaseError, state.StaleTaskError): - pass + binding, current, attempt.lease, **asdict(terminal), + expected_execution_generation=attempt.execution_generation) except state.StaleLeaseError: # Cancellation, disband, or authority transfer won the durable # race. The model result is intentionally discarded; never turn a @@ -1023,20 +780,12 @@ class HostedRoomRuntime: # A malformed terminal receipt must not escape the callback and # hold the profile lock until the deadline. self._settle_failure_if_current( - attempt, - RuntimeError(f"terminal result could not be committed: {exc}"), - ) + attempt, RuntimeError(f"terminal result could not be committed: {exc}")) self.wakeup() def _wait_for_terminal( - self, - binding: HostedRoomBinding, - *, - profile: str, - session_id: str, - attempt: state.TaskAttempt, - transport: InternalSessionRPC, - deadline_monotonic: float, + self, binding: HostedRoomBinding, *, profile: str, session_id: str, + attempt: state.TaskAttempt, transport: InternalSessionRPC, deadline_monotonic: float, ) -> _TerminalReceipt | None: lease = attempt.lease while not self._stop.is_set(): @@ -1046,13 +795,10 @@ class HostedRoomRuntime: if task["status"] == "stopping": try: lease = self._renew_lease_if_needed(binding, lease) - if self._settle_stopping_completion(binding, task, lease): - return None - if self._interrupt_stopping_task(binding, task): - self._complete_acknowledged_stop(binding, task, lease) + if self._finish_stop(binding, task, lease): return None except Exception as exc: - self._record_error(f"stop retry remains pending: {exc}") + self._record_error(_STOP_PENDING.format(exc=exc)) self._wake.wait(self.active_poll_interval_seconds) self._wake.clear() continue @@ -1062,19 +808,11 @@ class HostedRoomRuntime: return None lease = self._renew_lease_if_needed(binding, lease) - receipt = _find_terminal_receipt( - transport.history( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ), - attempt.identity, - attempt.execution_generation, - ) + receipt = self._terminal_from_history(transport, profile, session_id, task) if receipt is not None: return receipt - info = transport.info( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ) + info = transport.info(**_session_kw(profile, session_id)) self._report_pending_action(task, session_id=session_id, info=info) remaining = max(0.0, deadline_monotonic - time.monotonic()) self._wake.wait(min(self.active_poll_interval_seconds, remaining)) @@ -1082,51 +820,33 @@ class HostedRoomRuntime: return None # ------------------------------------------------------------------ deadline stops - @staticmethod def _is_deadline_stop(task: Mapping[str, Any]) -> bool: return str(task.get("cancel_id") or "").startswith("deadline:") def _complete_acknowledged_stop( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease ) -> dict[str, Any]: """Terminalize an acknowledged Stop: deadline stops publish an explicit failure.""" if not self._is_deadline_stop(task): return self._complete_cancel(task) return self._settle_stopping( - binding, - task, - lease, - settlement_id=f"deadline:{int(task['execution_generation'])}", - status="failed", + binding, task, lease, + settlement_id=f"deadline:{int(task['execution_generation'])}", status="failed", result={ - "error": ( - "This Group Chat turn exceeded its configured time limit and " - "was stopped." - ), + "error": "This Group Chat turn exceeded its configured time limit and was stopped.", "reason_code": "turn_deadline_exceeded", - "timeout_seconds": self.turn_timeout_seconds, - }, - ) + "timeout_seconds": self.turn_timeout_seconds}) def _expire_attempt_deadline( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - lease: state.DriverLease, + self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease ) -> None: """Fence, stop, and terminalize one exact attempt at its deadline.""" if task["status"] == "running": task = state.begin_task_cancel( - self.db_path, - task["identity"], + self.db_path, task["identity"], cancel_id=f"deadline:{int(task['execution_generation'])}", - expected_cancel_generation=int(task["cancel_generation"]), - clock=self.clock, - ) + expected_cancel_generation=int(task["cancel_generation"]), clock=self.clock) elif task["status"] != "stopping": return @@ -1134,30 +854,20 @@ class HostedRoomRuntime: if not self._is_deadline_stop(task): return lease = self._renew_lease_if_needed(binding, lease, force=True) - if self._settle_stopping_completion(binding, task, lease): - return - if self._interrupt_stopping_task(binding, task): - self._complete_acknowledged_stop(binding, task, lease) + if self._finish_stop(binding, task, lease): return self._record_error( - f"task {task['identity'].task_id} exceeded its deadline; stop remains pending" - ) + f"task {task['identity'].task_id} exceeded its deadline; stop remains pending") # ------------------------------------------------------------------ recovery - def _inspect_abandoned_attempts(self, binding: HostedRoomBinding) -> None: - running = state.list_tasks( - self.db_path, room_id=binding.room_id, status="running" - ) - for task in running: + for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="running"): if task["run_process_generation"] == self.process_generation: continue - transport = self._transport_for(binding, task) inspection = ( self._inspect_local_recovery_session(task) - if transport is self.rpc - else self._inspect_recovery_session(binding, task) - ) + if self._transport_for(binding, task) is self.rpc + else self._inspect_recovery_session(binding, task)) if inspection.terminal is not None: self._harvest_previous_attempt(binding, task, inspection.terminal) elif inspection.active: @@ -1165,52 +875,31 @@ class HostedRoomRuntime: # lease or submit a duplicate prompt. raise state.LeaseHeldError("recovered session turn is still active") - def _inspection_from_info( - self, - task: Mapping[str, Any], - *, - session_id: str, - info: Mapping[str, Any], - terminal: _TerminalReceipt | None, - ) -> _RecoveryInspection: + def _inspect_session( + self, transport: InternalSessionRPC, task: Mapping[str, Any], session_id: str, + *, read_history: bool) -> _RecoveryInspection: + """Probe one resolved session: optional terminal receipt from history, then live info.""" + profile = task["payload"]["target_profile"] + receipt = None + if read_history: + receipt = self._terminal_from_history(transport, profile, session_id, task) + info = transport.info(**_session_kw(profile, session_id)) self._report_pending_action(task, session_id=session_id, info=info) return _RecoveryInspection( - terminal=terminal, - active=_info_is_active_for(info, task["identity"]), - status=str(info.get("status") or "") or None, - ) + terminal=receipt, active=_info_is_active_for(info, task["identity"]), + status=str(info.get("status") or "") or None) def _inspect_recovery_session( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - ) -> _RecoveryInspection: + self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> _RecoveryInspection: profile = task["payload"]["target_profile"] transport = self._transport_for(binding, task) with self.turn_lock(profile): - session_id = self._resume_exact( - transport, task["identity"].room_id, profile - ) + session_id = self._resume_exact(transport, task["identity"].room_id, profile) if session_id is None: return _NO_INSPECTION - receipt = _find_terminal_receipt( - transport.history( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ), - task["identity"], - task["execution_generation"], - ) - info = transport.info( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ) - return self._inspection_from_info( - task, session_id=session_id, info=info, terminal=receipt - ) + return self._inspect_session(transport, task, session_id, read_history=True) - def _inspect_local_recovery_session( - self, - task: Mapping[str, Any], - ) -> _RecoveryInspection: + def _inspect_local_recovery_session(self, task: Mapping[str, Any]) -> _RecoveryInspection: """Check only live process state before explicit local recovery. A restart loses the in-process terminal callback identity. Session @@ -1222,28 +911,15 @@ class HostedRoomRuntime: profile = task["payload"]["target_profile"] with self.turn_lock(profile): session = self.rpc.resolve_exact( - profile=profile, - title=room_session_title(task["identity"].room_id), - source=ROOM_SESSION_SOURCE, - ) + profile=profile, title=room_session_title(task["identity"].room_id), + source=ROOM_SESSION_SOURCE) if session is None: return _NO_INSPECTION - session_id = _session_id(session) - info = self.rpc.info( - profile=profile, session_id=session_id, source=ROOM_SESSION_SOURCE - ) - return self._inspection_from_info( - task, session_id=session_id, info=info, terminal=None - ) + return self._inspect_session(self.rpc, task, _session_id(session), read_history=False) def _reconcile_indeterminate( - self, - binding: HostedRoomBinding, - lease: state.DriverLease, - ) -> bool: - unresolved = state.list_tasks( - self.db_path, room_id=binding.room_id, status="indeterminate" - ) + self, binding: HostedRoomBinding, lease: state.DriverLease) -> bool: + unresolved = state.list_tasks(self.db_path, room_id=binding.room_id, status="indeterminate") if not unresolved: self._set_blocked(binding.room_id, False) return False @@ -1251,10 +927,8 @@ class HostedRoomRuntime: for task in unresolved: generation = int(task["execution_generation"]) attempt_key = (binding.room_id, task["identity"].task_id, generation) - if ( - self._transport_for(binding, task) is self.rpc - and attempt_key not in inspected - ): + is_local = self._transport_for(binding, task) is self.rpc + if is_local and attempt_key not in inspected: inspection = self._inspect_local_recovery_session(task) inspected.add(attempt_key) if inspection.terminal is not None: @@ -1268,8 +942,7 @@ class HostedRoomRuntime: task.get("indeterminate_at") or task.get("updated_at") or task.get("created_at") - or self.clock() - ) + or self.clock()) deadline = deferred_at + self.indeterminate_defer_seconds inspection = _NO_INSPECTION if attempt_key not in inspected or self.clock() >= deadline: @@ -1278,8 +951,7 @@ class HostedRoomRuntime: inspection = self._inspect_recovery_session(binding, task) except Exception as exc: self._record_error( - f"task {task['identity'].task_id} recovery probe failed: {exc}" - ) + f"task {task['identity'].task_id} recovery probe failed: {exc}") inspected.add(attempt_key) if inspection.status == "cancelled": # Remote-probe resolutions are not republished here. @@ -1288,73 +960,43 @@ class HostedRoomRuntime: continue if inspection.terminal is not None: self._resolve_indeterminate( - binding, task, lease, inspection.terminal, publish=False - ) + binding, task, lease, inspection.terminal, publish=False) inspected.discard(attempt_key) continue if self.clock() < deadline: self._set_blocked(binding.room_id, True) return True - deferred = state.defer_indeterminate_task( - self.db_path, - task["identity"], - lease, - expected_execution_generation=generation, - expected_cancel_generation=task["cancel_generation"], - reason="member_unavailable", - clock=self.clock, - ) + deferred = self._fenced( + state.defer_indeterminate_task, None, task, lease, reason="member_unavailable") inspected.discard(attempt_key) self._publish(binding, deferred) self._set_blocked(binding.room_id, False) return False def _harvest_previous_attempt( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - receipt: _TerminalReceipt, + self, binding: HostedRoomBinding, task: Mapping[str, Any], receipt: _TerminalReceipt ) -> None: previous_attempt = state.TaskAttempt( identity=task["identity"], lease=state.DriverLease( - room_id=binding.room_id, - gateway_id=task["run_gateway_id"], + room_id=binding.room_id, gateway_id=task["run_gateway_id"], authority_epoch=binding.authority_epoch, process_generation=task["run_process_generation"], - lease_generation=task["run_lease_generation"], - expires_at=0.0, - ), + lease_generation=task["run_lease_generation"], expires_at=0.0), execution_generation=task["execution_generation"], - cancel_generation=task["cancel_generation"], - ) - try: - state.settle_task( - self.db_path, - previous_attempt, - settlement_id=receipt.settlement_id, - status=receipt.status, - result=receipt.result, - clock=self.clock, - ) - except (state.StaleLeaseError, state.StaleTaskError): - # Once the previous proof has expired there is deliberately no - # "trust this historical output" escape hatch; fenced recovery - # leaves the task indeterminate for explicit user action. - return + cancel_generation=task["cancel_generation"]) + # Once the previous proof has expired there is deliberately no "trust + # this historical output" escape hatch; fenced recovery leaves the task + # indeterminate for explicit user action. + with suppress(state.StaleLeaseError, state.StaleTaskError): + state.settle_task(self.db_path, previous_attempt, **asdict(receipt), clock=self.clock) # ------------------------------------------------------------------ misc - def _binding_for_room(self, room_id: str) -> HostedRoomBinding | None: - return next( - (b for b in self._rooms_provider() if b.room_id == room_id), None - ) + return next((b for b in self._rooms_provider() if b.room_id == room_id), None) def _transport_for( - self, - binding: HostedRoomBinding, - task: Mapping[str, Any], - ) -> InternalSessionRPC: + self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> InternalSessionRPC: if self.transport_resolver is not None: return self.transport_resolver(binding, task) if self.rpc is None: @@ -1362,37 +1004,20 @@ class HostedRoomRuntime: return self.rpc def _resolve_or_create( - self, - transport: InternalSessionRPC, - profile: str, - room_id: str, - ) -> Mapping[str, Any]: - title = room_session_title(room_id) - session = transport.resolve_exact( - profile=profile, title=title, source=ROOM_SESSION_SOURCE - ) + self, transport: InternalSessionRPC, profile: str, room_id: str) -> Mapping[str, Any]: + coords = { + "profile": profile, "title": room_session_title(room_id), "source": ROOM_SESSION_SOURCE} + session = transport.resolve_exact(**coords) if session is None: - return transport.create( - profile=profile, title=title, source=ROOM_SESSION_SOURCE - ) - return transport.resume( - profile=profile, session_id=_session_id(session), source=ROOM_SESSION_SOURCE - ) + return transport.create(**coords) + return transport.resume(**_session_kw(profile, _session_id(session))) - def _settle_failure_if_current( - self, attempt: state.TaskAttempt, exc: Exception - ) -> None: - try: + def _settle_failure_if_current(self, attempt: state.TaskAttempt, exc: Exception) -> None: + with suppress(state.DriverStateError, state.RoomUnavailableError): state.settle_task( - self.db_path, - attempt, + self.db_path, attempt, settlement_id=f"failure:{attempt.identity.task_id}:{attempt.execution_generation}", - status="failed", - result={"error": str(exc)}, - clock=self.clock, - ) - except (state.DriverStateError, state.RoomUnavailableError): - pass + status="failed", result={"error": str(exc)}, clock=self.clock) self._record_error(f"task {attempt.identity.task_id} failed: {exc}") def _record_error(self, message: str) -> None: @@ -1433,22 +1058,17 @@ def _truncate_utf8(value: Any, *, max_bytes: int) -> tuple[str, bool]: def _bounded_terminal_result(receipt: Mapping[str, Any]) -> dict[str, Any]: - text, truncated = _truncate_utf8( - receipt.get("text", ""), max_bytes=MAX_TERMINAL_TEXT_BYTES - ) + text, truncated = _truncate_utf8(receipt.get("text", ""), max_bytes=MAX_TERMINAL_TEXT_BYTES) error, error_truncated = _truncate_utf8(receipt.get("error", ""), max_bytes=4096) return { "message_id": receipt.get("message_id"), "text": text, **({"error": error} if error else {}), - **({"truncated": True} if truncated or error_truncated else {}), - } + **({"truncated": True} if truncated or error_truncated else {})} def _find_terminal_receipt( - history: Sequence[Mapping[str, Any]], - identity: state.TaskIdentity, - execution_generation: int, + history: Sequence[Mapping[str, Any]], identity: state.TaskIdentity, execution_generation: int ) -> _TerminalReceipt | None: for message in reversed(history): status = message.get("status") @@ -1456,8 +1076,7 @@ def _find_terminal_receipt( message.get("task_id") != identity.task_id or message.get("execution_generation") != execution_generation or message.get("role") != "assistant" - or status not in {"settled", "failed"} - ): + or status not in {"settled", "failed"}): continue receipt_id = message.get("message_id") if not isinstance(receipt_id, str) or not receipt_id: @@ -1466,9 +1085,7 @@ def _find_terminal_receipt( status=cast(state.TerminalStatus, status), settlement_id=receipt_id, result=_bounded_terminal_result( - {"message_id": receipt_id, "text": message.get("content", "")} - ), - ) + {"message_id": receipt_id, "text": message.get("content", "")})) return None @@ -1477,11 +1094,7 @@ def _info_active(info: Mapping[str, Any]) -> bool: def _info_is_active_for( - info: Mapping[str, Any], - identity: state.TaskIdentity, - *, - require_exact: bool = False, -) -> bool: + info: Mapping[str, Any], identity: state.TaskIdentity, *, require_exact: bool = False) -> bool: if not _info_active(info): return False active_task_id = info.get("task_id") diff --git a/tui_gateway/hosted_room_peer_http.py b/tui_gateway/hosted_room_peer_http.py index 53a9f56c22..2d48e7903a 100644 --- a/tui_gateway/hosted_room_peer_http.py +++ b/tui_gateway/hosted_room_peer_http.py @@ -13,14 +13,12 @@ import urllib.error import urllib.parse import urllib.request from collections.abc import Callable, Mapping, Sequence +from contextlib import suppress from pathlib import Path from typing import Any, NoReturn from gateway.hosted_room_peer import ( - GatewayRoomCatalog, - HostedMemberDispatch, - validate_room_link_url, -) + GatewayRoomCatalog, HostedMemberDispatch, validate_room_link_url) logger = logging.getLogger(__name__) @@ -29,8 +27,7 @@ logger = logging.getLogger(__name__) _NOT_ADMITTED_ERRNOS = frozenset( value for name in ("ECONNREFUSED", "ENETDOWN", "ENETUNREACH", "EHOSTDOWN", "EHOSTUNREACH") - if (value := getattr(errno, name, None)) is not None -) + if (value := getattr(errno, name, None)) is not None) _ERROR_CODE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") # A replay page may legitimately contain many bounded 64 KiB room events. Keep # enough room for the largest normal page while preventing peer-sized responses @@ -41,14 +38,8 @@ _PEER_RESPONSE_CHUNK_BYTES = 64 * 1024 # Receipt fields that must match between a stored receipt and a dispatch. _RECEIPT_SCOPE_FIELDS = ( - "room_id", - "home_install_id", - "authority_gateway_id", - "authority_epoch", - "member_id", - "target_install_id", - "target_profile", -) + "room_id", "home_install_id", "authority_gateway_id", "authority_epoch", + "member_id", "target_install_id", "target_profile") _TERMINAL_RUN_STATES = frozenset({"completed", "failed", "interrupted", "cancelled"}) _ACTIVE_RUN_STATES = frozenset({"queued", "running", "waiting_for_approval", "stopping"}) _KNOWN_RUN_STATES = _TERMINAL_RUN_STATES | _ACTIVE_RUN_STATES @@ -57,23 +48,23 @@ _KNOWN_RUN_STATES = _TERMINAL_RUN_STATES | _ACTIVE_RUN_STATES _LEGACY_DISPATCH_MESSAGE_CODES = ( ("room grant", "invalid_room_grant"), ("capability catalog changed", "room_capability_catalog_changed"), - ("execution policy changed", "room_execution_policy_changed"), -) -_REAUTHORIZATION_CODES = frozenset({ - "invalid_room_grant", - "room_capability_catalog_changed", - "room_execution_policy_changed", - "room_reauthorization_required", -}) + ("execution policy changed", "room_execution_policy_changed")) +_GRANT_RENEWAL_CODES = frozenset({"invalid_room_grant", "room_reauthorization_required"}) # (error_code, human message) for the two digest checks after a grant refresh. _EXECUTION_POLICY_CHANGED = ( - "room_execution_policy_changed", - "peer room execution policy needs reauthorization", -) + "room_execution_policy_changed", "peer room execution policy needs reauthorization") _CAPABILITY_CHANGED = ( - "room_capability_catalog_changed", - "peer room capabilities need reauthorization", -) + "room_capability_catalog_changed", "peer room capabilities need reauthorization") +_REAUTHORIZATION_CODES = _GRANT_RENEWAL_CODES | { + _EXECUTION_POLICY_CHANGED[0], _CAPABILITY_CHANGED[0]} +# Human messages for 401/403 reauthorization codes (any other status keeps the generic text). +_REAUTHORIZATION_MESSAGES = { + **dict.fromkeys(_GRANT_RENEWAL_CODES, "peer room authorization needs renewal"), + _EXECUTION_POLICY_CHANGED[0]: _EXECUTION_POLICY_CHANGED[1], + _CAPABILITY_CHANGED[0]: _CAPABILITY_CHANGED[1]} +_BUDGET_MESSAGES = { + "size": "peer{kind} response exceeded the RoomLink size limit", + "time": "peer{kind} response exceeded the RoomLink time budget"} class _PeerResponseTooLarge(ValueError): @@ -85,19 +76,15 @@ class _PeerResponseDeadlineExceeded(TimeoutError): def _content_length(response: Any) -> int | None: - headers = getattr(response, "headers", None) - if headers is None or not hasattr(headers, "get"): - return None try: - value = int(headers.get("Content-Length")) - except (TypeError, ValueError): + value = int(response.headers.get("Content-Length")) + except (AttributeError, TypeError, ValueError): return None return value if value >= 0 else None def _set_response_socket_timeout(response: Any, remaining: float) -> None: """Best-effort urllib socket timeout tightened to the remaining budget.""" - frontier = [response] seen: set[int] = set() for _depth in range(5): @@ -108,14 +95,10 @@ def _set_response_socket_timeout(response: Any, remaining: float) -> None: seen.add(id(value)) setter = getattr(value, "settimeout", None) if callable(setter): - try: + with suppress(OSError, ValueError): setter(max(0.001, remaining)) - except (OSError, ValueError): - pass return - next_frontier.extend( - getattr(value, field, None) for field in ("fp", "raw", "_sock") - ) + next_frontier.extend(getattr(value, field, None) for field in ("fp", "raw", "_sock")) frontier = next_frontier @@ -123,7 +106,6 @@ def _read_bounded_response(response: Any, *, max_bytes: int, deadline: float) -> declared = _content_length(response) if declared is not None and declared > max_bytes: raise _PeerResponseTooLarge - reader = getattr(response, "read1", None) if not callable(reader): reader = response.read @@ -184,15 +166,10 @@ def _response_error_code(detail: str) -> str | None: def _http_error_message(method: str, path: str, status: int, error_code: str | None) -> str: - if status in {401, 403} and error_code in { - "invalid_room_grant", - "room_reauthorization_required", - }: - return "peer room authorization needs renewal" - if status == 403 and error_code == _EXECUTION_POLICY_CHANGED[0]: - return _EXECUTION_POLICY_CHANGED[1] - if status == 403 and error_code == _CAPABILITY_CHANGED[0]: - return _CAPABILITY_CHANGED[1] + renewal = status in {401, 403} and error_code in _GRANT_RENEWAL_CODES + drift = status == 403 and error_code in {_EXECUTION_POLICY_CHANGED[0], _CAPABILITY_CHANGED[0]} + if renewal or drift: + return _REAUTHORIZATION_MESSAGES[error_code] return f"peer rejected {method} {path} with HTTP {status}" @@ -200,16 +177,9 @@ class PeerRunsHTTPError(RuntimeError): """Controlled peer HTTP failure.""" def __init__( - self, - message: str, - *, - retryable: bool = False, - ambiguous: bool = False, - not_admitted: bool = False, - status_code: int | None = None, - error_code: str | None = None, - error_message: str | None = None, - ) -> None: + self, message: str, *, retryable: bool = False, ambiguous: bool = False, + not_admitted: bool = False, status_code: int | None = None, error_code: str | None = None, + error_message: str | None = None) -> None: super().__init__(message) self.retryable = retryable self.ambiguous = ambiguous @@ -217,56 +187,41 @@ class PeerRunsHTTPError(RuntimeError): self.status_code = status_code self.error_code = error_code self.error_message = error_message - self.needs_reauthorization = bool( - status_code in {401, 403} and error_code in _REAUTHORIZATION_CODES - ) - self.needs_capability_refresh = bool( - status_code == 403 and error_code == _CAPABILITY_CHANGED[0] - ) - self.needs_execution_policy_refresh = bool( - status_code == 403 and error_code == _EXECUTION_POLICY_CHANGED[0] - ) + self.needs_reauthorization = ( + status_code in {401, 403} and error_code in _REAUTHORIZATION_CODES) + self.needs_capability_refresh = status_code == 403 and error_code == _CAPABILITY_CHANGED[0] + self.needs_execution_policy_refresh = ( + status_code == 403 and error_code == _EXECUTION_POLICY_CHANGED[0]) def digest_reauthorization_error( - catalog: GatewayRoomCatalog, - *, - capability_digest: str | None, - execution_policy_digest: str | None, -) -> PeerRunsHTTPError | None: + catalog: GatewayRoomCatalog, *, capability_digest: str | None, + execution_policy_digest: str | None) -> PeerRunsHTTPError | None: """Return the fail-closed error when a refreshed catalog drifted from the frozen digests. Execution policy is checked before capabilities; a ``None`` digest skips its check. """ for expected, actual, (code, message) in ( ( - execution_policy_digest, - catalog.execution_policy.policy_digest, - _EXECUTION_POLICY_CHANGED, - ), - (capability_digest, catalog.catalog_digest, _CAPABILITY_CHANGED), - ): + execution_policy_digest, catalog.execution_policy.policy_digest, + _EXECUTION_POLICY_CHANGED), + (capability_digest, catalog.catalog_digest, _CAPABILITY_CHANGED)): if expected is not None and actual != expected: - return PeerRunsHTTPError( - message, status_code=403, error_code=code, not_admitted=True - ) + return PeerRunsHTTPError(message, status_code=403, error_code=code, not_admitted=True) return None +def _run_path(record: Mapping[str, Any], *suffix: str) -> str: + return "/".join(("/v1/runs", urllib.parse.quote(str(record["run_id"]), safe=""), *suffix)) + + class PeerRunsHTTPClient: """Drive a peer's dedicated group session via scoped async Runs APIs.""" def __init__( - self, - *, - base_url: str, - api_key: str, - timeout_seconds: float = 30, - receipt_db_path: Path | str | None = None, - poll_min_seconds: float = 0.1, - poll_max_seconds: float = 2.0, - clock: Callable[[], float] = time.monotonic, - ) -> None: + self, *, base_url: str, api_key: str, timeout_seconds: float = 30, + receipt_db_path: Path | str | None = None, poll_min_seconds: float = 0.1, + poll_max_seconds: float = 2.0, clock: Callable[[], float] = time.monotonic) -> None: base_url, self.transport_security = validate_room_link_url(base_url) if api_key and len(api_key) < 16: raise ValueError("peer API key is missing or too short") @@ -294,18 +249,9 @@ class PeerRunsHTTPClient: self.receipt_db_path = path def bind_room_scope( - self, - *, - room_id: str, - home_install_id: str, - authority_gateway_id: str, - authority_epoch: int, - member_id: str, - target_install_id: str, - target_profile: str, - ) -> None: + self, *, room_id: str, home_install_id: str, authority_gateway_id: str, + authority_epoch: int, member_id: str, target_install_id: str, target_profile: str) -> None: """Fence every in-memory and durable receipt to one room authority.""" - scope = { "room_id": str(room_id or ""), "home_install_id": str(home_install_id or ""), @@ -313,8 +259,7 @@ class PeerRunsHTTPClient: "authority_epoch": int(authority_epoch or 0), "member_id": str(member_id or ""), "target_install_id": str(target_install_id or ""), - "target_profile": str(target_profile or ""), - } + "target_profile": str(target_profile or "")} if not all(value for key, value in scope.items() if key != "authority_epoch"): raise PeerRunsHTTPError("peer room receipt scope is incomplete") if scope["authority_epoch"] < 1: @@ -329,9 +274,7 @@ class PeerRunsHTTPClient: self._terminal_receipts.clear() def _bind_dispatch_scope(self, dispatch: HostedMemberDispatch) -> None: - self.bind_room_scope( - **{field: getattr(dispatch, field) for field in _RECEIPT_SCOPE_FIELDS} - ) + self.bind_room_scope(**{field: getattr(dispatch, field) for field in _RECEIPT_SCOPE_FIELDS}) def _receipt(self, task_id: str, execution_generation: int) -> dict[str, Any] | None: """Return the in-memory receipt, falling back to the durable store.""" @@ -343,11 +286,8 @@ class PeerRunsHTTPClient: return hosted_rooms.remote_run_receipt( self.receipt_db_path, record={ - **self._room_scope, - "task_id": task_id, - "execution_generation": execution_generation, - }, - ) + **self._room_scope, "task_id": task_id, "execution_generation": execution_generation + }) def bind_observation(self, *, task_id: str, execution_generation: int) -> None: """Pin history/status reads to one exact logical task attempt.""" @@ -363,36 +303,25 @@ class PeerRunsHTTPClient: self._recovery_backoff.clear() def _request( - self, - path: str, - *, - method: str = "GET", - body: Mapping[str, Any] | None = None, - headers: Mapping[str, str] | None = None, - room_grant: str | None = None, - ) -> dict[str, Any]: + self, path: str, *, method: str = "GET", body: Mapping[str, Any] | None = None, + headers: Mapping[str, str] | None = None, room_grant: str | None = None) -> dict[str, Any]: from hermes_cli.urllib_security import open_credentialed_url deadline = time.monotonic() + self.timeout_seconds ambiguous = method == "POST" - request_headers = { - "Authorization": ( - f"HermesRoom {room_grant}" if room_grant else f"Bearer {self.api_key}" - ), - "Content-Type": "application/json", - "User-Agent": "Hermes-RoomLink/1.0", - **(headers or {}), - } request = urllib.request.Request( f"{self.base_url}{path}", data=( json.dumps(body, separators=(",", ":")).encode("utf-8") if body is not None - else None - ), + else None), method=method, - headers=request_headers, - ) + headers={ + "Authorization": ( + f"HermesRoom {room_grant}" if room_grant else f"Bearer {self.api_key}"), + "Content-Type": "application/json", + "User-Agent": "Hermes-RoomLink/1.0", + **(headers or {})}) try: with open_credentialed_url(request, timeout=self.timeout_seconds) as response: raw = _read_bounded_response( @@ -400,23 +329,19 @@ class PeerRunsHTTPClient: ).decode("utf-8", "replace") except _PeerResponseTooLarge as exc: raise PeerRunsHTTPError( - "peer response exceeded the RoomLink size limit", ambiguous=ambiguous + _BUDGET_MESSAGES["size"].format(kind=""), ambiguous=ambiguous, ) from exc except _PeerResponseDeadlineExceeded as exc: raise PeerRunsHTTPError( - "peer response exceeded the RoomLink time budget", - retryable=True, - ambiguous=ambiguous, + _BUDGET_MESSAGES["time"].format(kind=""), retryable=True, ambiguous=ambiguous ) from exc except urllib.error.HTTPError as exc: self._raise_http_error(exc, method=method, path=path, deadline=deadline) except (urllib.error.URLError, TimeoutError, OSError) as exc: not_admitted = ambiguous and _is_proven_pre_admission_failure(exc) raise PeerRunsHTTPError( - "peer RoomLink endpoint is unreachable", - retryable=True, - ambiguous=ambiguous and not not_admitted, - not_admitted=not_admitted, + "peer RoomLink endpoint is unreachable", retryable=True, + ambiguous=ambiguous and not not_admitted, not_admitted=not_admitted, ) from exc try: payload = json.loads(raw) @@ -428,134 +353,89 @@ class PeerRunsHTTPClient: @staticmethod def _raise_http_error( - exc: urllib.error.HTTPError, *, method: str, path: str, deadline: float - ) -> NoReturn: + exc: urllib.error.HTTPError, *, method: str, path: str, deadline: float) -> NoReturn: """Raise the classified PeerRunsHTTPError for an HTTP error response.""" - ambiguous = method == "POST" and exc.code >= 500 # A 4xx on admission proves the peer never admitted the run. - pre_admission = method == "POST" and path == "/v1/runs" and 400 <= exc.code < 500 - flags = {"ambiguous": ambiguous, "not_admitted": pre_admission, "status_code": exc.code} + flags = { + "ambiguous": method == "POST" and exc.code >= 500, + "not_admitted": method == "POST" and path == "/v1/runs" and 400 <= exc.code < 500, + "status_code": exc.code} try: detail = _read_bounded_response( exc, max_bytes=MAX_PEER_ERROR_RESPONSE_BYTES, deadline=deadline ).decode("utf-8", "replace")[:500] except _PeerResponseTooLarge as body_exc: raise PeerRunsHTTPError( - "peer error response exceeded the RoomLink size limit", **flags + _BUDGET_MESSAGES["size"].format(kind=" error"), **flags, ) from body_exc except _PeerResponseDeadlineExceeded as body_exc: raise PeerRunsHTTPError( - "peer error response exceeded the RoomLink time budget", - retryable=True, - **flags, + _BUDGET_MESSAGES["time"].format(kind=" error"), retryable=True, **flags ) from body_exc except Exception: detail = "" error_code = _response_error_code(detail) logger.debug( - "Peer RoomLink request returned HTTP %s (%s)", exc.code, error_code or "no-code" - ) + "Peer RoomLink request returned HTTP %s (%s)", exc.code, error_code or "no-code") raise PeerRunsHTTPError( _http_error_message(method, path, exc.code, error_code), retryable=exc.code in {408, 425, 429} or exc.code >= 500, - error_code=error_code, - **flags, + error_code=error_code, **flags, ) from exc def prepare( - self, - *, - room_id: str, - profile: str, - source: str, - grant: str, - create: bool, - expected_session_id: str | None = None, - ) -> Mapping[str, Any] | None: + self, *, room_id: str, profile: str, source: str, grant: str, create: bool, + expected_session_id: str | None = None) -> Mapping[str, Any] | None: if source != "bot_room": raise PeerRunsHTTPError("peer room source must be bot_room") self._require_room_grant(grant) logical_session = ( - "roomlink_" - + hashlib.sha256(f"{room_id}\0{profile}".encode("utf-8")).hexdigest()[:32] - ) + "roomlink_" + hashlib.sha256(f"{room_id}\0{profile}".encode("utf-8")).hexdigest()[:32]) if expected_session_id and expected_session_id != logical_session: raise PeerRunsHTTPError("peer room session identity changed") - return { - "session_id": logical_session, - "title": f"Group: {room_id}", - "source": source, - } + return {"session_id": logical_session, "title": f"Group: {room_id}", "source": source} - def _checked_dispatch( - self, dispatch: Mapping[str, Any], grant: str - ) -> HostedMemberDispatch: + def _checked_dispatch(self, dispatch: Mapping[str, Any], grant: str) -> HostedMemberDispatch: """Validate a dispatch, its grant, and pin scope + observation to it.""" checked = HostedMemberDispatch.from_mapping(dispatch) self._require_room_grant(grant) self._bind_dispatch_scope(checked) self.bind_observation( - task_id=checked.task_id, - execution_generation=checked.execution_generation, - ) + task_id=checked.task_id, execution_generation=checked.execution_generation) return checked - def dispatch( - self, - *, - dispatch: Mapping[str, Any], - grant: str, - ) -> Mapping[str, Any]: - checked = self._checked_dispatch(dispatch, grant) - return self._admit_dispatch(checked, grant=grant) + def dispatch(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any]: + return self._admit_dispatch(self._checked_dispatch(dispatch, grant), grant=grant) - def recover_dispatch( - self, - *, - dispatch: Mapping[str, Any], - grant: str, - ) -> Mapping[str, Any]: + def recover_dispatch(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any]: """Recover one exact admission by receipt or idempotent POST replay.""" checked = self._checked_dispatch(dispatch, grant) - existing = self._receipt_for_dispatch(checked) + existing = self._receipt(checked.task_id, checked.execution_generation) if existing is not None: - if any( - existing[field] != getattr(checked, field) - for field in _RECEIPT_SCOPE_FIELDS - ): - raise PeerRunsHTTPError( - "peer run receipt conflicts with the recovered dispatch" - ) + if any(existing[field] != getattr(checked, field) for field in _RECEIPT_SCOPE_FIELDS): + raise PeerRunsHTTPError("peer run receipt conflicts with the recovered dispatch") return self._accepted( - checked, - run_id=str(existing["run_id"]), - session_id=str(existing["session_id"]), - replayed=True, - ) + checked, run_id=str(existing["run_id"]), + session_id=str(existing["session_id"]), replayed=True) key = (checked.task_id, checked.execution_generation) now = self.clock() backoff = self._recovery_backoff.get(key) if backoff is not None and now < float(backoff["next_attempt_at"]): raise PeerRunsHTTPError( - "peer admission recovery is backing off", - retryable=True, - ambiguous=True, - ) + "peer admission recovery is backing off", retryable=True, ambiguous=True) try: recovered = self._admit_dispatch(checked, grant=grant) except PeerRunsHTTPError as exc: if exc.retryable or exc.ambiguous: delay = self._next_poll_delay(backoff) - self._recovery_backoff = { - key: {"delay": delay, "next_attempt_at": now + delay} - } + self._recovery_backoff = {key: {"delay": delay, "next_attempt_at": now + delay}} raise self._recovery_backoff.pop(key, None) return recovered @staticmethod def _accepted( - checked: HostedMemberDispatch, *, run_id: str, session_id: str, replayed: bool + checked: HostedMemberDispatch, *, run_id: str, session_id: str, replayed: bool, ) -> dict[str, Any]: return { "status": "accepted", @@ -563,37 +443,25 @@ class PeerRunsHTTPClient: "execution_generation": checked.execution_generation, "run_id": run_id, "session_id": session_id, - "replayed": replayed, - } + "replayed": replayed} - def _admit_dispatch( - self, - checked: HostedMemberDispatch, - *, - grant: str, - ) -> Mapping[str, Any]: + def _admit_dispatch(self, checked: HostedMemberDispatch, *, grant: str) -> Mapping[str, Any]: session_id = self._session_id(checked, grant=grant) - idempotency_key = f"room:{checked.task_id}:{checked.execution_generation}" - def admit(dispatch: HostedMemberDispatch) -> dict[str, Any]: + def admit() -> dict[str, Any]: return self._request( - "/v1/runs", - method="POST", - body={ - "input": dispatch.prompt, - "hosted_room_dispatch": dispatch.as_mapping(), - }, - headers={"Idempotency-Key": idempotency_key}, - room_grant=grant, - ) + "/v1/runs", method="POST", + body={"input": checked.prompt, "hosted_room_dispatch": checked.as_mapping()}, + headers={ + "Idempotency-Key": f"room:{checked.task_id}:{checked.execution_generation}"}, + room_grant=grant) try: - result = admit(checked) + result = admit() except PeerRunsHTTPError as exc: - if exc.ambiguous: - result = admit(checked) - else: + if not exc.ambiguous: raise + result = admit() run_id = str(result.get("run_id") or "") if not run_id: raise PeerRunsHTTPError("peer did not return a run id") @@ -602,8 +470,7 @@ class PeerRunsHTTPClient: "session_id": session_id, **{field: getattr(checked, field) for field in _RECEIPT_SCOPE_FIELDS}, "task_id": checked.task_id, - "execution_generation": checked.execution_generation, - } + "execution_generation": checked.execution_generation} if self.receipt_db_path is not None: from gateway import hosted_rooms @@ -611,40 +478,29 @@ class PeerRunsHTTPClient: self._runs[(checked.task_id, checked.execution_generation)] = receipt self._status_cache.pop(run_id, None) return self._accepted( - checked, - run_id=run_id, - session_id=session_id, - replayed=bool(result.get("replayed", False)), - ) + checked, run_id=run_id, session_id=session_id, + replayed=bool(result.get("replayed", False))) def _session_id(self, dispatch: HostedMemberDispatch, *, grant: str) -> str: - existing = self._receipt_for_dispatch(dispatch) + existing = self._receipt(dispatch.task_id, dispatch.execution_generation) if existing: return str(existing["session_id"]) prepared = self.prepare( - room_id=dispatch.room_id, - profile=dispatch.target_profile, - source="bot_room", - grant=grant, - create=True, - ) + room_id=dispatch.room_id, profile=dispatch.target_profile, source="bot_room", + grant=grant, create=True) if prepared is None: raise PeerRunsHTTPError("peer room session is unavailable") return str(prepared.get("session_id") or prepared.get("id") or "") def _observation_receipt( - self, *, room_id: str, profile: str, session_id: str - ) -> dict[str, Any] | None: + self, *, room_id: str, profile: str, session_id: str) -> dict[str, Any] | None: if self._observation_key is None: return None record = self._receipt(*self._observation_key) if record is None: return None - if ( - record["room_id"] != room_id - or record["target_profile"] != profile - or record["session_id"] != session_id - ): + scope = (record["room_id"], record["target_profile"], record["session_id"]) + if scope != (room_id, profile, session_id): raise PeerRunsHTTPError("peer observation receipt changed scope") return record @@ -653,25 +509,17 @@ class PeerRunsHTTPClient: return { key: status[key] for key in ("run_id", "status", "output", "error", "approval", "last_event") - if key in status - } + if key in status} def _next_poll_delay(self, cached: Mapping[str, Any] | None) -> float: - previous = ( - float(cached["delay"]) if cached is not None else self.poll_min_seconds / 2 - ) + previous = float(cached["delay"]) if cached is not None else self.poll_min_seconds / 2 return min(self.poll_max_seconds, max(self.poll_min_seconds, previous * 2)) @staticmethod def _run_is_terminal(status: Mapping[str, Any]) -> bool: return status.get("status") in _TERMINAL_RUN_STATES - def _poll_receipt( - self, - record: Mapping[str, Any], - *, - grant: str, - ) -> dict[str, Any]: + def _poll_receipt(self, record: Mapping[str, Any], *, grant: str) -> dict[str, Any]: run_id = str(record["run_id"]) now = self.clock() cached = self._status_cache.get(run_id) @@ -685,81 +533,45 @@ class PeerRunsHTTPClient: raise error return status delay = self._next_poll_delay(cached) + entry = {"delay": delay, "next_poll_at": now + delay} try: status = self._compact_run_status( - self._request( - f"/v1/runs/{urllib.parse.quote(run_id, safe='')}", - room_grant=self._require_room_grant(grant), - ) - ) + self._request(_run_path(record), room_grant=self._require_room_grant(grant))) if ( str(status.get("run_id") or "") != run_id - or status.get("status") not in _KNOWN_RUN_STATES - ): + or status.get("status") not in _KNOWN_RUN_STATES): raise PeerRunsHTTPError("peer returned a mismatched run status") except PeerRunsHTTPError as exc: - self._status_cache = { - run_id: { - "status": cached["status"] if cached is not None else {}, - "error": exc, - "delay": delay, - "next_poll_at": now + delay, - } - } + previous = cached["status"] if cached is not None else {} + self._status_cache = {run_id: {"status": previous, "error": exc, **entry}} raise - self._status_cache = { - run_id: {"status": status, "delay": delay, "next_poll_at": now + delay} - } + self._status_cache = {run_id: {"status": status, **entry}} if self._run_is_terminal(status): self._terminal_receipts.add( - (str(record["task_id"]), int(record["execution_generation"])) - ) + (str(record["task_id"]), int(record["execution_generation"]))) return status - def _receipt_for_dispatch( - self, dispatch: HostedMemberDispatch - ) -> dict[str, Any] | None: - return self._receipt(dispatch.task_id, dispatch.execution_generation) - def history( - self, - *, - room_id: str, - profile: str, - session_id: str, - grant: str, + self, *, room_id: str, profile: str, session_id: str, grant: str, ) -> Sequence[Mapping[str, Any]]: - receipt = self._observation_receipt( - room_id=room_id, profile=profile, session_id=session_id - ) + receipt = self._observation_receipt(room_id=room_id, profile=profile, session_id=session_id) if receipt is None: return [] status = self._poll_receipt(receipt, grant=grant) state = str(status.get("status") or "") if state not in {"completed", "failed", "interrupted"}: return [] - return [ - { - "role": "assistant", - "task_id": receipt["task_id"], - "execution_generation": receipt["execution_generation"], - "status": "settled" if state == "completed" else "failed", - "message_id": f"peer-run:{status.get('run_id')}", - "content": status.get("output") or status.get("error") or "", - } - ] + return [{ + "role": "assistant", + "task_id": receipt["task_id"], + "execution_generation": receipt["execution_generation"], + "status": "settled" if state == "completed" else "failed", + "message_id": f"peer-run:{status.get('run_id')}", + "content": status.get("output") or status.get("error") or ""}] def status( - self, - *, - room_id: str, - profile: str, - session_id: str, - grant: str, - ) -> Mapping[str, Any]: - receipt = self._observation_receipt( - room_id=room_id, profile=profile, session_id=session_id - ) + self, *, room_id: str, profile: str, session_id: str, grant: str) -> Mapping[str, Any]: + receipt = self._observation_receipt(room_id=room_id, profile=profile, session_id=session_id) if receipt is None: return {"active": False, "task_id": None} status = self._poll_receipt(receipt, grant=grant) @@ -769,17 +581,10 @@ class PeerRunsHTTPClient: "execution_generation": receipt["execution_generation"], "status": status.get("status"), "run_id": status.get("run_id"), - "approval": status.get("approval"), - } + "approval": status.get("approval")} def approve_receipt( - self, - *, - task_id: str, - execution_generation: int, - request_id: str, - choice: str, - grant: str, + self, *, task_id: str, execution_generation: int, request_id: str, choice: str, grant: str ) -> Mapping[str, Any] | None: """Resolve approval for the exact durable remote run.""" record = self._receipt(task_id, execution_generation) @@ -790,74 +595,43 @@ class PeerRunsHTTPClient: raise PeerRunsHTTPError("an exact approval request_id is required") self._require_room_grant(grant) return self._post_run_action( - record, "approval", body={"choice": choice, "request_id": request_id}, grant=grant - ) + record, "approval", body={"choice": choice, "request_id": request_id}, grant=grant) def _post_run_action( self, record: Mapping[str, Any], action: str, *, body: dict[str, Any], grant: str ) -> dict[str, Any]: - run_id = str(record["run_id"]) result = self._request( - f"/v1/runs/{urllib.parse.quote(run_id, safe='')}/{action}", - method="POST", - body=body, - room_grant=grant, - ) - self._status_cache.pop(run_id, None) + _run_path(record, action), method="POST", body=body, room_grant=grant) + self._status_cache.pop(str(record["run_id"]), None) return result - def stop( - self, - *, - dispatch: Mapping[str, Any], - grant: str, - ) -> Mapping[str, Any] | None: + def stop(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any] | None: checked = HostedMemberDispatch.from_mapping(dispatch) self._bind_dispatch_scope(checked) return self.stop_receipt( - task_id=checked.task_id, - execution_generation=checked.execution_generation, - grant=grant, - ) + task_id=checked.task_id, execution_generation=checked.execution_generation, grant=grant) def stop_receipt( - self, - *, - task_id: str, - execution_generation: int, - grant: str, - ) -> Mapping[str, Any] | None: + self, *, task_id: str, execution_generation: int, grant: str) -> Mapping[str, Any] | None: """Stop the exact durable remote run after a home restart.""" record = self._receipt(task_id, execution_generation) if record is None: return None result = self._post_run_action( - record, "stop", body={}, grant=self._require_room_grant(grant) - ) + record, "stop", body={}, grant=self._require_room_grant(grant)) if self._run_is_terminal(result): self._terminal_receipts.add((str(task_id), int(execution_generation))) return result def issue_invitation( - self, - *, - room_id: str, - home_install_id: str, - authority_gateway_id: str, - authority_epoch: int, - member_id: str, - grant_id: str, - ttl_seconds: float = 3600, - status_ttl_seconds: float | None = None, - ) -> Mapping[str, Any]: + self, *, room_id: str, home_install_id: str, authority_gateway_id: str, + authority_epoch: int, member_id: str, grant_id: str, ttl_seconds: float = 3600, + status_ttl_seconds: float | None = None) -> Mapping[str, Any]: """Ask the target gateway to mint a scoped room-member grant.""" if not self.api_key: - raise PeerRunsHTTPError( - "issuing an invitation requires the target gateway API key" - ) + raise PeerRunsHTTPError("issuing an invitation requires the target gateway API key") return self._request( - "/v1/room-members/invitations", - method="POST", + "/v1/room-members/invitations", method="POST", body={ "room_id": room_id, "home_install_id": home_install_id, @@ -869,23 +643,15 @@ class PeerRunsHTTPClient: **( {"status_ttl_seconds": status_ttl_seconds} if status_ttl_seconds is not None - else {} - ), - }, - ) + else {})}) def refresh_grant( - self, - *, - grant: str, - ttl_seconds: float = 24 * 60 * 60, - capability_digest: str | None = None, - execution_policy_digest: str | None = None, + self, *, grant: str, ttl_seconds: float = 24 * 60 * 60, + capability_digest: str | None = None, execution_policy_digest: str | None = None, ) -> Mapping[str, Any]: """Renew dispatch access only while its frozen authority is unchanged.""" refreshed = self._scoped_post( - "/v1/room-members/grants/refresh", grant, body={"ttl_seconds": ttl_seconds} - ) + "/v1/room-members/grants/refresh", grant, body={"ttl_seconds": ttl_seconds}) replacement = str(refreshed.get("grant") or "") if not replacement: raise PeerRunsHTTPError("peer returned no refreshed room grant") @@ -894,9 +660,7 @@ class PeerRunsHTTPClient: probe = self.probe(grant=replacement) error = digest_reauthorization_error( GatewayRoomCatalog.from_mapping(probe.get("catalog")), - capability_digest=capability_digest, - execution_policy_digest=execution_policy_digest, - ) + capability_digest=capability_digest, execution_policy_digest=execution_policy_digest) if error is not None: raise error return {**refreshed, "catalog": probe.get("catalog")} @@ -907,12 +671,12 @@ class PeerRunsHTTPClient: def probe(self, *, grant: str) -> Mapping[str, Any]: """Verify gateway reachability and the live scoped capability catalog.""" - self._require_room_grant(grant) - return self._request("/v1/room-members/capabilities", room_grant=grant) + return self._request( + "/v1/room-members/capabilities", room_grant=self._require_room_grant(grant)) def _scoped_post(self, path: str, grant: str, *, body: dict[str, Any]) -> dict[str, Any]: - self._require_room_grant(grant) - return self._request(path, method="POST", body=body, room_grant=grant) + return self._request( + path, method="POST", body=body, room_grant=self._require_room_grant(grant)) @staticmethod def _require_room_grant(grant: str) -> str: diff --git a/tui_gateway/hosted_room_peer_transport.py b/tui_gateway/hosted_room_peer_transport.py index 6040b8f10f..1748fe5bc7 100644 --- a/tui_gateway/hosted_room_peer_transport.py +++ b/tui_gateway/hosted_room_peer_transport.py @@ -17,11 +17,7 @@ from typing import Any, Protocol from gateway.hosted_room_driver import TaskIdentity from gateway.hosted_room_peer import HostedMemberDispatch, PROTOCOL_VERSION from tui_gateway.hosted_room_driver import ( - ROOM_SESSION_SOURCE, - HostedRoomBinding, - InternalSessionRPC, - room_session_title, -) + ROOM_SESSION_SOURCE, HostedRoomBinding, InternalSessionRPC, room_session_title) class HostedRoomPeerClient(Protocol): @@ -30,13 +26,7 @@ class HostedRoomPeerClient(Protocol): def bind_room_scope(self, **scope: Any) -> None: ... def prepare( - self, - *, - room_id: str, - profile: str, - source: str, - grant: str, - create: bool, + self, *, room_id: str, profile: str, source: str, grant: str, create: bool, expected_session_id: str | None = None, ) -> Mapping[str, Any] | None: ... @@ -71,12 +61,8 @@ class FailoverHostedRoomPeerClient: """Try alternate links without changing target or logical task identity.""" def __init__( - self, - candidates: Sequence[RoomLinkCandidate], - *, - reprobe_interval_seconds: float = 60, - clock: Callable[[], float] = time.monotonic, - ) -> None: + self, candidates: Sequence[RoomLinkCandidate], *, reprobe_interval_seconds: float = 60, + clock: Callable[[], float] = time.monotonic) -> None: if not candidates: raise ValueError("at least one RoomLink candidate is required") if len({candidate.target_install_id for candidate in candidates}) != 1: @@ -158,17 +144,9 @@ class PeerMemberRoute: def build_member_dispatch( - *, - binding: HostedRoomBinding, - route: PeerMemberRoute, - room_id: str, - task_id: str, - target_profile: str, - execution_generation: int, - source_event_seq: int, - prompt: str, - trace_id: str, -) -> HostedMemberDispatch: + *, binding: HostedRoomBinding, route: PeerMemberRoute, room_id: str, task_id: str, + target_profile: str, execution_generation: int, source_event_seq: int, prompt: str, + trace_id: str) -> HostedMemberDispatch: """Build the fully fenced member dispatch shared by submit and recovery.""" return HostedMemberDispatch.from_mapping({ "protocol_version": PROTOCOL_VERSION, @@ -187,23 +165,16 @@ def build_member_dispatch( "prompt_digest": hashlib.sha256(prompt.encode("utf-8")).hexdigest(), "capability_digest": route.capability_digest, "execution_policy_digest": route.execution_policy_digest, - "trace_id": trace_id, - }) + "trace_id": trace_id}) class PeerHostedRoomTransport(InternalSessionRPC): """Translate runtime session operations into recipient-validated peer RPC.""" def __init__( - self, - *, - binding: HostedRoomBinding, - route: PeerMemberRoute, - client: HostedRoomPeerClient, - source_event_seq: int = 1, - task_id: str | None = None, - execution_generation: int | None = None, - ) -> None: + self, *, binding: HostedRoomBinding, route: PeerMemberRoute, + client: HostedRoomPeerClient, source_event_seq: int = 1, task_id: str | None = None, + execution_generation: int | None = None) -> None: self.binding = binding self.route = route self.client = client @@ -217,14 +188,11 @@ class PeerHostedRoomTransport(InternalSessionRPC): bind_scope = getattr(self.client, "bind_room_scope", None) if callable(bind_scope): bind_scope( - room_id=self.binding.room_id, - home_install_id=self.route.home_install_id, + room_id=self.binding.room_id, home_install_id=self.route.home_install_id, authority_gateway_id=self.binding.gateway_id, - authority_epoch=self.binding.authority_epoch, - member_id=self.route.member_id, + authority_epoch=self.binding.authority_epoch, member_id=self.route.member_id, target_install_id=self.route.target_install_id, - target_profile=self.route.target_profile, - ) + target_profile=self.route.target_profile) def _validate_coordinates(self, *, profile: str, source: str, title: str | None = None) -> None: if source != ROOM_SESSION_SOURCE: @@ -234,15 +202,13 @@ class PeerHostedRoomTransport(InternalSessionRPC): if title is not None and title != room_session_title(self.binding.room_id): raise ValueError("peer room transport title does not match room identity") + def _scoped(self, **extra: Any) -> dict[str, Any]: + """Room id + grant keyword arguments shared by every scoped client call.""" + return {"room_id": self.binding.room_id, "grant": self.route.grant, **extra} + def _prepare(self, *, profile: str, source: str, create: bool, **extra): return self.client.prepare( - room_id=self.binding.room_id, - profile=profile, - source=source, - grant=self.route.grant, - create=create, - **extra, - ) + **self._scoped(profile=profile, source=source, create=create, **extra)) def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None: self._validate_coordinates(profile=profile, source=source, title=title) @@ -259,38 +225,24 @@ class PeerHostedRoomTransport(InternalSessionRPC): def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: self._validate_coordinates(profile=profile, source=source) session = self._prepare( - profile=profile, source=source, create=False, expected_session_id=session_id - ) + profile=profile, source=source, create=False, expected_session_id=session_id) if session is None: raise RuntimeError("peer room session is unavailable") self._session_id = session_id return session def submit( - self, - *, - profile: str, - session_id: str, - prompt: str, - source: str, - task: TaskIdentity, - execution_generation: int, - on_terminal: Callable[[Mapping[str, Any]], None], + self, *, profile: str, session_id: str, prompt: str, source: str, task: TaskIdentity, + execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None], ) -> Mapping[str, Any]: self._validate_coordinates(profile=profile, source=source) if self._session_id not in {None, session_id}: raise ValueError("peer room session changed during admission") dispatch = build_member_dispatch( - binding=self.binding, - route=self.route, - room_id=task.room_id, - task_id=task.task_id, - target_profile=profile, - execution_generation=execution_generation, - source_event_seq=self.source_event_seq, - prompt=prompt, - trace_id=self.route.trace_id or f"trace-{uuid.uuid4().hex}", - ) + binding=self.binding, route=self.route, room_id=task.room_id, task_id=task.task_id, + target_profile=profile, execution_generation=execution_generation, + source_event_seq=self.source_event_seq, prompt=prompt, + trace_id=self.route.trace_id or f"trace-{uuid.uuid4().hex}") self._dispatch = dispatch self._session_id = session_id result = self.client.dispatch(dispatch=dispatch.as_mapping(), grant=self.route.grant) @@ -300,21 +252,11 @@ class PeerHostedRoomTransport(InternalSessionRPC): def history(self, *, profile: str, session_id: str, source: str) -> Sequence[Mapping[str, Any]]: self._validate_coordinates(profile=profile, source=source) - return self.client.history( - room_id=self.binding.room_id, - profile=profile, - session_id=session_id, - grant=self.route.grant, - ) + return self.client.history(**self._scoped(profile=profile, session_id=session_id)) def info(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: self._validate_coordinates(profile=profile, source=source) - return self.client.status( - room_id=self.binding.room_id, - profile=profile, - session_id=session_id, - grant=self.route.grant, - ) + return self.client.status(**self._scoped(profile=profile, session_id=session_id)) def interrupt( self, *, profile: str, session_id: str, source: str, expected_task_id: str @@ -325,14 +267,11 @@ class PeerHostedRoomTransport(InternalSessionRPC): if ( self.task_id != expected_task_id or not self.execution_generation - or not hasattr(self.client, "stop_receipt") - ): + or not hasattr(self.client, "stop_receipt")): return None return self.client.stop_receipt( - task_id=expected_task_id, - execution_generation=self.execution_generation, - grant=self.route.grant, - ) + task_id=expected_task_id, execution_generation=self.execution_generation, + grant=self.route.grant) if dispatch.task_id != expected_task_id: return None return self.client.stop(dispatch=dispatch.as_mapping(), grant=self.route.grant) diff --git a/tui_gateway/hosted_room_server_rpc.py b/tui_gateway/hosted_room_server_rpc.py index 97d3925ed7..edbe5b8eaa 100644 --- a/tui_gateway/hosted_room_server_rpc.py +++ b/tui_gateway/hosted_room_server_rpc.py @@ -36,17 +36,14 @@ class HostedRoomServerRPC: self._ids = itertools.count(1) def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: - handler = self.server._methods[method] - envelope = handler(f"hosted-room-{next(self._ids)}", params) + envelope = self.server._methods[method](f"hosted-room-{next(self._ids)}", params) if not isinstance(envelope, dict): envelope = {} error = envelope.get("error") if isinstance(error, dict): raise HostedRoomSessionError( - method, - int(error.get("code") or 5000), - str(error.get("message") or "gateway rejected the request"), - ) + method, int(error.get("code") or 5000), + str(error.get("message") or "gateway rejected the request")) result = envelope.get("result") if not isinstance(result, dict): raise HostedRoomSessionError(method, 5000, "gateway returned no result") @@ -55,16 +52,14 @@ class HostedRoomServerRPC: def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None: del source result = self._call( - "session.list", {"profile": profile, "title": title, "include_hidden": True} - ) + "session.list", {"profile": profile, "title": title, "include_hidden": True}) rows = result.get("sessions") if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict): return None row = rows[0] return { "session_id": row.get("resolved_id") or row.get("id"), - "title": row.get("title") or title, - } + "title": row.get("title") or title} def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]: return self._call( @@ -76,26 +71,16 @@ class HostedRoomServerRPC: "hidden": True, "room_plumbing": True, "follow_profile_config": True, - "close_on_disconnect": False, - }, - ) + "close_on_disconnect": False}) def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]: return self._call( "session.resume", - {"profile": profile, "session_id": session_id, "omit_messages": True, "source": source}, - ) + {"profile": profile, "session_id": session_id, "omit_messages": True, "source": source}) def submit( - self, - *, - profile: str, - session_id: str, - prompt: str, - source: str, - task: state.TaskIdentity, - execution_generation: int, - on_terminal: Callable[[Mapping[str, Any]], None], + self, *, profile: str, session_id: str, prompt: str, source: str, task: state.TaskIdentity, + execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None], ) -> Mapping[str, Any]: try: return self._call( @@ -110,11 +95,8 @@ class HostedRoomServerRPC: "task_id": task.task_id, "thread_id": task.thread_id, "turn_id": task.turn_id, - "execution_generation": execution_generation, - }, - "_hosted_terminal_callback": on_terminal, - }, - ) + "execution_generation": execution_generation}, + "_hosted_terminal_callback": on_terminal}) except HostedRoomSessionError as exc: # In-process prompt.submit error envelopes are returned before the # background turn is admitted. Preserve that proof so the driver @@ -150,14 +132,12 @@ class HostedRoomServerRPC: task = record.get("_hosted_room_task") result = { "active": bool(record.get("running")), - "task_id": task.get("task_id") if isinstance(task, dict) else None, - } + "task_id": task.get("task_id") if isinstance(task, dict) else None} pending_reader = getattr(self.server, "_pending_approval_request_payload", None) pending = ( pending_reader(str(record.get("session_key") or "")) if callable(pending_reader) - else None - ) + else None) if pending: result["status"] = "waiting_for_approval" result["pending_approval"] = pending @@ -167,8 +147,7 @@ class HostedRoomServerRPC: """Resolve one exact local room approval without broad policy changes.""" return self._call( "approval.respond", - {"session_id": session_id, "request_id": request_id, "choice": choice, "all": False}, - ) + {"session_id": session_id, "request_id": request_id, "choice": choice, "all": False}) def interrupt( self, *, profile: str, session_id: str, source: str, expected_task_id: str @@ -176,5 +155,7 @@ class HostedRoomServerRPC: del source return self._call( "session.interrupt", - {"profile": profile, "session_id": session_id, "expected_hosted_task_id": expected_task_id}, - ) + { + "profile": profile, + "session_id": session_id, + "expected_hosted_task_id": expected_task_id}) From 9ea83e7f96961a7632977d58a169b172912e4f5a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:34:06 -0700 Subject: [PATCH 2/2] refactor(tui_gateway): collapse hosted-room init/scope validation, share resolve-or-create, squeeze layout --- tui_gateway/hosted_room_driver.py | 89 ++++++++--------------- tui_gateway/hosted_room_peer_http.py | 31 ++++---- tui_gateway/hosted_room_peer_transport.py | 9 +-- 3 files changed, 48 insertions(+), 81 deletions(-) diff --git a/tui_gateway/hosted_room_driver.py b/tui_gateway/hosted_room_driver.py index 526bb3e5bf..3e89c3d879 100644 --- a/tui_gateway/hosted_room_driver.py +++ b/tui_gateway/hosted_room_driver.py @@ -128,24 +128,19 @@ class HostedRoomRuntime: if value <= 0: raise ValueError(f"{name} must be positive") if ( - isinstance(max_concurrent_rooms, bool) - or not isinstance(max_concurrent_rooms, int) - or max_concurrent_rooms < 1): + not isinstance(max_concurrent_rooms, int) + or isinstance(max_concurrent_rooms, bool) + or max_concurrent_rooms < 1 + ): raise ValueError("max_concurrent_rooms must be a positive integer") - if not ( - unavailable_retry_min_seconds > 0 - and unavailable_retry_max_seconds >= unavailable_retry_min_seconds): + if not 0 < unavailable_retry_min_seconds <= unavailable_retry_max_seconds: raise ValueError("unavailable retry bounds are invalid") if rpc is None and transport_resolver is None: raise ValueError("rpc or transport_resolver is required") self.db_path = Path(db_path) - self.rpc = rpc - self.transport_resolver = transport_resolver - self.turn_lock = turn_lock - self.prepare_room = prepare_room - self.publish_terminal = publish_terminal - self.pending_action = pending_action - self.clock = clock + self.rpc, self.transport_resolver, self.turn_lock = rpc, transport_resolver, turn_lock + self.prepare_room, self.publish_terminal = prepare_room, publish_terminal + self.pending_action, self.clock = pending_action, clock self.lease_ttl_seconds = float(lease_ttl_seconds) self.poll_interval_seconds = float(poll_interval_seconds) self.active_poll_interval_seconds = float(active_poll_interval_seconds) @@ -155,15 +150,12 @@ class HostedRoomRuntime: self.unavailable_retry_min_seconds = float(unavailable_retry_min_seconds) self.unavailable_retry_max_seconds = float(unavailable_retry_max_seconds) self.process_generation = process_generation or uuid.uuid4().hex - self._rooms_provider: Callable[[], Iterable[HostedRoomBinding]] - if callable(rooms): - self._rooms_provider = cast(Callable[[], Iterable[HostedRoomBinding]], rooms) - else: - room_bindings = tuple(rooms) - self._rooms_provider = lambda: room_bindings + self._rooms_provider: Callable[[], Iterable[HostedRoomBinding]] = ( + cast(Callable[[], Iterable[HostedRoomBinding]], rooms) + if callable(rooms) + else (lambda bindings=tuple(rooms): bindings)) - self._stop = threading.Event() - self._wake = threading.Event() + self._stop, self._wake = threading.Event(), threading.Event() self._thread: threading.Thread | None = None self._room_threads: dict[str, threading.Thread] = {} self._rooms_needing_reschedule: set[str] = set() @@ -175,9 +167,8 @@ class HostedRoomRuntime: self._blocked_rooms: set[str] = set() self._status_lock = threading.Lock() self._current_tasks: dict[str, state.TaskIdentity] = {} - self._room_schedule_cursor = 0 + self._room_schedule_cursor, self._cycles = 0, 0 self._last_error: str | None = None - self._cycles = 0 # ------------------------------------------------------------------ lifecycle def start(self) -> None: @@ -322,9 +313,8 @@ class HostedRoomRuntime: task: Mapping[str, Any], lease: state.DriverLease, *, publish: bool = True, **extra: Any, ) -> dict[str, Any]: """Run one lease-fenced state transition on ``task``; ``extra`` may override fences.""" - result = op( - self.db_path, task["identity"], lease, **{**_fences(task), "clock": self.clock, **extra} - ) + kwargs = {**_fences(task), "clock": self.clock, **extra} + result = op(self.db_path, task["identity"], lease, **kwargs) return self._publish(binding, result) if publish and binding is not None else result def _requeue( @@ -381,11 +371,8 @@ class HostedRoomRuntime: Callers must use the returned id (not the stored one) for every subsequent history/info probe; resume may hand back a different id. """ - session = transport.resolve_exact( - profile=profile, title=room_session_title(room_id), source=ROOM_SESSION_SOURCE) - if session is None: - return None - return _session_id(transport.resume(**_session_kw(profile, _session_id(session)))) + session = self._resolve_or_create(transport, profile, room_id, create=False) + return None if session is None else _session_id(session) def _open_session( self, binding: HostedRoomBinding, task: Mapping[str, Any], *, peer_only: bool = False @@ -468,9 +455,8 @@ class HostedRoomRuntime: approval = info.get("pending_approval") or info.get("approval") action = None if isinstance(approval, Mapping): - safe_approval = dict(approval) - choices = [c for c in safe_approval.get("choices") or () if c in {"once", "deny"}] - safe_approval["choices"] = choices or ["once", "deny"] + choices = [c for c in approval.get("choices") or () if c in {"once", "deny"}] + safe_approval = {**approval, "choices": choices or ["once", "deny"]} action = { "kind": "approval", "task_id": task["identity"].task_id, @@ -528,7 +514,6 @@ class HostedRoomRuntime: return self._run_room_once(binding) return - with self._status_lock: self._room_threads = { room_id: thread @@ -538,13 +523,11 @@ class HostedRoomRuntime: active_rooms = set(self._room_threads) if available <= 0: return - bindings = tuple(self._rooms_provider()) if not bindings: return start = self._room_schedule_cursor % len(bindings) self._room_schedule_cursor = (start + 1) % len(bindings) - for binding in bindings[start:] + bindings[:start]: if self._stop.is_set() or available <= 0: return @@ -564,11 +547,10 @@ class HostedRoomRuntime: self._process_room(binding) except state.LeaseHeldError: return - except (state.RoomUnavailableError, state.StaleLeaseError) as exc: - self._drop_lease(binding.room_id) - self._set_blocked(binding.room_id, False) - self._record_error(f"room {binding.room_id}: {exc}") except Exception as exc: + if isinstance(exc, (state.RoomUnavailableError, state.StaleLeaseError)): + self._drop_lease(binding.room_id) + self._set_blocked(binding.room_id, False) self._record_error(f"room {binding.room_id}: {exc}") finally: current = threading.current_thread() @@ -600,7 +582,6 @@ class HostedRoomRuntime: return if self._reconcile_indeterminate(binding, lease): return - for task in state.list_tasks(self.db_path, room_id=binding.room_id, status="queued"): if self._stop.is_set() or self._route_retry_is_deferred(task): return @@ -625,12 +606,8 @@ class HostedRoomRuntime: def _defer_unavailable_route(self, task: Mapping[str, Any]) -> float: key = self._route_retry_key(task) previous = self._unavailable_route_retries.get(key) - delay = ( - self.unavailable_retry_min_seconds - if previous is None - else min( - self.unavailable_retry_max_seconds, - max(self.unavailable_retry_min_seconds, previous["delay"] * 2))) + lo, hi = self.unavailable_retry_min_seconds, self.unavailable_retry_max_seconds + delay = lo if previous is None else min(hi, max(lo, previous["delay"] * 2)) self._unavailable_route_retries[key] = { "delay": delay, "next_attempt_at": self.clock() + delay} return delay @@ -677,11 +654,9 @@ class HostedRoomRuntime: def _release_idle_leases(self) -> None: for room_id, lease in tuple(self._leases.items()): - try: + with suppress(state.DriverStateError): state.release_lease(self.db_path, lease, clock=self.clock) - except state.DriverStateError: - continue - self._drop_lease(room_id) + self._drop_lease(room_id) # ------------------------------------------------------------------ attempt execution def _execute_attempt( @@ -802,16 +777,13 @@ class HostedRoomRuntime: self._wake.wait(self.active_poll_interval_seconds) self._wake.clear() continue - if time.monotonic() >= deadline_monotonic: self._expire_attempt_deadline(binding, task, lease) return None - lease = self._renew_lease_if_needed(binding, lease) receipt = self._terminal_from_history(transport, profile, session_id, task) if receipt is not None: return receipt - info = transport.info(**_session_kw(profile, session_id)) self._report_pending_action(task, session_id=session_id, info=info) remaining = max(0.0, deadline_monotonic - time.monotonic()) @@ -849,7 +821,6 @@ class HostedRoomRuntime: expected_cancel_generation=int(task["cancel_generation"]), clock=self.clock) elif task["status"] != "stopping": return - # A user Stop that won the race keeps its own cancellation semantics. if not self._is_deadline_stop(task): return @@ -1004,12 +975,14 @@ class HostedRoomRuntime: return self.rpc def _resolve_or_create( - self, transport: InternalSessionRPC, profile: str, room_id: str) -> Mapping[str, Any]: + self, transport: InternalSessionRPC, profile: str, room_id: str, *, create: bool = True + ) -> Mapping[str, Any] | None: + """Resolve + resume the canonical room session; create it (or return None) when absent.""" coords = { "profile": profile, "title": room_session_title(room_id), "source": ROOM_SESSION_SOURCE} session = transport.resolve_exact(**coords) if session is None: - return transport.create(**coords) + return transport.create(**coords) if create else None return transport.resume(**_session_kw(profile, _session_id(session))) def _settle_failure_if_current(self, attempt: state.TaskAttempt, exc: Exception) -> None: diff --git a/tui_gateway/hosted_room_peer_http.py b/tui_gateway/hosted_room_peer_http.py index 2d48e7903a..e3af7e3377 100644 --- a/tui_gateway/hosted_room_peer_http.py +++ b/tui_gateway/hosted_room_peer_http.py @@ -252,18 +252,17 @@ class PeerRunsHTTPClient: self, *, room_id: str, home_install_id: str, authority_gateway_id: str, authority_epoch: int, member_id: str, target_install_id: str, target_profile: str) -> None: """Fence every in-memory and durable receipt to one room authority.""" - scope = { - "room_id": str(room_id or ""), - "home_install_id": str(home_install_id or ""), - "authority_gateway_id": str(authority_gateway_id or ""), - "authority_epoch": int(authority_epoch or 0), - "member_id": str(member_id or ""), - "target_install_id": str(target_install_id or ""), - "target_profile": str(target_profile or "")} - if not all(value for key, value in scope.items() if key != "authority_epoch"): + epoch = int(authority_epoch or 0) + names = [ + str(value or "") + for value in ( + room_id, home_install_id, authority_gateway_id, member_id, target_install_id, + target_profile)] + if not all(names): raise PeerRunsHTTPError("peer room receipt scope is incomplete") - if scope["authority_epoch"] < 1: + if epoch < 1: raise PeerRunsHTTPError("peer room receipt authority epoch is invalid") + scope = dict(zip(_RECEIPT_SCOPE_FIELDS, names[:3] + [epoch] + names[3:])) if self._room_scope == scope: return self._room_scope = scope @@ -283,11 +282,9 @@ class PeerRunsHTTPClient: return record from gateway import hosted_rooms + identity = {"task_id": task_id, "execution_generation": execution_generation} return hosted_rooms.remote_run_receipt( - self.receipt_db_path, - record={ - **self._room_scope, "task_id": task_id, "execution_generation": execution_generation - }) + self.receipt_db_path, record={**self._room_scope, **identity}) def bind_observation(self, *, task_id: str, execution_generation: int) -> None: """Pin history/status reads to one exact logical task attempt.""" @@ -417,8 +414,7 @@ class PeerRunsHTTPClient: return self._accepted( checked, run_id=str(existing["run_id"]), session_id=str(existing["session_id"]), replayed=True) - key = (checked.task_id, checked.execution_generation) - now = self.clock() + key, now = (checked.task_id, checked.execution_generation), self.clock() backoff = self._recovery_backoff.get(key) if backoff is not None and now < float(backoff["next_attempt_at"]): raise PeerRunsHTTPError( @@ -520,8 +516,7 @@ class PeerRunsHTTPClient: return status.get("status") in _TERMINAL_RUN_STATES def _poll_receipt(self, record: Mapping[str, Any], *, grant: str) -> dict[str, Any]: - run_id = str(record["run_id"]) - now = self.clock() + run_id, now = str(record["run_id"]), self.clock() cached = self._status_cache.get(run_id) if cached is not None: status = cached["status"] diff --git a/tui_gateway/hosted_room_peer_transport.py b/tui_gateway/hosted_room_peer_transport.py index 1748fe5bc7..170c98a28b 100644 --- a/tui_gateway/hosted_room_peer_transport.py +++ b/tui_gateway/hosted_room_peer_transport.py @@ -188,11 +188,10 @@ class PeerHostedRoomTransport(InternalSessionRPC): bind_scope = getattr(self.client, "bind_room_scope", None) if callable(bind_scope): bind_scope( - room_id=self.binding.room_id, home_install_id=self.route.home_install_id, - authority_gateway_id=self.binding.gateway_id, - authority_epoch=self.binding.authority_epoch, member_id=self.route.member_id, - target_install_id=self.route.target_install_id, - target_profile=self.route.target_profile) + room_id=binding.room_id, home_install_id=route.home_install_id, + authority_gateway_id=binding.gateway_id, authority_epoch=binding.authority_epoch, + member_id=route.member_id, target_install_id=route.target_install_id, + target_profile=route.target_profile) def _validate_coordinates(self, *, profile: str, source: str, title: str | None = None) -> None: if source != ROOM_SESSION_SOURCE: