diff --git a/tests/tui_gateway/test_hosted_room_peer_http.py b/tests/tui_gateway/test_hosted_room_peer_http.py index 3f578607ff..c43e85e287 100644 --- a/tests/tui_gateway/test_hosted_room_peer_http.py +++ b/tests/tui_gateway/test_hosted_room_peer_http.py @@ -463,7 +463,6 @@ def test_invalid_room_dispatch_http_403_is_definitively_not_admitted(monkeypatch assert caught.value.error_code == "room_capability_catalog_changed" assert caught.value.not_admitted is True assert caught.value.ambiguous is False - assert caught.value.needs_capability_refresh is True def test_capability_mismatch_requires_reauthorization_without_retry(tmp_path): @@ -525,7 +524,6 @@ def test_peer_http_error_body_is_never_exposed_or_logged(monkeypatch, caplog): assert caught.value.status_code == 500 assert hostile not in str(caught.value) - assert caught.value.error_message is None assert hostile not in caplog.text diff --git a/tui_gateway/hosted_room_driver.py b/tui_gateway/hosted_room_driver.py index fb40791b14..1867c02f4b 100644 --- a/tui_gateway/hosted_room_driver.py +++ b/tui_gateway/hosted_room_driver.py @@ -53,7 +53,6 @@ class InternalSessionRPC(Protocol): ) -> Mapping[str, Any] | None: ... -# Resolve the local or peer session transport for one durable room task. MemberTransportResolver = Callable[["HostedRoomBinding", Mapping[str, Any]], InternalSessionRPC] @@ -109,18 +108,16 @@ class HostedRoomRuntime: 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)): + positive = dict( + 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) + for name, value in positive.items(): if value <= 0: raise ValueError(f"{name} must be positive") - if ( - not isinstance(max_concurrent_rooms, int) - or isinstance(max_concurrent_rooms, bool) - or max_concurrent_rooms < 1): + if (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 0 < unavailable_retry_min_seconds <= unavailable_retry_max_seconds: raise ValueError("unavailable retry bounds are invalid") @@ -130,11 +127,8 @@ class HostedRoomRuntime: 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, self.poll_interval_seconds = ( - float(lease_ttl_seconds), float(poll_interval_seconds)) - self.active_poll_interval_seconds = float(active_poll_interval_seconds) - self.turn_timeout_seconds = float(turn_timeout_seconds) - self.indeterminate_defer_seconds = float(indeterminate_defer_seconds) + for name, value in positive.items(): + setattr(self, name, float(value)) self.max_concurrent_rooms = max_concurrent_rooms self.unavailable_retry_min_seconds, self.unavailable_retry_max_seconds = ( float(unavailable_retry_min_seconds), float(unavailable_retry_max_seconds)) @@ -143,7 +137,7 @@ class HostedRoomRuntime: cast(Callable[[], Iterable[HostedRoomBinding]], rooms) if callable(rooms) else (lambda bindings=tuple(rooms): bindings)) self._stop, self._wake = threading.Event(), threading.Event() - self._thread: threading.Thread | None = None + self._thread = self._last_error = None self._room_threads: dict[str, threading.Thread] = {} self._rooms_needing_reschedule: set[str] = set() self._leases: dict[str, state.DriverLease] = {} @@ -154,7 +148,6 @@ class HostedRoomRuntime: self._blocked_rooms: set[str] = set() self._status_lock, self._current_tasks = threading.Lock(), {} self._room_schedule_cursor, self._cycles = 0, 0 - self._last_error: str | None = None # ------------------------------------------------------------------ lifecycle def start(self) -> None: @@ -182,7 +175,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 t.is_alive() for t in room_threads) + return not any(t.is_alive() for t in (thread, *room_threads)) def wakeup(self) -> None: """Wake the worker after task admission or a room-state change.""" @@ -219,32 +212,26 @@ class HostedRoomRuntime: if before["status"] in state.TERMINAL_STATUSES: raise state.InvalidTaskTransitionError( 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, **fenced) - except (state.InvalidTaskTransitionError, state.StaleTaskError): - continue # lost the race with the worker; re-route - self.wakeup() - return cancelled + direct = before["status"] in {"queued", "deferred"} try: - stopping = state.begin_task_cancel(self.db_path, identity, **fenced) + result = (state.cancel_task if direct else state.begin_task_cancel)( + self.db_path, identity, cancel_id=cancel_id, + expected_cancel_generation=before["cancel_generation"], clock=self.clock) 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) 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}") + continue # lost the race with the worker (settled or re-queued); re-route + if not direct: + 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, result) or ( + not self._settle_stopping_completion(binding, result, lease) + and self._interrupt_stopping_task(binding, result)): + self._complete_cancel(result, cancel_id=cancel_id) + except Exception as exc: + self._record_error(f"stop remains pending: {exc}") self.wakeup() - return state.get_task(self.db_path, identity) + return result if direct else state.get_task(self.db_path, identity) # Routing retries exhausted under contention: surface the live status honestly. final = state.get_task(self.db_path, identity) if final["status"] == "cancelled": @@ -264,20 +251,21 @@ class HostedRoomRuntime: lease = self._ensure_lease(binding) if task["status"] == "deferred": 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. + # Explicit Retry may resume the exact stored session; the automatic abandoned-attempt + # scan stays non-resuming for local sessions. inspection = self._inspect_recovery_session(binding, task) if inspection.terminal is not None: return self._resolve_indeterminate(binding, task, lease, inspection.terminal) if inspection.status == "cancelled": - return self._resolve_remote_cancel(binding, task, lease) + return self._fenced( + state.resolve_indeterminate_cancellation, binding, task, lease, + cancel_id=f"remote-cancel:{task['execution_generation']}") 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) - # ------------------------------------------------------------------ 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) @@ -307,9 +295,9 @@ class HostedRoomRuntime: def _complete_cancel( 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"], clock=self.clock, 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"]) def _resolve_indeterminate( self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease, @@ -318,13 +306,6 @@ class HostedRoomRuntime: 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]: - return self._fenced( - state.resolve_indeterminate_cancellation, binding, task, lease, publish=publish, - cancel_id=f"remote-cancel:{task['execution_generation']}") - def _finish_stop( self, binding: HostedRoomBinding, task: Mapping[str, Any], lease: state.DriverLease ) -> bool: @@ -336,7 +317,6 @@ class HostedRoomRuntime: return True return False - # ------------------------------------------------------------------ session probes def _resume_exact( self, transport: InternalSessionRPC, room_id: str, profile: str) -> str | None: """Resume the canonical room session and return its runtime id (None when absent). @@ -380,15 +360,13 @@ class HostedRoomRuntime: 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. + # A local turn cannot survive without its canonical session, so an authoritative + # absence is a safe Stop acknowledgement (errors raise); a peer stays uncertain. 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 - # a restart its process-local task marker is expected to be absent. + # History was checked just before this probe: an inactive exact session cannot + # keep executing, and after a restart its process-local task marker is absent. return True if not _info_is_active_for(info, task["identity"], require_exact=True): return False @@ -415,8 +393,7 @@ class HostedRoomRuntime: 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 + approval, action = info.get("pending_approval") or info.get("approval"), None if isinstance(approval, Mapping): choices = [c for c in approval.get("choices") or () if c in {"once", "deny"}] safe_approval = {**approval, "choices": choices or ["once", "deny"]} @@ -430,7 +407,7 @@ class HostedRoomRuntime: def _retry_stopping_tasks(self, binding: HostedRoomBinding, lease: state.DriverLease) -> bool: for task in self._tasks(binding, "stopping"): try: - lease = self._renew_lease_if_needed(binding, lease) + lease = self._renew_lease_if_needed(lease) if self._peer_stop_acknowledged(binding, task): self._complete_cancel(task) continue @@ -445,8 +422,7 @@ class HostedRoomRuntime: def _worker_loop(self) -> None: try: while not self._stop.is_set(): - # Clear before work so a write racing the cycle remains set and - # causes an immediate follow-up pass rather than being lost. + # Clear before work so a write racing the cycle forces a follow-up pass. self._wake.clear() try: self._run_cycle() @@ -511,9 +487,8 @@ class HostedRoomRuntime: self._set_blocked(binding.room_id, False) self._record_error(f"room {binding.room_id}: {exc}") finally: - current = threading.current_thread() with self._status_lock: - if self._room_threads.get(binding.room_id) is current: + if self._room_threads.get(binding.room_id) is threading.current_thread(): self._room_threads.pop(binding.room_id, None) should_wake = binding.room_id in self._rooms_needing_reschedule self._rooms_needing_reschedule.discard(binding.room_id) @@ -530,19 +505,21 @@ class HostedRoomRuntime: return self._ambiguous_rooms.pop(binding.room_id, None) lease = self._ensure_lease(binding) - recovery_key = (lease.room_id, lease.lease_generation) - if recovery_key not in self._recovered_leases: + if (lease.room_id, lease.lease_generation) not in self._recovered_leases: state.recover_room(self.db_path, lease, clock=self.clock) - self._recovered_leases.add(recovery_key) + self._recovered_leases.add((lease.room_id, lease.lease_generation)) if self._retry_stopping_tasks(binding, lease): self._set_blocked(binding.room_id, True) return if self._reconcile_indeterminate(binding, lease): return for task in self._tasks(binding, "queued"): - if self._stop.is_set() or self._route_retry_is_deferred(task): + retry = self._unavailable_route_retries.get( + (task["identity"].room_id, _member_id(task))) + if self._stop.is_set() or ( + retry is not None and self.clock() < retry["next_attempt_at"]): return - lease = self._renew_lease_if_needed(binding, lease) + lease = self._renew_lease_if_needed(lease) attempt = state.start_task( self.db_path, task["identity"], lease, expected_cancel_generation=task["cancel_generation"], clock=self.clock) @@ -551,17 +528,8 @@ class HostedRoomRuntime: 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) - - def _route_retry_is_deferred(self, task: Mapping[str, Any]) -> bool: - retry = self._unavailable_route_retries.get(self._route_retry_key(task)) - return retry is not None and self.clock() < retry["next_attempt_at"] - def _defer_unavailable_route(self, task: Mapping[str, Any]) -> float: - key = self._route_retry_key(task) + key = (task["identity"].room_id, _member_id(task)) previous = self._unavailable_route_retries.get(key) 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)) @@ -575,26 +543,21 @@ class HostedRoomRuntime: current = self._leases.get(binding.room_id) if current is not None: try: - return self._renew_lease_if_needed(binding, current) + return self._renew_lease_if_needed(current) except state.StaleLeaseError: 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) 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} + self._recovered_leases = {k for k in self._recovered_leases if k[0] != binding.room_id} return lease def _renew_lease_if_needed( - 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: + self, lease: state.DriverLease, *, force: bool = False) -> state.DriverLease: + if not force and self.clock() < lease.expires_at - (self.lease_ttl_seconds / 2): return lease renewed = state.renew_lease( self.db_path, lease, ttl_seconds=self.lease_ttl_seconds, clock=self.clock) @@ -616,9 +579,8 @@ class HostedRoomRuntime: def _execute_attempt( self, binding: HostedRoomBinding, task: Mapping[str, Any], attempt: state.TaskAttempt ) -> None: - profile = task["payload"]["target_profile"] + profile, submit_attempted = task["payload"]["target_profile"], False transport = self._transport_for(binding, task) - submit_attempted = False with self._status_lock: self._current_tasks[binding.room_id] = attempt.identity try: @@ -626,14 +588,14 @@ class HostedRoomRuntime: session = self._resolve_or_create(transport, profile, binding.room_id) # A submit should fail before admission or return after it; an unexpected # exception at that boundary is ambiguous, never a proven failure. - submit_attempted = True - session_id = _session_id(session) + submit_attempted, session_id = True, _session_id(session) deadline_monotonic = time.monotonic() + self.turn_timeout_seconds transport.submit( **_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._unavailable_route_retries.pop(self._route_retry_key(task), None) + self._unavailable_route_retries.pop( + (task["identity"].room_id, _member_id(task)), None) receipt = self._wait_for_terminal( binding, profile=profile, session_id=session_id, attempt=attempt, transport=transport, deadline_monotonic=deadline_monotonic) @@ -686,8 +648,9 @@ class HostedRoomRuntime: or f"reply:{attempt.identity.task_id}:{attempt.execution_generation}", result=_bounded_terminal_result(receipt)) try: - settled = state.settle_task(self.db_path, attempt, **asdict(terminal), clock=self.clock) - self._publish(binding, settled) + self._publish( + binding, + state.settle_task(self.db_path, attempt, **asdict(terminal), clock=self.clock)) except state.StaleTaskError: with suppress(state.StaleLeaseError, state.StaleTaskError): current = state.get_task(self.db_path, attempt.identity) @@ -717,7 +680,7 @@ class HostedRoomRuntime: return None if task["status"] == "stopping": try: - lease = self._renew_lease_if_needed(binding, lease) + lease = self._renew_lease_if_needed(lease) if self._finish_stop(binding, task, lease): return None except Exception as exc: @@ -728,7 +691,7 @@ class HostedRoomRuntime: if time.monotonic() >= deadline_monotonic: self._expire_attempt_deadline(binding, task, lease) return None - lease = self._renew_lease_if_needed(binding, lease) + lease = self._renew_lease_if_needed(lease) receipt = self._terminal_from_history(transport, profile, session_id, task) if receipt is not None: return receipt @@ -739,16 +702,11 @@ class HostedRoomRuntime: self._wake.clear() 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 ) -> dict[str, Any]: """Terminalize an acknowledged Stop: deadline stops publish an explicit failure.""" - if not self._is_deadline_stop(task): + if not str(task.get("cancel_id") or "").startswith("deadline:"): return self._complete_cancel(task) return self._fenced( state.settle_stopping_task, binding, task, lease, @@ -764,15 +722,15 @@ class HostedRoomRuntime: """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"], clock=self.clock, 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"])) elif task["status"] != "stopping": return # A user Stop that won the race keeps its own cancellation semantics. - if not self._is_deadline_stop(task): + if not str(task.get("cancel_id") or "").startswith("deadline:"): return - lease = self._renew_lease_if_needed(binding, lease, force=True) + lease = self._renew_lease_if_needed(lease, force=True) if not self._finish_stop(binding, task, lease): self._record_error( f"task {task['identity'].task_id} exceeded its deadline; stop remains pending") @@ -789,8 +747,7 @@ class HostedRoomRuntime: if inspection.terminal is not None: self._harvest_previous_attempt(binding, task, inspection.terminal) elif inspection.active: - # The prior session still owns the turn. Do not contend for its - # lease or submit a duplicate prompt. + # The prior session still owns the turn: no lease contention, no duplicate prompt. raise state.LeaseHeldError("recovered session turn is still active") def _inspect_session( @@ -798,9 +755,9 @@ class HostedRoomRuntime: *, 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) + receipt = ( + self._terminal_from_history(transport, profile, session_id, task) + if read_history else None) info = transport.info(**_session_kw(profile, session_id)) self._report_pending_action(task, session_id=session_id, info=info) return _RecoveryInspection( @@ -809,8 +766,7 @@ class HostedRoomRuntime: def _inspect_recovery_session( self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> _RecoveryInspection: - profile = task["payload"]["target_profile"] - transport = self._transport_for(binding, task) + profile, transport = task["payload"]["target_profile"], self._transport_for(binding, task) with self.turn_lock(profile): session_id = self._resume_exact(transport, task["identity"].room_id, profile) if session_id is None: @@ -841,8 +797,8 @@ class HostedRoomRuntime: return False inspected = self._inspected_indeterminate_attempts for task in unresolved: - generation = int(task["execution_generation"]) - attempt_key = (binding.room_id, task["identity"].task_id, generation) + attempt_key = ( + binding.room_id, task["identity"].task_id, int(task["execution_generation"])) 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) @@ -868,7 +824,9 @@ class HostedRoomRuntime: inspected.add(attempt_key) if inspection.status == "cancelled": # Remote-probe resolutions are not republished here. - self._resolve_remote_cancel(binding, task, lease, publish=False) + self._fenced( + state.resolve_indeterminate_cancellation, binding, task, lease, publish=False, + cancel_id=f"remote-cancel:{task['execution_generation']}") inspected.discard(attempt_key) continue if inspection.terminal is not None: @@ -902,7 +860,6 @@ class HostedRoomRuntime: with suppress(state.StaleLeaseError, state.StaleTaskError): state.settle_task(self.db_path, previous_attempt, **asdict(receipt), clock=self.clock) - # ------------------------------------------------------------------ misc def _tasks(self, binding: HostedRoomBinding, status: str) -> list[dict[str, Any]]: return state.list_tasks(self.db_path, room_id=binding.room_id, status=status) @@ -965,8 +922,7 @@ def _truncate_utf8(value: Any, *, max_bytes: int) -> tuple[str, bool]: text, encoded = str(value or ""), str(value or "").encode("utf-8") if len(encoded) <= max_bytes: return text, False - suffix = _TERMINAL_TRUNCATION_NOTICE.encode("utf-8") - prefix = encoded[: max(0, max_bytes - len(suffix))] + prefix = encoded[: max(0, max_bytes - len(_TERMINAL_TRUNCATION_NOTICE.encode("utf-8")))] while prefix: try: return prefix.decode("utf-8") + _TERMINAL_TRUNCATION_NOTICE, True @@ -979,8 +935,7 @@ def _bounded_terminal_result(receipt: Mapping[str, Any]) -> dict[str, Any]: 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, + "message_id": receipt.get("message_id"), "text": text, **({"error": error} if error else {}), **({"truncated": True} if truncated or error_truncated else {})} @@ -993,8 +948,7 @@ def _find_terminal_receipt( if ( 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 message.get("role") != "assistant" or status not in {"settled", "failed"}): continue receipt_id = message.get("message_id") if not isinstance(receipt_id, str) or not receipt_id: @@ -1012,8 +966,5 @@ 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: - if not _info_active(info): - return False - active_task_id = info.get("task_id") - return active_task_id == identity.task_id if require_exact else ( - active_task_id in {None, identity.task_id}) + accepted = (identity.task_id,) if require_exact else (None, identity.task_id) + return _info_active(info) and info.get("task_id") in accepted diff --git a/tui_gateway/hosted_room_peer_http.py b/tui_gateway/hosted_room_peer_http.py index 308352b64e..416b726226 100644 --- a/tui_gateway/hosted_room_peer_http.py +++ b/tui_gateway/hosted_room_peer_http.py @@ -73,18 +73,9 @@ class _PeerResponseDeadlineExceeded(TimeoutError): """A peer response exceeded the request's monotonic wall-clock budget.""" -def _content_length(response: Any) -> int | None: - try: - 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() + frontier, seen = [response], set() for _depth in range(5): next_frontier = [] for value in frontier: @@ -101,12 +92,14 @@ def _set_response_socket_timeout(response: Any, remaining: float) -> None: def _read_bounded_response(response: Any, *, max_bytes: int, deadline: float) -> bytes: - declared = _content_length(response) - if declared is not None and declared > max_bytes: + try: + declared = int(response.headers.get("Content-Length")) + except (AttributeError, TypeError, ValueError): + declared = -1 + if declared > max_bytes: raise _PeerResponseTooLarge reader = getattr(response, "read1", None) - if not callable(reader): - reader = response.read + reader = reader if callable(reader) else response.read body = bytearray() while len(body) <= max_bytes: remaining = deadline - time.monotonic() @@ -146,9 +139,8 @@ def _is_proven_pre_admission_failure(exc: BaseException) -> bool: reason: Any = exc while isinstance(reason, urllib.error.URLError): reason = reason.reason - if isinstance(reason, socket.gaierror): - return True - return isinstance(reason, OSError) and reason.errno in _NOT_ADMITTED_ERRNOS + return isinstance(reason, socket.gaierror) or ( + isinstance(reason, OSError) and reason.errno in _NOT_ADMITTED_ERRNOS) def _valid_code(code: Any) -> str | None: @@ -175,30 +167,18 @@ def _response_error_code(detail: str) -> str | None: return _valid_code(payload.get("code")) -def _http_error_message(method: str, path: str, status: int, error_code: str | None) -> str: - 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}" - - 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: + ) -> None: super().__init__(message) self.retryable, self.ambiguous, self.not_admitted = retryable, ambiguous, not_admitted self.status_code, self.error_code = status_code, error_code - self.error_message = error_message 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( @@ -258,10 +238,9 @@ class PeerRunsHTTPClient: 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.""" 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)] + names = [str(v or "") for v 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 epoch < 1: @@ -269,15 +248,10 @@ class PeerRunsHTTPClient: scope = dict(zip(_RECEIPT_SCOPE_FIELDS, names[:3] + [epoch] + names[3:])) if self._room_scope == scope: return - self._room_scope = scope - self._runs.clear() - self._observation_key = None - self._status_cache.clear() - self._recovery_backoff.clear() - 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._room_scope, self._observation_key = scope, None + for table in ( + self._runs, self._status_cache, self._recovery_backoff, self._terminal_receipts): + table.clear() def _receipt(self, task_id: str, execution_generation: int) -> dict[str, Any] | None: """Return the in-memory receipt, falling back to the durable store.""" @@ -285,7 +259,6 @@ class PeerRunsHTTPClient: if record is not None or self.receipt_db_path is None or self._room_scope is None: 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, **identity}) @@ -295,21 +268,20 @@ class PeerRunsHTTPClient: key = (str(task_id or ""), int(execution_generation or 0)) if not key[0] or key[1] < 1: raise PeerRunsHTTPError("peer observation identity is invalid") - if self._observation_key != key: - for terminal_key in self._terminal_receipts - {key}: - self._runs.pop(terminal_key, None) - self._terminal_receipts.intersection_update({key}) - self._observation_key = key - self._status_cache.clear() - self._recovery_backoff.clear() + if self._observation_key == key: + return + for terminal_key in self._terminal_receipts - {key}: + self._runs.pop(terminal_key, None) + self._terminal_receipts.intersection_update({key}) + self._observation_key = key + self._status_cache.clear() + 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]: from hermes_cli.urllib_security import open_credentialed_url - - deadline = time.monotonic() + self.timeout_seconds - ambiguous = method == "POST" + deadline, ambiguous = time.monotonic() + self.timeout_seconds, method == "POST" request = urllib.request.Request( f"{self.base_url}{path}", method=method, data=None if body is None else json.dumps(body, separators=(",", ":")).encode("utf-8"), @@ -345,9 +317,8 @@ class PeerRunsHTTPClient: """Raise the classified PeerRunsHTTPError for an HTTP error response.""" # A 4xx on admission proves the peer never admitted the run. 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} + "ambiguous": method == "POST" and exc.code >= 500, "status_code": exc.code, + "not_admitted": method == "POST" and path == "/v1/runs" and 400 <= exc.code < 500} try: detail = _read_body( exc, max_bytes=MAX_PEER_ERROR_RESPONSE_BYTES, deadline=deadline, kind=" error", @@ -359,8 +330,12 @@ class PeerRunsHTTPClient: error_code = _response_error_code(detail) logger.debug( "Peer RoomLink request returned HTTP %s (%s)", exc.code, error_code or "no-code") + renewal = exc.code in {401, 403} and error_code in _GRANT_RENEWAL_CODES + drift = exc.code == 403 and error_code in { + _EXECUTION_POLICY_CHANGED[0], _CAPABILITY_CHANGED[0]} raise PeerRunsHTTPError( - _http_error_message(method, path, exc.code, error_code), + _REAUTHORIZATION_MESSAGES[error_code] if renewal or drift + else f"peer rejected {method} {path} with HTTP {exc.code}", retryable=exc.code in {408, 425, 429} or exc.code >= 500, error_code=error_code, **flags, ) from exc @@ -371,8 +346,8 @@ class PeerRunsHTTPClient: 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]) + logical_session = "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} @@ -381,7 +356,7 @@ class PeerRunsHTTPClient: """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_room_scope(**{f: getattr(checked, f) for f in _RECEIPT_SCOPE_FIELDS}) self.bind_observation( task_id=checked.task_id, execution_generation=checked.execution_generation) return checked @@ -397,8 +372,8 @@ class PeerRunsHTTPClient: 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, 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"]): @@ -449,7 +424,6 @@ class PeerRunsHTTPClient: "task_id": checked.task_id, "execution_generation": checked.execution_generation} if self.receipt_db_path is not None: from gateway import hosted_rooms - hosted_rooms.upsert_remote_run_receipt(self.receipt_db_path, record=receipt) self._runs[(checked.task_id, checked.execution_generation)] = receipt self._status_cache.pop(run_id, None) @@ -566,7 +540,7 @@ class PeerRunsHTTPClient: def stop(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any] | None: checked = HostedMemberDispatch.from_mapping(dispatch) - self._bind_dispatch_scope(checked) + self.bind_room_scope(**{f: getattr(checked, f) for f in _RECEIPT_SCOPE_FIELDS}) return self.stop_receipt( task_id=checked.task_id, execution_generation=checked.execution_generation, grant=grant) diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 98d6a1977d..1b28717b36 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -22,7 +22,8 @@ from gateway.hosted_room_peer import ( GatewayRoomCatalog, HostedMemberDispatch, PROTOCOL_VERSION, room_grant_needs_dispatch_refresh) from tui_gateway.hosted_room_driver import HostedRoomBinding, HostedRoomRuntime from tui_gateway.hosted_room_server_rpc import HostedRoomServerRPC -from tui_gateway.hosted_room_peer_http import PeerRunsHTTPClient, PeerRunsHTTPError +from tui_gateway.hosted_room_peer_http import ( + PeerRunsHTTPClient, PeerRunsHTTPError, digest_reauthorization_error) from tui_gateway.hosted_room_peer_transport import ( HostedRoomPeerClient, PeerHostedRoomTransport, PeerMemberRoute, build_member_dispatch) @@ -34,10 +35,6 @@ _TERMINAL_STATUSES = ("deferred", "settled", "failed", "cancelled") _LIVE_STATUSES = ("queued", "running", "stopping") _STOPPABLE_STATUSES = ("queued", "running", "indeterminate", "deferred", "stopping") _RETRYABLE_STATUSES = ("indeterminate", "deferred") -# (message, error_code) for a refreshed peer catalog whose digests drift from the dispatch. -_POLICY_DRIFT = ( - "peer room execution policy needs reauthorization", "room_execution_policy_changed") -_CATALOG_DRIFT = ("peer room capabilities need reauthorization", "room_capability_catalog_changed") def _hosted_room_turn_timeout_seconds() -> float: @@ -105,16 +102,15 @@ class HostedRoomService: stored_links, load_errors = hosted_room_links.load_room_links_tolerant(self.db_path) errors = list(load_errors) for stored in stored_links: - key = (stored.room_id, stored.member_id) - if PROTOCOL_VERSION not in stored.catalog.protocol_versions: + key, catalog = (stored.room_id, stored.member_id), stored.catalog + if PROTOCOL_VERSION not in catalog.protocol_versions: errors.append(f"{stored.room_id}:{stored.member_id}:protocol-upgrade-required") continue self.peer_routes[key] = PeerMemberRoute( home_install_id=hosted_rooms.local_authority_gateway_id(), - member_id=stored.member_id, target_install_id=stored.catalog.installation_id, - target_profile=stored.target_profile, - capability_digest=stored.catalog.catalog_digest, - execution_policy_digest=stored.catalog.execution_policy.policy_digest, + member_id=stored.member_id, target_install_id=catalog.installation_id, + target_profile=stored.target_profile, capability_digest=catalog.catalog_digest, + execution_policy_digest=catalog.execution_policy.policy_digest, cancellation_scope_id=stored.cancellation_scope_id, trace_id=stored.trace_id, grant=stored.grant) self.peer_clients[key] = PeerRunsHTTPClient( @@ -151,11 +147,9 @@ class HostedRoomService: "This Group Chat is managed by another gateway.") return gateway_id, epoch - @contextlib.contextmanager - def _turn_lock(self, profile: str) -> Iterator[None]: + def _turn_lock(self, profile: str) -> contextlib.AbstractContextManager[Path]: from tools.bot_relay import acquire_turn_lock - with acquire_turn_lock(self.root, profile): - yield + return acquire_turn_lock(self.root, profile) def start(self) -> None: self.runtime.start() @@ -203,10 +197,9 @@ class HostedRoomService: def _publish_route(self, key: tuple[str, str], route: PeerMemberRoute, client=None) -> None: """Make a persisted route live as ``ready`` (and bind its client when given).""" with self._policy_lock: - self.peer_routes[key] = route + self.peer_routes[key], self._peer_route_status[key] = route, "ready" if client is not None: self.peer_clients[key] = client - self._peer_route_status[key] = "ready" def revoke_room_routes(self, room_id: str) -> int: """Revoke and forget every scoped peer route for one room; an unreachable target @@ -294,11 +287,10 @@ class HostedRoomService: return False def _set_route_status(self, room_id: str, member_id: str, status: str) -> None: - key = (room_id, member_id) with self._policy_lock: - if self._peer_route_status.get(key) == status: + if self._peer_route_status.get((room_id, member_id)) == status: return - self._peer_route_status[key] = status + self._peer_route_status[(room_id, member_id)] = status hosted_room_links.mark_room_link_status( self.db_path, room_id=room_id, member_id=member_id, status=status) @@ -318,8 +310,9 @@ class HostedRoomService: route = self.peer_routes.get(key) if route is None: raise RuntimeError("peer room route is unavailable") - links = hosted_room_links.load_room_links(self.db_path) - stored = next((l for l in links if (l.room_id, l.member_id) == key), None) + stored = next(( + l for l in hosted_room_links.load_room_links(self.db_path) + if (l.room_id, l.member_id) == key), None) if stored is None: raise RuntimeError("peer room route cannot be renewed before persistence") digests = {} @@ -371,24 +364,23 @@ class HostedRoomService: def _publish_terminal_tasks(self, room: Mapping[str, Any]) -> bool: changed, room_id, local_profiles = False, str(room["room_id"]), self.local_profiles() - for status in _TERMINAL_STATUSES: - for task in driver.list_tasks(self.db_path, room_id=room_id, status=status): - execution_generation = int(task["execution_generation"]) - if self.policy_checkpoint.publication_exists( - room_id=room_id, task_id=task["identity"].task_id, status=status, - execution_generation=execution_generation): - continue - task_events = self.policy_checkpoint.events_for_task( - room_id=room_id, source_event_seq=int(task["payload"]["source_event_seq"])) - plan = discussion.reconstruct_task_plan( - room, task_events, task, local_profiles=local_profiles) - publication = discussion.plan_publication( - room, task_events, plan, status=status, result=task.get("result"), - execution_generation=execution_generation if status == "deferred" else None, - local_profiles=local_profiles) - for event in publication.events: - hosted_rooms.append_event(self.db_path, **event.append_kwargs(room_id)) - changed = True + for task in self._list_tasks(room_id, _TERMINAL_STATUSES): + status, execution_generation = task["status"], int(task["execution_generation"]) + if self.policy_checkpoint.publication_exists( + room_id=room_id, task_id=task["identity"].task_id, status=status, + execution_generation=execution_generation): + continue + task_events = self.policy_checkpoint.events_for_task( + room_id=room_id, source_event_seq=int(task["payload"]["source_event_seq"])) + plan = discussion.reconstruct_task_plan( + room, task_events, task, local_profiles=local_profiles) + publication = discussion.plan_publication( + room, task_events, plan, status=status, result=task.get("result"), + execution_generation=execution_generation if status == "deferred" else None, + local_profiles=local_profiles) + for event in publication.events: + hosted_rooms.append_event(self.db_path, **event.append_kwargs(room_id)) + changed = True return changed def _append_room_status( @@ -461,8 +453,7 @@ class HostedRoomService: binding = next((b for b in self.bindings() if b.room_id == room_id), None) if binding is None: raise hosted_rooms.RoomNotFoundError("hosted room not found") - self.prepare_room(binding) - self.runtime.wakeup() + self.publish_terminal(binding, {}) return event def stop_room( @@ -506,11 +497,10 @@ class HostedRoomService: requested_approval_id = str(request_id or "") def matches(pending: Mapping[str, Any] | None) -> bool: - return ( - pending is not None - and str(pending.get("request_id") or "") == requested_approval_id - and pending.get("task_id") == task_id - and int(pending.get("execution_generation") or 0) == execution_generation) + return pending is not None and ( + str(pending.get("request_id") or ""), pending.get("task_id"), + int(pending.get("execution_generation") or 0), + ) == (requested_approval_id, task_id, execution_generation) if not requested_approval_id or not matches(action): raise RuntimeError("room approval is no longer pending") if choice not in {"once", "deny"}: @@ -544,13 +534,11 @@ class HostedRoomService: counts = Counter(str(task["status"]) for task in tasks) pending_actions = [ {"kind": "retry", "task_id": task["identity"].task_id} - for task in tasks - if task["status"] in _RETRYABLE_STATUSES] + for task in tasks if task["status"] in _RETRYABLE_STATUSES] with self._policy_lock: pending_actions.extend( - dict(action) - for (action_room_id, _member_id), action in self._pending_actions.items() - if action_room_id == room_id) + dict(action) for (action_room_id, _member_id), action + in self._pending_actions.items() if action_room_id == room_id) return { "running": runtime["running"], "working": any(counts.get(s) for s in _LIVE_STATUSES), "blocked": room_id in runtime["blocked_rooms"] @@ -595,15 +583,12 @@ class _RouteStatusPeerClient: refreshed_catalog = None if refreshed.get("catalog") is not None: refreshed_catalog = GatewayRoomCatalog.from_mapping(refreshed.get("catalog")) - policy_digest = refreshed_catalog.execution_policy.policy_digest - drift = ( - _POLICY_DRIFT if policy_digest != checked.execution_policy_digest - else _CATALOG_DRIFT if refreshed_catalog.catalog_digest != checked.capability_digest - else None) + drift = digest_reauthorization_error( + refreshed_catalog, capability_digest=checked.capability_digest, + execution_policy_digest=checked.execution_policy_digest) if drift is not None: self._on_reauthorization() - raise PeerRunsHTTPError( - drift[0], status_code=403, error_code=drift[1], not_admitted=True) + raise drift self._on_refreshed(replacement, refreshed_catalog) return {**kwargs, "grant": replacement}